Aller au contenu principal

Exemple Complet : Serveur Local avec Agent

Un projet MCP complet de A à Z

Dans cette leçon, vous allez assembler tous les concepts vus précédemment en un projet fonctionnel : un serveur MCP local qui fournit des informations météo, connecté à un agent Mistral capable de répondre à des questions en langage naturel.

Le serveur MCP : weather_server.py

# 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)

Le client avec agent : main.py

# 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())

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

Le flux d’exécution sera :

  1. Le client crée un agent Mistral
  2. Le RunContext lance weather_server.py en processus STDIO
  3. Il découvre les 3 tools : get_weather, compare_weather, list_cities
  4. Pour chaque question, le modèle décide quels tools appeler
  5. Les résultats sont intégrés dans la réponse

Mode streaming

Pour une meilleure expérience utilisateur, utilisez le mode streaming :

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 streaming vous permet de voir en temps réel les appels de tools et la construction progressive de la réponse.

Gestion des erreurs en pratique

Avec continue_on_fn_error=True, votre agent reste fonctionnel même si un tool échoue :

# 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..."

Structure du projet final

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