第 19 章 · MCP 集成

本章目标：理解 MCP 协议在多智能体体系中的位置，掌握 CrewAI 1.15 推荐的 mcps DSL 写法与进阶的 MCPServerAdapter，能把本地/远程 MCP server 的工具安全地挂给你的 Agent。

19.1 MCP 协议一分钟回顾

MCP（Model Context Protocol）是 Anthropic 发起的开放协议，它把"模型怎么调用外部能力"标准化成三件事：

Tools（工具）：模型可以调用的函数（本教程关心的核心）；
Resources（资源）：可读取的数据上下文；
Prompts（提示模板）：预置的提示词。

MCP 采用客户端-服务端架构：你的 CrewAI 应用是 MCP 客户端，通过三种传输方式连接 MCP 服务端：

传输方式	适用场景	通信方式
Stdio	本地进程（同一台机器）	标准输入/输出
Streamable HTTP	远程服务（推荐默认）	HTTPS，可双向
SSE	远程服务（实时流）	HTTP 单向推送

版本提示

CrewAI 1.15 提供两套 MCP 接入方式：官方推荐的 mcps 字段 DSL（自动管理连接），以及 crewai-tools 中的 MCPServerAdapter（手动管理连接，适合复杂场景）。老教程里"只能用 MCPServerAdapter"的说法已过时。

19.2 方式一：`mcps` DSL 字段（推荐）

最简单的方式是直接在 Agent 上加 mcps 字段，CrewAI 会自动发现工具、处理连接生命周期、给工具名加前缀防冲突：

python

import os
from crewai import Agent, Task, Crew, LLM

# 三方 OpenAI 兼容模型
llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.7,
)

researcher = Agent(
    role="研究分析师",
    goal="利用外部搜索工具调研 AI Agent 框架的最新进展",
    backstory="资深研究员，擅长使用多种数据源交叉验证信息",
    llm=llm,
    mcps=[
        # 远程 MCP server（字符串直连，可带鉴权参数）
        "https://mcp.exa.ai/mcp?api_key=your_exa_key",
        # 用 # 语法只取该 server 的某一个工具
        "https://api.weather.com/mcp#get_forecast",
    ],
)

task = Task(
    description="调研 2025 年多智能体框架的关键演进方向",
    expected_output="带引用来源的中文研究简报，500 字以内",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()
print(result.raw)

不需要手动启动/关闭连接，不需要手动传 tools=——工具自动挂载。

19.3 结构化配置：三种传输 + 工具过滤

需要完全控制连接参数时，用结构化配置对象。它们来自 crewai.mcp 模块：

python

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

# 静态过滤：只放行白名单工具，屏蔽危险工具
file_filter = create_static_tool_filter(
    allowed_tool_names=["read_file", "list_directory"],
    blocked_tool_names=["delete_file"],
)

agent = Agent(
    role="文档整理员",
    goal="安全地整理本地与远程文档",
    backstory="谨慎细致的文档管理员",
    llm=llm,
    mcps=[
        # 1) Stdio：本地 filesystem server（npx 启动官方示例 server）
        MCPServerStdio(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp/docs"],
            tool_filter=file_filter,       # 只暴露读类工具
            cache_tools_list=True,         # 缓存工具列表，加速后续连接
        ),
        # 2) Streamable HTTP：远程服务（默认 streamable=True）
        MCPServerHTTP(
            url="https://api.example.com/mcp",
            headers={"Authorization": f"Bearer {os.getenv('MCP_TOKEN')}"},
            cache_tools_list=True,
        ),
        # 3) SSE：实时流式远程服务
        MCPServerSSE(
            url="https://stream.example.com/mcp/sse",
            headers={"Authorization": f"Bearer {os.getenv('MCP_TOKEN')}"},
        ),
    ],
)

三种配置对象的公共参数：

tool_filter：None（全部工具）/ 静态过滤（allow/block 列表）/ 动态过滤函数（可按 context.agent.role 决定放行哪些工具）；
cache_tools_list：首次发现工具后缓存，避免每次连接重复拉取；
连接失败优雅降级：某个 server 挂掉只会记警告日志，Agent 继续用其余工具；无效配置则在 Agent 创建时直接报校验错误。

19.4 方式二：`MCPServerAdapter` 手动管理

需要精细控制连接生命周期（比如复用连接、精确到单个工具）时，用 crewai-tools 的 MCPServerAdapter。推荐配合 with 上下文管理器，自动启停连接：

python

import os
from crewai import Agent, Task, Crew
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters

# 本地 stdio server 的启动参数
server_params = StdioServerParameters(
    command="python3",
    args=["servers/filesystem_server.py"],
    env={"UV_PYTHON": "3.12", **os.environ},
)

# 远程 server 也可以用字典形式：
# server_params = {"url": "http://localhost:8000/sse", "transport": "sse"}
# server_params = {"url": "http://localhost:8001/mcp", "transport": "streamable-http"}

with MCPServerAdapter(server_params, connect_timeout=60) as mcp_tools:
    print(f"可用工具: {[t.name for t in mcp_tools]}")

    agent = Agent(
        role="文件管理员",
        goal="用 MCP 工具完成文件整理",
        backstory="熟悉文件系统的自动化助手",
        llm=llm,
        tools=[mcp_tools["read_file"]],  # 字典式索引：只取一个工具
    )
    task = Task(
        description="读取 /tmp/docs/notes.md 并总结要点",
        expected_output="3-5 条要点列表",
        agent=agent,
    )
    crew = Crew(agents=[agent], tasks=[task])
    result = crew.kickoff()
    print(result.raw)
# with 块结束时连接自动关闭

两种过滤方式：字典式索引 mcp_tools["tool_name"]，或在构造时传工具名列表 MCPServerAdapter(server_params, "tool_1", "tool_2")。

19.5 在 `@CrewBase` 工程化项目中使用

第 17 章的 CLI 工程结构中，@CrewBase 类提供了 get_mcp_tools() 方法，连接生命周期由框架托管（kickoff 结束后自动关闭）：

python

import os
from mcp import StdioServerParameters
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task


@CrewBase
class DocCrew:
    """带 MCP 文件工具的文档处理 Crew"""

    # 支持单配置或配置列表（可混合三种传输）
    mcp_server_params = [
        StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp/docs"],
        ),
        {"url": "http://localhost:8001/mcp", "transport": "streamable-http"},
    ]
    mcp_connect_timeout = 60  # 默认 30 秒，可按需调大

    @agent
    def reader(self) -> Agent:
        return Agent(
            config=self.agents_config["reader"],
            tools=self.get_mcp_tools("read_file", "list_directory"),  # 按名取工具
        )

    @task
    def summarize(self) -> Task:
        return Task(config=self.tasks_config["summarize"])

    @crew
    def crew(self) -> Crew:
        return Crew(agents=self.agents, tasks=self.tasks, process=Process.sequential)

get_mcp_tools() 不传参数则返回全部工具；首次调用会懒创建共享的 adapter，全 crew 复用同一个连接。

19.6 MCP 安全注意事项

MCP 把"能执行代码/读文件的外部服务"直接接进了你的 Agent，必须像对待"给陌生人的服务器开账号"一样谨慎：

只连接可信 server：官方文档反复强调——使用前必须信任该 MCP server。恶意 server 可以通过工具描述注入提示、诱导模型泄露数据；
防 DNS 重绑定（SSE 场景）：本地 SSE server 要校验 Origin 头、只绑定 127.0.0.1 而不是 0.0.0.0、并加鉴权；
最小权限原则：用 tool_filter 白名单只暴露必需工具，坚决屏蔽 delete_*、execute_* 类高危工具；
密钥不落盘：env= 与 headers= 中的凭证一律从环境变量读取（见第 25 章）；
超时兜底：远程 server 一定配置 connect_timeout，避免一个慢 server 拖死整个 crew（DSL 默认 30 秒超时并优雅跳过）。

已知限制

MCPServerAdapter 目前主要适配 MCP 的 tools 原语，prompts 和 resources 尚未直接映射为 CrewAI 组件；复杂/多模态的工具输出通常只取 .content[0].text。

本章小结

MCP 标准化了工具/资源/提示三类能力，CrewAI 作为客户端支持 Stdio、Streamable HTTP、SSE 三种传输；
推荐用 mcps DSL：字符串引用（支持 #工具名 精确选取）或结构化配置（MCPServerStdio/HTTP/SSE）；
进阶用 crewai-tools 的 MCPServerAdapter + with 上下文手动管理；@CrewBase 项目里用 get_mcp_tools()；
工具过滤（静态白名单/动态函数）+ 连接超时 + 优雅降级是生产必备；
安全面：只连可信 server、防 DNS 重绑定、最小权限、密钥走环境变量。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 在 CrewAI 1.15 中，官方推荐的 MCP 接入方式是？

2. mcps 字符串 "https://api.weather.com/mcp#get_forecast" 中的 # 语法作用是？

3. 关于 MCPServerAdapter 的连接管理，正确做法是？

4. 以下哪项不是官方给出的 MCP 安全建议？

🛠️ 动手实践

用 npx -y @modelcontextprotocol/server-filesystem /tmp/mcp-lab 起一个本地 filesystem server，用 DSL MCPServerStdio 挂载，配置白名单只允许 read_file 和 list_directory，让 Agent 列出并总结某个文本文件。
把实践 1 改造成 @CrewBase 项目结构（mcp_server_params + get_mcp_tools("read_file")），验证 kickoff 结束后连接被自动关闭（在 server 端打印连接日志观察）。
给实践 1 的配置增加一个不可达的远程 mcps 条目（如 https://unreachable.example.com/mcp），运行并观察日志：确认 crew 不会崩溃，而是警告后继续执行。

下一章我们解决"怎么科学地评估一个 crew 的好坏"：第 20 章 · 测试与评估。

第 19 章 · MCP 集成 ​

19.1 MCP 协议一分钟回顾 ​

19.2 方式一：mcps DSL 字段（推荐） ​

19.3 结构化配置：三种传输 + 工具过滤 ​

19.4 方式二：MCPServerAdapter 手动管理 ​

19.5 在 @CrewBase 工程化项目中使用 ​

19.6 MCP 安全注意事项 ​

本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 19 章 · MCP 集成

19.1 MCP 协议一分钟回顾

19.2 方式一：`mcps` DSL 字段（推荐）

19.3 结构化配置：三种传输 + 工具过滤

19.4 方式二：`MCPServerAdapter` 手动管理

19.5 在 `@CrewBase` 工程化项目中使用

19.6 MCP 安全注意事项

本章小结

🛠️ 动手实践