news 2026/4/14 17:40:03

使用 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/4/7 5:46:26

EmotiVoice是否支持多人对话场景下的角色区分？

EmotiVoice 是否支持多人对话场景下的角色区分？ 在虚拟助手逐渐走进千家万户、游戏NPC开始拥有“人格”、有声书制作趋向自动化生产的今天，用户早已不再满足于单调的“机器音”。他们期待的是更具情感温度、角色辨识度清晰的声音交互体验。尤其是在涉及多…

作者头像 李华
网站建设 2026/4/12 16:30:38

情感语音合成技术前沿：EmotiVoice背后的科研力量

情感语音合成技术前沿：EmotiVoice背后的科研力量 在虚拟助手轻声细语地安慰用户、游戏角色因剧情转折而情绪激荡的今天，我们早已不再满足于“能说话”的机器。人们期待的是会共情、有温度、能表达悲喜的语音交互体验。正是在这一需求驱动下，情…

作者头像 李华
网站建设 2026/4/8 4:47:51

Java RESTful 接口开发 核心指南

RESTful（Representational State Transfer）是基于 HTTP 协议的接口设计规范，Java 作为主流后端开发语言，凭借 Spring 生态等工具能高效实现符合 RESTful 规范的接口开发，以下是核心知识点与实操要点：一、RE…

作者头像 李华
网站建设 2026/4/14 0:54:15

AI塔罗占卜工具：智能解读每日运势与人生疑问

一、工具概览与核心理念 “AI塔罗占卜”是一款结合传统塔罗文化与人工智能技术的数字占卜工具，旨在为用户提供随时可及的运势分析与问题解答。该工具不仅保留塔罗占卜的神秘感和仪式感，还通过AI技术实现个性化解读，让用户在日常生活中获得心…

作者头像 李华
网站建设 2026/4/14 12:27:28

隐私与安全工具集：纯客户端安全解决方案

一、工具定位与核心理念 “隐私与安全工具集”是一款专注于用户隐私保护的纯客户端工具合集，其核心设计理念是“数据永不离开您的浏览器”。在当今数据泄露事件频发、隐私保护日益重要的数字化时代，该工具集为用户提供了一个完全在本机环境中运行的安全…

作者头像 李华
网站建设 2026/4/9 6:45:25

EmotiVoice能否替代专业配音演员？行业专家这样说

EmotiVoice：当AI开始“动情”说话 在某档热门播客的后期制作间里，编辑正为一段主角情绪崩溃的独白发愁——真人配音演员因档期冲突无法补录，而现有素材又缺乏感染力。他尝试输入一句台词：“我没想到事情会变成这样”，选…

作者头像 李华