Aller au contenu principal

Exemple Complet : Serveur Local avec Agent

Mis à jour le 29 juillet 2026

Assembler les pièces

Les leçons précédentes ont posé les briques séparément : le serveur et ses tools, le client et son RunContext. Vous allez maintenant les assembler en un projet qui tourne réellement — un serveur MCP local fournissant des informations météo, piloté par un agent Mistral capable de répondre à des questions posées en langage naturel. Lisez le code, mais surtout notez ce que chaque partie décide : le serveur décide de ce qui est possible, l’agent décide de ce qui est pertinent.

Le serveur : weather_server.py

Trois tools composent ce serveur. Le premier renvoie la météo d’une ville, le deuxième compare deux villes, le troisième liste ce que la base contient. Ce dernier n’est pas décoratif : il donne au modèle un moyen de découvrir le périmètre disponible au lieu de deviner des noms de villes.

# weather_server.py
"""Serveur MCP local fournissant des outils météo."""

from mcp.server.fastmcp import FastMCP
import json

app = FastMCP("Weather Service")

# Base de données simulée
WEATHER_DATA = {
    "paris": {"temp": 18.5, "condition": "Nuageux", "humidity": 72},
    "lyon": {"temp": 22.0, "condition": "Ensoleillé", "humidity": 45},
    "marseille": {"temp": 25.3, "condition": "Ensoleillé", "humidity": 38},
    "lille": {"temp": 14.2, "condition": "Pluvieux", "humidity": 88},
    "bordeaux": {"temp": 20.1, "condition": "Partiellement nuageux", "humidity": 55},
}

@app.tool()
def get_weather(city: str) -> str:
    """Récupère la météo actuelle pour une ville française.

    Args:
        city: Nom de la ville en minuscules (paris, lyon, marseille, etc.)

    Returns:
        JSON avec température, condition et humidité.
        Retourne une erreur si la ville est inconnue.
    """
    city_lower = city.lower().strip()
    if city_lower in WEATHER_DATA:
        data = WEATHER_DATA[city_lower]
        return json.dumps({
            "city": city.capitalize(),
            "temperature_celsius": data["temp"],
            "condition": data["condition"],
            "humidity_percent": data["humidity"],
        }, ensure_ascii=False)
    return json.dumps({"error": f"Ville {city} non trouvée. Villes disponibles : {', '.join(WEATHER_DATA.keys())}"})

@app.tool()
def compare_weather(city_a: str, city_b: str) -> str:
    """Compare la météo entre deux villes françaises.

    Args:
        city_a: Première ville (en minuscules)
        city_b: Deuxième ville (en minuscules)

    Returns:
        Comparaison des températures et conditions des deux villes
    """
    a = WEATHER_DATA.get(city_a.lower().strip())
    b = WEATHER_DATA.get(city_b.lower().strip())
    if not a:
        return f"Ville inconnue : {city_a}"
    if not b:
        return f"Ville inconnue : {city_b}"

    diff = a["temp"] - b["temp"]
    warmer = city_a if diff > 0 else city_b
    return json.dumps({
        "city_a": {"name": city_a.capitalize(), **a},
        "city_b": {"name": city_b.capitalize(), **b},
        "temperature_difference": abs(diff),
        "warmer_city": warmer.capitalize(),
    }, ensure_ascii=False)

@app.tool()
def list_cities() -> str:
    """Liste toutes les villes disponibles dans la base météo.

    Returns:
        Liste des villes avec leur température actuelle
    """
    cities = [
        {"city": k.capitalize(), "temp": v["temp"], "condition": v["condition"]}
        for k, v in WEATHER_DATA.items()
    ]
    return json.dumps(cities, ensure_ascii=False)

Deux détails de conception méritent l’attention. Les tools normalisent l’entrée avec .lower().strip(), parce que le modèle écrira parfois « Paris » et parfois « paris » selon la formulation de l’utilisateur. Et lorsqu’une ville est absente, get_weather ne lève pas d’exception : il renvoie un JSON d’erreur qui énumère les villes connues, transformant l’échec en information exploitable par le modèle.

Le client : main.py

Le client fait quatre choses dans l’ordre : instancier Mistral, créer l’agent, décrire comment lancer le serveur, puis ouvrir le RunContext pour enchaîner les conversations.

# main.py
"""Client MCP avec agent Mistral pour les requêtes météo."""

import asyncio
import os
from pathlib import Path

from mistralai import Mistral
from mistralai.extra.run.context import RunContext
from mistralai.extra.mcp.stdio import MCPClientSTDIO
from mcp import StdioServerParameters
from pydantic import BaseModel

# Configuration
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
SERVER_PATH = Path(__file__).parent / "weather_server.py"

# Format de sortie structuré (optionnel)
class WeatherResponse(BaseModel):
    summary: str
    details: dict | None = None

async def main():
    # 1. Initialiser le client Mistral
    client = Mistral(api_key=MISTRAL_API_KEY)

    # 2. Créer l'agent
    agent = client.beta.agents.create(
        model="mistral-medium-latest",
        name="assistant-meteo",
        instructions=(
            "Vous êtes un assistant météo pour la France. "
            "Utilisez les outils disponibles pour répondre aux questions. "
            "Répondez toujours en français avec les unités métriques."
        ),
    )

    # 3. Configurer le serveur MCP local
    server_params = StdioServerParameters(
        command="python",
        args=[str(SERVER_PATH)],
        env=None,
    )
    mcp_client = MCPClientSTDIO(stdio_params=server_params)

    # 4. Ouvrir le contexte et enregistrer le client MCP
    async with RunContext(
        agent_id=agent.id,
        continue_on_fn_error=True,
    ) as run_ctx:
        await run_ctx.register_mcp_client(mcp_client=mcp_client)

        # 5. Lancer des conversations
        questions = [
            "Quel temps fait-il à Paris ?",
            "Compare la météo entre Marseille et Lille.",
            "Quelle ville est la plus chaude en ce moment ?",
        ]

        for question in questions:
            print(f"\n{'='*60}")
            print(f"Question : {question}")
            print(f"{'='*60}")

            result = await client.beta.conversations.run_async(
                run_ctx=run_ctx,
                inputs=question,
            )

            print(f"Réponse : {result.output}")

    # 6. Nettoyage automatique à la sortie du context manager
    print("\nAgent et serveur MCP fermés proprement.")

if __name__ == "__main__":
    asyncio.run(main())

Remarquez le chemin du serveur : Path(__file__).parent / "weather_server.py" situe le fichier par rapport au script plutôt que par rapport au répertoire courant. C’est la seule façon de garantir que le lancement fonctionne quel que soit l’endroit d’où vous exécutez python main.py.

Exécuter le projet

# Définir la clé API
export MISTRAL_API_KEY="votre_cle_api_mistral"

# Lancer le client (il lancera le serveur automatiquement)
python main.py

Vous ne démarrez qu’un seul processus. Le client crée l’agent Mistral, le RunContext lance weather_server.py comme processus STDIO, découvre les trois tools — get_weather, compare_weather et list_cities — puis, question après question, laisse le modèle choisir lesquels appeler avant d’intégrer les résultats dans sa réponse. La troisième question de l’exemple est la plus intéressante à observer : pour savoir quelle ville est la plus chaude, le modèle n’a pas d’autre choix que de passer par list_cities, alors qu’aucun mot de la question ne le suggère.

Voir la réponse se construire

Attendre la réponse complète convient à un script ; face à un utilisateur, le mode streaming offre une bien meilleure expérience.

async with RunContext(agent_id=agent.id) as run_ctx:
    await run_ctx.register_mcp_client(mcp_client=mcp_client)

    events = await client.beta.conversations.run_stream_async(
        run_ctx=run_ctx,
        inputs="Quel temps fait-il à Lyon ?",
    )

    async for event in events:
        if hasattr(event, "output"):
            print(event.output)
        else:
            # Événements intermédiaires (appels de tools, etc.)
            print(f"[Event] {event}")

Le flux d’événements ne sert pas qu’à l’affichage : c’est aussi votre meilleur outil de débogage, puisqu’il expose en temps réel les appels de tools et les arguments choisis par le modèle.

Quand un tool échoue

Grâce à continue_on_fn_error=True, une requête hors périmètre ne casse rien. Si l’utilisateur demande la météo à Tokyo, get_weather renvoie son JSON d’erreur, le modèle le lit et rebondit en proposant les villes réellement disponibles.

# Si l'utilisateur demande une ville inexistante :
# → get_weather retourne {"error": "Ville tokyo non trouvée..."}
# → Le modèle lit l'erreur et reformule sa réponse
# → "Je suis désolé, Tokyo n'est pas dans ma base. Voici les villes disponibles..."

Au final, le projet tient en quatre fichiers, dont deux seulement contiennent de la logique :

weather-mcp-project/
├── weather_server.py    # Serveur MCP (3 tools)
├── main.py              # Client avec agent Mistral
├── requirements.txt     # mcp, fastmcp, mistralai, pydantic
└── .env                 # MISTRAL_API_KEY=...

Points clés à retenir

  • Un projet MCP complet = serveur (tools) + client (agent + RunContext)
  • Le serveur définit les tools avec FastMCP et des décorateurs @app.tool()
  • Le client crée l’agent, ouvre le RunContext, et lance les conversations
  • Le mode streaming permet un affichage progressif
  • continue_on_fn_error=True gère gracieusement les erreurs de tools
  • Tout le code est asynchrone — utilisez asyncio.run() comme point d’entrée
← Lecon precedenteRunContext : Gérer le Client MCP et l'AgentLecon suivante →Configurer un Serveur SSE Distant