Exemple Complet : Serveur Distant avec Auth — MCP pour Développeurs — Serveurs et Intégrations

Scénario : un serveur MCP de gestion de tâches

Les leçons précédentes ont posé les pièces séparément : le transport HTTP, le middleware, l’authentification par token. Vous allez maintenant les assembler dans un projet qui tourne de bout en bout. Le serveur expose quatre tools de gestion de tâches — créer, lister, compléter, supprimer — et le client s’y connecte à distance en présentant un bearer token à chaque requête. Le même serveur servira ensuite depuis Le Chat, sans une ligne de code supplémentaire : c’est tout l’intérêt d’un serveur distant par rapport à un serveur STDIO.

Le serveur : task_server.py

Trois éléments cohabitent dans ce fichier unique. FastAPI porte le transport HTTP et le middleware d’authentification, FastMCP porte le protocole MCP et les tools, et un dictionnaire Python tient lieu de base de données. L’ordre du fichier n’est pas indifférent : le serveur MCP et ses tools se déclarent d’abord, puis l’application FastAPI est créée avec le lifespan qui démarre le gestionnaire de sessions, et le montage vient en dernier. Observez au passage que le middleware protège le chemin /tasks, celui-là même sur lequel le serveur MCP est monté : les deux valeurs doivent rester cohérentes, sinon vous obtenez soit une porte ouverte, soit une porte murée.

# task_server.py
"""Serveur MCP distant avec authentification token."""

import json
import uuid
from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, Request, HTTPException
from mcp.server.fastmcp import FastMCP

# --- Configuration ---
API_TOKEN = "mcp-secret-token-2026"

# --- Base de données en mémoire ---
tasks_db: dict[str, dict] = {}

# --- Serveur MCP ---
mcp = FastMCP("Task Manager")

@mcp.tool()
def create_task(title: str, description: str = "", priority: str = "medium") -> str:
    """Crée une nouvelle tâche dans le gestionnaire.

    Args:
        title: Titre de la tâche (obligatoire, max 200 caractères)
        description: Description détaillée (optionnel)
        priority: Niveau de priorité : low, medium, high (défaut: medium)

    Returns:
        JSON avec l'identifiant et les détails de la tâche créée
    """
    task_id = str(uuid.uuid4())[:8]
    task = {
        "id": task_id,
        "title": title[:200],
        "description": description,
        "priority": priority if priority in ("low", "medium", "high") else "medium",
        "status": "open",
        "created_at": datetime.now().isoformat(),
    }
    tasks_db[task_id] = task
    return json.dumps({"success": True, "task": task}, ensure_ascii=False)

@mcp.tool()
def list_tasks(status: str = "all", priority: str = "all") -> str:
    """Liste les tâches avec filtres optionnels.

    Args:
        status: Filtrer par statut : all, open, completed (défaut: all)
        priority: Filtrer par priorité : all, low, medium, high (défaut: all)

    Returns:
        JSON avec la liste des tâches et le nombre total
    """
    filtered = list(tasks_db.values())
    if status != "all":
        filtered = [t for t in filtered if t["status"] == status]
    if priority != "all":
        filtered = [t for t in filtered if t["priority"] == priority]

    return json.dumps({
        "total": len(filtered),
        "tasks": filtered,
    }, ensure_ascii=False)

@mcp.tool()
def complete_task(task_id: str) -> str:
    """Marque une tâche comme terminée.

    Args:
        task_id: L'identifiant unique de la tâche

    Returns:
        Confirmation de la mise à jour ou message d'erreur
    """
    if task_id not in tasks_db:
        return json.dumps({"error": f"Tâche {task_id} introuvable"})

    tasks_db[task_id]["status"] = "completed"
    tasks_db[task_id]["completed_at"] = datetime.now().isoformat()
    return json.dumps({"success": True, "task": tasks_db[task_id]}, ensure_ascii=False)

@mcp.tool()
def delete_task(task_id: str) -> str:
    """Supprime une tâche du gestionnaire.

    Args:
        task_id: L'identifiant unique de la tâche à supprimer

    Returns:
        Confirmation de la suppression ou message d'erreur
    """
    if task_id not in tasks_db:
        return json.dumps({"error": f"Tâche {task_id} introuvable"})

    deleted = tasks_db.pop(task_id)
    return json.dumps({"success": True, "deleted_task": deleted["title"]}, ensure_ascii=False)

# --- Monter le serveur MCP dans FastAPI ---
# streamable_http_app() produit l'application ASGI du serveur MCP.
# Son gestionnaire de sessions doit être démarré par le lifespan de
# FastAPI, sinon le serveur répond mais chaque appel d'outil échoue.
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with mcp.session_manager.run():
        yield

api = FastAPI(title="Task Manager MCP", lifespan=lifespan)

# --- Middleware d'authentification ---
@api.middleware("http")
async def auth_middleware(request: Request, call_next):
    """Vérifie le token pour toutes les routes /tasks."""
    if request.url.path.startswith("/tasks"):
        auth = request.headers.get("Authorization", "")
        if auth != f"Bearer {API_TOKEN}":
            raise HTTPException(status_code=401, detail="Token invalide")
    return await call_next(request)

api.mount("/tasks", mcp.streamable_http_app())

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(api, host="0.0.0.0", port=8000)

Deux détails d’écriture méritent votre attention. Dans create_task, le titre est tronqué à 200 caractères et la priorité est ramenée à medium si la valeur reçue n’appartient pas à la liste autorisée : le modèle propose parfois des valeurs plausibles mais hors contrat, et le tool doit rester robuste sans échouer. Dans complete_task et delete_task, un identifiant inconnu ne lève pas d’exception mais retourne un message d’erreur en JSON — le modèle peut le lire, comprendre son erreur et reformuler, ce qu’un crash lui interdirait.

Lancer et exposer le serveur

# Lancer le serveur
python task_server.py

# Dans un autre terminal, exposer via ngrok
ngrok http 8000
# URL publique : https://xyz789.ngrok-free.app

L’endpoint MCP se trouve alors à https://xyz789.ngrok-free.app/tasks/mcp. Notez la composition de cette adresse : l’URL publique fournie par ngrok, puis le chemin de montage /tasks, puis le suffixe /mcp ajouté par FastMCP. Une confusion sur cette concaténation est la première cause d’échec de connexion.

Le client : client_tasks.py

Côté client, rien ne change par rapport à un serveur local sinon le type de client MCP et l’ajout du header d’authentification. L’agent, lui, reçoit des instructions qui décrivent son rôle en langage naturel — c’est ce qui l’incite à confirmer chaque action plutôt qu’à enchaîner silencieusement les appels.

# client_tasks.py
"""Client MCP pour le serveur de tâches distant avec authentification."""

import asyncio
import os
from mistralai import Mistral
from mistralai.extra.run.context import RunContext
from mistralai.extra.mcp.streamable_http import (
    MCPClientStreamableHTTP,
    StreamableHTTPServerParams,
)

API_TOKEN = "mcp-secret-token-2026"
SERVER_URL = "https://xyz789.ngrok-free.app/tasks/mcp"

async def main():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

    # Créer l'agent avec des instructions adaptées
    agent = client.beta.agents.create(
        model="mistral-medium-latest",
        name="task-manager",
        instructions=(
            "Vous êtes un gestionnaire de tâches intelligent. "
            "Utilisez les outils disponibles pour créer, lister et gérer "
            "les tâches de l'utilisateur. Répondez en français. "
            "Confirmez chaque action effectuée."
        ),
    )

    # Configurer le client MCP distant avec authentification
    mcp_client = MCPClientStreamableHTTP(
        params=StreamableHTTPServerParams(
            url=SERVER_URL,
            headers={"Authorization": f"Bearer {API_TOKEN}"},
            timeout=30,
        )
    )

    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)

        # Conversation avec le gestionnaire
        conversations = [
            "Crée une tâche haute priorité : Préparer la présentation MCP",
            "Crée une tâche moyenne priorité : Relire la documentation",
            "Liste toutes les tâches en cours",
            "Marque la première tâche comme terminée",
            "Montre-moi les tâches restantes",
        ]

        for msg in conversations:
            print(f"\n> {msg}")
            result = await client.beta.conversations.run_async(
                run_ctx=run_ctx,
                inputs=msg,
            )
            print(f"Agent : {result.output}")

asyncio.run(main())

La séquence de messages n’a rien d’anodin : elle enchaîne deux créations, une lecture, une mise à jour puis une relecture. C’est le scénario minimal pour vérifier que l’état survit d’un appel à l’autre et que le modèle sait retrouver l’identifiant d’une tâche créée plus tôt dans la conversation. Si la quatrième instruction échoue, votre problème vient rarement du serveur : il vient de ce que la réponse de list_tasks ne rend pas les identifiants assez visibles.

Tester dans Le Chat de Mistral

Le même endpoint s’ajoute directement comme connecteur custom, ce qui vous donne une interface de test sans écrire de client. C’est d’ailleurs la meilleure façon de valider vos docstrings : Le Chat les affiche telles quelles au moment de la découverte, et vous voyez immédiatement ce que le modèle aura sous les yeux pour décider quel tool appeler.

Aller dans l’onglet Connectors de Le Chat
Cliquer Add Custom Connector
Renseigner Name avec Task Manager, URL avec https://xyz789.ngrok-free.app/tasks/mcp, puis choisir Auth : Token et y coller mcp-secret-token-2026
Cliquer Connect

Le Chat découvre alors automatiquement les quatre tools et vous conversez en langage naturel avec votre propre serveur.

Organisation du projet et suite

task-mcp-project/
├── task_server.py       # Serveur MCP + FastAPI + auth
├── client_tasks.py      # Client Python avec agent
├── requirements.txt     # fastapi, uvicorn, mcp, fastmcp, mistralai
└── .env                 # MISTRAL_API_KEY, API_TOKEN

Ce squelette convient au prototype, mais deux choses le disqualifient pour la production. Le token en clair dans le code source doit passer dans le .env, et l’URL ngrok change à chaque redémarrage, ce qui casse tous les connecteurs déjà configurés. ngrok reste l’outil idéal pour valider une idée en quelques minutes ; l’hébergement permanent, avec ses options Docker, cloud et Hugging Face Spaces, fait l’objet de la leçon 12.

Points clés à retenir

Un serveur distant combine FastAPI (HTTP) + FastMCP (MCP) + middleware (auth)
L’authentification par token se fait via le header Authorization: Bearer ...
Le client MCPClientStreamableHTTP accepte des headers personnalisés pour l’authentification
Le même serveur est utilisable depuis un SDK Python ou Le Chat de Mistral
ngrok permet de prototyper rapidement avant un déploiement production