在 CrewAI 中将 MCP 服务器用作工具

本页面包含以下内容：

• 概述
• 安装步骤
• 核心概念与入门指南
• 工具筛选方法
• 采用字典式索引访问特定工具
• 向 MCPServerAdapter 构造函数传递工具名称列表
• 与 CrewBase 结合使用
• 探索 MCP 集成方案
• MCP 使用安全注意事项
• 安全警告：DNS 重绑定攻击
• 局限性
• MCP 集成详情

在 CrewAI 中将 MCP 服务器用作工具

了解如何借助 crewai-tools 库，在 CrewAI 智能体中集成 MCP 服务器并将其用作工具。

概述

模型上下文协议（Model Context Protocol，简称 MCP）提供了一种标准化方式，能让人工智能智能体通过与外部服务（即 MCP 服务器）通信，向大型语言模型（LLMs）提供上下文信息。crewai-tools 库对 CrewAI 的功能进行了扩展，允许你将这些 MCP 服务器中的工具无缝集成到智能体中。这使得你的智能体团队（crews）能够获取丰富的功能生态。

目前，我们支持以下几种传输方式：

• 标准输入输出（Stdio）：适用于本地服务器（同一台机器上的进程之间通过标准输入/输出进行通信）
• 服务器发送事件（SSE）：适用于远程服务器（通过 HTTP 实现从服务器到客户端的单向实时数据流传输）
• 可流式传输 HTTP（Streamable HTTP）：适用于远程服务器（通过 HTTP 实现灵活的、潜在的双向通信，通常利用 SSE 实现服务器到客户端的流传输）

安装步骤

在开始使用 crewai-tools 结合 MCP 之前，你需要通过以下命令安装 crewai-tools 依赖中的 mcp 扩展组件：

uv pip install 'crewai-tools[mcp]'

核心概念与入门指南

crewai-tools 库中的 MCPServerAdapter 类是连接 MCP 服务器并将其工具提供给 CrewAI 智能体的主要方式。该类支持多种传输方式，并简化了连接管理流程。

推荐使用 Python 的上下文管理器（即 with 语句）来使用 MCPServerAdapter。它能自动处理与 MCP 服务器连接的建立和关闭操作。

from crewai import Agent
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters  # 用于标准输入输出（Stdio）服务器

# 示例 server_params（根据你的服务器类型选择一种）：
# 1. 标准输入输出（Stdio）服务器：
server_params = StdioServerParameters(
    command="python3",
    args=["servers/your_server.py"],
    env={"UV_PYTHON": "3.12", **os.environ},
)

# 2. 服务器发送事件（SSE）服务器：
server_params = {
    "url": "http://localhost:8000/sse",
    "transport": "sse"
}

# 3. 可流式传输 HTTP（Streamable HTTP）服务器：
server_params = {
    "url": "http://localhost:8001/mcp",
    "transport": "streamable-http"
}

# 示例用法（设置好 server_params 后，取消注释并根据需求调整）：
with MCPServerAdapter(server_params) as mcp_tools:
    print(f"可用工具：{[tool.name for tool in mcp_tools]}")
    my_agent = Agent(
        role="MCP 工具使用者",
        goal="使用来自 MCP 服务器的工具",
        backstory="我能够连接 MCP 服务器并使用其提供的工具",
        tools=mcp_tools,  # 将加载的工具传递给智能体
        reasoning=True,
        verbose=True
    )
    # ... 智能体团队（crew）的其他设置代码 ...

上述通用模式展示了工具集成的方法。如需针对每种传输方式的具体示例，请参考下方的详细指南（文档中未提供具体详细指南链接，此处保留原提示位置）。

工具筛选方法

筛选工具主要有两种方式：

1. 采用字典式索引访问特定工具
2. 向 MCPServerAdapter 构造函数传递工具名称列表

采用字典式索引访问特定工具

with MCPServerAdapter(server_params) as mcp_tools:
    print(f"可用工具：{[tool.name for tool in mcp_tools]}")
    my_agent = Agent(
        role="MCP 工具使用者",
        goal="使用来自 MCP 服务器的工具",
        backstory="我能够连接 MCP 服务器并使用其提供的工具",
        tools=[mcp_tools["tool_name"]],  # 将加载的工具传递给智能体（tool_name 需替换为实际工具名称）
        reasoning=True,
        verbose=True
    )
    # ... 智能体团队（crew）的其他设置代码 ...

向 MCPServerAdapter 构造函数传递工具名称列表

with MCPServerAdapter(server_params, "tool_name") as mcp_tools:  # tool_name 需替换为实际工具名称
    print(f"可用工具：{[tool.name for tool in mcp_tools]}")
    my_agent = Agent(
        role="MCP 工具使用者",
        goal="使用来自 MCP 服务器的工具",
        backstory="我能够连接 MCP 服务器并使用其提供的工具",
        tools=mcp_tools,  # 将加载的工具传递给智能体
        reasoning=True,
        verbose=True
    )
    # ... 智能体团队（crew）的其他设置代码 ...

与 CrewBase 结合使用

若要在 CrewBase 类中使用 MCP 服务器工具，可借助 mcp_tools 方法。服务器配置需通过 mcp_server_params 属性提供，你可以传递单个配置，也可以传递包含多个服务器配置的列表。

@CrewBase
class CrewWithMCP:
    # ... 定义你的智能体和任务配置文件 ...
    mcp_server_params = [
        # 可流式传输 HTTP（Streamable HTTP）服务器配置
        {
            "url": "http://localhost:8001/mcp",
            "transport": "streamable-http"
        },
        # 服务器发送事件（SSE）服务器配置
        {
            "url": "http://localhost:8000/sse",
            "transport": "sse"
        },
        # 标准输入输出（Stdio）服务器配置
        StdioServerParameters(
            command="python3",
            args=["servers/your_stdio_server.py"],
            env={"UV_PYTHON": "3.12", **os.environ},
        )
    ]

    @agent
    def your_agent(self):
        return Agent(config=self.agents_config["your_agent"], tools=self.get_mcp_tools())  # 获取所有可用工具
    # ... 智能体团队（crew）的其他设置代码 ...

你可以通过向 get_mcp_tools 方法传递工具名称列表，筛选可供智能体使用的工具：

@agent
def another_agent(self):
    return Agent(
        config=self.agents_config["your_agent"],
        tools=self.get_mcp_tools("tool_1", "tool_2")  # 获取指定的工具（tool_1、tool_2 需替换为实际工具名称）
    )

MCP 使用安全注意事项

在使用任何 MCP 服务器之前，务必确保你信任该服务器。

安全警告：DNS 重绑定攻击

如果未采取适当的安全措施，服务器发送事件（SSE）传输方式可能容易受到 DNS 重绑定攻击。为防范此类攻击，请采取以下措施：

1. 始终验证传入 SSE 连接的 Origin 标头，确保其来自预期的来源
2. 在本地运行服务器时，避免将服务器绑定到所有网络接口（0.0.0.0），而是仅绑定到本地主机（127.0.0.1）
3. 为所有 SSE 连接实现适当的身份验证机制

若不采取这些防护措施，攻击者可能会利用 DNS 重绑定技术，通过远程网站与本地 MCP 服务器进行交互。

局限性

1. 支持的基本组件：目前，MCPServerAdapter 主要支持适配 MCP 的“工具（tools）”组件。其他 MCP 基本组件（如“提示词（prompts）”或“资源（resources）”）目前无法通过该适配器直接集成为 CrewAI 组件。
2. 输出处理：该适配器通常会处理来自 MCP 工具的主要文本输出（例如 .content[0].text）。对于不符合此模式的复杂输出或多模态输出，可能需要进行自定义处理。

MCP DSL 集成

MCP 集成

MCP DSL 集成

了解如何使用 CrewAI 简洁的 DSL（领域特定语言）语法，通过 mcps 字段将 MCP 服务器直接与智能体集成。

概述

CrewAI 的 MCP DSL 集成功能提供了最简单的方式，让智能体连接到 MCP（模型上下文协议）服务器。只需在智能体中添加一个 mcps 字段，CrewAI 就会自动处理所有复杂的底层操作。

对于大多数 MCP 使用场景，推荐优先采用此方法。若需手动管理连接的高级场景，请参考 MCPServerAdapter（MCP 服务器适配器）相关内容。

基础用法

通过 mcps 字段将 MCP 服务器添加到智能体中：

from crewai import Agent

agent = Agent(
    role="研究助手",
    goal="协助完成研究与分析任务",
    backstory="拥有高级研究工具访问权限的专业助手",
    mcps=[
        "https://mcp.exa.ai/mcp?api_key=your_key&profile=research"  # MCP 服务器地址及配置
    ]
)
# 此时 MCP 工具已自动可用！
# 无需手动管理连接或配置工具

支持的引用格式

外部 MCP 远程服务器

可通过不同格式的 URL 配置外部远程 MCP 服务器，满足基础连接、认证、自定义路径等需求：

# 基础 HTTPS 服务器（无认证）
"https://api.example.com/mcp"

# 带认证的服务器（通过 URL 参数传递 API 密钥和配置文件）
"https://mcp.exa.ai/mcp?api_key=your_key&profile=your_profile"

# 自定义路径的服务器（工具接口位于非根路径）
"https://services.company.com/api/v1/mcp"

特定工具筛选

使用 # 语法从 MCP 服务器中筛选所需的特定工具（而非加载全部工具），提升效率：

# 仅从天气服务器加载 "get_forecast"（获取天气预报）工具
"https://weather.api.com/mcp#get_forecast"

# 仅从 Exa 服务器加载 "web_search_exa"（Exa 网页搜索）工具
"https://mcp.exa.ai/mcp?api_key=your_key#web_search_exa"

CrewAI AMP 市场

通过特定格式的引用，直接访问 CrewAI AMP（应用市场）中的工具服务，支持加载完整服务或单个工具：

# 加载金融数据服务的全部工具
"crewai-amp:financial-data"

# 仅加载研究工具服务中的 "pubmed_search"（PubMed 学术搜索）工具
"crewai-amp:research-tools#pubmed_search"

# 同时加载多个 AMP 服务
mcps=[
    "crewai-amp:weather-insights",        # 天气洞察服务
    "crewai-amp:market-analysis",         # 市场分析服务
    "crewai-amp:social-media-monitoring"  # 社交媒体监控服务
]

完整示例

以下示例展示如何集成多个 MCP 服务器（外部远程服务器 + AMP 市场服务），实现多数据源的综合研究任务：

from crewai import Agent, Task, Crew, Process

# 1. 创建集成多 MCP 源的智能体
multi_source_agent = Agent(
    role="多源研究分析师",
    goal="通过多个数据源开展全面研究",
    backstory="""具备网页搜索、天气数据、金融信息及学术研究工具访问权限的专业研究员，
    擅长整合多领域数据形成系统性分析结果""",
    mcps=[
        # 外部 MCP 服务器（带认证的 Exa 研究工具、天气工具）
        "https://mcp.exa.ai/mcp?api_key=your_exa_key&profile=research",
        "https://weather.api.com/mcp#get_current_conditions",
        # CrewAI AMP 市场服务（金融洞察、学术研究、市场情报）
        "crewai-amp:financial-insights",
        "crewai-amp:academic-research#pubmed_search",
        "crewai-amp:market-intelligence#competitor_analysis"
    ]
)

# 2. 创建综合研究任务（明确需求与输出格式）
research_task = Task(
    description="""研究 AI 智能体对企业生产力的影响，需包含：
    - 天气对远程办公的当前影响
    - 金融市场与 AI 相关的趋势
    - 近期关于 AI 智能体框架的学术出版物""",
    expected_output="""结构化综合报告，涵盖以下 5 部分：
    1. AI 智能体对企业的影响分析
    2. 天气因素对远程办公的影响考量
    3. 与 AI 相关的金融市场趋势
    4. 学术研究引用及核心见解
    5. 竞争格局分析""",
    agent=multi_source_agent  # 将任务分配给多源研究智能体
)

# 3. 创建并执行智能体团队（Crew）
research_crew = Crew(
    agents=[multi_source_agent],  # 团队成员（仅多源研究智能体）
    tasks=[research_task],        # 团队任务（综合研究）
    process=Process.sequential,   # 执行流程（顺序执行）
    verbose=True                  # 启用详细日志（便于调试）
)

# 启动任务并获取结果
result = research_crew.kickoff()
# 输出结果统计信息
print(f"研究完成，共使用 {len(multi_source_agent.mcps)} 个 MCP 数据源")

工具命名与组织

CrewAI 会自动对 MCP 工具进行命名处理，通过添加服务器来源前缀避免不同服务器间的工具名称冲突，确保工具调用无歧义：

# 示例：假设原始 MCP 服务器工具名称存在重复（如均有 "search" "analyze" 工具）
# CrewAI 会自动添加前缀，生成唯一名称：
# - Exa 服务器工具 → "mcp_exa_ai_search", "mcp_exa_ai_analyze"
# - 天气服务器工具 → "weather_service_com_search", "weather_service_com_analyze"

agent = Agent(
    role="工具组织演示",
    goal="展示工具自动命名机制",
    backstory="演示 CrewAI 如何避免 MCP 工具名称冲突",
    mcps=[
        "https://mcp.exa.ai/mcp?api_key=key",      # 工具前缀：mcp_exa_ai_*
        "https://weather.service.com/mcp",         # 工具前缀：weather_service_com_*
        "crewai-amp:financial-data"                # 工具前缀：financial_data_*
    ]
)
# 核心逻辑：每个服务器的工具会基于服务器名称生成唯一前缀，彻底解决命名冲突问题

错误处理与韧性

MCP DSL 设计注重鲁棒性与用户友好性，能自动处理服务器故障、超时等异常场景，确保智能体稳定运行。

优雅处理服务器故障

当部分 MCP 服务器不可用时，智能体会自动跳过故障服务器，继续使用可用工具，避免整体任务崩溃：

agent = Agent(
    role="韧性研究员",
    goal="即使部分服务器故障，仍能完成研究任务",
    backstory="经验丰富的研究员，可根据可用工具灵活调整研究方案",
    mcps=[
        "https://primary-server.com/mcp",         # 主数据源（优先使用）
        "https://backup-server.com/mcp",          # 备用数据源（主服务器故障时启用）
        "https://unreachable-server.com/mcp",     # 不可达服务器（会被跳过并记录警告）
        "crewai-amp:reliable-service"             # 高可靠性 AMP 服务（兜底选项）
    ]
)
# 智能体的故障处理逻辑：
# 1. 成功连接可用服务器
# 2. 对故障服务器记录警告日志（不中断任务）
# 3. 基于可用工具继续执行任务
# 4. 不会因服务器故障导致崩溃或挂起

超时保护

所有 MCP 操作均内置超时限制，防止因服务器响应缓慢或无响应导致任务阻塞，具体超时配置如下：

• 连接超时：10 秒（建立 MCP 服务器连接的最长等待时间）
• 工具执行超时：30 秒（工具调用后等待结果的最长时间）
• 发现超时：15 秒（获取服务器工具列表的最长等待时间）

# 示例：对可能超时的服务器，智能体会自动触发超时保护
mcps=[
    "https://slow-server.com/mcp",        # 响应缓慢的服务器（10 秒内无响应则超时）
    "https://overloaded-api.com/mcp"      # 负载过高的服务器（15 秒内未返回工具列表则超时）
]

性能优化特性

自动缓存

MCP 工具的 schema（结构定义）会被缓存 5 分钟，避免重复从服务器获取相同信息，大幅提升智能体创建效率：

# 第一次创建智能体：从服务器获取工具 schema（耗时较长）
agent1 = Agent(
    role="First", 
    goal="Test", 
    backstory="Test",
    mcps=["https://api.example.com/mcp"]
)

# 5 分钟内创建第二个智能体：直接使用缓存的 schema（速度极快）
agent2 = Agent(
    role="Second", 
    goal="Test", 
    backstory="Test",
    mcps=["https://api.example.com/mcp"]  # 无需再次请求服务器
)

按需连接

MCP 服务器连接仅在工具实际被调用时建立，而非智能体创建时，减少不必要的连接开销，提升智能体启动速度：

# 创建智能体时：不建立 MCP 连接（启动速度快）
agent = Agent(
    role="按需连接智能体",
    goal="高效使用 MCP 工具",
    backstory="仅在需要时建立服务器连接，减少资源消耗",
    mcps=["https://api.example.com/mcp"]
)

# 核心逻辑：
# - 智能体创建阶段无 MCP 连接操作
# - 仅当调用该服务器的工具时，才建立连接
# - 显著降低启动开销，尤其适用于多 MCP 源场景

与现有功能的集成

MCP 工具可与 CrewAI 的其他功能（自定义工具、平台集成应用）无缝协作，形成完整的工具生态：

from crewai.tools import BaseTool

# 1. 定义自定义工具（继承 BaseTool）
class CustomTool(BaseTool):
    name: str = "custom_analysis"  # 工具名称
    description: str = "自定义分析工具，返回特定格式的分析结果"  # 工具描述

    def _run(self, **kwargs):
        # 工具核心逻辑（示例：返回固定分析结果）
        return "Custom analysis result"

# 2. 创建集成多类型工具的智能体
agent = Agent(
    role="全功能智能体",
    goal="使用所有可用类型的工具完成任务",
    backstory="可灵活调用自定义工具、平台应用和 MCP 工具的综合型智能体",
    
    # 同时集成三种类型的工具
    tools=[CustomTool()],                          # 自定义工具
    apps=["gmail", "slack"],                       # 平台集成应用（Gmail、Slack）
    mcps=[                                         # MCP 服务器工具
        "https://mcp.exa.ai/mcp?api_key=key",
        "crewai-amp:research-tools"
    ],
    
    verbose=True,  # 启用详细日志
    max_iter=15    # 最大任务迭代次数（防止无限循环）
)

最佳实践

遵循以下最佳实践，可确保 MCP 集成的效率、安全性和稳定性：

1. 尽可能使用特定工具

优先通过 # 语法筛选所需工具，避免加载服务器全部工具，减少资源消耗和命名冲突风险：

# 推荐：仅加载所需的 "get_forecast" 工具
mcps=["https://weather.api.com/mcp#get_forecast"]

# 不推荐：加载服务器所有工具（可能包含无用工具，增加开销）
mcps=["https://weather.api.com/mcp"]

2. 安全处理认证信息

避免在代码中硬编码 API 密钥等敏感信息，优先使用环境变量存储，降低泄露风险：

import os

# 从环境变量中读取认证信息（需提前在系统/环境中配置）
exa_key = os.getenv("EXA_API_KEY")
exa_profile = os.getenv("EXA_PROFILE")

# 动态拼接 MCP 服务器地址（敏感信息不暴露在代码中）
agent = Agent(
    role="安全智能体",
    goal="安全使用 MCP 工具",
    backstory="具备安全意识，严格保护认证信息的智能体",
    mcps=[f"https://mcp.exa.ai/mcp?api_key={exa_key}&profile={exa_profile}"]
)

3. 为服务器故障制定预案

始终配置主备 MCP 服务器或 AMP 服务，确保单一服务器故障不影响整体任务：

mcps=[
    "https://primary-api.com/mcp",       # 主服务器（优先使用）
    "https://backup-api.com/mcp",        # 备用服务器（主服务器故障时启用）
    "crewai-amp:reliable-service"        # AMP 服务（兜底选项，可靠性更高）
]

4. 使用描述性的智能体角色

明确智能体的角色和背景，使其更合理地选择 MCP 工具，提升任务执行准确性：

agent = Agent(
    role="天气增强型市场分析师",  # 角色明确（结合天气与市场分析）
    goal="结合天气影响分析市场趋势",
    backstory="金融分析师，擅长利用天气数据评估农业、零售等行业的市场波动",
    mcps=[
        "https://weather.service.com/mcp#get_forecast",  # 天气工具（核心需求）
        "crewai-amp:financial-data#stock_analysis"       # 股票分析工具（核心需求）
    ]
)

故障排除

常见问题与解决方案

问题 1：未发现任何工具

可能原因：MCP 服务器 URL 错误、认证失败、服务器未运行或网络不可达。
解决方案：

1. 检查 MCP 服务器 URL 是否正确（如路径、域名拼写）；
2. 验证 API 密钥、令牌等认证信息是否有效；
3. 确认服务器已启动且可通过当前网络访问；
4. 示例（修正后的配置）：

   mcps=["https://mcp.example.com/mcp?api_key=valid_key"]  # 确保 URL 和密钥正确
   

问题 2：连接超时

可能原因：服务器响应缓慢、负载过高、网络延迟大。
解决方案：

1. 检查服务器状态（如是否过载、是否存在网络瓶颈）；
2. 增加备用服务器或 AMP 服务（避免依赖单一数据源）；
3. CrewAI 会自动记录超时警告并跳过故障服务器，无需额外代码处理。

问题 3：认证失败

可能原因：API 密钥无效、权限不足、认证参数格式错误。
解决方案：

1. 核对 API 密钥、令牌等认证信息是否与服务器要求一致；
2. 参考服务器文档，确认所需的认证参数（如是否需要 profile、token 等）；
3. 确保 URL 中的查询参数已正确 URL 编码（如特殊字符处理）。

高级用法：MCPServerAdapter

对于需要手动管理连接的复杂场景（如自定义连接生命周期、精细控制工具加载），可使用 crewai-tools 库中的 MCPServerAdapter 类。推荐通过 Python 上下文管理器（with 语句）使用，它会自动处理 MCP 服务器连接的建立与关闭，避免资源泄漏。

详细使用方法请参考 MCP 服务器作为 CrewAI 工具 中的相关章节。