Stdio 전송
Stdio 전송 (Stdio Transport)
Stdio(Standard Input/Output) 전송은 MCPServerAdapter를 표준 입출력 스트림으로 통신하는 로컬 MCP 서버에 연결하기 위한 방식이에요. 주로 MCP 서버가 CrewAI 애플리케이션과 같은 머신에서 실행되는 스크립트나 실행 파일일 때 사용해요.
출처: 문서
본문
Overview (개요)
Key Concepts (핵심 개념)
- Local Execution — Stdio 전송은 MCP 서버를 위해 로컬에서 실행되는 프로세스를 관리해요.
- StdioServerParameters — mcp 라이브러리에서 제공하는 이 클래스로 Stdio 서버를 실행하기 위한 명령, 인자, 환경 변수를 구성해요.
Connecting via 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 수명주기를 수동으로 관리할 수 있어요.
서버 프로세스가 종료되고 리소스가 해제되도록 하려면 반드시 mcp_server_adapter.stop()을 호출해야 해요. try...finally 블록을 사용하는 것을 강력히 권장해요.
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)를 제공할 때 유용해요.