news 2026/5/23 18:54:15

使用 Python 快速构建一个 MCP Server(基于本地Stdio方式)

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
使用 Python 快速构建一个 MCP Server(基于本地Stdio方式)

使用 Python 快速构建一个 MCP Server

目标:完成三方接口调用/Mysql数据库查询的MCPServer

要求:python版本大于等于3.11 版本 太低记得有些东西不支持

最终结果-完整代码

fromfastmcpimportFastMCPimportpymysqlimportrequests mcp=FastMCP(name="Submission Content Server")DB_CONFIG={"host":"10.0.6.1","user":"root","password":"root","database":"work_state","charset":"utf8mb4"}@mcp.tool(description="获取公司列表")defget_company_list()->list[dict]:""" 获取公司列表 返回: - list of dict: 每个字典包含公司ID和公司名称 调用本地接口: http://localhost:8087/record/logrecord/getAllCompany """try:url="http://localhost:8087/record/logrecord/getAllCompany"response=requests.get(url,timeout=5)response.raise_for_status()returnresponse.json()# 假设接口返回 JSON 数组exceptExceptionase:return{"error":str(e)}@mcp.tool(description="获取指定用户在某年月的汇报内容及提交次数")defget_user_submissions(nick_name:str,year:int,month:int)->dict:""" 获取指定用户昵称在某年月的汇报内容及提交次数 参数: - nick_name: 用户昵称，例如 "张一山" - year: 年份，例如 2025 - month: 月份，例如 10 返回: - dict: { "count": 提交次数, "records": [ {"submission_date": "2025-10-01", "work_item": "..."}, {"submission_date": "2025-10-02", "work_item": "..."}, ... ] } """try:conn=pymysql.connect(**DB_CONFIG)cursor=conn.cursor(pymysql.cursors.DictCursor)sql=""" SELECT lr.submission_date, lr.work_item FROM ws_log_record lr JOIN sys_user u ON lr.create_by = u.user_name WHERE u.nick_name = %s AND YEAR(lr.submission_date) = %s AND MONTH(lr.submission_date) = %s ORDER BY lr.submission_date """cursor.execute(sql,(nick_name,year,month))records=cursor.fetchall()return{"count":len(records),"records":records}exceptrequests.exceptions.RequestExceptionase:return{"error":f"网络请求失败:{str(e)}"}exceptValueErrorase:return{"error":f"JSON解析失败:{str(e)}"}exceptExceptionase:return{"error":f"获取部门统计失败:{str(e)}"}@mcp.tool(description="获取指定公司在某年月的部门月度统计")defget_dept_month_statistics(company_id:int,year:int,month:int)->dict:""" 获取指定公司在某年月的部门月度统计 参数: - company_id: 公司ID - year: 年份，例如 2025 - month: 月份，例如 11 返回: - dict: 接口返回的部门统计结果 调用本地接口: http://localhost:8087/back/api/logRecord/getDeptMonthStatisticsByCompanyId queryFirstDept 参数固定为 false """try:url="http://localhost:8087/back/api/logRecord/getDeptMonthStatisticsByCompanyId"params={"companyId":company_id,"year":year,"month":month,"queryFirstDept":"false"# 固定参数}response=requests.get(url,params=params,timeout=5)response.raise_for_status()returnresponse.json()# 假设接口返回 JSONexceptExceptionase:return{"error":str(e)}# 3. 添加静态资源@mcp.resource("resource://config")defget_config()->dict:"""Provides the application's configuration."""return{"version":"1.0","author":"MyTeam"}# 4. 添加动态资源模板@mcp.resource("greetings://{name}")defpersonalized_greeting(name:str)->str:"""Generates a personalized greeting for the given name."""returnf"Hello,{name}! Welcome to the MCP server."# 5. 启动服务器if__name__=="__main__":# mcp.run(host='0.0.0.0', port=8000)mcp.run()

测试：

ChatBox测试

配置如图:

测试通过✅

整个项目测试

这里通过Java中使用langchain4j配置MCPServer调用测试: 如图

控制台打印结果:

通过日志可以看到,tools中的工具已经加载, 并且通过提问已经找到了对应的tool, 调用了接口

测试通过✅

-body:{"model":"qwen-plus","messages":[{"role":"system","content":"你是一个管理助手, 可以回答所有问题"},{"role":"user","content":"回答我的问题:都有哪些公司"}],"stream":true,"stream_options":{"include_usage":true},"tools":[{"type":"function","function":{"name":"get_user_submissions","description":"获取指定用户在某年月的汇报内容及提交次数","parameters":{"type":"object","properties":{"nick_name":{"type":"string"},"year":{"type":"integer"},"month":{"type":"integer"}},"required":["nick_name","year","month"]}}},{"type":"function","function":{"name":"get_dept_month_statistics","description":"获取指定公司在某年月的部门月度统计","parameters":{"type":"object","properties":{"company_id":{"type":"integer"},"year":{"type":"integer"},"month":{"type":"integer"}},"required":["company_id","year","month"]}}},{"type":"function","function":{"name":"get_company_list","description":"获取公司列表","parameters":{"type":"object","properties":{},"required":[]}}}]}--------------------------------------------------2--------------------------------------body:{"model":"qwen-plus","messages":[{"role":"system","content":"你是一个管理助手, 可以回答所有问题"},{"role":"user","content":"回答我的问题:都有哪些公司"},{"role":"assistant","tool_calls":[{"id":"call_bdc468be7e4f46f28d26f5","type":"function","function":{"name":"get_company_list","arguments":"{}"}}]},{"role":"tool","tool_call_id":"call_bdc468be7e4f46f28d26f5","content":"There was an error executing the tool. The tool returned: Output validation error: {'msg': '请求访问：/record/logrecord/getAllCompany，认证失败，无法访问系统资源', 'code': 401} is not of type 'array'"}],"stream":true,"stream_options":{"include_usage":true},"tools":[{"type":"function","function":{"name":"get_user_submissions","description":"获取指定用户在某年月的汇报内容及提交次数","parameters":{"type":"object","properties":{"nick_name":{"type":"string"},"year":{"type":"integer"},"month":{"type":"integer"}},"required":["nick_name","year","month"]}}},{"type":"function","function":{"name":"get_dept_month_statistics","description":"获取指定公司在某年月的部门月度统计","parameters":{"type":"object","properties":{"company_id":{"type":"integer"},"year":{"type":"integer"},"month":{"type":"integer"}},"required":["company_id","year","month"]}}},{"type":"function","function":{"name":"get_company_list","description":"获取公司列表","parameters":{"type":"object","properties":{},"required":[]}}}]}

MCP（Model Control Protocol）是一种新兴的协议标准，旨在让大语言模型（LLM）或智能体（Agent）能够安全、结构化地调用外部工具与资源。本文将带你从零开始，使用 Python 和fastmcp库快速搭建一个功能完整的 MCP Server，用于提供企业工作汇报数据的查询能力。

什么是 MCP？

MCP（Model Control Protocol）是一种为 LLM 设计的“工具调用”协议。它允许模型通过标准化的方式：

  • 调用远程函数（称为Tools）
  • 获取静态或动态Resources
  • 与后端系统（如数据库、API）交互

通过 MCP Server，我们可以将内部业务逻辑“暴露”给 AI 智能体，使其具备真实世界的数据操作能力。

详细学习指路（全网最详细没有之一）:
https://oigi8odzc5w.feishu.cn/wiki/LWqEwXNkBibT0ykrbI0cvptBnAf

项目目标

我们希望构建一个名为工作汇报的 MCP 服务，支持以下功能：

  1. 获取公司列表(调用接口)
  2. 查询某用户在指定年月的工作汇报内容及提交次数(查询数据库)
  3. 获取某公司在指定年月的部门月度统计(调用接口)
  4. 提供配置信息和个性化欢迎语（作为资源）

代码实现详解

1. 初始化 MCP Server

fromfastmcpimportFastMCP mcp=FastMCP(name="Submission Content Server")

这一步创建了一个 MCP 服务实例，并命名为 “Submission Content Server”，便于识别和管理。

2. 定义数据库配置

DB_CONFIG={"host":"10.0.6.1","user":"root","password":"root","database":"work_state","charset":"utf8mb4"}

连接到内部的工作状态数据库work_state，用于查询用户提交记录。

3. 实现核心工具（Tools）

✅ 工具一：获取公司列表
@mcp.tool(description="获取公司列表")defget_company_list()->list[dict]:# 调用已有内部 APIresponse=requests.get("http://localhost:8087/record/logrecord/getAllCompany",timeout=5)returnresponse.json()
✅ 工具二：查询用户月度汇报
@mcp.tool(description="获取指定用户在某年月的汇报内容及提交次数")defget_user_submissions(nick_name:str,year:int,month:int)->dict:# 直连数据库查询sql=""" SELECT lr.submission_date, lr.work_item FROM ws_log_record lr JOIN sys_user u ON lr.create_by = u.user_name WHERE u.nick_name = %s AND YEAR(lr.submission_date) = %s AND MONTH(lr.submission_date) = %s ORDER BY lr.submission_date """# 执行查询并返回结构化结果return{"count":len(records),"records":records}

支持自然语言参数（如“张一山 2025年10月”），由 LLM 自动提取nick_name,year,month并调用。

✅ 工具三：获取部门月度统计
@mcp.tool(description="获取指定公司在某年月的部门月度统计")defget_dept_month_statistics(company_id:int,year:int,month:int)->dict:.....

4. 添加资源（Resources）

MCP 不仅支持“执行动作”的工具，还支持“读取信息”的资源。

🔹 静态资源：应用配置
@mcp.resource("resource://config")defget_config()->dict:return{"version":"1.0","author":"MyTeam"}

智能体可主动读取此资源以了解服务版本。

🔹 动态资源：个性化问候
@mcp.resource("greetings://{name}")defpersonalized_greeting(name:str)->str:returnf"Hello,{name}! Welcome to the MCP server."

当请求greetings://Alice时，返回"Hello, Alice! ..."—— 类似 RESTful 路径参数。

5. 启动服务

if__name__=="__main__":mcp.run()# 默认 localhost:8000# 或 mcp.run(host='0.0.0.0', port=8000) 以允许外部访问

一行代码启动 MCP 服务！底层会自动注册所有@tool和@resource。

如何与 LLM 集成？

一旦 MCP Server 运行起来，你的 LLM Agent 就可以通过 MCP 协议：

  1. 发现可用工具（如get_user_submissions）
  2. 根据用户问题生成工具调用参数
  3. 执行调用并获取结构化结果
  4. 将结果自然语言化返回给用户

例如，用户问：

“张一山在2025年10月提交了几次工作汇报？”

Agent 会自动调用：

{"tool":"get_user_submissions","args":{"nick_name":"张一山","year":2025,"month":10}}

然后收到：

{"count":22,"records":[...]}

最终回答：“张一山在2025年10月共提交了22次工作汇报。”

总结

通过fastmcp，我们用不到 100 行代码就构建了一个具备数据库查询 + API 调用 + 资源服务能力的 MCP Server。它的优势在于：

  • ✅低侵入：复用现有系统，无需改造
  • ✅高可读：每个工具/资源职责清晰
  • ✅易扩展：新增功能只需加一个@mcp.tool
  • ✅AI 友好：结构化输入输出，完美适配 LLM 工具调用
版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/5/16 18:56:53

PostgreSQL 16 + pgvector 完整安装和内网访问指南（Ubuntu 20.04）

1. 更新系统并安装必要工具 sudo apt update sudo apt install -y wget ca-certificates gnupg lsb-release2. 导入 PostgreSQL archive 仓库 GPG 密钥 wget -qO- https://www.postgresql.org/media/keys/ACCC4CF8.asc | sudo gpg --dearmor -o /usr/share/keyrings/postgresql…

作者头像 李华
网站建设 2026/5/13 22:37:06

dify v1.11.1 离线安装本地插件怎么报错了？！

Dify v1.11.1离线插件安装失败？ 最近Dify v1.11.1版本发布后，不少开发者反馈离线插件安装频频碰壁——界面提示"安装失败"却无具体原因，进度条卡在90%一动不动，后台日志疯狂刷屏"依赖下载超时"。这些问题看似…

作者头像 李华
网站建设 2026/5/23 9:41:21

EmotiVoice实战指南：从文本到富有情感的语音只需三步

EmotiVoice实战指南：从文本到富有情感的语音只需三步 在智能语音助手越来越“懂人心”的今天，用户早已不满足于机械地播报天气或导航路线。我们期待的是一个能共情、会安慰、甚至带点小情绪的对话伙伴——这正是情感化语音合成（Emotional TTS…

作者头像 李华
网站建设 2026/5/22 18:30:01

灵活用工平台注册，亲测合规要点

灵活用工平台行业分析：天语灵活用工平台的合规要点行业痛点分析在当前的灵活用工平台领域，技术挑战主要体现在数据安全、算薪准确性和合规性等方面。随着灵活用工需求的增加，平台需要处理大量的用户数据和薪资计算，这对系统的技术…

作者头像 李华
网站建设 2026/5/22 1:49:35

EmotiVoice能否用于盲文转换辅助系统？触觉听觉协同设计

EmotiVoice能否用于盲文转换辅助系统？触觉听觉协同设计 在视障人群的信息获取世界里，声音和触觉是两根最重要的支柱。传统的盲文阅读依赖指尖的触觉辨识，学习曲线陡峭、信息更新缓慢；而纯语音播报虽然便捷，却难以传达细…

作者头像 李华
网站建设 2026/5/10 5:46:04

LobeChat春节营销主题内容生成

LobeChat：构建春节智能营销的AI内容引擎 在年味渐浓的节日前夕，品牌运营团队往往面临一个共同挑战：如何在短时间内产出大量富有节日氛围、风格统一且创意十足的文案？从朋友圈推送、商品标题到客户祝福语，传统人工创作模…

作者头像 李华