Comment connecter l'SDK d'agent OpenAI aux serveurs MCP

Le SDK des Agents OpenAI offre un moyen puissant de construire des applications AI avec des capacités d'agent. L'une de ses fonctionnalités clés est la prise en charge du Protocole de Contexte de Modèle (MCP), qui standardise la façon dont les applications fournissent contexte et outils aux Modèles de Langage de Grande Taille (LLMs). Cet article vous guidera dans la connexion du SDK des Agents OpenAI aux serveurs MCP, avec des étapes détaillées et un code exemple.

Comprendre le Protocole de Contexte de Modèle (MCP)

Le MCP est un protocole ouvert qui standardise la façon dont les applications fournissent contexte aux LLMs. Pensez à MCP comme à un "port USB-C" pour les applications IA - il fournit un moyen standardisé de connecter des modèles IA à différentes sources de données et outils. Tout comme l'USB-C connecte vos appareils à divers périphériques, le MCP connecte les modèles IA à différents outils et sources de données.

Types de Serveurs MCP

La spécification MCP définit deux types de serveurs en fonction de leur mécanisme de transport :

1. Serveurs Stdio : Ceux-ci fonctionnent en tant que sous-processus de votre application, s'exécutant essentiellement "localement".
2. Serveurs SSE : Ceux-ci fonctionnent à distance, et vous vous connectez à eux via une URL en utilisant HTTP sur les Événements Envoyés par le Serveur (SSE).

Le SDK des Agents OpenAI fournit des classes correspondantes pour ces deux types de serveurs :

• MCPServerStdio pour les serveurs stdio locaux
• MCPServerSse pour les serveurs SSE distants

Prérequis

Avant de commencer, assurez-vous d'avoir :

Python 3.10 ou supérieur installé

Le SDK des Agents OpenAI installé :

pip install openai-agents

Pour utiliser des serveurs stdio locaux, vous pourriez avoir besoin d'outils supplémentaires comme npx (pour les serveurs MCP basés sur JavaScript)

Connexion à un Serveur MCP Stdio

Commençons par connecter à un serveur MCP stdio local. Nous utiliserons le serveur MCP de système de fichiers comme exemple.

Étape 1 : Installer les Dépendances Requises

Si vous prévoyez d'utiliser le serveur MCP de système de fichiers, vous aurez besoin de Node.js et NPX :

# Installer Node.js (si ce n'est pas déjà fait)
# Pour Ubuntu/Debian
sudo apt update
sudo apt install nodejs npm

# Pour macOS
brew install node

# Vérifier l'installation
node --version
npx --version

Étape 2 : Configurer la Structure de Votre Projet

Créez une structure de projet basique :

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

Étape 3 : Se Connecter à un Serveur MCP Stdio

Voici un exemple complet de connexion au serveur MCP de système de fichiers en utilisant 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():
    # Obtenez le chemin du répertoire pour les données d'exemple
    current_dir = os.path.dirname(os.path.abspath(__file__))
    sample_data_dir = os.path.join(current_dir, "sample_data")

    # Créez et configurez le serveur MCP
    async with MCPServerStdio(
        name="Serveur de Système de Fichiers",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
        },
    ) as mcp_server:
        # Créez un agent qui utilise le serveur MCP
        agent = Agent(
            name="Assistant de Fichiers",
            instructions="Utilisez les outils du système de fichiers pour lire et analyser les fichiers dans le répertoire sample_data.",
            mcp_servers=[mcp_server]
        )

        # Exécutez l'agent avec une requête utilisateur
        message = "Listez tous les fichiers dans le répertoire et résumez leur contenu."
        print(f"Exécution de la requête : {message}\\\\n")

        # Générez un ID de trace pour le débogage
        trace_id = gen_trace_id()
        with trace(workflow_name="Exemple de Système de Fichiers MCP", trace_id=trace_id):
            print(f"Voir la trace : <https://platform.openai.com/traces/{trace_id}\\\\n>")
            result = await Runner.run(starting_agent=agent, input=message)
            print(result.final_output)

if __name__ == "__main__":
    # Vérifiez si npx est installé
    if not shutil.which("npx"):
        raise RuntimeError("npx n'est pas installé. Veuillez l'installer avec `npm install -g npx`.")

    asyncio.run(run_agent_with_mcp())

Dans cet exemple :

1. Nous créons une instance MCPServerStdio qui exécute le serveur MCP de système de fichiers en tant que sous-processus.
2. Nous passons ce serveur au constructeur Agent via le paramètre mcp_servers.
3. Lorsque l'agent s'exécute, il appelle automatiquement list_tools() sur le serveur MCP pour rendre le LLM conscient des outils disponibles.
4. Si le LLM décide d'utiliser l'un des outils MCP, le SDK appelle call_tool() sur le serveur.

Connexion à un Serveur MCP SSE

Maintenant, voyons comment se connecter à un serveur MCP distant en utilisant SSE :

Étape 1 : Comprendre les Serveurs MCP SSE

Les serveurs MCP SSE (Événements Envoyés par le Serveur) fonctionnent à distance et exposent leur fonctionnalité via des points de terminaison HTTP. Contrairement aux serveurs stdio, ils ne fonctionnent pas en tant que sous-processus de votre application.

Étape 2 : Se Connecter à un Serveur MCP SSE

Voici un exemple de code connectant à un serveur 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():
    # Créez et configurez la connexion au serveur MCP SSE
    async with MCPServerSse(
        name="Service Météo",
        params={
            "url": "<https://example.com/mcp/sse>",
            # Paramètres d'authentification optionnels
            "headers": {
                "Authorization": "Bearer votre_clé_api_ici"
            }
        },
    ) as mcp_server:
        # Créez un agent utilisant le serveur MCP distant
        agent = Agent(
            name="Assistant Météo",
            instructions="Utilisez les outils météorologiques pour répondre à des questions sur les conditions météorologiques actuelles.",
            mcp_servers=[mcp_server],
            # Forcez l'agent à utiliser les outils lorsque disponibles
            model_settings=ModelSettings(tool_choice="required")
        )

        # Exécutez l'agent avec une requête utilisateur
        message = "Quel temps fait-il à Tokyo aujourd'hui ?"
        print(f"Exécution de la requête : {message}\\\\n")

        trace_id = gen_trace_id()
        with trace(workflow_name="Exemple MCP Météo", trace_id=trace_id):
            print(f"Voir la trace : <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())

Créer un Serveur MCP SSE Local Simple

Pour bien comprendre comment fonctionne le MCP, il est utile de mettre en œuvre un serveur MCP simple. Voici un exemple d'un serveur MCP SSE minimal :

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

app = FastAPI()

# Définir les outils que notre serveur MCP fournira
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "add",
            "description": "Additionnez deux nombres",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "Premier nombre"},
                    "b": {"type": "number", "description": "Deuxième nombre"}
                },
                "required": ["a", "b"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Obtenez la météo pour un emplacement",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "Nom de la ville"}
                },
                "required": ["location"]
            }
        }
    }
]

# Implémenter la fonctionnalité réelle de l'outil
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":
        # Dans une implémentation réelle, vous feriez appel à une API météo réelle
        weather_data = {
            "Tokyo": {"condition": "Ensoleillé", "temperature": 25},
            "New York": {"condition": "Nuageux", "temperature": 18},
            "Londres": {"condition": "Pluvieux", "temperature": 15}
        }
        location = parameters["location"]
        if location in weather_data:
            return {"weather": weather_data[location]}
        return {"error": f"Données météorologiques non disponibles pour {location}"}

    return {"error": f"Outil inconnu : {tool_name}"}

async def sse_event_generator(request: Request):
    # Lire le corps de la requête
    body_bytes = await request.body()
    body = json.loads(body_bytes)

    # Gérer différentes opérations 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)

Fonctionnalités Avancées

Mise en Cache des Listes d'Outils

Pour améliorer les performances, vous pouvez mettre en cache la liste des outils des serveurs MCP :

# Créer un serveur MCP avec mise en cache des outils
async with MCPServerSse(
    name="Service Météo",
    params={"url": "<https://example.com/mcp/sse>"},
    cache_tools_list=True  # Activer la mise en cache
) as mcp_server:
    # Utiliser le serveur comme avant...

Lorsque cache_tools_list=True est défini, le SDK n'appellera list_tools() sur le serveur MCP qu'une seule fois et réutilisera le résultat pour les exécutions d'agent suivantes. Cela réduit la latence, en particulier pour les serveurs distants.

Pour invalider le cache si nécessaire :

mcp_server.invalidate_tools_cache()

Traçage des Opérations MCP

Le SDK des Agents OpenAI inclut des capacités de traçage intégrées qui capturent automatiquement les opérations MCP :

1. Appels au serveur MCP pour lister les outils
2. Informations liées au MCP sur les appels de fonction

Vous pouvez voir ces traces sur https://platform.openai.com/traces/ lorsque vous utilisez le gestionnaire de contexte trace comme montré dans les exemples ci-dessus.

Utilisation de Plusieurs Serveurs MCP

Vous pouvez connecter votre agent à plusieurs serveurs MCP simultanément, lui donnant accès à une gamme plus large d'outils :

async def run_with_multiple_servers():
    async with MCPServerStdio(
        name="Système de Fichiers",
        params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
    ) as fs_server, MCPServerSse(
        name="API Météo",
        params={"url": "<https://example.com/weather/mcp/sse>"}
    ) as weather_server:
        # Créer un agent avec les deux serveurs MCP
        agent = Agent(
            name="Assistant Multi-Outils",
            instructions="Utilisez tous les outils disponibles pour aider l'utilisateur.",
            mcp_servers=[fs_server, weather_server]
        )

        # Exécutez l'agent
        result = await Runner.run(
            starting_agent=agent,
            input="Vérifiez d'abord la météo à Tokyo, puis lisez le contenu du fichier report.txt."
        )
        print(result.final_output)

Gestion des Erreurs et Débogage

Lorsque vous travaillez avec des serveurs MCP, vous pourriez rencontrer divers problèmes. Voici quelques problèmes courants et comment les gérer :

Problèmes de Connexion

Si votre serveur MCP ne répond pas :

try:
    async with MCPServerSse(
        name="Service Météo",
        params={"url": "<https://example.com/mcp/sse>"}
    ) as mcp_server:
        # Utiliser le serveur...
except Exception as e:
    print(f"Échec de la connexion au serveur MCP : {e}")
    # Implémentez une stratégie de secours

Erreurs d'Exécution des Outils

Lorsque l'exécution d'un outil échoue, gérez-la gracieusement :

try:
    result = await Runner.run(starting_agent=agent, input=user_query)
    print(result.final_output)
except Exception as e:
    print(f"Erreur lors de l'exécution de l'agent : {e}")
    # Vous pourriez vouloir vérifier les journaux de trace pour des informations détaillées sur l'erreur

Conclusion

Le support du SDK des Agents OpenAI pour le MCP vous permet d'étendre vos agents avec une large gamme d'outils et de capacités. Que vous utilisiez des serveurs stdio locaux ou des points de terminaison SSE distants, l'intégration est simple et puissante.

En vous connectant aux serveurs MCP, vos agents peuvent accéder aux systèmes de fichiers, aux API météo, aux bases de données, et virtuellement à tout outil externe ou source de données qui expose une interface MCP. Cette flexibilité fait du SDK des Agents OpenAI une base puissante pour la création d'applications IA sophistiquées.

N'oubliez pas de tirer parti des fonctionnalités comme la mise en cache des outils pour optimiser les performances, et d'utiliser les capacités de traçage intégrées pour déboguer et surveiller les interactions de votre agent avec les serveurs MCP.

Bonne programmation avec les Agents OpenAI et le MCP !