CrewAI 中作为工具的 MCP 服务器

概述

模型上下文协议 (MCP) 提供了一种标准化的方式，使 AI 代理通过与外部服务（称为 MCP 服务器）通信来向 LLM 提供上下文。 CrewAI 为 MCP 集成提供了两种方法：

🚀 简单的 DSL 集成（推荐）

直接在代理上使用 mcps 字段，实现无缝 MCP 工具集成。DSL 支持字符串引用（用于快速设置）和结构化配置（用于完全控制）。

基于字符串的引用（快速设置）

非常适合远程 HTTPS 服务器和 CrewAI AOP 市场

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 AOP marketplace
        "crewai-amp:research-tools#pubmed_search"            # Specific AMP tool
    ]
)
# MCP tools are now automatically available to your agent!

结构化配置（完全控制）

用于完全控制连接设置、工具过滤和所有传输类型

from crewai import Agent
from crewai.mcp import MCPServerStdio, MCPServerHTTP, MCPServerSSE
from crewai.mcp.filters import create_static_tool_filter

agent = Agent(
    role="Advanced Research Analyst",
    goal="Research with full control over MCP connections",
    backstory="Expert researcher with advanced tool access",
    mcps=[
        # Stdio transport for local servers
        MCPServerStdio(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem"],
            env={"API_KEY": "your_key"},
            tool_filter=create_static_tool_filter(
                allowed_tool_names=["read_file", "list_directory"]
            ),
            cache_tools_list=True,
        ),
        # HTTP/Streamable HTTP transport for remote servers
        MCPServerHTTP(
            url="https://api.example.com/mcp",
            headers={"Authorization": "Bearer your_token"},
            streamable=True,
            cache_tools_list=True,
        ),
        # SSE transport for real-time streaming
        MCPServerSSE(
            url="https://stream.example.com/mcp/sse",
            headers={"Authorization": "Bearer your_token"},
        ),
    ]
)

🔧 高级：MCPServerAdapter（用于复杂场景）

对于需要手动连接管理的高级用例，crewai-tools 库提供了 MCPServerAdapter 类。我们目前支持以下传输机制：

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

视频教程

观看此视频教程，获取 CrewAI 与 MCP 集成的全面指南

安装

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 using string references
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()

使用结构化配置的快速入门

from crewai import Agent, Task, Crew
from crewai.mcp import MCPServerStdio, MCPServerHTTP, MCPServerSSE

# Create agent with structured MCP configurations
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=[
        # Local stdio server
        MCPServerStdio(
            command="python",
            args=["local_server.py"],
            env={"API_KEY": "your_key"},
        ),
        # Remote HTTP server
        MCPServerHTTP(
            url="https://api.research.com/mcp",
            headers={"Authorization": "Bearer your_token"},
        ),
    ]
)

# 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 AOP 市场

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"
]

结构化配置

标准输入输出 (Stdio) 传输（本地服务器）

非常适合作为进程运行的本地 MCP 服务器

from crewai.mcp import MCPServerStdio
from crewai.mcp.filters import create_static_tool_filter

mcps=[
    MCPServerStdio(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem"],
        env={"API_KEY": "your_key"},
        tool_filter=create_static_tool_filter(
            allowed_tool_names=["read_file", "write_file"]
        ),
        cache_tools_list=True,
    ),
    # Python-based server
    MCPServerStdio(
        command="python",
        args=["path/to/server.py"],
        env={"UV_PYTHON": "3.12", "API_KEY": "your_key"},
    ),
]

HTTP/可流式 HTTP 传输（远程服务器）

用于通过 HTTP/HTTPS 的远程 MCP 服务器

from crewai.mcp import MCPServerHTTP

mcps=[
    # Streamable HTTP (default)
    MCPServerHTTP(
        url="https://api.example.com/mcp",
        headers={"Authorization": "Bearer your_token"},
        streamable=True,
        cache_tools_list=True,
    ),
    # Standard HTTP
    MCPServerHTTP(
        url="https://api.example.com/mcp",
        headers={"Authorization": "Bearer your_token"},
        streamable=False,
    ),
]

SSE 传输（实时流）

用于使用服务器发送事件的远程服务器

from crewai.mcp import MCPServerSSE

mcps=[
    MCPServerSSE(
        url="https://stream.example.com/mcp/sse",
        headers={"Authorization": "Bearer your_token"},
        cache_tools_list=True,
    ),
]

混合引用

您可以结合字符串引用和结构化配置

from crewai.mcp import MCPServerStdio, MCPServerHTTP

mcps=[
    # String references
    "https://external-api.com/mcp",              # External server
    "crewai-amp:financial-insights",             # AMP service

    # Structured configurations
    MCPServerStdio(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem"],
    ),
    MCPServerHTTP(
        url="https://api.example.com/mcp",
        headers={"Authorization": "Bearer token"},
    ),
]

工具过滤

结构化配置支持高级工具过滤

from crewai.mcp import MCPServerStdio
from crewai.mcp.filters import create_static_tool_filter, create_dynamic_tool_filter, ToolFilterContext

# Static filtering (allow/block lists)
static_filter = create_static_tool_filter(
    allowed_tool_names=["read_file", "write_file"],
    blocked_tool_names=["delete_file"],
)

# Dynamic filtering (context-aware)
def dynamic_filter(context: ToolFilterContext, tool: dict) -> bool:
    # Block dangerous tools for certain agent roles
    if context.agent.role == "Code Reviewer":
        if "delete" in tool.get("name", "").lower():
            return False
    return True

mcps=[
    MCPServerStdio(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem"],
        tool_filter=static_filter,  # or dynamic_filter
    ),
]

配置参数

每种传输类型都支持特定的配置选项

MCPServerStdio 参数

command（必需）：要执行的命令（例如，"python"、"node"、"npx"、"uvx"）
args（可选）：命令参数列表（例如，["server.py"] 或 ["-y", "@mcp/server"]）
env（可选）：要传递给进程的环境变量字典
tool_filter（可选）：用于过滤可用工具的工具过滤函数
cache_tools_list（可选）：是否缓存工具列表以便后续更快访问（默认值：False）

MCPServerHTTP 参数

url（必需）：服务器 URL（例如，"https://api.example.com/mcp"）
headers（可选）：用于身份验证或其他目的的 HTTP 头字典
streamable（可选）：是否使用可流式 HTTP 传输（默认值：True）
tool_filter（可选）：用于过滤可用工具的工具过滤函数
cache_tools_list（可选）：是否缓存工具列表以便后续更快访问（默认值：False）

MCPServerSSE 参数

url（必需）：服务器 URL（例如，"https://api.example.com/mcp/sse"）
headers（可选）：用于身份验证或其他目的的 HTTP 头字典
tool_filter（可选）：用于过滤可用工具的工具过滤函数
cache_tools_list（可选）：是否缓存工具列表以便后续更快访问（默认值：False）

通用参数

所有传输类型都支持

tool_filter：过滤函数，用于控制哪些工具可用。可以是
- None（默认）：所有工具都可用
- 静态过滤器：使用 create_static_tool_filter() 创建，用于允许/阻止列表
- 动态过滤器：使用 create_dynamic_tool_filter() 创建，用于上下文感知过滤
cache_tools_list：当 True 时，在首次发现后缓存工具列表，以提高后续连接的性能

主要特点

🔄 自动工具发现：工具会自动发现并集成
🏷️ 名称冲突预防：服务器名称会作为工具名称的前缀
⚡ 性能优化：按需连接，带模式缓存
🛡️ 错误恢复能力：优雅处理不可用服务器
⏱️ 超时保护：内置超时，防止连接挂起
📊 透明集成：与现有 CrewAI 功能无缝协作
🔧 完整传输支持：Stdio、HTTP/可流式 HTTP 和 SSE 传输
🎯 高级过滤：静态和动态工具过滤功能
🔐 灵活认证：支持头、环境变量和查询参数

错误处理

MCP DSL 集成设计为具有弹性，并能优雅地处理故障

from crewai import Agent
from crewai.mcp import MCPServerStdio, MCPServerHTTP

agent = Agent(
    role="Resilient Agent",
    goal="Continue working despite server issues",
    backstory="Agent that handles failures gracefully",
    mcps=[
        # String references
        "https://reliable-server.com/mcp",        # Will work
        "https://unreachable-server.com/mcp",     # Will be skipped gracefully
        "crewai-amp:working-service",             # Will work

        # Structured configs
        MCPServerStdio(
            command="python",
            args=["reliable_server.py"],          # Will work
        ),
        MCPServerHTTP(
            url="https://slow-server.com/mcp",     # Will timeout gracefully
        ),
    ]
)
# Agent will use tools from working servers and log warnings for failing ones

所有连接错误都会得到优雅处理

连接失败：记录为警告，代理继续使用可用工具
超时错误：连接在 30 秒后超时（可配置）
认证错误：清晰记录以便调试
无效配置：在代理创建时引发验证错误

高级：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 ...

这种通用模式展示了如何集成工具。有关针对每种传输的特定示例，请参阅下面的详细指南。

过滤工具

有两种过滤工具的方法

使用字典式索引访问特定工具。
将工具名称列表传递给 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 中的每个代理重用。
由于 @CrewBase 注入的隐式后启动钩子，适配器在 .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 集成的重要安全最佳实践，以确保您的代理安全。

查看此存储库，获取 CrewAI 与 MCP 集成的完整演示和示例！👇

GitHub 存储库

CrewAI MCP 演示

使用 MCP 保持安全

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

安全警告：DNS 重绑定攻击

如果未正确保护，SSE 传输可能容易受到 DNS 重绑定攻击。为防止这种情况发生：

始终验证传入 SSE 连接上的 Origin 头，以确保它们来自预期来源
在本地运行时，避免将服务器绑定到所有网络接口 (0.0.0.0) - 请仅绑定到 localhost (127.0.0.1)
为所有 SSE 连接实施适当的身份验证

如果没有这些保护措施，攻击者可能会利用 DNS 重绑定从远程网站与本地 MCP 服务器进行交互。有关更多详细信息，请参阅 Anthropic 的 MCP 传输安全文档。

局限性

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

开始使用

指南

核心概念

MCP 集成

工具

可观测性

学习

遥测

​概述

​🚀 简单的 DSL 集成（推荐）

​基于字符串的引用（快速设置）

​结构化配置（完全控制）

​🔧 高级：MCPServerAdapter（用于复杂场景）

​视频教程

​安装

​快速入门：简单的 DSL 集成

​使用字符串引用的快速入门

​使用结构化配置的快速入门

​MCP 引用格式

​基于字符串的引用

​外部 MCP 服务器

​CrewAI AOP 市场

​结构化配置

​标准输入输出 (Stdio) 传输（本地服务器）

​HTTP/可流式 HTTP 传输（远程服务器）

​SSE 传输（实时流）

​混合引用

​工具过滤

​配置参数

​MCPServerStdio 参数

​MCPServerHTTP 参数

​MCPServerSSE 参数

​通用参数

​主要特点

​错误处理

​高级：MCPServerAdapter

​连接配置

​过滤工具

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

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

​与 CrewBase 一起使用

​连接超时配置

​过滤工具

​探索 MCP 集成

简单的 DSL 集成

标准输入输出 (Stdio) 传输

SSE 传输

可流式 HTTP 传输

连接多个服务器

安全注意事项

GitHub 存储库

​使用 MCP 保持安全

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

​局限性

概述

🚀 简单的 DSL 集成（推荐）

基于字符串的引用（快速设置）

结构化配置（完全控制）

🔧 高级：MCPServerAdapter（用于复杂场景）

视频教程

安装

快速入门：简单的 DSL 集成

使用字符串引用的快速入门

使用结构化配置的快速入门

MCP 引用格式

基于字符串的引用

外部 MCP 服务器

CrewAI AOP 市场

结构化配置

标准输入输出 (Stdio) 传输（本地服务器）

HTTP/可流式 HTTP 传输（远程服务器）

SSE 传输（实时流）

混合引用

工具过滤

配置参数

MCPServerStdio 参数

MCPServerHTTP 参数

MCPServerSSE 参数

通用参数

主要特点

错误处理

高级：MCPServerAdapter

连接配置

过滤工具

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

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

与 CrewBase 一起使用

连接超时配置

过滤工具

探索 MCP 集成

使用 MCP 保持安全

安全警告：DNS 重绑定攻击

局限性