做pc端网站如何,友情链接出售,网站开发的工具,加猛挣钱免费做网站软件使用 Python 快速构建一个 MCP Server目标: 完成三方接口调用/Mysql数据库查询的MCPServer
要求: python版本大于等于3.11 版本 太低记得有些东西不支持最终结果-完整代码
from fastmcp import FastMCP
import pymysql
import requestsmcp FastMCP(nameSubmission Conte…使用 Python 快速构建一个 MCP Server目标:完成三方接口调用/Mysql数据库查询的MCPServer要求:python版本大于等于3.11 版本 太低记得有些东西不支持最终结果-完整代码fromfastmcpimportFastMCPimportpymysqlimportrequests mcpFastMCP(nameSubmission 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:urlhttp://localhost:8087/record/logrecord/getAllCompanyresponserequests.get(url,timeout5)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:connpymysql.connect(**DB_CONFIG)cursorconn.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))recordscursor.fetchall()return{count:len(records),records:records}exceptrequests.exceptions.RequestExceptionase:return{error:f网络请求失败:{str(e)}}exceptValueErrorase:return{error:fJSON解析失败:{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:urlhttp://localhost:8087/back/api/logRecord/getDeptMonthStatisticsByCompanyIdparams{companyId:company_id,year:year,month:month,queryFirstDept:false# 固定参数}responserequests.get(url,paramsparams,timeout5)response.raise_for_status()returnresponse.json()# 假设接口返回 JSONexceptExceptionase:return{error:str(e)}# 3. 添加静态资源mcp.resource(resource://config)defget_config()-dict:Provides the applications 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.returnfHello,{name}! Welcome to the MCP server.# 5. 启动服务器if__name____main__:# mcp.run(host0.0.0.0, port8000)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:[]}}}]}MCPModel Control Protocol是一种新兴的协议标准旨在让大语言模型LLM或智能体Agent能够安全、结构化地调用外部工具与资源。本文将带你从零开始使用 Python 和fastmcp库快速搭建一个功能完整的 MCP Server用于提供企业工作汇报数据的查询能力。什么是 MCPMCPModel Control Protocol是一种为 LLM 设计的“工具调用”协议。它允许模型通过标准化的方式调用远程函数称为Tools获取静态或动态Resources与后端系统如数据库、API交互通过 MCP Server我们可以将内部业务逻辑“暴露”给 AI 智能体使其具备真实世界的数据操作能力。详细学习指路全网最详细没有之一:https://oigi8odzc5w.feishu.cn/wiki/LWqEwXNkBibT0ykrbI0cvptBnAf项目目标我们希望构建一个名为工作汇报的 MCP 服务支持以下功能获取公司列表(调用接口)查询某用户在指定年月的工作汇报内容及提交次数(查询数据库)获取某公司在指定年月的部门月度统计(调用接口)提供配置信息和个性化欢迎语作为资源代码实现详解1. 初始化 MCP ServerfromfastmcpimportFastMCP mcpFastMCP(nameSubmission 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]:# 调用已有内部 APIresponserequests.get(http://localhost:8087/record/logrecord/getAllCompany,timeout5)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. 添加资源ResourcesMCP 不仅支持“执行动作”的工具还支持“读取信息”的资源。 静态资源应用配置mcp.resource(resource://config)defget_config()-dict:return{version:1.0,author:MyTeam}智能体可主动读取此资源以了解服务版本。 动态资源个性化问候mcp.resource(greetings://{name})defpersonalized_greeting(name:str)-str:returnfHello,{name}! Welcome to the MCP server.当请求greetings://Alice时返回Hello, Alice! ...—— 类似 RESTful 路径参数。5. 启动服务if__name____main__:mcp.run()# 默认 localhost:8000# 或 mcp.run(host0.0.0.0, port8000) 以允许外部访问一行代码启动 MCP 服务底层会自动注册所有tool和resource。如何与 LLM 集成一旦 MCP Server 运行起来你的 LLM Agent 就可以通过 MCP 协议发现可用工具如get_user_submissions根据用户问题生成工具调用参数执行调用并获取结构化结果将结果自然语言化返回给用户例如用户问“张一山在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 工具调用