CrewAI에서 MCP 서버를 도구로 활용하기

uv pip install 'crewai-tools[mcp]'

# 사용자 지정 연결 타임아웃 예시
with MCPServerAdapter(server_params, connect_timeout=60) as tools:
    # 60초 이내에 연결이 설정되지 않으면 타임아웃 발생
    pass

from crewai import Agent
from crewai_tools import MCPServerAdapter
from 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 설정 ...

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
class 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
class CrewWithMCP:
  mcp_server_params = [...]
  mcp_connect_timeout = 60  # 모든 MCP 연결에 60초 타임아웃

  @agent
  def your_agent(self):
      return Agent(config=self.agents_config["your_agent"], tools=self.get_mcp_tools())

@CrewBase
class CrewWithDefaultTimeout:
  mcp_server_params = [...]
  # mcp_connect_timeout 지정하지 않음 - 기본 30초 사용

  @agent
  def your_agent(self):
      return Agent(config=self.agents_config["your_agent"], tools=self.get_mcp_tools())

@agent
def another_agent(self):
    return Agent(
      config=self.agents_config["your_agent"],
      tools=self.get_mcp_tools("tool_1", "tool_2") # 특정 도구만 가져오기
    )

@CrewBase
class CrewWithCustomTimeout:
  mcp_server_params = [...]
  mcp_connect_timeout = 90  # 모든 MCP 연결에 90초 타임아웃

  @agent
  def filtered_agent(self):
      return Agent(
        config=self.agents_config["your_agent"],
        tools=self.get_mcp_tools("tool_1", "tool_2") # 사용자 지정 타임아웃으로 특정 도구
      )