PydanticAI 的 MCP 集成实践

PydanticAI 的 MCP 集成实践

学习目标

补充理解：

• MCPToolset 基于 FastMCP 客户端的实现
• 传输层自动推断（SSE vs StreamableHttp）
• 缓存失效与通知机制
• 旧 MCPServer 体系的弃用路径

前置知识

• MCP 协议与生态集成 — MCP 通用概念

---

1. MCPToolset 架构

PydanticAI 的 MCP 集成经历了两代架构：

新一代：MCPToolset（基于 FastMCP 客户端）

MCPToolset
    ├── 构造函数：接受 URL / 脚本路径 / ClientTransport / FastMCP Server 实例
    ├── _build_transport()：自动推断传输类型
    │   ├── URL 匹配 SSE 模式 → SSETransport
    │   ├── URL 匹配 StreamableHttp 模式 → StreamableHttpTransport
    │   └── 本地脚本路径 → StdioTransport
    ├── get_tools()：list_tools() → 转换为 ToolsetTool 字典
    ├── call_tool()：委托给 client.call_tool()
    ├── 缓存：tools/resources/prompts + 通知失效
    └── 生命周期：引用计数 __aenter__/__aexit__ + anyio.Lock

旧一代：MCPServer 层次结构（已弃用）

• MCPServer（抽象基类）→ MCPServerStdio、MCPServerSSE、MCPServerStreamableHTTP
• 使用原始 MCP SDK ClientSession + 手动流管理
• _MCPSessionState 管理单个后台会话任务
• 将在 v2 移除

2. 传输层自动推断

_build_transport() 方法根据 URL 模式自动推断传输类型：

URL 包含 /sse 或 /sse/messages → SSETransport
URL 包含 /mcp 或其他 → StreamableHttpTransport
本地文件路径 → StdioTransport

这减少了用户配置负担——用户只需提供 URL，框架自动选择正确的传输层。

3. 缓存与通知失效

MCP 工具集使用三层缓存：

tools_cache: dict[str, ToolsetTool]
resources_cache: dict[str, ...]
prompts_cache: dict[str, ...]

失效机制：通过 MCP 通知触发缓存失效：

• notifications/tools/list_changed → 工具缓存失效
• notifications/resources/list_changed → 资源缓存失效
• notifications/prompts/list_changed → 提示缓存失效

潜在问题：如果 MCP 服务器不发送 notifications，缓存可能过期。用户需要手动刷新或配置失效策略。

4. 结构化内容优先

工具结果处理优先使用 result.structured_content（当所有部分都是文本时）：

• JSON 编码的结构化内容直接解析
• 回退到通过 _map_mcp_tool_results() 映射 content blocks

这确保了当 MCP 服务器提供结构化输出时，PydanticAI 能直接使用类型化数据而非原始文本。

5. MCP 任务执行

支持 SEP-1686 持久任务（use_task=True）：

• 工具调用作为持久任务执行
• 支持长时间运行的工具调用
• 任务状态追踪与恢复

6. load_mcp_toolsets()

从 JSON 配置文件加载 MCP 服务器（Claude Desktop / Cursor 格式）：

{
  "mcpServers": {
    "filesystem": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path"] }
  }
}

服务器名称作为工具名称的前缀（如 filesystem_read_file），避免多服务器间的工具命名冲突。