Tutoriel : notifications en pratique

Ce tutoriel vous montre comment implémenter les logs et les notifications de progression dans votre code.

Étape 1 : L’argument Context dans les fonctions d’outils

Les fonctions d’outils reçoivent automatiquement un argument Context en dernier paramètre. Cet objet a des méthodes pour logger des messages et signaler la progression au client.

from mcp.server.fastmcp import Context

@mcp.tool()
async def long_operation(
    data: str,
    *,
    context: Context  # ← reçu automatiquement
):
    # Vous pouvez maintenant utiliser context pour communiquer
    await context.info("Opération démarrée")
    # ... votre logique ...

Note : L’argument Context doit être déclaré avec * avant lui (keyword-only) pour que le SDK le reconnaisse.

Étape 2 : Créer des logs et de la progression

Utilisez le contexte pour communiquer l’état à intervalles réguliers :

@mcp.tool()
async def process_large_file(filename: str, *, context: Context):
    await context.info(f"Chargement du fichier {filename}...")
    await context.report_progress(10, 100)
    
    data = load_file(filename)
    
    await context.info("Traitement des données...")
    await context.report_progress(40, 100)
    
    results = process_data(data)
    
    await context.info("Génération du rapport final...")
    await context.report_progress(80, 100)
    
    report = generate_report(results)
    
    await context.info("Terminé !")
    await context.report_progress(100, 100)
    
    return report

Étape 3 : Définir les callbacks côté client

Définissez vos fonctions de callback dans votre code client :

# Callback pour les messages de log
async def on_log(params: LoggingMessageNotificationParams):
    level = params.level.upper()
    print(f"[{level}] {params.data}")

# Callback pour la progression  
async def on_progress(progress: float, total: float | None, message: str | None):
    if total:
        bar_length = 30
        filled = int(bar_length * progress / total)
        bar = "█" * filled + "░" * (bar_length - filled)
        pct = progress / total * 100
        print(f"\r[{bar}] {pct:.0f}%", end="", flush=True)

Étape 4 : Passer les callbacks aux bonnes fonctions

Le callback de logging va sur la session ; le callback de progression va sur l’appel d’outil :

async with ClientSession(
    read,
    write,
    logging_callback=on_log        # ← sur la session
) as session:
    await session.initialize()
    
    result = await session.call_tool(
        name="process_large_file",
        arguments={"filename": "data.csv"},
        progress_callback=on_progress  # ← sur l'appel d'outil
    )
    
    print(f"\nRésultat : {result}")

Résumé du flux

Outil s'exécute sur le serveur
    ↓
context.info("message") → LoggingMessageNotification → client
context.report_progress(n, total) → ProgressNotification → client
    ↓
logging_callback(params) appelé → affichage
progress_callback(progress, total, msg) appelé → barre de progression

Points clés

• Context est keyword-only (*) — placez-le après * dans la signature
• Le logging callback est fourni à la session (une fois)
• Le progress callback est fourni à chaque appel d’outil (flexible)
• Vous pouvez choisir d’ignorer l’un ou l’autre selon vos besoins
• Ces notifications n’affectent pas le résultat de l’outil — elles sont informatives uniquement