通过 Stdio（标准输入/输出）传输机制，实现 CrewAI 与本地 MCP 服务器的连接
Stdio 传输（Stdio Transport）内容总结
该文档聚焦于如何通过 Stdio（标准输入/输出）传输机制，实现 CrewAI 与本地 MCP 服务器的连接，核心内容可分为概述、核心概念、连接方法及注意事项四部分，具体如下：
一、概述
Stdio 传输是专为 本地 MCP 服务器 设计的通信方式，通过进程间的标准输入（stdin）和输出（stdout）流实现数据交互，适用于 MCP 服务器为脚本或可执行文件、且与 CrewAI 应用运行在 同一台机器 的场景。
二、核心概念
- 本地执行（Local Execution）：Stdio 传输会管理本地运行的 MCP 服务器进程，所有通信均在本地完成，无需网络传输。
- StdioServerParameters 类：来自
mcp库的配置类，用于定义启动本地 MCP 服务器的关键参数，包括：command：启动服务器的命令（如python3）；args：命令参数（如服务器脚本路径["servers/your_stdio_server.py"]）；env：服务器进程的环境变量（如指定 Python 版本{"UV_PYTHON": "3.12"}，或配置PYTHONPATH）。
三、两种 Stdio 连接方式
文档提供了两种管理 MCP 服务器连接生命周期的方法，分别适用于不同需求场景：
1. 完全托管连接（推荐）
通过 Python 上下文管理器（with 语句） 自动管理连接，无需手动启动/停止服务器进程，安全性和便捷性更高，是大多数场景的首选。
核心逻辑
- 进入
with块时：自动启动 MCP 服务器进程，建立连接并加载工具； - 退出
with块时：自动停止服务器进程，释放资源。
关键代码示例
from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters
import os# 1. 配置本地 MCP 服务器参数
server_params = StdioServerParameters(command="python3", # 启动命令args=["servers/your_stdio_server.py"], # 服务器脚本路径env={"UV_PYTHON": "3.12", **os.environ} # 环境变量
)# 2. 上下文管理器托管连接，自动启停服务器
with MCPServerAdapter(server_params) as tools:# 3. 创建智能体、任务和团队，使用本地 MCP 工具research_agent = Agent(role="本地数据处理器",goal="通过本地 Stdio 工具处理数据",backstory="利用本地脚本完成专项任务的 AI",tools=tools, # 加载 MCP 服务器工具reasoning=True,verbose=True)processing_task = Task(description="处理 data.txt 文件并总结内容",expected_output="数据处理总结",agent=research_agent,markdown=True)data_crew = Crew(agents=[research_agent], tasks=[processing_task], process=Process.sequential)result = data_crew.kickoff() # 执行任务
2. 手动管理连接生命周期
适用于需要 精细控制服务器启停时机 的场景（如自定义启动条件、复杂错误处理），但需手动调用 API 管理进程，风险较高（需确保资源释放）。
核心要求
- 必须调用
mcp_server_adapter.stop()终止服务器进程，避免资源泄漏； - 推荐使用
try...finally块，确保异常场景下仍能停止进程。
关键代码示例
from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters
import os# 1. 配置服务器参数
stdio_params = StdioServerParameters(command="python3",args=["servers/your_stdio_server.py"],env={"UV_PYTHON": "3.12", **os.environ}
)mcp_server_adapter = MCPServerAdapter(server_params=stdio_params)try:# 2. 手动启动服务器和连接mcp_server_adapter.start()tools = mcp_server_adapter.tools # 加载工具# 3. 配置智能体、任务并执行manual_agent = Agent(role="本地任务执行器", goal="执行手动管理的 Stdio 工具", tools=tools, verbose=True)manual_task = Task(description="通过 Stdio 工具执行 perform_analysis 命令", agent=manual_agent)manual_crew = Crew(agents=[manual_agent], tasks=[manual_task], process=Process.sequential)result = manual_crew.kickoff()except Exception as e:print(f"手动连接出错：{e}")finally:# 4. 无论是否异常，均手动停止服务器（关键步骤）if mcp_server_adapter and mcp_server_adapter.is_connected:mcp_server_adapter.stop()print("手动停止 Stdio MCP 服务器")
四、重要注意事项
- 替换占位符：需将代码中的
servers/your_stdio_server.py等占位路径/命令，替换为实际的本地 MCP 服务器脚本路径和启动命令； - 环境变量配置：
StdioServerParameters的env参数可用于传递服务器所需的配置（如 API 密钥、路径变量），灵活控制服务器行为； - 资源释放优先：手动管理连接时，
stop()是核心操作，必须在进程结束前调用，否则可能导致僵尸进程占用资源。