Come collegare l'OpenAI Agent SDK ai server MCP

Il SDK degli agenti OpenAI fornisce un modo potente per costruire applicazioni AI con capacità di agenti. Una delle sue caratteristiche principali è il supporto per il Protocollo di Contesto Modello (MCP), che standardizza il modo in cui le applicazioni forniscono contesto e strumenti ai Modelli di Linguaggio di Grandi Dimensioni (LLM). Questo articolo ti guiderà nella connessione del SDK degli Agenti OpenAI ai Server MCP, con passi dettagliati e codice di esempio.

Comprendere il Protocollo di Contesto Modello (MCP)

Il MCP è un protocollo aperto che standardizza il modo in cui le applicazioni forniscono contesto ai LLM. Pensa al MCP come a una "porta USB-C" per le applicazioni di intelligenza artificiale: fornisce un modo standardizzato per collegare i modelli AI a diverse fonti di dati e strumenti. Proprio come la USB-C collega i tuoi dispositivi a vari periferici, il MCP collega i modelli AI a diversi strumenti e fonti di dati.

Tipi di Server MCP

La specifica MCP definisce due tipi di server in base al loro meccanismo di trasporto:

1. Server Stdio: Questi vengono eseguiti come un sottoprocesso della tua applicazione, essenzialmente eseguendosi "localmente".
2. Server SSE: Questi vengono eseguiti remotamente e ti connetti a loro tramite un URL utilizzando HTTP tramite Eventi Inviati dal Server (SSE).

Il SDK degli Agenti OpenAI fornisce classi corrispondenti per entrambi questi tipi di server:

• MCPServerStdio per server stdio locali
• MCPServerSse per server SSE remoti

Prerequisiti

Prima di iniziare, assicurati di avere:

Python 3.10 o superiore installato

Il SDK degli Agenti OpenAI installato:

pip install openai-agents

Per utilizzare server stdio locali, potresti aver bisogno di strumenti aggiuntivi come npx (per server MCP basati su JavaScript)

Collegarsi a un Server MCP Stdio

Iniziamo collegandoci a un server MCP stdio locale. Utilizzeremo il server MCP filesystem come esempio.

Passo 1: Installa le Dipendenze Necessarie

Se prevedi di utilizzare il server MCP filesystem, avrai bisogno di Node.js e NPX:

# Installa Node.js (se non già installato)
# Per Ubuntu/Debian
sudo apt update
sudo apt install nodejs npm

# Per macOS
brew install node

# Verifica l'installazione
node --version
npx --version

Passo 2: Imposta la Struttura del Tuo Progetto

Crea una struttura di progetto di base:

my_agent_project/
├── sample_data/
│   ├── file1.txt
│   └── file2.md
├── main.py
└── README.md

Passo 3: Connettersi a un Server MCP Stdio

Qui c'è un esempio completo di collegamento al server MCP filesystem utilizzando 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():
    # Ottieni il percorso della directory per i dati di esempio
    current_dir = os.path.dirname(os.path.abspath(__file__))
    sample_data_dir = os.path.join(current_dir, "sample_data")

    # Crea e configura il server MCP
    async with MCPServerStdio(
        name="Server Filesystem",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
        },
    ) as mcp_server:
        # Crea un agente che utilizza il server MCP
        agent = Agent(
            name="FileAssistant",
            instructions="Usa gli strumenti del filesystem per leggere e analizzare i file nella directory sample_data.",
            mcp_servers=[mcp_server]
        )

        # Esegui l'agente con una query dell'utente
        message = "Elenca tutti i file nella directory e riassumi il loro contenuto."
        print(f"Esecuzione della query: {message}\\\\n")

        # Genera un ID di traccia per il debug
        trace_id = gen_trace_id()
        with trace(workflow_name="MCP Filesystem Example", trace_id=trace_id):
            print(f"Visualizza traccia: <https://platform.openai.com/traces/{trace_id}\\\\n>")
            result = await Runner.run(starting_agent=agent, input=message)
            print(result.final_output)

if __name__ == "__main__":
    # Controlla se npx è installato
    if not shutil.which("npx"):
        raise RuntimeError("npx non è installato. Si prega di installarlo con `npm install -g npx`.")

    asyncio.run(run_agent_with_mcp())

In questo esempio:

1. Creamo un'istanza di MCPServerStdio che esegue il server MCP filesystem come un sottoprocesso.
2. Passiamo questo server al costruttore di Agent tramite il parametro mcp_servers.
3. Quando l'agente viene eseguito, chiama automaticamente list_tools() sul server MCP per far sapere all'LLM quali strumenti sono disponibili.
4. Se l'LLM decide di utilizzare uno degli strumenti MCP, il SDK chiama call_tool() sul server.

Collegarsi a un Server MCP SSE

Ora vediamo come collegarsi a un server MCP remoto utilizzando SSE:

Passo 1: Comprendere i Server MCP SSE

I server MCP SSE (Eventi Inviati dal Server) vengono eseguiti remotamente ed espongono le loro funzionalità tramite endpoint HTTP. A differenza dei server stdio, non vengono eseguiti come sottoprocessi della tua applicazione.

Passo 2: Connettersi a un Server MCP SSE

Ecco un codice di esempio che si collega a un server 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():
    # Crea e configura la connessione al server SSE MCP
    async with MCPServerSse(
        name="Servizio Meteo",
        params={
            "url": "<https://example.com/mcp/sse>",
            # Parametri di autenticazione opzionali
            "headers": {
                "Authorization": "Bearer your_api_key_here"
            }
        },
    ) as mcp_server:
        # Crea un agente utilizzando il server MCP remoto
        agent = Agent(
            name="WeatherAssistant",
            instructions="Utilizza gli strumenti per il meteo per rispondere a domande sulle condizioni meteorologiche attuali.",
            mcp_servers=[mcp_server],
            # Forza l'agente a utilizzare gli strumenti quando disponibili
            model_settings=ModelSettings(tool_choice="required")
        )

        # Esegui l'agente con una query dell'utente
        message = "Che tempo fa a Tokyo oggi?"
        print(f"Esecuzione della query: {message}\\\\n")

        trace_id = gen_trace_id()
        with trace(workflow_name="Weather MCP Example", trace_id=trace_id):
            print(f"Visualizza traccia: <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())

Creare un Semplice Server MCP SSE Locale

Per comprendere appieno come funziona il MCP, è utile implementare un semplice server MCP. Ecco un esempio di un server MCP SSE minimo:

import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from typing import Dict, Any, List, Optional

app = FastAPI()

# Definisci gli strumenti che il nostro server MCP fornirà
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "add",
            "description": "Aggiungi due numeri insieme",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "Primo numero"},
                    "b": {"type": "number", "description": "Secondo numero"}
                },
                "required": ["a", "b"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Ottieni il meteo per una località",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "Nome della città"}
                },
                "required": ["location"]
            }
        }
    }
]

# Implementa la funzionalità effettiva dello strumento
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":
        # In una realizzazione reale, chiameresti un'API meteo reale
        weather_data = {
            "Tokyo": {"condition": "Soleggiato", "temperature": 25},
            "New York": {"condition": "Nuvoloso", "temperature": 18},
            "Londra": {"condition": "Piovoso", "temperature": 15}
        }
        location = parameters["location"]
        if location in weather_data:
            return {"weather": weather_data[location]}
        return {"error": f"Dati meteo non disponibili per {location}"}

    return {"error": f"Strumento sconosciuto: {tool_name}"}

async def sse_event_generator(request: Request):
    # Leggi il corpo della richiesta
    body_bytes = await request.body()
    body = json.loads(body_bytes)

    # Gestisci diverse operazioni 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)

Funzionalità Avanzate

Cache delle Liste degli Strumenti

Per migliorare le prestazioni, puoi memorizzare nella cache la lista degli strumenti dai server MCP:

# Crea un server MCP con caching degli strumenti
async with MCPServerSse(
    name="Servizio Meteo",
    params={"url": "<https://example.com/mcp/sse>"},
    cache_tools_list=True  # Abilita il caching
) as mcp_server:
    # Usa il server come prima...

Quando cache_tools_list=True è impostato, il SDK chiamerà solo list_tools() sul server MCP una volta e riutilizzerà il risultato per le esecuzioni successive dell'agente. Questo riduce la latenza, specialmente per i server remoti.

Per invalidare la cache se necessario:

mcp_server.invalidate_tools_cache()

Tracciamento delle Operazioni MCP

Il SDK degli Agenti OpenAI include capacità di tracciamento integrate che registrano automaticamente le operazioni MCP:

1. Chiamate al server MCP per elencare gli strumenti
2. Informazioni relative al MCP sulle chiamate di funzione

Puoi visualizzare queste tracce su https://platform.openai.com/traces/ quando utilizzi il gestore di contesto trace come mostrato negli esempi sopra.

Utilizzare più Server MCP

Puoi collegare il tuo agente a più server MCP contemporaneamente, dandogli accesso a un'ampia gamma di strumenti:

async def run_with_multiple_servers():
    async with MCPServerStdio(
        name="Filesystem",
        params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
    ) as fs_server, MCPServerSse(
        name="API Meteo",
        params={"url": "<https://example.com/weather/mcp/sse>"}
    ) as weather_server:
        # Crea l'agente con entrambi i server MCP
        agent = Agent(
            name="MultiToolAssistant",
            instructions="Usa tutti gli strumenti disponibili per aiutare l'utente.",
            mcp_servers=[fs_server, weather_server]
        )

        # Esegui l'agente
        result = await Runner.run(
            starting_agent=agent,
            input="Controlla innanzitutto il meteo a Tokyo, poi leggi il contenuto del file report.txt."
        )
        print(result.final_output)

Gestione degli Errori e Debugging

Quando lavori con i server MCP, potresti incontrare vari problemi. Ecco alcuni problemi comuni e come affrontarli:

Problemi di Connessione

Se il tuo server MCP non risponde:

try:
    async with MCPServerSse(
        name="Servizio Meteo",
        params={"url": "<https://example.com/mcp/sse>"}
    ) as mcp_server:
        # Usa il server...
except Exception as e:
    print(f"Impossibile connettersi al server MCP: {e}")
    # Implementa la strategia di fallback

Errori di Esecuzione degli Strumenti

Quando un'esecuzione dello strumento fallisce, gestiscila in modo elegante:

try:
    result = await Runner.run(starting_agent=agent, input=user_query)
    print(result.final_output)
except Exception as e:
    print(f"Errore durante l'esecuzione dell'agente: {e}")
    # Potresti voler controllare i log delle tracce per dettagli sugli errori

Conclusione

Il supporto del SDK degli Agenti OpenAI per il MCP ti consente di estendere i tuoi agenti con un'ampia gamma di strumenti e capacità. Che tu stia utilizzando server stdio locali o endpoint SSE remoti, l'integrazione è semplice e potente.

Collegandoti ai server MCP, i tuoi agenti possono accedere a file system, API meteo, database e praticamente qualsiasi strumento o fonte di dati esterna che espone un'interfaccia MCP. Questa flessibilità rende il SDK degli Agenti OpenAI una base potente per costruire applicazioni AI sofisticate.

Ricorda di sfruttare funzionalità come il caching degli strumenti per ottimizzare le prestazioni e usa le capacità di tracciamento integrate per debug e monitorare le interazioni del tuo agente con i server MCP.

Buon coding con OpenAI Agents e MCP!