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

了解如何使用 crewai-tools 库将 MCP 服务器作为工具集成到你的 CrewAI 智能体中。

概述

Model Context Protocol（MCP）为 AI 智能体提供了一种标准化方式，通过与被称作 MCP 服务器的外部服务通信来为 LLM 提供上下文。

CrewAI 为 MCP 集成提供两种方法：

🚀 简单 DSL 集成（推荐）

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

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

非常适合远程 HTTPS 服务器以及来自 CrewAI 目录的已连接 MCP 集成：

python

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!

结构化配置（完全控制）

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

python

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 实现服务器到客户端的流）

视频教程

观看本视频教程，全面了解如何将 MCP 与 CrewAI 集成：

安装

CrewAI MCP 集成需要 mcp 库：

shell

# For Simple DSL Integration (Recommended)
uv add mcp

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

快速开始：简单 DSL 集成

集成 MCP 服务器最简单的方式是在智能体上使用 mcps 字段。你可以使用字符串引用或结构化配置。

使用字符串引用快速开始

python

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()

使用结构化配置快速开始

python

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 服务器

python

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 引用它们：

python

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 服务器：

python

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 服务器：

python

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 的远程服务器：

python

from crewai.mcp import MCPServerSSE

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

混合引用

你可以组合使用字符串引用和结构化配置：

python

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"},
    ),
]

工具过滤

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

python

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 时，会在首次发现后缓存工具列表，以提升后续连接的性能

核心特性

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

错误处理

MCP DSL 集成在设计上具备容错能力，并能优雅地处理故障：

python

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 秒。这对于响应时间可能变化的远程服务器特别有用。

python

# Example with custom connection timeout
with MCPServerAdapter(server_params, connect_timeout=60) as tools:
    # Connection will timeout after 60 seconds if not established
    pass

python

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 ...

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

过滤工具

有两种过滤工具的方式：

使用字典式索引访问特定工具。
将工具名称列表传递给 MCPServerAdapter 构造函数。

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

python

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` 构造函数。

python

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 属性提供。你可以传入单个配置或多个服务器配置组成的列表。

python

@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

当团队类被 @CrewBase 装饰时，适配器生命周期会为你管理：

第一次调用 get_mcp_tools() 时会延迟创建一个共享的 MCPServerAdapter，供团队中的每个智能体复用。
得益于 @CrewBase 注入的隐式 kickoff 后钩子，适配器会在 .kickoff() 完成后自动关闭，因此无需手动清理。
如果未定义 mcp_server_params，get_mcp_tools() 只会返回空列表，从而允许在配置或未配置 MCP 的情况下运行相同的代码路径。

这样可以安全地从多个智能体方法中调用 get_mcp_tools()，或按环境选择性启用 MCP。

连接超时配置

你可以通过设置 mcp_connect_timeout 类属性来配置 MCP 服务器的连接超时。如果未指定超时，则默认为 30 秒。

python

@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())

python

@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 方法来过滤智能体可用的工具。

python

@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
    )

超时配置适用于团队中的所有 MCP 工具调用：

python

@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 集成

Simple DSL Integration

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

Stdio Transport

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

SSE Transport

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

Streamable HTTP Transport

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

Connecting to Multiple Servers

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

Security Considerations

查阅 MCP 集成的重要安全最佳实践，确保你的智能体安全。

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

GitHub Repository

CrewAI MCP 演示

使用 MCP 时保持安全

WARNING

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

安全警告：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）。如果不符此模式，复杂或多模态输出可能需要自定义处理。

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

概述

🚀 简单 DSL 集成（推荐）

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

结构化配置（完全控制）

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

视频教程

安装

快速开始：简单 DSL 集成

使用字符串引用快速开始

使用结构化配置快速开始

MCP 引用格式

基于字符串的引用

外部 MCP 服务器

已连接的 MCP 集成

结构化配置

Stdio 传输（本地服务器）

HTTP/Streamable HTTP 传输（远程服务器）

SSE 传输（实时流式传输）

混合引用

工具过滤

配置参数

MCPServerStdio 参数

MCPServerHTTP 参数

MCPServerSSE 参数

通用参数

核心特性

错误处理

高级：MCPServerAdapter

连接配置

过滤工具

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

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

与 CrewBase 一起使用

连接超时配置

过滤工具

探索 MCP 集成

使用 MCP 时保持安全

安全警告：DNS 重绑定攻击

限制

友情链接

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