Agents-SDK智能体开发[5]之集成MCP进阶

一 Agents SDK+MCP进阶

1.1 Agents SDK接入开源MCP服务器

• 若要采用MCP技术栈，最核心的便利就在于可以快速接入海量MCP开源服务器，无需反复开发，即可快速丰富当前Agent功能。
• 热门MCP server合集地址。
  • Model Context Protocol servers
  • Awesome MCP Servers
  • MCP导航

---

1.2 MCP-Server-Git

• 尝试接入的MCP服务器——mcp-server-git。
• mcp-server-git是一个遵循Model Context Protocol (MCP)的Git操作服务器，专为大语言模型与Git仓库的交互而设计。通过该服务，模型可以安全、结构化地完成 Git 操作，包括状态查询、差异比较、提交更改、分支管理等，从而实现自动化代码管理与协作。

---

• ✨ 核心功能包括：
  • 查询仓库状态：获取当前工作区和暂存区的变动情况（git_status、git_diff_unstaged、git_diff_staged）
  • 版本比较：支持分支或提交之间的差异查看（git_diff）
  • 代码提交与暂存管理：支持新增、暂存、撤销暂存、更改提交（git_add、git_reset、git_commit）
  • 日志查询与历史查看：获取提交历史、查看具体提交内容（git_log、git_show）
  • 分支操作：新建分支、切换分支（git_create_branch、git_checkout）
  • 仓库初始化：支持新建空 Git 仓库（git_init）

---

🚀 调用方式
✅ 使用 uvenv 快速启动，无需安装，只需一行命令即可运行：

• 首次运行会从 PyPI 下载并缓存，后续启动速度更快。

uvenv run mcp-server-git

📡 接口调用格式，以 git_status 为例，MCP 工具调用格式如下：

{"tool": "git_status","input": {"repo_path": "/path/to/your/git/repo"}
}

• 服务将返回 Git 工作目录当前状态的文本描述。

1. 安装uv

   # Ubuntu安装uvenv
   curl -Ls https://astral.sh/uv/install.sh | sh
   # Ubuntu 使用wget命令安装
   wget -qO- https://astral.sh/uv/install.sh | sh
   # window安装venv
   powershell -ExecutionPolicy Bypass -c "irm https://github.com/astral-sh/uv/releases/download/0.7.12/uv-installer.ps1 | iex"
   

2. 测试是否安装成功

   uv --version
   uvenv --help
   

3. python脚本(未测试脚本)

   import asyncio
   import shutilfrom agents import Agent, Runner, trace
   from agents.mcp import MCPServer, MCPServerStdioasync def run(mcp_server: MCPServer, directory_path: str):agent = Agent(name="Assistant",instructions=f"Answer questions about the git repository at {directory_path}, use that for repo_path",mcp_servers=[mcp_server],model=deepseek_model)message = "请帮我介绍下这个项目。"print("\n" + "-" * 40)print(f"Running: {message}")result = await Runner.run(starting_agent=agent, input=message)print(result.final_output)async def main():async with MCPServerStdio(cache_tools_list=True, params={"command": "uvenv", "args": ["run", "mcp-server-git"]},) as server:await run(server, directory_path)if __name__ == "__main__":if not shutil.which("uvenv"):raise RuntimeError("uvx is not installed. Please install it with `pip install uvx`.")asyncio.run(main())
   

二 Agents SDK接入多个MCP服务器

• 理论上，MCP一个服务器能同时运行多个外部函数，而一个MCP Client则可以连接多个MCP服务器。Agents SDK本身也是可以作为MCP Client的，因此是完全可以连接多个MCP server。
  

2.1 项目结构

• 图片中项目单词误拼，不在更改，但代码文字描述已修改
  

1. 初始化项目，并创建项目虚拟环境

   uv init agents_multi_server
   cd agents_multi_server
   uv venv
   

2. 创建依赖文件requirements.txt

   mcp
   httpx
   openai
   openai-agents
   

3. 安装项目所需依赖

   uv pip install -r requirements.txt
   

2.2 单一客户端

• client_mulit_agent.py

from openai import AsyncOpenAI 
from agents import OpenAIChatCompletionsModel, Agent, Runner, set_default_openai_client 
from agents.mcp  import MCPServer, MCPServerStdio
from agents.model_settings  import ModelSettings
import asyncio 
from contextlib import AsyncExitStack
from agents import set_tracing_disabledOPENAI_API_KEY = "hk-xxx"
OPENAI_API_BASE = "https://api.openai-hk.com/v1" 
MODEL = "deepseek-v3"external_client = AsyncOpenAI(base_url=OPENAI_API_BASE,api_key=OPENAI_API_KEY,
)set_default_openai_client(external_client)
set_tracing_disabled(True)
deepseek_model = OpenAIChatCompletionsModel(model=MODEL,openai_client=external_client)async def mcp_run_multi(servers_params, message):# 使用 AsyncExitStack 自动管理多个上下文退出async with AsyncExitStack() as stack:servers = []# 创建并进入所有 server 上下文 for p in servers_params:server = MCPServerStdio(name=p.get("name", "Unnamed Server"),cache_tools_list=True,params={"command": "uv","args": ["run", p["script"]],},)entered_server = await stack.enter_async_context(server)servers.append(entered_server)# 构造 agent，传入多个 serveragent = Agent(name="Assistant",instructions=("你是一名助人为乐的助手。请先调用query_weather工具查询北京天气，""然后将查询结果通过write_file工具写入res.md文件中。"),mcp_servers=servers,model_settings=ModelSettings(tool_choice="required"),model=deepseek_model)print(f"Running: {message}")result = await Runner.run(starting_agent=agent, input=message)print(result.final_output)return resultif __name__ == "__main__":# 调用：传入多个 server 的配置 result = asyncio.run(mcp_run_multi( servers_params=[{"name": "Weather Server", "script": "weather_server.py"}, {"name": "Writer Server", "script": "write_server.py"} ],message="请帮我查询Beijing天气，并将查询的结果写入本地res.md文档。"))print(result)

2.3 天气查询服务端

• api测试，appid请到openweathermap.org自行获取

curl -s "https://api.openweathermap.org/data/2.5/weather?q=北京&units=metric&appid=xxx"

---

• weather_server.py

import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP# 初始化mcp服务器
mcp=FastMCP("Weather Server")#OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY ="xxx"
USER_AGENT = "weather-app/1.0"async def fetch_weather(city: str) -> dict[str, Any]|None:"""获取天气信息"""params={"q": city,"appid": API_KEY,"units": "metric","lang": "zh_cn"}headers={"User-Agent": USER_AGENT}async with httpx.AsyncClient() as client:response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers,timeout=1000)if response.status_code == 200:return response.json()else:print(f"Error fetching weather data: {response.status_code}, {response.text}")  # 增加日志输return Nonedef format_weather(data: dict[str,Any] | str)->str:"""解析天气数据字典，提取关键信息并格式化输出。功能：对可能缺失的嵌套数据字段进行容错处理，确保返回内容完整。参数：data: 天气API返回的原始数据字典返回：格式化后的天气信息字符串"""# 基础位置信息（城市、国家）- 缺失时显示"未知"city = data.get("name", "未知")  # 城市名称（顶层字段）country = data.get("sys", {}).get("country", "未知")  # 国家代码（嵌套在sys字段中）# 天气核心指标 - 缺失时显示"N/A"（Not Available）main_data = data.get("main", {})  # 提取main字段（包含温度、湿度等）temperature = main_data.get("temp", "N/A")  # 温度humidity = main_data.get("humidity", "N/A")  # 湿度wind_data = data.get("wind", {})  # 提取wind字段（包含风速等）wind_speed = wind_data.get("speed", "N/A")  # 风速# 天气描述 - weather字段可能为空列表，默认返回第一个元素的描述weather_list = data.get("weather", [{}])  # 提取weather数组（默认空字典避免索引错误）weather_description = weather_list[0].get("description", "未知")  # 天气状况描述# 格式化输出字符串（使用f-string拼接，添加emoji直观展示）weather_info = (f"🌍 {city}, {country}\n"f"🌡️ 温度：{temperature}℃\n"f"💧 湿度：{humidity}%\n"f"💨 风速：{wind_speed} m/s\n"f"☁️ 天气：{weather_description}\n")return [e-mail]()
async def query_weather(city: str) -> str:"""查询天气信息并返回结果注意：当大模型调用此工具时，必须使用参数名 'city' 来传递城市名称。city 参数只支持英文城市名或中文城市的拼音形式，不支持中文城市名直接输入。Args:city (str): 要查询天气的城市名称，仅支持英文或中文拼音例如：Beijing, Shanghai, london, tokyoReturns:str: 格式化的天气信息字符串，包含温度、湿度、风速和天气描述等信息Example:query_weather(city="Beijing")query_weather(city="Shanghai")query_weather(city="london")"""weather_data = await fetch_weather(city)if weather_data:return format_weather(weather_data)else:return "无法获取天气信息。请检查城市名称是否正确，确保使用英文或拼音格式。"if __name__=="__main__":mcp.run(transport='stdio')

2.4 文件写入服务端

import json
import httpx
from typing import Any
from pathlib import Path 
from mcp.server.fastmcp  import FastMCP# 初始化 MCP 服务器
mcp = FastMCP("Write Server")@mcp.tool() 
async def write_file(content: str) -> str:"""将指定内容写入本地文件。:param content: 必要参数，字符串类型，用于表示需要写入文档的具体内容:return：字符串，表示是否成功写入"""try:with open("res.md", "w", encoding="utf-8") as file:file.write(content)return f"已成功写入本地文件(res.md)，内容长度：{len(content)} 字符"except Exception as e:return f"写入文件失败: {str(e)}"if __name__ == "__main__":# 以标准 I/O 方式运行 MCP 服务器mcp.run(transport='stdio')

2.5 测试结果

Running: 请帮我查询Beijing天气，并将查询的结果写入本地res.md文档。
北京的天气信息已成功写入到 `res.md` 文件中，内容如下：🌍 Beijing, CN
🌡️ 温度：33.77℃
💧 湿度：59%
💨 风速：3.86 m/s
☁️ 天气：小雨如果需要进一步帮助，请随时告诉我！
RunResult:
- Last agent: Agent(name="Assistant", ...)
- Final output (str):北京的天气信息已成功写入到 `res.md` 文件中，内容如下：🌍 Beijing, CN🌡️ 温度：33.77℃💧 湿度：59%💨 风速：3.86 m/s☁️ 天气：小雨如果需要进一步帮助，请随时告诉我！
- 7 new item(s)
- 3 raw response(s)
- 0 input guardrail result(s)
- 0 output guardrail result(s)
(See `RunResult` for more details)

2.6 双地区天气查询保存执行流程

• Agents SDK对于MCP实现过程基本遵照Function calling来执行。整体执行流程如下：
  

• 修改client_multi_server.py文件，同时查询两个地区天气，进行写入。

if __name__ == "__main__":# 调用：传入多个 server 的配置 result = asyncio.run(mcp_run_multi( servers_params=[{"name": "Weather Server", "script": "weather_server.py"}, {"name": "Writer Server", "script": "write_server.py"} ],message="请帮我查询北京和南京天气，并将查询的结果写入本地res.md文档。"))print(result)