Agents-SDK智能體開發[4]之集成MCP入門

文章目錄

說明
一 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
- **天氣狀況**：晴天氣晴朗，但溫度較高，請注意防曬和補水！