Stdio 传输

概览

Stdio（标准输入/输出）传输用于将 MCPServerAdapter 连接到本地 MCP 服务器，这些服务器通过标准输入和输出流进行通信。通常用于 MCP 服务器是与你的 CrewAI 应用运行在同一台机器上的脚本或可执行文件时。

关键概念

本地执行：Stdio 传输会管理 MCP 服务器的本地运行进程。
StdioServerParameters：来自 mcp 库的这个类用于配置启动 Stdio 服务器所需的命令、参数和环境变量。

通过 Stdio 连接

你可以通过两种主要方式管理与 Stdio 型 MCP 服务器的连接生命周期：

1. 完全托管连接（推荐）

推荐使用 Python 上下文管理器（with 语句）。它会在上下文进入时自动启动 MCP 服务器进程，并在退出时停止。

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

# Create a StdioServerParameters object
server_params=StdioServerParameters(
    command="python3",
    args=["servers/your_stdio_server.py"],
    env={"UV_PYTHON": "3.12", **os.environ},
)

with MCPServerAdapter(server_params) as tools:
    print(f"Available tools from Stdio MCP server: {[tool.name for tool in tools]}")

    # Example: Using the tools from the Stdio MCP server in a CrewAI Agent
    research_agent = Agent(
        role="Local Data Processor",
        goal="Process data using a local Stdio-based tool.",
        backstory="An AI that leverages local scripts via MCP for specialized tasks.",
        tools=tools,
        reasoning=True,
        verbose=True,
    )

    processing_task = Task(
        description="Process the input data file 'data.txt' and summarize its contents.",
        expected_output="A summary of the processed data.",
        agent=research_agent,
        markdown=True
    )

    data_crew = Crew(
        agents=[research_agent],
        tasks=[processing_task],
        verbose=True,
        process=Process.sequential
    )

    result = data_crew.kickoff()
    print("\nCrew Task Result (Stdio - Managed):\n", result)

2. 手动管理连接生命周期

如果你需要更精细地控制 Stdio MCP 服务器进程的启动和停止，可以手动管理 MCPServerAdapter 生命周期。

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

# Create a StdioServerParameters object
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:
    mcp_server_adapter.start()  # Manually start the connection and server process
    tools = mcp_server_adapter.tools
    print(f"Available tools (manual Stdio): {[tool.name for tool in tools]}")

    # Example: Using the tools with your Agent, Task, Crew setup
    manual_agent = Agent(
        role="Local Task Executor",
        goal="Execute a specific local task using a manually managed Stdio tool.",
        backstory="An AI proficient in controlling local processes via MCP.",
        tools=tools,
        verbose=True
    )

    manual_task = Task(
        description="Execute the 'perform_analysis' command via the Stdio tool.",
        expected_output="Results of the analysis.",
        agent=manual_agent
    )

    manual_crew = Crew(
        agents=[manual_agent],
        tasks=[manual_task],
        verbose=True,
        process=Process.sequential
    )


    result = manual_crew.kickoff() # Actual inputs depend on your tool
    print("\nCrew Task Result (Stdio - Manual):\n", result)

except Exception as e:
    print(f"An error occurred during manual Stdio MCP integration: {e}")
finally:
    if mcp_server_adapter and mcp_server_adapter.is_connected: # Check if connected before stopping
        print("Stopping Stdio MCP server connection (manual)...")
        mcp_server_adapter.stop()  # **Crucial: Ensure stop is called**
    elif mcp_server_adapter: # If adapter exists but not connected (e.g. start failed)
        print("Stdio MCP server adapter was not connected. No stop needed or start failed.")

请记得将占位路径和命令替换为你实际的 Stdio 服务器细节。StdioServerParameters 中的 env 参数可用于为服务器进程设置环境变量，这在配置其行为或提供必要路径（如 PYTHONPATH）时很有用。