MCP 与工具集成

学习目标

理解 CrewAI 如何集成 MCP 协议，包括三种传输方式、工具解析机制、Schema 缓存，以及 MCP 工具与本地工具的混用方式。

前置知识

• 工具调用协议与执行模型 — Function Calling 协议与工具系统
• MCP 协议与生态集成 — Model Context Protocol 架构
• 多服务器 MCP 架构 — 多客户端与工具分组

下文假设你已理解上述概念，直接聚焦 CrewAI 的具体实现。

---

项目实践

MCPClient 连接管理

MCPClient 管理 BaseTransport，使用 AsyncExitStack 进行上下文管理：

# 伪代码
client = MCPClient()
await client.connect(transport_config)  # Stdio/SSE/HTTP
tools = await client.list_tools()
result = await client.call_tool("tool_name", {"arg": "value"})
await client.disconnect()

支持指数退避重试（最多 3 次），30 秒连接超时，事件总线集成。

三种传输方式

传输	实现	特点
StdioTransport	启动本地进程	合并 os.environ 与自定义环境变量
SSETransport	SSE 客户端	适用于远程 MCP 服务器
HTTPTransport	流式 HTTP	可流式传输，适合低延迟场景

MCPToolResolver 工具解析

MCPToolResolver 解析三种 MCP 引用类型：

1. 原生配置：MCPServerStdio/HTTP/SSE 配置对象
2. HTTPS URL：直接解析为 HTTP 传输的 MCP 服务器
3. AMP 引用：如 "notion" 或 "notion#search"

JSON Schema 到 Pydantic Model 的自动转换，5 分钟 Schema 缓存。

MCP 工具与本地工具混用

Agent 的工具可来自多个渠道：

• self.tools — 直接配置的本地工具
• get_mcp_tools(mcps) — MCP 服务器工具
• get_platform_tools(apps) — CrewAI Platform 的 AMP 工具
• create_memory_tools(memory) — 记忆检索工具

所有工具统一转换为 CrewStructuredTool，通过 ToolUsage 编排器执行。

工具执行核心

ToolUsage 通过 4 种尝试解析 LLM 输出中的工具调用：JSON → AST → json5 → repair_json。最多 3 次重试，包含缓存查找、重复使用检测、遥测上报。

钩子系统：before_tool_call 可阻止工具执行；after_tool_call 可修改结果。

---

问题与规避

MCP 服务器启动失败时的降级

• 如果 MCP 服务器进程启动失败，Agent 会缺少该服务器的工具
• 对策：在 Agent 初始化时检查 MCP 连接状态，不可用时使用本地工具替代

Schema 缓存过期

• 5 分钟 TTL 的 Schema 缓存可能在 MCP 服务器更新工具后返回过时的 Schema
• 对策：在 MCP 服务器更新工具后主动调用 disconnect() + connect() 刷新缓存

HTTPS URL 引用的认证

• HTTPS URL 引用的 MCP 服务器可能需要认证，CrewAI 默认不传递凭证
• 对策：使用 MCPServerConfig 显式配置认证信息

---

设计取舍

client_factory 模式

• MCPClient 支持 client_factory 模式：每次调用创建独立客户端
• 优势：避免多 Agent 共享连接导致的状态污染
• 代价：每次调用都需要重新建立连接，增加延迟

Schema 缓存 vs 实时获取

• 缓存降低延迟但可能返回过时信息
• 5 分钟 TTL 是平衡点——MCP 服务器工具定义通常不频繁变化

---

参考来源

• 源码验证: lib/crewai/src/crewai/mcp/client.py
• 源码验证: lib/crewai/src/crewai/mcp/tool_resolver.py
• 源码验证: lib/crewai/src/crewai/mcp/transports/stdio.py