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

Mise à jour du 2 novembre 2026 — Le Chat devient Vibe. Depuis le 28 mai 2026, Le Chat s’appelle Vibe (toujours sur chat.mistral.ai) : un seul agent pour le travail et le code, avec vos conversations, réglages et abonnements repris. L’ancienne expérience Le Chat reste accessible dans le mode Chat ; dans le mode Work, les agents sont devenus des skills (invoquées avec @), les mémoires la base de connaissances et la recherche approfondie la skill Deep Research. Les offres s’appellent désormais Mistral Free, Pro, Team et Enterprise. Cette leçon emploie désormais le nom Vibe ; certains menus décrits ici ont pu changer de nom ou de place. Complément du 3 novembre : le résultat d’un run_async se lit dans result.output_as_text ; le SDK mistralai 3.7.0 n’a pas d’attribut output.

Mise à jour du 2 novembre 2026 — SDK Python MCP v2. Ce cours suit désormais la version 2 du SDK mcp (2.3.0 au 2 octobre 2026) : MCPServer remplace FastMCP, les réglages de transport passent à run() ou à streamable_http_app(), et une application exposée sous un vrai nom d’hôte doit déclarer ce nom (transport_security). C’est aussi la version qu’exige le SDK Mistral : l’extra mistralai[agents] demande mcp>=2.2.0,<3.0. Sources : What’s new in v2 · Guide de migration · Deploy & scale.

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 Vibe, 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, MCPServer 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 os
import uuid
from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, Request, HTTPException
from mcp.server.mcpserver import MCPServer
from mcp.server.transport_security import TransportSecuritySettings

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

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

# --- Serveur MCP ---
mcp = MCPServer("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)

# Nom d'hôte public (ngrok, domaine de production) déclaré au serveur MCP
PUBLIC_HOST = os.environ.get("MCP_PUBLIC_HOST", "localhost")
securite = TransportSecuritySettings(
    allowed_hosts=[PUBLIC_HOST, f"{PUBLIC_HOST}:*", "localhost:*", "127.0.0.1:*"]
)
api.mount("/tasks", mcp.streamable_http_app(transport_security=securite))

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

# Dans un premier terminal, ouvrir le tunnel ngrok
ngrok http 8000
# URL publique : https://xyz789.ngrok-free.app

# Dans un second terminal, lancer le serveur en déclarant ce nom d'hôte
MCP_PUBLIC_HOST=xyz789.ngrok-free.app python task_server.py

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 le SDK. 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.client 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_as_text}")

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 Vibe 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 : Vibe 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 Vibe
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

Vibe 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>=2.2,<3, mistralai[agents]
└── .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) + MCPServer (MCP) + middleware (auth), avec le nom d’hôte public déclaré dans transport_security
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 Vibe de Mistral
ngrok permet de prototyper rapidement avant un déploiement production