El SDK de Agentes de OpenAI proporciona una forma poderosa de construir aplicaciones de IA con capacidades de agente. Una de sus características clave es el soporte para el Protocolo de Contexto de Modelo (MCP), que estandariza cómo las aplicaciones proporcionan contexto y herramientas a los Modelos de Lenguaje Grande (LLMs). Este artículo te guiará a través de la conexión del SDK de Agentes de OpenAI a los Servidores MCP, con pasos detallados y código de ejemplo.
El MCP es un protocolo abierto que estandariza cómo las aplicaciones proporcionan contexto a los LLMs. Piensa en el MCP como un "puerto USB-C" para aplicaciones de IA: proporciona una forma estandarizada de conectar modelos de IA a diferentes fuentes de datos y herramientas. Así como USB-C conecta tus dispositivos a varios periféricos, el MCP conecta modelos de IA a diferentes herramientas y fuentes de datos.
La especificación MCP define dos tipos de servidores según su mecanismo de transporte:
El SDK de Agentes de OpenAI proporciona clases correspondientes para estos dos tipos de servidores:
MCPServerStdio para servidores stdio localesMCPServerSse para servidores SSE remotosAntes de comenzar, asegúrate de tener:
Python 3.10 o superior instalado
El SDK de Agentes de OpenAI instalado:
pip install openai-agents
Para usar servidores stdio locales, es posible que necesites herramientas adicionales como npx (para servidores MCP basados en JavaScript)
Comencemos conectando a un servidor MCP stdio local. Usaremos el servidor MCP de sistema de archivos como ejemplo.
Si planeas usar el servidor MCP de sistema de archivos, necesitarás Node.js y NPX:
# Instalar Node.js (si no está ya instalado)
# Para Ubuntu/Debian
sudo apt update
sudo apt install nodejs npm
# Para macOS
brew install node
# Verificar la instalación
node --version
npx --version
Crea una estructura básica de proyecto:
my_agent_project/
├── sample_data/
│ ├── file1.txt
│ └── file2.md
├── main.py
└── README.md
Aquí tienes un ejemplo completo de conexión al servidor MCP de sistema de archivos 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():
# Obtén la ruta del directorio para los datos de muestra
current_dir = os.path.dirname(os.path.abspath(__file__))
sample_data_dir = os.path.join(current_dir, "sample_data")
# Crea y configura el servidor MCP
async with MCPServerStdio(
name="Servidor de Sistema de Archivos",
params={
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
},
) as mcp_server:
# Crea un agente que utiliza el servidor MCP
agent = Agent(
name="AsistenteDeArchivos",
instructions="Usa las herramientas del sistema de archivos para leer y analizar archivos en el directorio sample_data.",
mcp_servers=[mcp_server]
)
# Ejecuta el agente con una consulta de usuario
message = "Lista todos los archivos en el directorio y resume su contenido."
print(f"Ejecutando consulta: {message}\\\\n")
# Genera un ID de trazado para depuración
trace_id = gen_trace_id()
with trace(workflow_name="Ejemplo de MCP Sistema de Archivos", trace_id=trace_id):
print(f"Ver traza: <https://platform.openai.com/traces/{trace_id}\\\\n>")
result = await Runner.run(starting_agent=agent, input=message)
print(result.final_output)
if __name__ == "__main__":
# Verifica si npx está instalado
if not shutil.which("npx"):
raise RuntimeError("npx no está instalado. Por favor instálalo con `npm install -g npx`.")
asyncio.run(run_agent_with_mcp())
En este ejemplo:
MCPServerStdio que ejecuta el servidor MCP de sistema de archivos como un subprocesso.Agent a través del parámetro mcp_servers.list_tools() en el servidor MCP para hacer que el LLM sea consciente de las herramientas disponibles.call_tool() en el servidor.Ahora veamos cómo conectarnos a un servidor MCP remoto utilizando SSE:
Los servidores MCP SSE (Eventos Enviados por el Servidor) se ejecutan de forma remota y exponen su funcionalidad a través de puntos finales HTTP. A diferencia de los servidores stdio, no se ejecutan como subprocessos de tu aplicación.
Aquí tienes un código de ejemplo que se conecta a un 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():
# Crear y configurar la conexión al servidor MCP SSE
async with MCPServerSse(
name="Servicio Meteorológico",
params={
"url": "<https://example.com/mcp/sse>",
# Parámetros de autenticación opcionales
"headers": {
"Authorization": "Bearer your_api_key_here"
}
},
) as mcp_server:
# Crear un agente utilizando el servidor MCP remoto
agent = Agent(
name="AsistenteMeteorológico",
instructions="Usa las herramientas meteorológicas para responder preguntas sobre las condiciones meteorológicas actuales.",
mcp_servers=[mcp_server],
# Obliga al agente a usar herramientas cuando estén disponibles
model_settings=ModelSettings(tool_choice="required")
)
# Ejecuta el agente con una consulta de usuario
message = "¿Cómo está el clima en Tokio hoy?"
print(f"Ejecutando consulta: {message}\\\\n")
trace_id = gen_trace_id()
with trace(workflow_name="Ejemplo de MCP Meteorológico", trace_id=trace_id):
print(f"Ver traza: <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 cómo funciona el MCP, es útil implementar un servidor MCP simple. Aquí tienes un ejemplo de un 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 herramientas que nuestro servidor MCP proporcionará
TOOLS = [
{
"type": "function",
"function": {
"name": "add",
"description": "Suma dos números",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number", "description": "Primer número"},
"b": {"type": "number", "description": "Segundo número"}
},
"required": ["a", "b"]
}
}
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Obtener el clima para una ubicación",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "Nombre de la ciudad"}
},
"required": ["location"]
}
}
}
]
# Implementar la funcionalidad real de la herramienta
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":
# En una implementación real, llamarías a una API de clima real
weather_data = {
"Tokio": {"condition": "Soleado", "temperature": 25},
"Nueva York": {"condition": "Nublado", "temperature": 18},
"Londres": {"condition": "Lluvioso", "temperature": 15}
}
location = parameters["location"]
if location in weather_data:
return {"weather": weather_data[location]}
return {"error": f"Datos del clima no disponibles para {location}"}
return {"error": f"Herramienta desconocida: {tool_name}"}
async def sse_event_generator(request: Request):
# Leer el cuerpo de la solicitud
body_bytes = await request.body()
body = json.loads(body_bytes)
# Manejar diferentes operaciones 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 mejorar el rendimiento, puedes almacenar en caché la lista de herramientas de los servidores MCP:
# Crear un servidor MCP con almacenamiento en caché de herramientas
async with MCPServerSse(
name="Servicio Meteorológico",
params={"url": "<https://example.com/mcp/sse>"},
cache_tools_list=True # Habilitar almacenamiento en caché
) as mcp_server:
# Usar el servidor como antes...
Cuando cache_tools_list=True está configurado, el SDK solo llamará a list_tools() en el servidor MCP una vez y reutilizará el resultado para ejecuciones posteriores del agente. Esto reduce la latencia, especialmente para servidores remotos.
Para invalidar la caché si es necesario:
mcp_server.invalidate_tools_cache()
El SDK de Agentes de OpenAI incluye capacidades de trazado incorporadas que capturan automáticamente las operaciones MCP:
Puedes ver estas trazas en https://platform.openai.com/traces/ cuando usas el administrador de contexto trace como se muestra en los ejemplos anteriores.
Puedes conectar tu agente a múltiples servidores MCP simultáneamente, dándole acceso a una gama más amplia de herramientas:
async def run_with_multiple_servers():
async with MCPServerStdio(
name="Sistema de Archivos",
params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
) as fs_server, MCPServerSse(
name="API Meteorológica",
params={"url": "<https://example.com/weather/mcp/sse>"}
) as weather_server:
# Crear agente con ambos servidores MCP
agent = Agent(
name="AsistenteMultiherramientas",
instructions="Usa todas las herramientas disponibles para ayudar al usuario.",
mcp_servers=[fs_server, weather_server]
)
# Ejecutar el agente
result = await Runner.run(
starting_agent=agent,
input="Primero verifica el clima en Tokio, luego lee el contenido del archivo report.txt."
)
print(result.final_output)
Al trabajar con servidores MCP, es posible que encuentres varios problemas. Aquí hay algunos problemas comunes y cómo manejarlos:
Si tu servidor MCP no responde:
try:
async with MCPServerSse(
name="Servicio Meteorológico",
params={"url": "<https://example.com/mcp/sse>"}
) as mcp_server:
# Usar el servidor...
except Exception as e:
print(f"Error al conectar al servidor MCP: {e}")
# Implementar estrategia de fallback
Cuando una ejecución de herramienta falla, maneja el error de forma elegante:
try:
result = await Runner.run(starting_agent=agent, input=user_query)
print(result.final_output)
except Exception as e:
print(f"Error durante la ejecución del agente: {e}")
# Podrías querer verificar los registros de trazado para información de error detallada
El soporte del SDK de Agentes de OpenAI para MCP te permite extender tus agentes con una amplia gama de herramientas y capacidades. Ya sea que estés utilizando servidores stdio locales o puntos finales SSE remotos, la integración es simple y poderosa.
Al conectarte a servidores MCP, tus agentes pueden acceder a sistemas de archivos, APIs meteorológicas, bases de datos y prácticamente cualquier herramienta o fuente de datos externa que exponga una interfaz MCP. Esta flexibilidad hace que el SDK de Agentes de OpenAI sea una base poderosa para construir aplicaciones de IA sofisticadas.
Recuerda aprovechar funciones como el almacenamiento en caché de herramientas para optimizar el rendimiento, y utiliza las capacidades de trazado incorporadas para depurar y monitorear las interacciones de tu agente con los servidores MCP.
¡Feliz codificación con OpenAI Agents y MCP!
