Tutoriel : notifications en pratique
Un outil MCP qui met trente secondes à répondre pose un problème que son résultat ne résout pas : pendant ces trente secondes, le client n’a aucun moyen de savoir si le traitement avance ou s’il est bloqué. Les notifications comblent ce vide en ouvrant un canal secondaire — des logs et une progression qui remontent pendant l’exécution, sans attendre la fin.
Ce tutoriel les implémente des deux côtés : côté serveur pour émettre, côté client pour recevoir. Retenez dès maintenant l’asymétrie qui surprend le plus et que l’étape 4 détaille : le callback de logs se déclare une fois sur la session, celui de progression à chaque appel d’outil.
Étape 1 : L’argument Context dans les fonctions d’outils
Le Context est injecté par le SDK : vous le déclarez dans la signature, et il arrive rempli. C’est lui qui porte le lien avec le client appelant, donc les méthodes pour lui envoyer des logs et de la progression. Sans lui, votre outil n’a aucun moyen de communiquer avant d’avoir terminé.
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
Le bon rythme d’émission est celui qui rassure sans noyer : un message par étape significative, pas un par itération de boucle. Un client qui reçoit trois cents notifications pour un traitement de trois cents lignes ne saura pas plus où en est le travail — et vous aurez saturé le canal pour rien.
@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
Contextest 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