Aller au contenu principal

Exemple Complet : Serveur Distant avec Auth

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

Dans cette leçon, vous allez créer un serveur MCP distant complet avec authentification par token. Le serveur expose des tools de gestion de tâches (créer, lister, compléter) et le client s’y connecte via HTTP avec un bearer token.

Le serveur : task_server.py

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

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

# --- Configuration ---
API_TOKEN = "mcp-secret-token-2026"
api = FastAPI(title="Task Manager MCP")

# --- Middleware d'authentification ---
@api.middleware("http")
async def auth_middleware(request: Request, call_next):
    """Vérifie le token pour toutes les routes /mcp."""
    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)

# --- 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 ---
mcp.mount(api, path="/tasks")

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

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 est accessible à : https://xyz789.ngrok-free.app/tasks/mcp

Le client : client_tasks.py

# 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.sse import MCPClientSSE, SSEServerParams

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 = MCPClientSSE(
        sse_params=SSEServerParams(
            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())

Tester dans Le Chat de Mistral

Vous pouvez aussi ajouter ce serveur directement dans Le Chat :

  1. Aller dans l’onglet Connectors de Le Chat
  2. Cliquer Add Custom Connector
  3. Remplir les champs :
    • Name : Task Manager
    • URL : https://xyz789.ngrok-free.app/tasks/mcp
    • Auth : Token, coller mcp-secret-token-2026
  4. Cliquer Connect

Le Chat découvre automatiquement les 4 tools et vous pouvez interagir en langage naturel.

Architecture du projet

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

De ngrok à la production

ngrok est parfait pour le développement, mais pour la production, vous aurez besoin d’un hébergement permanent. La leçon 12 couvre les options de déploiement (Docker, cloud, Hugging Face Spaces).

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 MCPClientSSE 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
← Lecon precedenteOAuth 2.0 pour Serveurs MCPLecon suivante →Architecture d'un Serveur MCP Custom