from crewai import Agentagent = Agent( role="연구 분석가", goal="정보를 연구하고 분석", backstory="외부 도구에 접근할 수 있는 전문가 연구원", mcps=[ "https://mcp.exa.ai/mcp?api_key=your_key", # 외부 MCP 서버 "https://api.weather.com/mcp#get_forecast", # 서버의 특정 도구 "snowflake", # 카탈로그에서 연결된 MCP "stripe#list_invoices" # 연결된 MCP의 특정 도구 ])# MCP 도구들이 이제 자동으로 에이전트에서 사용 가능합니다!
crewai-tools의 MCPServerAdapter 클래스는 MCP 서버에 연결하고 해당 도구들을 CrewAI 에이전트에서 사용할 수 있도록 하는 기본 방법입니다. 다양한 전송 메커니즘을 지원하며 연결 관리를 간소화합니다.파이썬 컨텍스트 매니저(with 문)를 사용하는 것이 MCPServerAdapter를 위한 권장 방법입니다. 이를 통해 MCP 서버와의 연결 시작 및 종료가 자동으로 처리됩니다.
MCPServerAdapter는 연결 동작을 맞춤화할 수 있는 여러 구성 옵션을 지원합니다:
connect_timeout (선택 사항): MCP 서버에 연결을 설정하기 위해 대기할 최대 시간(초 단위)입니다. 명시하지 않으면 기본값은 30초입니다. 응답 시간이 가변적인 원격 서버에 특히 유용합니다.
# 사용자 지정 연결 타임아웃 예시with MCPServerAdapter(server_params, connect_timeout=60) as tools: # 60초 이내에 연결이 설정되지 않으면 타임아웃 발생 pass
from crewai import Agentfrom crewai_tools import MCPServerAdapterfrom mcp import StdioServerParameters # Stdio 서버용# 예시 server_params (서버 유형에 따라 하나 선택):# 1. Stdio 서버:server_params=StdioServerParameters( command="python3", args=["servers/your_server.py"], env={"UV_PYTHON": "3.12", **os.environ},)# 2. SSE 서버:server_params = { "url": "http://localhost:8000/sse", "transport": "sse"}# 3. 스트림 가능 HTTP 서버:server_params = { "url": "http://localhost:8001/mcp", "transport": "streamable-http"}# 예시 사용법 (server_params 설정 후 주석 해제 및 적용):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="MCP 서버의 도구를 활용합니다.", backstory="나는 MCP 서버에 연결하여 해당 도구를 사용할 수 있습니다.", tools=mcp_tools, # 불러온 도구를 Agent에 전달 reasoning=True, verbose=True ) # ... 나머지 crew 설정 ...
이 일반적인 패턴은 도구를 통합하는 방법을 보여줍니다. 각 transport에 맞춘 구체적인 예시는 아래의 상세 가이드를 참고하세요.
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 ...
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 클래스 내에서 MCPServer 도구를 사용하려면 get_mcp_tools 메서드를 사용하세요. 서버 구성은 mcp_server_params 속성을 통해 제공되어야 합니다. 단일 구성 또는 여러 서버 구성을 리스트 형태로 전달할 수 있습니다.
@CrewBaseclass CrewWithMCP: # ... 에이전트 및 작업 구성 파일 정의 ... mcp_server_params = [ # 스트리머블 HTTP 서버 { "url": "http://localhost:8001/mcp", "transport": "streamable-http" }, # SSE 서버 { "url": "http://localhost:8000/sse", "transport": "sse" }, # StdIO 서버 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()) # 모든 사용 가능한 도구 가져오기 # ... 나머지 crew 설정 ...
@CrewBase로 데코레이션된 클래스에서는 어댑터 수명 주기가 자동으로 관리됩니다.
get_mcp_tools()가 처음 호출될 때 공유 MCPServerAdapter가 지연 생성되며 crew 내 모든 에이전트가 이를 재사용합니다.
.kickoff()가 끝나면 @CrewBase가 주입한 after-kickoff 훅이 어댑터를 종료하므로 별도의 정리 코드가 필요 없습니다.
mcp_server_params를 지정하지 않으면 get_mcp_tools()는 빈 리스트를 반환하여 MCP 설정 여부와 상관없이 동일한 코드 경로를 사용할 수 있습니다.
따라서 여러 에이전트에서 get_mcp_tools()를 호출하거나 환경에 따라 MCP 사용을 토글하더라도 안전하게 동작합니다.
