CrewAI에서 MCP 서버를 툴로 사용하기

CrewAI에서 MCP 서버를 툴로 사용하기 (MCP Servers as Tools in CrewAI)

MCP(Model Context Protocol)는 AI 에이전트가 MCP 서버라고 불리는 외부 서비스와 통신해 LLM에 컨텍스트를 제공하는 표준화된 방식을 제공해요. CrewAI는 MCP 통합을 위한 두 가지 접근 방식을 제공합니다.

출처: 문서

본문

🚀 Simple DSL Integration (권장)

에이전트에 직접 mcps 필드를 사용해 매끄러운 MCP 툴 통합을 할 수 있어요. DSL은 문자열 참조(빠른 설정용)와 구조화된 구성(전체 제어용)을 모두 지원해요.

String-Based References (문자열 기반 참조 — 빠른 설정)

원격 HTTPS 서버와 CrewAI 카탈로그의 연결된 MCP 통합에 딱 맞아요:

from crewai import Agent

agent = Agent(
    role="Research Analyst",
    goal="Research and analyze information",
    backstory="Expert researcher with access to external tools",
    mcps=[
        "https://mcp.exa.ai/mcp?api_key=your_key",           # External MCP server
        "https://api.weather.com/mcp#get_forecast",          # Specific tool from server
        "snowflake",                                         # Connected MCP from catalog
        "stripe#list_invoices"                               # Specific tool from connected MCP
    ]
)
# MCP tools are now automatically available to your agent!

Structured Configurations (구조화된 구성 — 전체 제어)

연결 설정, 툴 필터링, 모든 전송 유형을 완전히 제어하려면:

from crewai import Agent
from crewai.mcp import MCPServerStdio, MCPServerHTTP, MCPServerSSE
from crewai.mcp.filters import create_static_tool_filter

agent = Agent(
    role="Advanced Research Analyst",
    goal="Research with full control over MCP connections",
    backstory="Expert researcher with advanced tool access",
    mcps=[
        # Stdio transport for local servers
        MCPServerStdio(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem"],
            env={"API_KEY": "your_key"},
            tool_filter=create_static_tool_filter(
                allowed_tool_names=["read_file", "list_directory"]
            ),
            cache_tools_list=True,
        ),
        # HTTP/Streamable HTTP transport for remote servers
        MCPServerHTTP(
            url="https://api.example.com/mcp",
            headers={"Authorization": "Bearer your_token"},
            streamable=True,
            cache_tools_list=True,
        ),
        # SSE transport for real-time streaming
        MCPServerSSE(
            url="https://stream.example.com/mcp/sse",
            headers={"Authorization": "Bearer your_token"},
        ),
    ]
)

🔧 Advanced: MCPServerAdapter (복잡한 시나리오용)

수동 연결 관리가 필요한 고급 사용 사례에서는 crewai-tools 라이브러리가 MCPServerAdapter 클래스를 제공해요. 현재 지원하는 전송 메커니즘은 다음과 같아요:

• Stdio — 로컬 서버용 (같은 머신의 프로세스 간 표준 입출력 통신)
• Server-Sent Events (SSE) — 원격 서버용 (HTTP를 통한 서버→클라이언트 단방향 실시간 데이터 스트리밍)
• Streamable HTTPS — 원격 서버용 (HTTPS를 통한 유연한, 잠재적 양방향 통신. 종종 서버→클라이언트 스트림에 SSE 활용)

Video Tutorial (비디오 튜토리얼)

CrewAI와의 MCP 통합에 대한 종합적인 가이드는 비디오 튜토리얼을 시청하세요.

Installation (설치)

CrewAI MCP 통합은 mcp 라이브러리가 필요해요:

# For Simple DSL Integration (Recommended)
uv add mcp

# For Advanced MCPServerAdapter usage
uv pip install 'crewai-tools[mcp]'

Quick Start: Simple DSL Integration

MCP 서버를 통합하는 가장 쉬운 방법은 에이전트의 mcps 필드를 사용하는 거예요. 문자열 참조나 구조화된 구성을 사용할 수 있어요.

문자열 참조로 빠른 시작

from crewai import Agent, Task, Crew

# Create agent with MCP tools using string references
research_agent = Agent(
    role="Research Analyst",
    goal="Find and analyze information using advanced search tools",
    backstory="Expert researcher with access to multiple data sources",
    mcps=[
        "https://mcp.exa.ai/mcp?api_key=your_key&profile=your_profile",
        "snowflake#run_query"
    ]
)

# Create task
research_task = Task(
    description="Research the latest developments in AI agent frameworks",
    expected_output="Comprehensive research report with citations",
    agent=research_agent
)

# Create and run crew
crew = Crew(agents=[research_agent], tasks=[research_task])
result = crew.kickoff()

구조화된 구성으로 빠른 시작

from crewai import Agent, Task, Crew
from crewai.mcp import MCPServerStdio, MCPServerHTTP, MCPServerSSE

# Create agent with structured MCP configurations
research_agent = Agent(
    role="Research Analyst",
    goal="Find and analyze information using advanced search tools",
    backstory="Expert researcher with access to multiple data sources",
    mcps=[
        # Local stdio server
        MCPServerStdio(
            command="python",
            args=["local_server.py"],
            env={"API_KEY": "your_key"},
        ),
        # Remote HTTP server
        MCPServerHTTP(
            url="https://api.research.com/mcp",
            headers={"Authorization": "Bearer your_token"},
        ),
    ]
)

# Create task
research_task = Task(
    description="Research the latest developments in AI agent frameworks",
    expected_output="Comprehensive research report with citations",
    agent=research_agent
)

# Create and run crew
crew = Crew(agents=[research_agent], tasks=[research_task])
result = crew.kickoff()

이게 전부예요! MCP 툴이 자동으로 발견되어 에이전트가 사용할 수 있게 돼요.

MCP Reference Formats (MCP 참조 형식)

mcps 필드는 문자열 참조(빠른 설정용)와 구조화된 구성(전체 제어용)을 모두 지원해요. 같은 목록에서 두 형식을 섞을 수도 있어요.

String-Based References (문자열 기반 참조)

외부 MCP 서버:

mcps=[
    # Full server - get all available tools
    "https://mcp.example.com/api",

    # Specific tool from server using # syntax
    "https://api.weather.com/mcp#get_current_weather",

    # Server with authentication parameters
    "https://mcp.exa.ai/mcp?api_key=your_key&profile=your_profile"
]

연결된 MCP 통합:

mcps=[
    # Connected MCP - get all available tools
    "snowflake",

    # Specific tool from a connected MCP using # syntax
    "stripe#list_invoices",

    # Multiple connected MCPs
    "snowflake",
    "stripe",
    "github"
]

Structured Configurations (구조화된 구성)

Stdio Transport (로컬 서버) — 로컬 프로세스로 실행되는 MCP 서버에 딱 맞아요:

from crewai.mcp import MCPServerStdio
from crewai.mcp.filters import create_static_tool_filter

mcps=[
    MCPServerStdio(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem"],
        env={"API_KEY": "your_key"},
        tool_filter=create_static_tool_filter(
            allowed_tool_names=["read_file", "write_file"]
        ),
        cache_tools_list=True,
    ),
    # Python-based server
    MCPServerStdio(
        command="python",
        args=["path/to/server.py"],
        env={"UV_PYTHON": "3.12", "API_KEY": "your_key"},
    ),
]

HTTP/Streamable HTTP Transport (원격 서버) — HTTP/HTTPS상의 원격 MCP 서버용:

from crewai.mcp import MCPServerHTTP

mcps=[
    # Streamable HTTP (default)
    MCPServerHTTP(
        url="https://api.example.com/mcp",
        headers={"Authorization": "Bearer your_token"},
        streamable=True,
        cache_tools_list=True,
    ),
    # Standard HTTP
    MCPServerHTTP(
        url="https://api.example.com/mcp",
        headers={"Authorization": "Bearer your_token"},
        streamable=False,
    ),
]

SSE Transport (실시간 스트리밍) — Server-Sent Events를 사용하는 원격 서버용:

from crewai.mcp import MCPServerSSE

mcps=[
    MCPServerSSE(
        url="https://stream.example.com/mcp/sse",
        headers={"Authorization": "Bearer your_token"},
        cache_tools_list=True,
    ),
]

Mixed References (혼합 참조)

문자열 참조와 구조화된 구성을 결합할 수 있어요:

from crewai.mcp import MCPServerStdio, MCPServerHTTP

mcps=[
    # String references
    "https://external-api.com/mcp",              # External server
    "snowflake",                                 # Connected MCP from catalog

    # Structured configurations
    MCPServerStdio(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem"],
    ),
    MCPServerHTTP(
        url="https://api.example.com/mcp",
        headers={"Authorization": "Bearer token"},
    ),
]

Tool Filtering (툴 필터링)

구조화된 구성은 고급 툴 필터링을 지원해요:

from crewai.mcp import MCPServerStdio
from crewai.mcp.filters import create_static_tool_filter, create_dynamic_tool_filter, ToolFilterContext

# Static filtering (allow/block lists)
static_filter = create_static_tool_filter(
    allowed_tool_names=["read_file", "write_file"],
    blocked_tool_names=["delete_file"],
)

# Dynamic filtering (context-aware)
def dynamic_filter(context: ToolFilterContext, tool: dict) -> bool:
    # Block dangerous tools for certain agent roles
    if context.agent.role == "Code Reviewer":
        if "delete" in tool.get("name", "").lower():
            return False
    return True

mcps=[
    MCPServerStdio(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem"],
        tool_filter=static_filter,  # or dynamic_filter
    ),
]

Configuration Parameters (구성 파라미터)

각 전송 유형은 특정 구성 옵션을 지원해요.

MCPServerStdio Parameters

• command (required): 실행할 명령 (예: "python", "node", "npx", "uvx")
• args (optional): 명령 인자 목록 (예: ["server.py"] 또는 ["-y", "@mcp/server"])
• env (optional): 프로세스에 전달할 환경 변수 딕셔너리
• tool_filter (optional): 사용 가능한 툴을 필터링하는 툴 필터 함수
• cache_tools_list (optional): 이후 접근 속도를 위해 툴 목록을 캐시할지 여부 (기본값: False)

MCPServerHTTP Parameters

• url (required): 서버 URL (예: "https://api.example.com/mcp")
• headers (optional): 인증 또는 기타 목적의 HTTP 헤더 딕셔너리
• streamable (optional): streamable HTTP 전송을 사용할지 여부 (기본값: True)
• tool_filter (optional): 툴 필터 함수
• cache_tools_list (optional): 툴 목록 캐시 여부 (기본값: False)

MCPServerSSE Parameters

• url (required): 서버 URL (예: "https://api.example.com/mcp/sse")
• headers (optional): HTTP 헤더 딕셔너리
• tool_filter (optional): 툴 필터 함수
• cache_tools_list (optional): 툴 목록 캐시 여부 (기본값: False)

Common Parameters (공통 파라미터)

모든 전송 유형이 지원해요:

• tool_filter: 어떤 툴을 사용할 수 있는지 제어하는 필터 함수. 다음 중 하나가 돼요:
  • None (default): 모든 툴 사용 가능
  • Static filter: create_static_tool_filter()로 만들며 allow/block 목록용
  • Dynamic filter: create_dynamic_tool_filter()로 만들며 컨텍스트 인지 필터링용
• cache_tools_list: True일 때, 첫 발견 후 툴 목록을 캐시해 이후 연결 성능을 향상시켜요.

Key Features (주요 기능)

• 🔄 Automatic Tool Discovery — 툴이 자동으로 발견·통합돼요.
• 🏷️ Name Collision Prevention — 서버 이름이 툴 이름에 접두사로 붙어요.
• ⚡ Performance Optimized — 스키마 캐싱을 통한 온디맨드 연결.
• 🛡️ Error Resilience — 사용 불가능한 서버를 우아하게 처리.
• ⏱️ Timeout Protection — 내장 타임아웃이 연결이 멈추는 것을 방지.
• 📊 Transparent Integration — 기존 CrewAI 기능과 매끄럽게 동작.
• 🔧 Full Transport Support — Stdio, HTTP/Streamable HTTP, SSE 전송.
• 🎯 Advanced Filtering — 정적·동적 툴 필터링 기능.
• 🔐 Flexible Authentication — 헤더, 환경 변수, 쿼리 파라미터 지원.

Error Handling (에러 처리)

MCP DSL 통합은 복원력 있게 설계되어 실패를 우아하게 처리해요:

from crewai import Agent
from crewai.mcp import MCPServerStdio, MCPServerHTTP

agent = Agent(
    role="Resilient Agent",
    goal="Continue working despite server issues",
    backstory="Agent that handles failures gracefully",
    mcps=[
        # String references
        "https://reliable-server.com/mcp",        # Will work
        "https://unreachable-server.com/mcp",     # Will be skipped gracefully
        "snowflake",                              # Connected MCP from catalog

        # Structured configs
        MCPServerStdio(
            command="python",
            args=["reliable_server.py"],          # Will work
        ),
        MCPServerHTTP(
            url="https://slow-server.com/mcp",     # Will timeout gracefully
        ),
    ]
)
# Agent will use tools from working servers and log warnings for failing ones

모든 연결 에러는 우아하게 처리돼요:

• 연결 실패 — 경고로 기록되고, 에이전트는 사용 가능한 툴로 계속 진행.
• 타임아웃 에러 — 연결이 30초 후 타임아웃 (설정 가능).
• 인증 에러 — 디버깅을 위해 명확하게 기록.
• 잘못된 구성 — 에이전트 생성 시점에 검증 에러가 raise됨.

Advanced: MCPServerAdapter

수동 연결 관리가 필요한 복잡한 시나리오에서는 crewai-tools의 MCPServerAdapter 클래스를 사용하세요. Python 컨텍스트 매니저(with 문)를 사용하는 것이 권장되며, MCP 서버 연결의 시작·중지를 자동으로 처리해 줘요.

Connection Configuration (연결 구성)

MCPServerAdapter는 연결 동작을 커스터마이즈하는 여러 구성 옵션을 지원해요.

• connect_timeout (optional): MCP 서버에 연결을 수립할 때까지 기다리는 최대 시간(초). 지정하지 않으면 기본값 30초. 응답 시간이 가변적인 원격 서버에 특히 유용해요.

# Example with custom connection timeout
with MCPServerAdapter(server_params, connect_timeout=60) as tools:
    # Connection will timeout after 60 seconds if not established
    pass

from crewai import Agent
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters # For Stdio Server

# Example server_params (choose one based on your server type):
# 1. Stdio Server:
server_params=StdioServerParameters(
    command="python3",
    args=["servers/your_server.py"],
    env={"UV_PYTHON": "3.12", **os.environ},
)

# 2. SSE Server:
server_params = {
    "url": "http://localhost:8000/sse",
    "transport": "sse"
}

# 3. Streamable HTTP Server:
server_params = {
    "url": "http://localhost:8001/mcp",
    "transport": "streamable-http"
}

# Example usage (uncomment and adapt once server_params is set):
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, # Pass the loaded tools to your agent
        reasoning=True,
        verbose=True
    )
    # ... rest of your crew setup ...

이 일반적인 패턴으로 툴을 통합할 수 있어요. 각 전송에 맞춘 구체적인 예시는 아래 상세 가이드를 참조하세요.

Filtering Tools (툴 필터링)

툴을 필터링하는 두 가지 방법이 있어요:

• 딕셔너리 스타일 인덱싱으로 특정 툴에 접근.
• MCPServerAdapter 생성자에 툴 이름 목록을 전달.

딕셔너리 스타일 인덱싱으로 특정 툴 접근

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 ...

MCPServerAdapter 생성자에 툴 이름 목록 전달

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 ...

Using with CrewBase (CrewBase와 함께 사용)

CrewBase 클래스 안에서 MCPServer 툴을 사용하려면 get_mcp_tools 메서드를 사용하세요. 서버 구성은 mcp_server_params 속성으로 제공해야 해요. 단일 구성이나 여러 서버 구성의 리스트를 전달할 수 있어요.

@CrewBase
class CrewWithMCP:
  # ... define your agents and tasks config file ...

  mcp_server_params = [
    # Streamable HTTP Server
    {
        "url": "http://localhost:8001/mcp",
        "transport": "streamable-http"
    },
    # SSE Server
    {
        "url": "http://localhost:8000/sse",
        "transport": "sse"
    },
    # StdIO Server
    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()) # get all available tools

    # ... rest of your crew setup ...

crew 클래스가 @CrewBase로 데코레이팅되면 어댑터 수명주기가 관리돼요:

• get_mcp_tools()의 첫 호출이 crew의 모든 에이전트가 재사용하는 공유 MCPServerAdapter를 지연 생성해요.
• @CrewBase가 주입한 암시적 after-kickoff 훅 덕분에 어댑터가 .kickoff() 완료 후 자동으로 종료돼요 — 수동 정리가 필요 없어요.
• mcp_server_params가 정의되지 않으면 get_mcp_tools()는 빈 목록을 반환해, MCP가 구성되었든 아니든 같은 코드 경로가 실행될 수 있어요. 여러 에이전트 메서드에서 get_mcp_tools()를 안전하게 호출하거나 환경별로 MCP를 선택적으로 활성화할 수 있어요.

Connection Timeout Configuration (연결 타임아웃 구성)

mcp_connect_timeout 클래스 속성을 설정해 MCP 서버의 연결 타임아웃을 구성할 수 있어요. 타임아웃을 지정하지 않으면 기본값 30초예요.

@CrewBase
class CrewWithMCP:
  mcp_server_params = [...]
  mcp_connect_timeout = 60  # 60 seconds timeout for all MCP connections

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

@CrewBase
class CrewWithDefaultTimeout:
  mcp_server_params = [...]
  # No mcp_connect_timeout specified - uses default 30 seconds

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

Filtering Tools (툴 필터링)

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") # get specific tools
    )

타임아웃 구성은 crew 내의 모든 MCP 툴 호출에 적용돼요:

@CrewBase
class CrewWithCustomTimeout:
  mcp_server_params = [...]
  mcp_connect_timeout = 90  # 90 seconds timeout for all MCP connections

  @agent
  def filtered_agent(self):
      return Agent(
        config=self.agents_config["your_agent"],
        tools=self.get_mcp_tools("tool_1", "tool_2") # specific tools with custom timeout
      )

Explore MCP Integrations (MCP 통합 살펴보기)

• Simple DSL Integration — 권장: 간단한 mcps=[] 필드 구문으로 손쉬운 MCP 통합.
• Stdio Transport — 표준 입출력으로 로컬 MCP 서버에 연결. 스크립트와 로컬 실행 파일에 이상적.
• SSE Transport — 실시간 데이터 스트리밍을 위해 Server-Sent Events로 원격 MCP 서버 통합.
• Streamable HTTP Transport — 유연한 Streamable HTTP로 원격 MCP 서버와 견고한 통신.
• Connecting to Multiple Servers — 단일 어댑터로 여러 MCP 서버에서 툴 집계.
• Security Considerations — 에이전트를 안전하게 유지하기 위한 MCP 통합 보안 모범 사례 검토.

전체 데모와 예시는 GitHub 저장소를 확인하세요! 👇

• GitHub Repository — CrewAI MCP Demo

Staying Safe with MCP (MCP로 안전하게 지내기)

MCP 서버를 사용하기 전에 항상 그 서버를 신뢰할 수 있는지 확인하세요.

Security Warning: DNS Rebinding Attacks

SSE 전송은 제대로 보호하지 않으면 DNS rebinding 공격에 취약할 수 있어요. 방지하려면:

• 들어오는 SSE 연결의 Origin 헤더를 항상 검증해 예상된 출처에서 왔는지 확인하세요.
• 로컬에서 실행할 때 서버를 모든 네트워크 인터페이스(0.0.0.0)에 바인딩하지 말고, localhost(127.0.0.1)에만 바인딩하세요.
• 모든 SSE 연결에 적절한 인증을 구현하세요.

이런 보호가 없으면 공격자가 DNS rebinding으로 원격 웹사이트에서 로컬 MCP 서버와 상호작용할 수 있어요. 자세한 내용은 Anthropic의 MCP Transport Security 문서를 참조하세요.

Limitations (제한 사항)

• 지원되는 프리미티브 — 현재 MCPServerAdapter는 주로 MCP 툴을 어댑팅해요. prompts나 resources 같은 다른 MCP 프리미티브는 이 어댑터를 통해 CrewAI 컴포넌트로 직접 통합되지 않아요.
• 출력 처리 — 어댑터는 보통 MCP 툴의 주요 텍스트 출력(예: .content[0].text)을 처리해요. 복잡하거나 멀티모달 출력이 이 패턴에 맞지 않으면 커스텀 처리가 필요할 수 있어요.

더 알아보기 (Learn more)

• MCP DSL 통합
• Stdio 전송