当前位置： 首页 > ds >正文

Python-MCPServer开发

ds 2025/5/7 3:00:00

Python-MCPServer开发

使用FastMCP开发【SSE模式的MCPServer】，熟悉【McpServer编码过程】+【McpServer调试方法】

1-核心知识点

  • 1-熟悉【SSE模式的MCPServer】开发
  • 2-熟悉【stdio模式的MCPServer】开发
  • 3-熟悉【启动MCPServer】的三种方式
    • 3.1-直接启动:python mcp_server.py
    • 3.2-代码中uv使用：subprocess.Popen([“uv”, “run”, server_file])
    • 3.3-使用mcpInspector网页：mcp dev mcp_server.py
  • 4-熟悉【Agent增强外部能力】的两种方式
    • 4.1-调用tool:Agent(name=“Assistant”, instructions=“用中文回复”, tools=[get_weather])
    • 4.2-调用tool:Agent(name=“Assistant”, instructions=“用中文回复”, mcp_servers=[mcp_server])

2-思路整理

1-McpServer须知

  • 1-MCPServer已经可以对外提供服务了，这个不同于传统的API，虽然也运行在特殊端口了，但是不能通用postman进行调用

  • 2-那有人质疑这个说法了，为什么使用mcpInspector网页调试就可以连通？

    • 可以理解为：mcpInspector联通的是Agent，只是查看Agent上有哪些Resources/Prompts/Tools…并且可以通过Agent和McpServer进行交互，而不是Rest请求

  • 3-MCPServer_SSE和MCPServer_stdio如何进行快速区分：

    • 1-stdio-适用于同一台机器上（进程间通讯，通讯不经过HTTP协议，超级快）->要求：你要有可运行文件【Python文件】或者【Java的Jar包】
    • 2-SSE适用于所有环境，通过IP进行HTTP通讯->基本没有要求，无脑就全部使用这种

  • 4-再次澄清：MCPServer_SSE虽然可以部署，有大网IP或者域名，但是依然不能通用postman进行调用

    • post调用的接口->要求是@Controller标记或者FastAPI开发的API接口
    • agent调用的接口->要求是@mcp.tool标记（可能对外就不是API接口）
    • 总结：可以理解@mcp.tool标记后是一种特殊的API，只能被Agent调用

  • 5-SSE和stdio在代码上有什么区别？

    • 1-McpServer的编写上只有一个区别【mcp.run(transport=“sse/stdio”)】
    • 2-Agent在连接的时候使用【McpServerstdio】和【McpServerSSE】进行McpServer构建
    • 3-【mcp.run(transport=“sse”)】后，服务【可以被任何远程】的Agent进行访问（此时先不谈鉴权的问题）
    • 4-【mcp.run(transport=“stdio”)】后，服务【只可以被本地进程】的Agent进行访问（此时先不谈鉴权的问题）

2-McpServer核心思路

1-【mcp.run(transport=“sse”)】后，服务【可以被任何远程】的Agent进行访问（此时先不谈鉴权的问题）

2-【mcp.run(transport=“stdio”)】后，服务【只可以被本地进程】的Agent进行访问（此时先不谈鉴权的问题）

  • 1-编写传统的Service业务代码
  • 2-在Service业务代码头上添加@tool装饰器，即可实现FastMCP的Tool功能
  • 3-在Service业务代码头上添加@mcp.tool()装饰器，即可实现FastMCP的McpServer功能
  • 4-主程序指定运行方法-stdio进程启动（此时MCPServer已经可以对外提供服务了，这个不同于传统的API，虽然也运行在特殊端口了，但是不能通用postman进行调用）
    • 1-直接启动:python mcp_server.py
    • 2-代码中uv使用：subprocess.Popen([“uv”, “run”, server_file])
    • 3-使用mcpInspector网页：mcp dev mcp_server.py
  • 5-Agent调试McpServer
    • 1-指定LLM
    • 2-指定Agent(McpServerList+角色定位)
    • 3-Runner.run(Agent+LLM+问题)交互

3-McpServer和McpClient关系

  • 1-McpClient只是一个概念，在使用Python进行开发的时候根本就没有任何关于McpClient的类
  • 2-McpClient就是McpServer连接过程的过程

4-Agent+LLM+McpServers

  • 1-LLM和Agent最初是独立创建的，两者之间没有关系

    • LLM指明用哪家大模型
    • Agent指明用哪些McpServerList

  • 2-LLM和Agent是通过Runner.run(Agent+LLM+问题)进行关联

  • 3-Agent+LLM如何进行交互（要不要进行最后的总结）是通用（result.output获取的时候进行逻辑编排的）

    • 如果我们自己做MCP编排，可能就是在这个地方进行核心逻辑编写

5-MCPServer核心代码

  • 1-在Service业务代码头上添加@tool装饰器，即可实现FastMCP的Tool功能
# 假设 mcp 已经正确导入
try:from mcp import tool
except ImportError:# 如果 mcp 未找到，模拟一个 tool 装饰器def tool(func):return func# 在 Service 业务代码头上添加 @tool 装饰器
@tool
async def get_city_list(self) -> list:"""获取所有的城市信息。返回:str: 所有的城市信息列表"""logging.info(f"获取所有的城市信息")city_list = []for city in self.CITY_WEATHER_DATA:city_list.append(city)return city_list
  • 2-在Service业务代码头上添加@mcp.tool()装饰器，即可实现FastMCP的McpServer功能
from mcp.server.fastmcp import FastMCPfrom city_01_service import CityDataServer# 1-初始化 MCP 服务器
mcp = FastMCP("CityDataServer")# 2-初始化城市信息服务器(业务代码+@tool装饰器)
city_server = CityDataServer()# 3-在 Service 业务代码头上添加@mcp.tool()装饰器
@mcp.tool()
# 获取所有城市列表
async def get_city_list():"""获取所有城市列表。返回:str: 所有城市列表"""city_list = await city_server.get_city_list()return city_list# 4-主程序指定运行方法-stdio/sse进程启动
if __name__ == "__main__":mcp.run(transport='stdio/sse')

3-参考网址

  • MCP官网的开发样例：https://github.com/openai/openai-agents-python/blob/main/examples/mcp/sse_example/main.py
  • 个人代码实现仓库：https://gitee.com/enzoism/python_mcp_client_server

4-上手实操

1-空工程初始化环境

mkdir my_project
cd my_project
python -m venv .venv

2-激活环境

# Windows
source .venv/Scripts/activate# Mac
source .venv/bin/activate

3-添加依赖

对应的依赖是在激活的环境中

# uv用于后续MCP Inspector的连接
pip install uv httpx mcp

4-ToolFunction核心代码

import asyncio
import osfrom agents import (Agent,RunConfig,Runner,function_tool,set_tracing_disabled,
)
from dotenv import load_dotenvfrom model_providers.deepseek import DeepSeekModelProvider# 1-环境变量加载相关
load_dotenv()
BASE_URL = os.getenv("BASE_URL") or ""
API_KEY = os.getenv("API_KEY") or ""
MODEL_NAME = os.getenv("MODEL_NAME") or ""
if not BASE_URL or not API_KEY or not MODEL_NAME:raise ValueError("请通过环境变量或代码设置EXAMPLE_BASE_URL、EXAMPLE_API_KEY、EXAMPLE_MODEL_NAME。")# 2-跳过大模型的链路追踪
set_tracing_disabled(disabled=True)"""
本例使用自定义提供程序调用Runner.run()的部分，并直接调用OpenAI进行其他操作。
步骤：
1. 【实例化LLM】ModelProvider对象-并构建RunConfig
2. 【实例化Agent】创建一个Agent。
3. 在调用Runner.run()结合【LLM】+【Agent】进行问答
注意，在本例中，我们假设您没有从platform.openai.com获取API密钥，因此禁用了跟踪。
如果您有API密钥，您可以选择设置`OPENAI_API_KEY`环境变量或调用set_tracing_export_api_key()来设置跟踪特定的密钥。
"""# 3-定义一个工具函数
@function_tool
def init_weather_tool_function(city: str):print(f"[debug] getting weather for {city}")return f"The weather in {city} is sunny."async def run_mcp_tool_function():# 1-【实例化LLM】ModelProvider对象-并构建RunConfigrun_config = RunConfig(model_provider=DeepSeekModelProvider(BASE_URL, API_KEY, MODEL_NAME))# 2-【实例化Agent】创建一个Agentagent = Agent(name="Assistant",instructions="使用工具回答大模型的问题",tools=[init_weather_tool_function])# 3.1-调用工具回答问题message = "杭州的天气怎么样？"print(f"\n\n【大模型请求案例】-> {message}")result = await Runner.run(starting_agent=agent, input=message, run_config=run_config)print(result.final_output)# 3.2-获取Agent对话的结果-没有配置RunConfig使用的是OpenAI的默认模型# result = await Runner.run(#     agent,#     "给我讲一个笑话吧!",# )# print(result.final_output)if __name__ == "__main__":asyncio.run(run_mcp_tool_function())

5-McpServer核心代码

1-【mcp.run(transport=“sse”)】后，服务【可以被任何远程】的Agent进行访问（此时先不谈鉴权的问题）

2-【mcp.run(transport=“stdio”)】后，服务【只可以被本地进程】的Agent进行访问（此时先不谈鉴权的问题）

import random
import randomfrom mcp.server.fastmcp import FastMCP# Create server
mcp = FastMCP("Echo Server")# 1-MCP工具1-数字加和
@mcp.tool()
def add(a: int, b: int) -> int:"""Add two numbers"""print(f"[MCP工具1-数字加和] add({a}, {b})")return a + b# 2-MCP工具2-随机选定的字符
@mcp.tool()
def get_secret_word() -> str:"""获取随机单词"""print("[MCP工具2-随机选定的字符] get_secret_word()")return random.choice(["apple", "banana", "cherry"])if __name__ == "__main__":mcp.run(transport="sse/stdio")

6-Agent调试McpServer

hitokoto_02_api.py：API直接调用service对外暴露rest请求

import asyncio
import osfrom agents import Agent, Runner, RunConfig, set_tracing_disabled
from agents.mcp import MCPServer, MCPServerStdio
from agents.model_settings import ModelSettings
from dotenv import load_dotenvfrom model_providers.deepseek import DeepSeekModelProvider# 1-环境变量加载相关
load_dotenv()
BASE_URL = os.getenv("BASE_URL") or ""
API_KEY = os.getenv("API_KEY") or ""
MODEL_NAME = os.getenv("MODEL_NAME") or ""
if not BASE_URL or not API_KEY or not MODEL_NAME:raise ValueError("请通过环境变量或代码设置EXAMPLE_BASE_URL、EXAMPLE_API_KEY、EXAMPLE_MODEL_NAME。")# 2-跳过大模型的链路追踪
set_tracing_disabled(disabled=True)"""
本例使用自定义提供程序调用Runner.run()的部分，并直接调用OpenAI进行其他操作。
步骤：
1. 【实例化LLM】ModelProvider对象-并构建RunConfig
2. 【实例化Agent】创建一个Agent。
3. 在调用Runner.run()结合【LLM】+【Agent】进行问答
- 1）直接和大模型对话
- 2）调用MCPServer_SSE模式-[MCP工具1-数字加和]
- 3）调用MCPServer_SSE模式-[MCP工具2-随机选定的字符]
"""async def run_mcp_server(mcp_server: MCPServer):# 1-【实例化LLM】ModelProvider对象-并构建RunConfigrun_config = RunConfig(model_provider=DeepSeekModelProvider(BASE_URL, API_KEY, MODEL_NAME))# 2-【实例化Agent】创建一个Agentagent = Agent(name="Assistant",instructions="使用工具回答大模型的问题",mcp_servers=[mcp_server],model_settings=ModelSettings(tool_choice="required"),)# 3.1-直接和大模型对话message = "给我讲一个笑话吧！"print(f"\n\n【大模型请求案例】-> {message}")result = await Runner.run(starting_agent=agent, input=message, run_config=run_config)print(result.final_output)# 3.2-调用MCPServer_SSE模式-[MCP工具1-数字加和]message = "What's the weather in Tokyo?"print(f"\n\n【大模型请求案例】-> {message}")result = await Runner.run(starting_agent=agent, input=message, run_config=run_config)print(result.final_output)# 3.3-调用MCPServer_SSE模式-[MCP工具2-随机选定的字符]message = "What's the secret word?"print(f"\n\n【大模型请求案例】-> {message}")result = await Runner.run(starting_agent=agent, input=message, run_config=run_config)print(result.final_output)async def init_mcp_server() -> MCPServerStdio:# 1-创建 MCP 服务器连接实例，但不立即运行(python mcp_server_xx.py)this_dir = os.path.dirname(os.path.abspath(__file__))python_exec_path = os.path.join(this_dir, ".venv/Scripts/python.exe")mcp_server_stdio_file = os.path.join(this_dir, "mcp05_stdio.py")weather_server = MCPServerStdio(name="weather",params={"command": python_exec_path,"args": [mcp_server_stdio_file],"env": {},},# 缓存工具列表以减少延迟，不需要每次连接时重新获取工具列表cache_tools_list=True)# 2-手动连接到MCP服务器print("正在连接到MCP服务器...")await weather_server.connect()print("MCP服务器连接成功！")# 3-等待服务器连接成功并获取MCP服务可用工具列表tools = await weather_server.list_tools()print("\n可用工具列表: ")for tool in tools:print(f" - {tool.name}: {tool.description}")return weather_serverif __name__ == "__main__":# 1-获取McpServerserver = asyncio.run(init_mcp_server())# 2-运行McpServerasyncio.run(run_mcp_server(server))

http://www.xdnf.cn/news/2317.html

相关文章：

  • 产业观察：哈飞空客2025.4.26
  • 【MATLAB】基于RSSI原理的Wi-Fi定位程序，N个锚点（数量可自适应）、三维空间，轨迹使用UKF进行滤波，附完整代码（订阅专栏后可直接复制粘贴）
  • 100亿补贴不是终点：京东外卖在下一盘颠覆即时零售的大棋
  • w307MES生产制造执行系统设计与实现
  • SEO新手快速上手核心步骤
  • 【Android Compose】焦点管理
  • AWS中国区ICP备案全攻略:流程、注意事项与最佳实践
  • python 画折线统计图
  • 华为OD机试真题——二维伞的雨滴效应（2025A卷：200分）Java/python/JavaScript/C/C++/GO最佳实现
  • element ui el-col的高度不一致导致换行
  • AQS条件队列源码详细剖析
  • FPGA 100G UDP纯逻辑协议栈
  • 【C语言】柔性数组
  • 单片机-89C51部分：3、创建工程
  • 【MCP 应用】CherryStudio 配置和应用 MCP
  • node入门和npm
  • std::mutex底层实现原理
  • 使用命令关闭Redis服务端
  • 【Castle-X机器人】一、模块安装与调试：机器人底盘
  • 前端学习笔记（四）自定义组件控制自己的css
  • QT跨平台软件开发要点
  • 0803分页_加载更多-网络ajax请求2-react-仿低代码平台项目
  • OSPF中DR/BDR的选举
  • WPF程序使用Sugar操作数据库
  • 从 Vue 到 React：React 合成事件
  • WPF定义扩展属性和依赖属性
  • LeetCode热题100——70. 爬楼梯
  • 解锁生成式AI潜力的金钥匙
  • WSL 中 nvidia-smi: command not found的解决办法
  • OpenAI o3和o4 - mini模型发布：创新乏力与竞争困境