跳转到主要内容

​
概述

模型上下文协议 (MCP) 为 AI 智能体提供了一种标准化的方式，通过与外部服务（称为 MCP 服务器）通信，向 LLM 提供上下文。 CrewAI 提供了两种方法进行 MCP 集成： 直接在智能体上使用 mcps 字段，实现无缝的 MCP 工具集成。 
from crewai import Agent

agent = Agent(
    role="Research Analyst",
    goal="Research and analyze information",
    backstory="Expert researcher with access to external tools",
    mcps=[
        "https://mcp.exa.ai/mcp?api_key=your_key",           # External MCP server
        "https://api.weather.com/mcp#get_forecast",          # Specific tool from server
        "crewai-amp:financial-data",                         # CrewAI AMP marketplace
        "crewai-amp:research-tools#pubmed_search"            # Specific AMP tool
    ]
)
# MCP tools are now automatically available to your agent!

​
🔧 高级：MCPServerAdapter (适用于复杂场景)

对于需要手动管理连接的高级用例，crewai-tools 库提供了 MCPServerAdapter 类。我们目前支持以下传输机制：
  • Stdio：用于本地服务器（通过同一台机器上进程间的标准输入/输出进行通信）
  • 服务器发送事件 (SSE)：用于远程服务器（通过 HTTP 从服务器到客户端的单向、实时数据流）
  • 可流式 HTTPS：用于远程服务器（通过 HTTPS 进行灵活的、可能双向的通信，通常利用 SSE 实现服务器到客户端的流）

​
视频教程

观看此视频教程，全面了解 MCP 与 CrewAI 的集成

​
安装

CrewAI MCP 集成需要 mcp 库 
# For Simple DSL Integration (Recommended)
uv add mcp

# For Advanced MCPServerAdapter usage
uv pip install 'crewai-tools[mcp]'

​
快速入门：简单 DSL 集成

集成 MCP 服务器最简单的方法是在您的智能体上使用 mcps 字段。 
from crewai import Agent, Task, Crew

# Create agent with MCP tools
research_agent = Agent(
    role="Research Analyst",
    goal="Find and analyze information using advanced search tools",
    backstory="Expert researcher with access to multiple data sources",
    mcps=[
        "https://mcp.exa.ai/mcp?api_key=your_key&profile=your_profile",
        "crewai-amp:weather-service#current_conditions"
    ]
)

# Create task
research_task = Task(
    description="Research the latest developments in AI agent frameworks",
    expected_output="Comprehensive research report with citations",
    agent=research_agent
)

# Create and run crew
crew = Crew(agents=[research_agent], tasks=[research_task])
result = crew.kickoff()
就是这样！MCP 工具会自动被发现并可供您的智能体使用。

​
MCP 引用格式

mcps 字段支持多种引用格式，以实现最大灵活性。

​
外部 MCP 服务器

mcps=[
    # Full server - get all available tools
    "https://mcp.example.com/api",

    # Specific tool from server using # syntax
    "https://api.weather.com/mcp#get_current_weather",

    # Server with authentication parameters
    "https://mcp.exa.ai/mcp?api_key=your_key&profile=your_profile"
]

​
CrewAI AMP 市场

mcps=[
    # Full AMP MCP service - get all available tools
    "crewai-amp:financial-data",

    # Specific tool from AMP service using # syntax
    "crewai-amp:research-tools#pubmed_search",

    # Multiple AMP services
    "crewai-amp:weather-service",
    "crewai-amp:market-analysis"
]

​
混合引用

mcps=[
    "https://external-api.com/mcp",              # External server
    "https://weather.service.com/mcp#forecast",  # Specific external tool
    "crewai-amp:financial-insights",             # AMP service
    "crewai-amp:data-analysis#sentiment_tool"    # Specific AMP tool
]

​
主要特点

  • 🔄 自动工具发现：工具被自动发现和集成
  • 🏷️ 名称冲突预防：服务器名称会作为工具名称的前缀
  • ⚡ 性能优化：按需连接并缓存 schema
  • 🛡️ 错误恢复能力：优雅地处理不可用的服务器
  • ⏱️ 超时保护：内置超时功能防止连接挂起
  • 📊 透明集成：与现有的 CrewAI 功能无缝协作

​
错误处理

MCP DSL 集成设计为具有弹性 
agent = Agent(
    role="Resilient Agent",
    goal="Continue working despite server issues",
    backstory="Agent that handles failures gracefully",
    mcps=[
        "https://reliable-server.com/mcp",        # Will work
        "https://unreachable-server.com/mcp",     # Will be skipped gracefully
        "https://slow-server.com/mcp",            # Will timeout gracefully
        "crewai-amp:working-service"              # Will work
    ]
)
# Agent will use tools from working servers and log warnings for failing ones

​
高级：MCPServerAdapter

对于需要手动管理连接的复杂场景，请使用 crewai-tools 中的 MCPServerAdapter 类。推荐使用 Python 上下文管理器（with 语句），因为它会自动处理与 MCP 服务器的连接启动和停止。

​
连接配置

MCPServerAdapter 支持多种配置选项以自定义连接行为。
  • connect_timeout (可选)：等待与 MCP 服务器建立连接的最大时间（秒）。如果未指定，默认为 30 秒。这对于响应时间可能变化的远程服务器特别有用。
# Example with custom connection timeout
with MCPServerAdapter(server_params, connect_timeout=60) as tools:
    # Connection will timeout after 60 seconds if not established
    pass

from crewai import Agent
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters # For Stdio Server

# Example server_params (choose one based on your server type):
# 1. Stdio Server:
server_params=StdioServerParameters(
    command="python3",
    args=["servers/your_server.py"],
    env={"UV_PYTHON": "3.12", **os.environ},
)

# 2. SSE Server:
server_params = {
    "url": "https://:8000/sse",
    "transport": "sse"
}

# 3. Streamable HTTP Server:
server_params = {
    "url": "https://:8001/mcp",
    "transport": "streamable-http"
}

# Example usage (uncomment and adapt once server_params is set):
with MCPServerAdapter(server_params, connect_timeout=60) as mcp_tools:
    print(f"Available tools: {[tool.name for tool in mcp_tools]}")

    my_agent = Agent(
        role="MCP Tool User",
        goal="Utilize tools from an MCP server.",
        backstory="I can connect to MCP servers and use their tools.",
        tools=mcp_tools, # Pass the loaded tools to your agent
        reasoning=True,
        verbose=True
    )
    # ... rest of your crew setup ...
这个通用模式展示了如何集成工具。有关针对每种传输方式的具体示例，请参阅下面的详细指南。

​
筛选工具

有两种方法可以筛选工具
  1. 使用字典式索引访问特定工具。
  2. 将工具名称列表传递给 MCPServerAdapter 构造函数。

​
使用字典式索引访问特定工具。

with MCPServerAdapter(server_params, connect_timeout=60) as mcp_tools:
    print(f"Available tools: {[tool.name for tool in mcp_tools]}")

    my_agent = Agent(
        role="MCP Tool User",
        goal="Utilize tools from an MCP server.",
        backstory="I can connect to MCP servers and use their tools.",
        tools=[mcp_tools["tool_name"]], # Pass the loaded tools to your agent
        reasoning=True,
        verbose=True
    )
    # ... rest of your crew setup ...

​
将工具名称列表传递给 MCPServerAdapter 构造函数。

with MCPServerAdapter(server_params, "tool_name", connect_timeout=60) as mcp_tools:
    print(f"Available tools: {[tool.name for tool in mcp_tools]}")

    my_agent = Agent(
        role="MCP Tool User",
        goal="Utilize tools from an MCP server.",
        backstory="I can connect to MCP servers and use their tools.",
        tools=mcp_tools, # Pass the loaded tools to your agent
        reasoning=True,
        verbose=True
    )
    # ... rest of your crew setup ...

​
与 CrewBase 一起使用

要在 CrewBase 类中使用 MCPServer 工具，请使用 get_mcp_tools 方法。服务器配置应通过 mcp_server_params 属性提供。您可以传递单个配置或多个服务器配置的列表。 
@CrewBase
class CrewWithMCP:
  # ... define your agents and tasks config file ...

  mcp_server_params = [
    # Streamable HTTP Server
    {
        "url": "https://:8001/mcp",
        "transport": "streamable-http"
    },
    # SSE Server
    {
        "url": "https://:8000/sse",
        "transport": "sse"
    },
    # StdIO Server
    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()) # get all available tools

    # ... rest of your crew setup ...
当一个 crew 类被 @CrewBase 装饰器修饰时，适配器的生命周期将为您管理。
  • 首次调用 get_mcp_tools() 会延迟创建一个共享的 MCPServerAdapter，该适配器会被 crew 中的每个智能体重用。
  • 适配器在 .kickoff() 完成后会自动关闭，这要归功于 @CrewBase 注入的隐式 after-kickoff 钩子，因此不需要手动清理。
  • 如果未定义 mcp_server_params，get_mcp_tools() 将简单地返回一个空列表，从而允许相同的代码路径在配置或未配置 MCP 的情况下运行。
这使得可以安全地从多个智能体方法中调用 get_mcp_tools()，或根据不同环境选择性地启用 MCP。

​
连接超时配置

您可以通过设置 mcp_connect_timeout 类属性来配置 MCP 服务器的连接超时。如果未指定超时，则默认为 30 秒。 
@CrewBase
class CrewWithMCP:
  mcp_server_params = [...]
  mcp_connect_timeout = 60  # 60 seconds timeout for all MCP connections

  @agent
  def your_agent(self):
      return Agent(config=self.agents_config["your_agent"], tools=self.get_mcp_tools())

@CrewBase
class CrewWithDefaultTimeout:
  mcp_server_params = [...]
  # No mcp_connect_timeout specified - uses default 30 seconds

  @agent
  def your_agent(self):
      return Agent(config=self.agents_config["your_agent"], tools=self.get_mcp_tools())

​
筛选工具

您可以通过向 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") # get specific tools
    )
超时配置适用于 crew 内的所有 MCP 工具调用。 
@CrewBase
class CrewWithCustomTimeout:
  mcp_server_params = [...]
  mcp_connect_timeout = 90  # 90 seconds timeout for all MCP connections

  @agent
  def filtered_agent(self):
      return Agent(
        config=self.agents_config["your_agent"],
        tools=self.get_mcp_tools("tool_1", "tool_2") # specific tools with custom timeout
      )

​
探索 MCP 集成

简单 DSL 集成

推荐：使用简单的 mcps=[] 字段语法，轻松实现 MCP 集成。

Stdio 传输

通过标准输入/输出连接到本地 MCP 服务器。非常适合脚本和本地可执行文件。

SSE 传输

使用服务器发送事件与远程 MCP 服务器集成，以实现实时数据流。

可流式 HTTP 传输

利用灵活的可流式 HTTP 与远程 MCP 服务器进行稳健的通信。

连接到多个服务器

使用单个适配器同时聚合来自多个 MCP 服务器的工具。

安全注意事项

查看重要的 MCP 集成安全最佳实践，以确保您的智能体安全。
查看此存储库以获取 MCP 与 CrewAI 集成的完整演示和示例！ 👇

GitHub 存储库

CrewAI MCP 演示

​
使用 MCP 保持安全

在使用 MCP 服务器之前，请务必确保您信任它。

​
安全警告：DNS 重绑定攻击

如果未正确保护，SSE 传输可能容易受到 DNS 重绑定攻击。为防止这种情况：
  1. 始终验证传入 SSE 连接的 Origin 标头，以确保它们来自预期的源
  2. 在本地运行时，避免将服务器绑定到所有网络接口 (0.0.0.0) - 应仅绑定到 localhost (127.0.0.1)
  3. 为所有 SSE 连接实施适当的身份验证
如果没有这些保护措施，攻击者可能会利用 DNS 重绑定从远程网站与本地 MCP 服务器进行交互。有关更多详细信息，请参阅 Anthropic 的 MCP 传输安全文档。

​
限制

  • 支持的原语：目前，MCPServerAdapter 主要支持适配 MCP tools。其他 MCP 原语，如 prompts 或 resources，目前尚未通过此适配器直接集成为 CrewAI 组件。
  • 输出处理：适配器通常处理来自 MCP 工具的主要文本输出（例如 .content[0].text）。如果不符合此模式，复杂或多模态的输出可能需要自定义处理。
⌘我