Agent MCP

🚀 MCP 服务器

本项目参考 Antropic 的 模型上下文协议文档 编写了本地 Web 搜索 MCP 服务器，并使用 Google SerpAPI 作为搜索 API。

🚀 快速开始

MCP 服务器

本项目参考 Antropic 的 模型上下文协议文档 编写了本地 Web 搜索 MCP 服务器，使用 Google SerpAPI 作为搜索 API。

MCP 客户端

这是一个本地 MCP 客户端，它可以调用本地部署的 MCP 服务器，也能调用外部平台 Smithery 提供的公开 MCP 服务器服务。Smithery 平台。该客户端参考 Antropic MCP Client Quick Start，将调用接口（原文使用 Anthropic API）修改为 OpenAI 的接口，并调整了后处理流程，还增加了一些 Json 验证。经过测试，使用 Qwen2.5 系列模型和 DeepSeek - V3 时效果都比较稳定。

💻 使用示例

基础用法

以下是在 2025.4.22 更新中，custom_agents 文件夹规划 + Researcher Agent 模仿深度研究功能，使用 MCP 服务器的代码示例：

async with MCPServerStdio(
    params={"command": "python", "args": [LOCAL_SERVER_PATH]}
) as server:
    # 列出工具
    tools = await server.list_tools()
    print(f"连接到服务器的工具：{tools}")
    # 规划代理
    planning_agent = PlanningAgent(
        name="规划代理",
        model=model,
        handoffs=[
            ResearchAgent(
                name="研究代理",
                model=model,
                mcp_servers=[server],
            )
        ],
    )
    # 运行
    result_stream = Runner.run_streamed(
        starting_agent=planning_agent,
        input="2025 年 AI 的主要焦点是什么？给我一些趋势学术论文的关键点和 2025 年新技术的概述",
        max_turns=15,
    )
    async for event in result_stream.stream_events():
        if event.type == "agent_updated_stream_event":
            print(f"代理更新：{event.new_agent.name}")
        elif event.type == "run_item_stream_event":
            if isinstance(event.item, ToolCallItem):
                print(f"调用工具：{event.item.raw_item.name}，参数：{event.item.raw_item.arguments}")
                await asyncio.sleep(0.05)
            elif isinstance(event.item, ToolCallOutputItem):
                text = json.loads(event.item.output)["text"]
                print(f"工具调用输出：{text[:100]} ...")
                await asyncio.sleep(0.05)
            elif isinstance(event.item, MessageOutputItem):
                text = ItemHelpers.text_message_output(event.item)
                if text:
                    print(f"运行步骤：{text}")
                    await asyncio.sleep(0.05)
        elif event.type == "raw_response_event":
            try:
                if event.data.delta:
                    continue
            except Exception as e:
                print(f"事件数据：{event.data}")