Pular para o conteúdo

Servidores MCP como Ferramentas no CrewAI

O Model Context Protocol (MCP) fornece uma maneira padronizada para agentes de IA fornecerem contexto para LLMs comunicando-se com serviços externos, conhecidos como Servidores MCP.

O CrewAI oferece duas abordagens para integração MCP:

Use o campo mcps diretamente nos agentes para integração perfeita de ferramentas MCP:

from crewai import Agent
agent = Agent(
role="Analista de Pesquisa",
goal="Pesquisar e analisar informações",
backstory="Pesquisador especialista com acesso a ferramentas externas",
mcps=[
"https://mcp.exa.ai/mcp?api_key=sua_chave", # Servidor MCP externo
"https://api.weather.com/mcp#get_forecast", # Ferramenta específica do servidor
"snowflake", # MCP conectado do catálogo
"stripe#list_invoices" # Ferramenta específica de MCP conectado
]
)
# Ferramentas MCP agora estão automaticamente disponíveis para seu agente!

🔧 Avançado: MCPServerAdapter (Para Cenários Complexos)

Seção intitulada “🔧 Avançado: MCPServerAdapter (Para Cenários Complexos)”

Para casos de uso avançados que requerem gerenciamento manual de conexão, a biblioteca crewai-tools fornece a classe MCPServerAdapter.

Atualmente, suportamos os seguintes mecanismos de transporte:

  • HTTPS: para servidores remotos (comunicação segura via HTTPS)
  • Server-Sent Events (SSE): para servidores remotos (transmissão de dados unidirecional em tempo real do servidor para o cliente via HTTP)
  • Streamable HTTP: para servidores remotos (comunicação flexível e potencialmente bidirecional via HTTP, geralmente utilizando SSE para streams do servidor para o cliente)

Assista a este tutorial em vídeo para um guia abrangente sobre a integração do MCP com o CrewAI:

Antes de começar a usar MCP com crewai-tools, é necessário instalar a dependência extra mcp do crewai-tools com o seguinte comando:

Terminal window
uv pip install 'crewai-tools[mcp]'

A classe MCPServerAdapter da crewai-tools é a principal forma de conectar-se a um servidor MCP e disponibilizar suas ferramentas aos seus agentes CrewAI. Ela suporta diferentes mecanismos de transporte e simplifica o gerenciamento de conexões.

O uso de um gerenciador de contexto Python (with) é a abordagem recomendada para o MCPServerAdapter. Ele lida automaticamente com a abertura e o fechamento da conexão com o servidor MCP.

O MCPServerAdapter suporta várias opções de configuração para personalizar o comportamento da conexão:

  • connect_timeout (opcional): Tempo máximo em segundos para aguardar o estabelecimento de uma conexão com o servidor MCP. O padrão é 30 segundos se não especificado. Isso é particularmente útil para servidores remotos que podem ter tempos de resposta variáveis.
# Exemplo com timeout personalizado para conexão
with MCPServerAdapter(server_params, connect_timeout=60) as tools:
# A conexão terá timeout após 60 segundos se não estabelecida
pass
from crewai import Agent
from crewai_tools import MCPServerAdapter
from mcp import StdioServerParameters # Para servidor Stdio
# Exemplo de server_params (escolha um baseado no seu tipo de servidor):
# 1. Servidor Stdio:
server_params=StdioServerParameters(
command="python3",
args=["servers/your_server.py"],
env={"UV_PYTHON": "3.12", **os.environ},
)
# 2. Servidor SSE:
server_params = {
"url": "http://localhost:8000/sse",
"transport": "sse"
}
# 3. Servidor Streamable HTTP:
server_params = {
"url": "http://localhost:8001/mcp",
"transport": "streamable-http"
}
# Exemplo de uso (descomente e adapte após definir server_params):
with MCPServerAdapter(server_params, connect_timeout=60) as mcp_tools:
print(f"Available tools: {[tool.name for tool in mcp_tools]}")
meu_agente = Agent(
role="Usuário de Ferramentas MCP",
goal="Utilizar ferramentas de um servidor MCP.",
backstory="Posso conectar a servidores MCP e usar suas ferramentas.",
tools=mcp_tools, # Passe as ferramentas carregadas para o seu agente
reasoning=True,
verbose=True
)
# ... restante da configuração do seu crew ...

Este padrão geral mostra como integrar ferramentas. Para exemplos específicos para cada transporte, consulte os guias detalhados abaixo.

with MCPServerAdapter(server_params, connect_timeout=60) as mcp_tools:
print(f"Available tools: {[tool.name for tool in mcp_tools]}")
meu_agente = Agent(
role="Usuário de Ferramentas MCP",
goal="Utilizar ferramentas de um servidor MCP.",
backstory="Posso conectar a servidores MCP e usar suas ferramentas.",
tools=mcp_tools["tool_name"], # Passe as ferramentas filtradas para o seu agente
reasoning=True,
verbose=True
)
# ... restante da configuração do seu crew ...

Para usar ferramentas de servidores MCP dentro de uma classe CrewBase, utilize o método get_mcp_tools. As configurações dos servidores devem ser fornecidas via o atributo mcp_server_params. Você pode passar uma configuração única ou uma lista com múltiplas configurações.

@CrewBase
class CrewWithMCP:
# ... defina o arquivo de configuração de agentes e tasks ...
mcp_server_params = [
# Servidor Streamable HTTP
{
"url": "http://localhost:8001/mcp",
"transport": "streamable-http"
},
# Servidor SSE
{
"url": "http://localhost:8000/sse",
"transport": "sse"
},
# Servidor 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()) # obter todas as ferramentas disponíveis
# ... restante da configuração do seu crew ...

Você pode configurar o timeout de conexão para servidores MCP definindo o atributo de classe mcp_connect_timeout. Se nenhum timeout for especificado, o padrão é 30 segundos.

@CrewBase
class CrewWithMCP:
mcp_server_params = [...]
mcp_connect_timeout = 60 # timeout de 60 segundos para todas as conexões MCP
@agent
def your_agent(self):
return Agent(config=self.agents_config["your_agent"], tools=self.get_mcp_tools())
@CrewBase
class CrewWithDefaultTimeout:
mcp_server_params = [...]
# Nenhum mcp_connect_timeout especificado - usa padrão de 30 segundos
@agent
def your_agent(self):
return Agent(config=self.agents_config["your_agent"], tools=self.get_mcp_tools())

Você pode filtrar quais ferramentas estão disponíveis para seu agente passando uma lista de nomes de ferramentas para o método 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") # obter ferramentas específicas
)

A configuração de timeout se aplica a todas as chamadas de ferramentas MCP dentro do crew:

@CrewBase
class CrewWithCustomTimeout:
mcp_server_params = [...]
mcp_connect_timeout = 90 # timeout de 90 segundos para todas as conexões MCP
@agent
def filtered_agent(self):
return Agent(
config=self.agents_config["your_agent"],
tools=self.get_mcp_tools("tool_1", "tool_2") # ferramentas específicas com timeout personalizado
)

Confira este repositório para demonstrações completas e exemplos de integração MCP com CrewAI! 👇

Repositório GitHub

Demo MCP do CrewAI

Transportes SSE podem ser vulneráveis a ataques de DNS rebinding se não forem devidamente protegidos. Para prevenir isso:

  1. Sempre valide os cabeçalhos Origin das conexões SSE recebidas para garantir que venham de fontes esperadas
  2. Evite vincular servidores a todas as interfaces de rede (0.0.0.0) quando executando localmente – faça o bind apenas para localhost (127.0.0.1)
  3. Implemente autenticação adequada para todas as conexões SSE

Sem essas proteções, invasores podem usar DNS rebinding para interagir com servidores MCP locais via sites remotos.

Para mais detalhes, consulte a documentação de Segurança de Transporte da MCP da Anthropic.

  • Primitivas Suportadas: Atualmente, o MCPServerAdapter suporta principalmente a adaptação de tools MCP. Outras primitivas MCP como prompts ou resources não são integradas diretamente como componentes CrewAI através deste adaptador por enquanto.
  • Manipulação de Saída: O adaptador normalmente processa a saída principal de texto de uma ferramenta MCP (por exemplo, .content[0].text). Saídas complexas ou multimodais podem exigir tratamento customizado caso não se encaixem nesse padrão.