文章目录
- 说明
- 一 Agents SDK接入MCP
- 1.1 MCP技术回顾
- 1.2 MCP基础实践流程
- 1.2.1 天气查询服务器Server创建流程
- 1.2.2 服务器依赖安装和代码编写
- 1.2.3 环境配置文件
- 1.2.4 客户端代码编写
- 1.3 测试运行
- 二 MCP+Agents SDK基础调用
- 2.1 weather_server.py
- 2.2 client_agent.py
- 2.3 运行测试
说明
- 本文学自赋范社区公开课，仅供学习和交流使用，不用作任何商业用途！
一 Agents SDK接入MCP
1.1 MCP技术回顾
- 开篇：MCP理论理解和学习
- 启程：MCP开发环境配置和旅游攻略案例体验
- https://yuanyou.blog.csdn.net/article/details/148222368
1.2 MCP基础实践流程
- 先尝试手动实现一遍
MCP实践流程，再考虑将已经部署好的server带入Agents SDK中，作为tools进行调用。 - 一个极简的天气查询MCP基本执行流程：
1.2.1 天气查询服务器Server创建流程
- 创建一个天气查询的服务器，通过openweather，创建一个能够实时查询天气的服务器（server）。
curl -s "https://api.openweathermap.org/data/2.5/weather?q=Beijing&units=metric&appid=xxx" - 执行结果：
{"coord": {"lon": 116.3972,"lat": 39.9075},"weather": [{"id": 804,"main": "Clouds","description": "阴，多云","icon": "04n"}],"base": "stations","main": {"temp": 22.36,"feels_like": 22.77,"temp_min": 22.36,"temp_max": 22.36,"pressure": 1007,"humidity": 81,"sea_level": 1007,"grnd_level": 1002},"visibility": 10000,"wind": {"speed": 1.42,"deg": 26,"gust": 3.23},"clouds": {"all": 100},"dt": 1753014180,"sys": {"country": "CN","sunrise": 1752958921,"sunset": 1753011546},"timezone": 28800,"id": 1816670,"name": "Beijing","cod": 200 }
1.2.2 服务器依赖安装和代码编写
- 创建项目目录，创建并激活虚拟环境。
uv init mcp-weather cd mcp-weather uv venv .venv\Scripts\activate - 在当前虚拟环境中添加如下依赖：
pip install mcp httpx openai python-dotenv pypinyin openai-agents - 创建server服务器代码文件
server.py：import json import httpx from typing import Any from mcp.server.fastmcp import FastMCP# 初始化mcp服务器 mcp=FastMCP("WeatherServer")#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: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 weather_info@mcp.tool() async def query_weather(city: str) -> str:"""查询天气信息并返回结果"""weather_data = await fetch_weather(city)if weather_data:return format_weather(weather_data)else:return "无法获取天气信息。请检查城市名称是否正确。"if __name__=="__main__":mcp.run(transport='stdio')
1.2.3 环境配置文件
- 创建.env文件
BASE_URL="https://api.siliconflow.cn/v1/chat/completions"
MODEL=deepseek-ai/DeepSeek-V3
OPENAI_API_KEY="sk-xxx"
1.2.4 客户端代码编写
- 创建一个可以和
server进行通信的客户端，需要注意的是，该客户端需要包含大模型调用的基础信息。我们需要编写一个client.py脚本。import asyncio import os import json import sys from typing import Optional from contextlib import AsyncExitStack from openai.types.chat import ChatCompletionToolParam from openai import OpenAI from dotenv import load_dotenvfrom mcp import ClientSession,StdioServerParameters from mcp.client.stdio import stdio_clientfrom pypinyin import lazy_pinyin, Style# 加载env文件，确保配置正确 load_dotenv()class MCPClient:def __init__(self):"""初始化MCP客户端"""self.write = Noneself.stdio = Noneself.exit_stack = AsyncExitStack()self.base_url=os.getenv("BASE_URL")self.model = os.getenv("MODEL")self.openai_api_key = os.getenv("OPENAI_API_KEY")if not self.openai_api_key:raise ValueError("OPENAI_API_KEY未设置")self.client=OpenAI(api_key=self.openai_api_key, base_url=self.base_url)self.session: Optional[ClientSession] = Noneself.exit_stack=AsyncExitStack()async def connect_to_server(self, server_script_path:str):"""连接MCP服务器并列出可用工具"""is_python=server_script_path.endswith(".py")is_js=server_script_path.endswith(".js")if not (is_python or is_js):raise ValueError("服务器脚本必须以.py或.js结尾")command="python" if is_python else "node"server_params=StdioServerParameters(command=command,args=[server_script_path],env=None)# 启动MCP服务器并建立通信stdio_transport=await self.exit_stack.enter_async_context(stdio_client(server_params))self.stdio,self.write=stdio_transportself.session= await self.exit_stack.enter_async_context(ClientSession(self.stdio,self.write))await self.session.initialize()# 列出MCP服务器上的工具response=await self.session.list_tools()tools=response.toolsprint("\n已连接到服务器，支持以下工具：",[tool.name for tool in tools])async def process_query(self, query:str)-> str:"""使用大模型处理查询并调用可用的MCP工具（Function Calling）"""messages=[{"role": "user","content": query}]response=await self.session.list_tools()available_tools = [ChatCompletionToolParam(type="function",function={"name": tool.name,"description": tool.description,"parameters": tool.inputSchema})for tool in response.tools]response= self.client.chat.completions.create(model=self.model,messages=messages,tools=available_tools)# 处理返回的内容content=response.choices[0]# 检查是否使用了工具if content.finish_reason == "tool_calls":# 何时需要使用工具就解析工具tool_call = content.message.tool_calls[0]tool_name=tool_call.function.nametool_args=json.loads(tool_call.function.arguments)# 如果调用的是 query_weather 工具，处理城市名称if tool_name == "query_weather" and "city" in tool_args:city = tool_args["city"]# 简单判断是否为中文城市名if any('\u4e00' <= c <= '\u9fff' for c in city):# 转换为拼音，首字母大写pinyin_city = ''.join([word.capitalize() for word in lazy_pinyin(city)])tool_args["city"] = pinyin_city# 执行工具result=await self.session.call_tool(tool_name, tool_args)print(f"\n\n[Calling Tool: {tool_name} with args: {tool_args}]")# 将工具调用和结果添加到消息历史中messages.append(content.message.model_dump())messages.append({"role": "tool","content": result.content[0].text,"tool_call_id": tool_call.id})# 将上面的结果再返回给大模型用于最终的效果response=self.client.chat.completions.create(model=self.model,messages=messages)return response.choices[0].message.content# 如果调用工具直接返回结果return content.message.contentasync def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP客户端已启动！输入'quit'退出")while True:try:query=input("\n you:").strip()if query.lower() == "quit":breakresponse=await self.process_query(query)print(f"\n ai: {response}")except Exception as e:print(f"\n Error: {e}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():if len(sys.argv)<2:print("Usage: python client.py <server_address>")sys.exit(1)client=MCPClient()try:await client.connect_to_server(sys.argv[1])await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":asyncio.run(main())
1.3 测试运行
-
进入项目目录，激活虚拟环境
cd ./mcp-weather source .venv/bin/activate -
运行MCP客户端和服务器
uv run client.py server.py(mcp-weather) D:\Code\mcp-study\mcp-weather>uv run client.py server.py已连接到服务器，支持以下工具： ['query_weather']MCP客户端已启动！输入'quit'退出you:请问北京今天天气如何？[Calling Tool: query_weather with args: {'city': 'BeiJing'}]ai: 北京今天的天气情况如下：🌍 **北京，中国** 🌡️ **温度**：22.85℃ 💧 **湿度**：74% 💨 **风速**：2.14 m/s ☁️ **天气状况**：阴天，多云请根据实际需要增减衣物，出行注意安全！
二 MCP+Agents SDK基础调用
Agents SDK可以将某个对应的Agent封装为client与外部定义好的server进行通信。客户端为client_agent.py、服务端weather_server.py
2.1 weather_server.py
import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP# 初始化mcp服务器
mcp=FastMCP("WeatherServer")#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 weather_info@mcp.tool()
async def query_weather(city: str) -> str:"""查询天气信息并返回结果"""weather_data = await fetch_weather(city)if weather_data:return format_weather(weather_data)else:return "无法获取天气信息。请检查城市名称是否正确。"if __name__=="__main__":mcp.run(transport='stdio')
2.2 client_agent.py
import asyncio
import time
from openai import AsyncOpenAI
from agents import OpenAIChatCompletionsModel,Agent, Runner, gen_trace_id, trace, set_default_openai_client
from agents.mcp import MCPServer, MCPServerStdio
from agents.model_settings import ModelSettings
from agents import set_tracing_disabled # or from your framework's module
OPENAI_API_KEY="hk-xxx"
OPENAI_API_BASE="https://api.openai-hk.com/v1"
MODEL="deepseek-v3"# 创建一个Agent对象并调用DeepSeek模型
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 run(mcp_server: MCPServer):agent = Agent(name="Assistant",instructions="你是一名助人为乐的助手",mcp_servers=[mcp_server],model=deepseek_model)message = "请帮我查询Beijing天气如何？"print(f"Running: {message}")result = await Runner.run(starting_agent=agent, input=message)print(result.final_output)async def mcp_run():async with MCPServerStdio(name="Weather Server",cache_tools_list=True,params = {"command": "uv","args": ["run", "weather_server.py"]} ) as server:await run(server)if __name__ == "__main__":asyncio.run(mcp_run())
2.3 运行测试
(mcp-weather) D:\Code\mcp-study\mcp-weather>uv run client_agent.py
Running: 请帮我查询Beijing天气如何？
北京当前的天气情况如下：- **温度**：34.66℃
- **湿度**：54%
- **风速**：2.78 m/s
- **天气状况**：晴天气晴朗，但温度较高，请注意防晒和补水！
