O SDK de Agentes da OpenAI oferece uma maneira poderosa de criar aplicativos de IA com capacidades de agente. Uma das suas principais funcionalidades é o suporte ao Protocolo de Contexto do Modelo (MCP), que padroniza como os aplicativos fornecem contexto e ferramentas para Modelos de Linguagem Grande (LLMs). Este artigo irá guiá-lo na conexão do SDK de Agentes da OpenAI aos Servidores MCP, com etapas detalhadas e código de exemplo.
O MCP é um protocolo aberto que padroniza como os aplicativos fornecem contexto para os LLMs. Pense no MCP como uma "porta USB-C" para aplicativos de IA - ele fornece uma maneira padronizada de conectar modelos de IA a diferentes fontes de dados e ferramentas. Assim como a USB-C conecta seus dispositivos a vários periféricos, o MCP conecta modelos de IA a diferentes ferramentas e fontes de dados.
A especificação MCP define dois tipos de servidores baseados em seu mecanismo de transporte:
O SDK de Agentes da OpenAI fornece classes correspondentes para ambos os tipos de servidores:
MCPServerStdio para servidores stdio locaisMCPServerSse para servidores SSE remotosAntes de começar, certifique-se de que você tem:
Python 3.10 ou superior instalado
O SDK de Agentes da OpenAI instalado:
pip install openai-agents
Para usar servidores stdio locais, você pode precisar de ferramentas adicionais como npx (para servidores MCP baseados em JavaScript)
Vamos começar conectando a um servidor MCP stdio local. Usaremos o servidor MCP de sistema de arquivos como exemplo.
Se você planeja usar o servidor MCP de sistema de arquivos, você precisará do Node.js e NPX:
# Instalar Node.js (se ainda não estiver instalado)
# Para Ubuntu/Debian
sudo apt update
sudo apt install nodejs npm
# Para macOS
brew install node
# Verificar instalação
node --version
npx --version
Crie uma estrutura de projeto básica:
meu_projeto_agente/
├── dados_exemplo/
│ ├── arquivo1.txt
│ └── arquivo2.md
├── main.py
└── README.md
Aqui está um exemplo completo de conexão ao servidor MCP de sistema de arquivos usando MCPServerStdio:
import asyncio
import os
import shutil
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerStdio
async def run_agent_with_mcp():
# Obter o caminho do diretório para dados de exemplo
current_dir = os.path.dirname(os.path.abspath(__file__))
sample_data_dir = os.path.join(current_dir, "dados_exemplo")
# Criar e configurar o servidor MCP
async with MCPServerStdio(
name="Servidor de Sistema de Arquivos",
params={
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
},
) as mcp_server:
# Criar um agente que usa o servidor MCP
agent = Agent(
name="AssistenteDeArquivo",
instructions="Use as ferramentas do sistema de arquivos para ler e analisar arquivos no diretório de dados_exemplo.",
mcp_servers=[mcp_server]
)
# Executar o agente com uma consulta do usuário
message = "Liste todos os arquivos no diretório e resuma seu conteúdo."
print(f"Executando consulta: {message}\\\\n")
# Gerar ID de rastreamento para depuração
trace_id = gen_trace_id()
with trace(workflow_name="Exemplo MCP Sistema de Arquivos", trace_id=trace_id):
print(f"Veja o rastreio: <https://platform.openai.com/traces/{trace_id}\\\\n>")
result = await Runner.run(starting_agent=agent, input=message)
print(result.final_output)
if __name__ == "__main__":
# Verifique se o npx está instalado
if not shutil.which("npx"):
raise RuntimeError("npx não está instalado. Por favor, instale-o com `npm install -g npx`.")
asyncio.run(run_agent_with_mcp())
No exemplo acima:
MCPServerStdio que roda o servidor MCP de sistema de arquivos como um subprocesso.Agent através do parâmetro mcp_servers.list_tools() no servidor MCP para fazer com que o LLM esteja ciente das ferramentas disponíveis.call_tool() no servidor.Agora vamos ver como se conectar a um servidor MCP remoto usando SSE:
Servidores MCP SSE (Eventos Enviados pelo Servidor) funcionam remotamente e expõem sua funcionalidade via endpoints HTTP. Ao contrário dos servidores stdio, eles não funcionam como subprocessos do seu aplicativo.
Aqui está um código de exemplo conectando a um servidor MCP SSE:
import asyncio
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerSse
from agents.model_settings import ModelSettings
async def run_agent_with_remote_mcp():
# Criar e configurar a conexão do servidor MCP SSE
async with MCPServerSse(
name="Serviço de Clima",
params={
"url": "<https://example.com/mcp/sse>",
# Parâmetros de autenticação opcionais
"headers": {
"Authorization": "Bearer seu_api_key_aqui"
}
},
) as mcp_server:
# Criar um agente usando o servidor MCP remoto
agent = Agent(
name="AssistenteDoClima",
instructions="Use as ferramentas de clima para responder perguntas sobre as condições climáticas atuais.",
mcp_servers=[mcp_server],
# Forçar o agente a usar ferramentas quando disponíveis
model_settings=ModelSettings(tool_choice="required")
)
# Executar o agente com uma consulta do usuário
message = "Como está o tempo em Tóquio hoje?"
print(f"Executando consulta: {message}\\\\n")
trace_id = gen_trace_id()
with trace(workflow_name="Exemplo MCP do Clima", trace_id=trace_id):
print(f"Veja o rastreio: <https://platform.openai.com/traces/{trace_id}\\\\n>")
result = await Runner.run(starting_agent=agent, input=message)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(run_agent_with_remote_mcp())
Para entender completamente como o MCP funciona, ajuda implementar um servidor MCP simples. Aqui está um exemplo de um servidor MCP SSE mínimo:
import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from typing import Dict, Any, List, Optional
app = FastAPI()
# Definir ferramentas que nosso servidor MCP fornecerá
TOOLS = [
{
"type": "function",
"function": {
"name": "add",
"description": "Adicionar dois números juntos",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number", "description": "Primeiro número"},
"b": {"type": "number", "description": "Segundo número"}
},
"required": ["a", "b"]
}
}
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Obter clima para uma localização",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "Nome da cidade"}
},
"required": ["location"]
}
}
}
]
# Implementar a funcionalidade real da ferramenta
async def call_tool(tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
if tool_name == "add":
return {"result": parameters["a"] + parameters["b"]}
elif tool_name == "get_weather":
# Em uma implementação real, você chamaria uma API de tempo real
weather_data = {
"Tóquio": {"condition": "Ensunado", "temperature": 25},
"Nova York": {"condition": "Nublado", "temperature": 18},
"Londres": {"condition": "Chuvoso", "temperature": 15}
}
location = parameters["location"]
if location in weather_data:
return {"weather": weather_data[location]}
return {"error": f"Dados climáticos não disponíveis para {location}"}
return {"error": f"Ferramenta desconhecida: {tool_name}"}
async def sse_event_generator(request: Request):
# Ler o corpo da solicitação
body_bytes = await request.body()
body = json.loads(body_bytes)
# Manipular diferentes operações MCP
if body["action"] == "list_tools":
yield f"data: {json.dumps({'tools': TOOLS})}\\\\n\\\\n"
elif body["action"] == "call_tool":
tool_name = body["tool_name"]
parameters = body["parameters"]
result = await call_tool(tool_name, parameters)
yield f"data: {json.dumps({'result': result})}\\\\n\\\\n"
@app.post("/sse")
async def sse_endpoint(request: Request):
return StreamingResponse(
sse_event_generator(request),
media_type="text/event-stream"
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Para melhorar o desempenho, você pode armazenar em cache a lista de ferramentas dos servidores MCP:
# Criar um servidor MCP com cache de ferramentas
async with MCPServerSse(
name="Serviço de Clima",
params={"url": "<https://example.com/mcp/sse>"},
cache_tools_list=True # Ativar cache
) as mcp_server:
# Usar o servidor como antes...
Quando cache_tools_list=True está definido, o SDK chamará list_tools() no servidor MCP apenas uma vez e reutilizará o resultado para execuções subsequentes do agente. Isso reduz a latência, especialmente para servidores remotos.
Para invalidar o cache, se necessário:
mcp_server.invalidate_tools_cache()
O SDK de Agentes da OpenAI inclui recursos de rastreamento embutidos que capturam automaticamente operações MCP:
Você pode visualizar esses rastros em https://platform.openai.com/traces/ quando você usar o gerenciador de contexto trace como mostrado nos exemplos acima.
Você pode conectar seu agente a vários servidores MCP simultaneamente, dando a ele acesso a uma gama mais ampla de ferramentas:
async def run_with_multiple_servers():
async with MCPServerStdio(
name="Sistema de Arquivos",
params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
) as fs_server, MCPServerSse(
name="API do Clima",
params={"url": "<https://example.com/weather/mcp/sse>"}
) as weather_server:
# Criar agente com ambos os servidores MCP
agent = Agent(
name="AssistenteMultiFerramenta",
instructions="Use todas as ferramentas disponíveis para ajudar o usuário.",
mcp_servers=[fs_server, weather_server]
)
# Executar o agente
result = await Runner.run(
starting_agent=agent,
input="Primeiro verifique o clima em Tóquio, depois leia o conteúdo do arquivo report.txt."
)
print(result.final_output)
Ao trabalhar com servidores MCP, você pode encontrar vários problemas. Aqui estão alguns problemas comuns e como lidar com eles:
Se o seu servidor MCP não estiver respondendo:
try:
async with MCPServerSse(
name="Serviço de Clima",
params={"url": "<https://example.com/mcp/sse>"}
) as mcp_server:
# Usar o servidor...
except Exception as e:
print(f"Falha ao conectar ao servidor MCP: {e}")
# Implementar estratégia de fallback
Quando uma execução de ferramenta falha, lide com isso de forma elegante:
try:
result = await Runner.run(starting_agent=agent, input=user_query)
print(result.final_output)
except Exception as e:
print(f"Erro durante a execução do agente: {e}")
# Você pode querer verificar os logs de rastreamento para informações de erro detalhadas
O suporte do SDK de Agentes da OpenAI para MCP permite que você estenda seus agentes com uma ampla gama de ferramentas e capacidades. Seja usando servidores stdio locais ou endpoints SSE remotos, a integração é direta e poderosa.
Ao conectar-se aos servidores MCP, seus agentes podem acessar sistemas de arquivos, APIs de clima, bancos de dados e virtualmente qualquer ferramenta externa ou fonte de dados que exponha uma interface MCP. Essa flexibilidade torna o SDK de Agentes da OpenAI uma base poderosa para construir aplicativos de IA sofisticados.
Lembre-se de aproveitar recursos como armazenamento em cache de ferramentas para otimizar o desempenho e usar as capacidades de rastreamento embutidas para depurar e monitorar as interações do seu agente com os servidores MCP.
Feliz codificação com Agentes da OpenAI e MCP!
