MCP AI Infra Real Time Agent

第 1 部分：使用 MCP 工具执行的实时 LangGraph Agent

本项目展示了一种解耦的实时 Agent 架构，可将 LangGraph Agent 连接到由自定义 MCP（模块化命令协议）服务提供的远程工具。该架构支持灵活、可扩展的多 Agent 系统，每种工具都可以独立托管（通过 SSE 或 STDIO），实现模块化和可部署于云端的执行。

• 解耦架构： 构建了一个模块化系统，由基于 LangGraph 的 Agent 编排大模型工作流，同时通过 SSE 和 STDIO 传输将工具执行委托给远程 MCP 服务。
• 高级异步编程： 使用 Python 的 async/await 实现非阻塞 I/O，确保多个工具并发执行，并支持可扩展的实时通信。
• MCP 与 LangGraph 集成： 展示了将模块化命令协议（MCP）与 LangGraph 和 LangChain 集成的深厚专业能力，支持对分布式工具进行无缝转换和调用。
• 灵活的多服务连接： 设计了一个 MultiServerMCPClient，支持与不同工具服务建立一对一绑定，体现了系统集成多种环境（本地、云端、容器化）的能力。
• 可靠的 Agent 与工具通信： 实现了详细的客户端会话、握手协议和动态工具发现，确保 Agent 与 MCP 服务之间能够可靠地执行和交互。
• 面向未来的互操作性： 为 Agent2Agent 协议奠定了基础，旨在构建一个 AI Agent 可共享能力、协调操作并安全交换上下文和数据的生态系统。

---

🚀 项目目的

本项目旨在：

• 将基于大模型的 Agent 编排（LangGraph）与工具执行（通过 MCP 服务）解耦。
• 使用 MCP 协议实现实时、多服务且与语言无关的工具集成。
• 展示如何：
  • 启动兼容 LangChain 的 MCP 工具服务（例如 math_server.py、weather_server.py）
  • 将其与 LangGraph ReAct Agent 集成
  • 使用 async/await 编程，在 Agent 和工具服务之间实现非阻塞 I/O

---

📦 项目结构

.
├── servers/
│   ├── math_server.py         # STDIO-based MCP tool server
│   └── weather_server.py      # SSE-based MCP tool server
├── client/
│   ├── multiserver_client.py  # LangGraph agent using MultiServer MCP client
│   └── stdio_client.py        # LangGraph agent using STDIO transport

---

🔧 技术栈

• 🧠 LangGraph：ReAct Agent 编排
• 🔗 LangChain：大模型流水线与工具抽象
• 🧰 MCP（模块化命令协议）：
  • FastMCP – 基于 FastAPI 的服务抽象
  • ClientSession、StdioServerParameters、MultiServerMCPClient
• 🌐 SSE 与 STDIO：传输协议
• 🔁 AsyncIO：异步并发
• ☁️ OpenAI：后端大模型（通过 langchain_openai）
• 🧪 dotenv：API 密钥管理

---

📜 源代码说明

1. math_server.py 和 weather_server.py

使用 FastMCP 的工具服务：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    return "Cold in Chicago"

if __name__ == "__main__":
    mcp.run(transport="sse")

✅ 亮点：

• 每个服务都通过 @mcp.tool() 定义一个异步工具
• 服务与传输方式无关：可以通过 SSE 或 STDIO 运行
• 专为模块化部署而设计：支持本地、云端和容器化部署

2. multiserver_client.py

可与多个 MCP 服务并发通信的 Agent：

async with MultiServerMCPClient({
    "math": {"command": "python", "args": ["math_server.py"]},
    "weather": {"url": "http://localhost:8000/sse", "transport": "sse"}
}) as client:
    agent = create_react_agent(llm, client.get_tools())
    result = await agent.ainvoke({"message":"what's 1+1?"})

✅ 亮点：

• MultiServerMCPClient 支持与多个服务建立一对一绑定
• 所有工具调用均为异步，并通过相应传输方式流式传输
• 工具会自动转换为兼容 LangChain 的格式

3. stdio_client.py

使用原始 ClientSession 连接到一个 STDIO MCP 服务的 Agent：

async with stdio_client(StdioServerParameters(...)) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await load_mcp_tools(session)
        agent = create_react_agent(llm, tools)
        result = await agent.invoke({"messages": [HumanMessage(content="What's 1+1?")]})

✅ 亮点：

• ClientSession.initialize() 会建立符合 MCP 规范的握手
• 通过 session.list_tools() 和 load_mcp_tools() 动态发现工具
• 展示对 I/O 流的底层控制能力

🔄 LangGraph 与 MCP 如何交互

Agent 工作流：

1. 使用 MCP 客户端中的工具创建 Agent
2. Agent 接收用户查询
3. Agent 确定所需工具
4. MCP 客户端向远程服务发送工具调用请求
5. 远程服务执行工具，并将结果发回
6. Agent 使用结果并返回最终响应

架构图：

+----------------+         +----------------------+        +------------------------+
| LangGraph App  |<------->| MultiServerMCPClient |<-----> |   Remote MCP Servers   |
| (ReAct Agent)  |         |  (Tool Wrapper Layer)|        | (math_server, weather) |
+----------------+         +----------------------+        +------------------------+
                                 |                                ^
                                 |                                |
                                 +--------------------------------+
                                       Async I/O (SSE or STDIO)

---

🧠 异步用法与优势

整个系统使用 async/await 来：

• 避免在等待工具响应时阻塞
• 支持并发执行多个工具
• 实现可扩展的实时工具编排

所有 I/O——无论是基于网络（SSE）还是基于管道（STDIO）——都以异步方式处理，从而最大限度地提高响应能力和吞吐量。

---

✅ 优点

• 模块化且可扩展：工具服务可以独立扩展
• 语言无关：MCP 规范支持 Python、Node.js 和容器化服务
• 实时执行：SSE 和 STDIO 传输方式可实现实时交互
• 兼容 LangChain：全面支持 LangChain 和 LangGraph 工作流

---

⚠️ 挑战

• 工具发现延迟：初始握手会增加少量开销
• 传输复杂性：管理多种传输类型（SSE 与 STDIO）可能并非易事
• 错误处理：必须妥善处理工具服务故障或超时
• 部署策略：生产环境需要编排层（例如 Docker Compose、K8s）

---

🔮 未来方向

• 使用 StdioServerParameters 并支持容器的容器化工具服务
• 用于安全访问服务的身份验证层
• 可观测性：工具使用情况的日志、追踪和实时仪表板
• LangGraph 并行节点：在并行子图中运行多个 MCP 调用
• 多模态工具：扩展 MCP 工具以支持图像／音频输入

---

🧩 相关概念

• MCP：用于定义和调用模块化工具的协议
• LangGraph：受状态机启发的 Agent 推理框架
• ReAct：用于结构化决策的推理与行动范式
• LangChain 工具：将外部函数／API 转换为可供大模型调用的工具

---

📌 结论

本项目展示了大模型应用开发中清晰的职责划分：

• LangGraph 专注于 Agent 逻辑和编排
• MCP 服务负责实际任务执行
• 异步客户端连接两者，提供实时通信

这是一个面向未来的架构，可通过模块化、可观测且易于维护的组件构建企业级大模型应用。

✨ Agent2Agent 协议

AI Agent 往往孤立于特定应用或平台中，缺乏与其他厂商或使用不同框架构建的 Agent 通信、共享信息或协调行动的通用方式。A2A 定义了 Agent 发现能力的标准方式；Agent 可以公布其功能，让其他 Agent 了解它们能做什么。Agent 可以分配和跟踪任务，包括复杂的长期运行任务，并交换状态更新和结果。Agent 可以安全地交换包含上下文、指令或数据的消息。Agent 可以根据用户界面能力，就展示信息的最佳格式（文本、图像）达成一致。我们的目标是创建一个可互操作的生态系统，让 AI Agent 能够在不同企业应用之间无缝协作。

---

第 2 部分。MCP-AI-Infra-Real-Time-Agent

构建了基于 MCP 的 AI 基础设施，为 Claude 和 Cursor 等 AI 客户端提供实时工具调用、结构化知识检索和动态 Agent 交互。

• 设计并实现了基于 MCP 的 AI 系统，为 Claude 和 Cursor 等 AI 客户端提供实时工具调用、结构化知识检索和 Agent 工作流。
• 开发了 MCP 服务—客户端架构，支持无缝的大模型交互，并提供工具（get_forecast、get_alerts）、资源（API 响应、文件内容）和提示词（结构化任务模板）。
• 构建了动态工具执行框架，允许 AI 模型在获得用户批准后调用外部 API 服务，从而改进 AI 辅助决策和自动化。
• 将 MCP 集成到由 LangGraph 驱动的检索增强生成（RAG）工作流中，优化了上下文文档检索和结构化响应生成。
• 通过设计 AI 组件既可作为客户端又可作为服务端的 MCP 协议，实现了可组合的 AI Agent，支持多层 Agent 交互和模块化扩展。
• 将 MCP 协议用作通用 AI 接口，增强系统互操作性，使不同宿主环境都能即插即用地使用 AI 功能。
• 构建了可自我演进的工具注册 API，支持动态能力发现和运行时工具注册，以适应不断变化的 AI 工作流和 Agent 系统。
• 通过缓存和并行请求处理优化 AI 工具执行，提升 MCP 服务响应速度和大模型推理效率。
• 使用 Anthropic 的 MCP Inspector 进行交互式调试和测试，改进 AI Agent 的行为和工具执行流程。
• 开发了可扩展的框架，便于未来在 MCP 生态系统中集成更多 AI 驱动的实用工具、自动化 Agent 和外部 API 服务。

项目概述

MCP-Servers 项目致力于实现和扩展 MCP（模型控制协议）服务，为 Claude 和 Cursor 等 AI 系统提供实时且以文档为依据的响应。项目目标是集成 MCP 客户端—服务架构，使 AI 模型能够访问结构化知识并动态调用指定工具。

核心目标

1. MCP 客户端—服务集成

• 实现 MCP 服务，连接 Claude 3.7 Sonnet Desktop 和 Cursor 等 AI 客户端。
• 使用现有的 MCP 框架（例如 mcpdoc），避免重复实现核心功能。

2. 扩展 MCP 服务功能

• 为 MCP 服务开发自定义工具，特别是用于获取天气预报和警报等外部数据的工具。
• 将这些功能作为 MCP 工具（get_forecast、get_alerts）开放，供 AI 客户端使用。

3. 增强 AI 工具执行能力

• 使 AI 模型能够在获得用户批准后调用工具，与 MCP 服务交互。
• 确保妥善处理资源（例如 API 响应、文件内容）和提示词（用于结构化任务的预置模板）。

---

MCP 架构与工作流

1. MCP：通用 AI 接口

• MCP 充当互操作层，让外部 AI 应用（Claude、Cursor 等）能够与结构化数据源和可执行函数交互。
• MCP 采用类似 USB-C 的架构，MCP 服务就像外部插件，可连接到各种 AI 系统。

2. MCP 客户端-服务端角色

MCP 客户端（嵌入在 Claude 或 Cursor 等 AI 宿主中）

• 请求工具、查询资源并处理提示词。
• 充当 AI 系统与 MCP 服务之间的桥梁。

MCP 服务（本地实现）

• 向 AI 客户端公开工具（例如天气 API），供其动态调用。
• 提供资源（例如 API 响应、数据库查询结果）。
• 处理提示词，以支持结构化的用户交互。

---

主要功能与未来增强

• Agent 式组合能力：该架构支持多层 Agent 式交互，AI Agent 既可充当 MCP 客户端，也可充当 MCP 服务。这使模块化、专用的 Agent 能够处理不同任务。
• 通过 Registry API 实现自我进化的 AI：未来版本可以支持动态发现工具，让 AI 客户端能够实时注册和发现新的 MCP 功能。
• 开发与调试支持：使用 Anthropic 的 MCP Inspector 以交互方式测试和调试 MCP 交互，无需完整部署。

---

结论

本项目构建了由 MCP 驱动的 AI 基础设施，将 AI 模型与实时结构化知识相连，通过自定义工具执行扩展其能力，并增强 Agent 式组合能力。我们的目标是创建一个自适应、类似插件的 AI 系统，能够集成到多个宿主中，并通过工具注册和运行时发现动态演进。

附录

• 不要重复造轮子

MCP 就像 USB-C，MCP 服务就像可以连接到 AI（Claude Desktop）或云应用的外部设备。我们只需编写一次功能，就能将其接入多个 MCP 宿主。MCP 客户端位于 MCP 宿主内部，通过 MCP 协议与 MCP 服务进行一对一交互。MCP 客户端调用工具、查询资源、插入提示词；MCP 服务则公开工具（由模型控制：检索、更新数据库、发送）、资源（由应用控制：数据库记录、API）和提示词（由用户控制：文档）。

MCP + 容器化

使用 UV 初始化项目，使用 UV 创建虚拟环境，安装依赖项（MCP [CLI]），使用 Cursor 为官方 MCP 文档建立索引，并使用 Cursor 规则更新项目

氛围编程

• @server.py 根据 @MCP 帮我实现一个简单的 MCP 服务。请使用 Python SDK @MCP Python SDK 编写代码。该服务应公开一个名为 terminal tool 的工具，让用户能够运行终端命令，尽量保持简单。
• 帮我在我的 MCP 服务 @MCP 中公开一个资源，同样使用 @MCP Python SDK 编写代码。我想公开 Desktop 目录下的 mcpreadme.md。

致谢

MCP Crash Course: Complete Model Context Protocol in a Day