MCP 服务器作为 CrewAI 工具

概览

Model Context Protocol（MCP）为 AI 代理提供了一种标准化方式，通过与外部服务（即 MCP Servers）通信，为 LLM 提供上下文。

CrewAI 提供两种 MCP 集成方式：

🚀 简洁 DSL 集成（推荐）

直接在代理上使用 mcps 字段即可无缝集成 MCP 工具。该 DSL 同时支持字符串引用（快速配置）和结构化配置（完全控制）。

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

非常适合远程 HTTPS 服务器以及 CrewAI 目录中的已连接 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
        "snowflake",                                         # Connected MCP from catalog
        "stripe#list_invoices"                               # Specific tool from connected MCP
    ]
)
# 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：用于本地服务器（通过同一台机器上进程之间的标准输入/输出通信）
• Server-Sent Events（SSE）：用于远程服务器（通过 HTTP 从服务器到客户端的单向实时数据流）
• 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",
        "snowflake#run_query"
    ]
)

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

已连接的 MCP 集成

从 CrewAI 目录连接 MCP 服务器，或者接入你自己的服务器。连接到账户后，可通过 slug 引用：

mcps=[
    # Connected MCP - get all available tools
    "snowflake",

    # Specific tool from a connected MCP using # syntax
    "stripe#list_invoices",

    # Multiple connected MCPs
    "snowflake",
    "stripe",
    "github"
]

结构化配置

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/Streamable 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 传输（实时流式）

适用于使用 Server-Sent Events 的远程服务器：

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
    "snowflake",                                 # Connected MCP from catalog

    # 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（可选）：是否使用 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 时，会在首次发现后缓存工具列表，以提升后续连接性能

关键特性

• 🔄 自动工具发现：工具会自动发现并集成
• 🏷️ 名称冲突防护：服务器名称会作为前缀加到工具名上
• ⚡ 性能优化：按需连接并缓存 schema
• 🛡️ 错误韧性：优雅处理不可用服务器
• ⏱️ 超时保护：内置超时可防止连接挂起
• 📊 透明集成：与现有 CrewAI 功能无缝协作
• 🔧 完整传输支持：支持 Stdio、HTTP/Streamable 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
        "snowflake",                              # Connected MCP from catalog

        # 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": "http://localhost:8000/sse",
    "transport": "sse"
}

# 3. Streamable HTTP Server:
server_params = {
    "url": "http://localhost: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": "http://localhost:8001/mcp",
        "transport": "streamable-http"
    },
    # SSE Server
    {
        "url": "http://localhost: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 ...

Tip

当一个 crew 类使用 @CrewBase 装饰时，适配器生命周期会自动为你管理：

• 第一次调用 get_mcp_tools() 时，会延迟创建一个共享的 MCPServerAdapter，并供 crew 中所有代理复用。
• 由于 @CrewBase 注入了一个隐式的 after-kickoff hook，kickoff() 完成后适配器会自动关闭，因此无需手动清理。
• 如果未定义 mcp_server_params，get_mcp_tools() 会直接返回空列表，这样同一套代码既可在有 MCP 配置时运行，也可在没有 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 传输

使用 Server-Sent Events 与远程 MCP 服务器集成，实现实时数据流传输。

Streamable HTTP 传输

使用灵活的 Streamable HTTP，与远程 MCP 服务器进行稳健通信。

连接多个服务器

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

安全注意事项

查看 MCP 集成的重要安全最佳实践，保护你的代理安全。

到这个仓库查看完整的 MCP 集成演示和示例！👇

GitHub 仓库

CrewAI MCP 演示

使用 MCP 的安全建议

Warning

在使用 MCP Server 之前，务必确保你信任它。

安全警告：DNS 重绑定攻击

如果未妥善保护，SSE 传输可能会受到 DNS 重绑定攻击。 为防止这种情况：

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

如果没有这些保护，攻击者可能会利用 DNS 重绑定从远程网站与本地 MCP 服务器交互。

更多详情，请参阅 Anthropic 的 MCP Transport Security 文档。

局限性

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