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

文章目录

一 Agents SDK+MCP进阶
- 1.1 Agents SDK接入开源MCP服务器
- 1.2 MCP-Server-Git
二 Agents SDK接入多个MCP服务器

一 Agents SDK+MCP进阶

1.1 Agents SDK接入开源MCP服务器

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

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 工作目录当前状态的文本描述。

安装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"

测试是否安装成功
```
uv --version
uvenv --help
```

python脚本(未测试脚本)

import asyncio
import shutil

from agents import Agent, Runner, trace
from agents.mcp import MCPServer, MCPServerStdio


async 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 项目结构

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

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

uv init agents_multi_server
cd agents_multi_server
uv venv

创建依赖文件requirements.txt
```
mcp
httpx
openai
openai-agents
```
安装项目所需依赖
```
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_disabled

OPENAI_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，传入多个 server
        agent = 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 result

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="请帮我查询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 None

def 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:
    """
    查询天气信息并返回结果
    
    注意：当大模型调用此工具时，必须使用参数名 'city' 来传递城市名称。
    city 参数只支持英文城市名或中文城市的拼音形式，不支持中文城市名直接输入。
    
    Args:
        city (str): 要查询天气的城市名称，仅支持英文或中文拼音
                    例如：Beijing, Shanghai, london, tokyo
        
    Returns:
        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)

在这里插入图片描述

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

文章目录

一 Agents SDK+MCP进阶

1.1 Agents SDK接入开源MCP服务器

1.2 MCP-Server-Git

二 Agents SDK接入多个MCP服务器

2.1 项目结构

2.2 单一客户端

2.3 天气查询服务端

2.4 文件写入服务端

2.5 测试结果

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

网站公告

今日签到

热门文章

最新发布