MCP

Créer un MCP Server en Python

Tutoriel pour créer un MCP Server avec le SDK Python : décorateurs, tools, resources et test avec uvx.

Python et MCP : un duo naturel

Le SDK Python pour MCP utilise des décorateurs qui rendent le code très lisible. Si vous êtes plus à l'aise en Python qu'en TypeScript, c'est votre point d'entrée. Le résultat est identique : un serveur MCP que Claude Code peut utiliser via le transport stdio.

Pré-requis

  • Python 3.10+ installé (python --version ou python3 --version)
  • pip ou uv disponible
  • Claude Code installé et fonctionnel
  • Connaissances de base en Python (fonctions async, type hints, décorateurs)

Scaffolding du projet

1

Créer le projet

mkdir mcp-internal-db && cd mcp-internal-db
python -m venv .venv
source .venv/bin/activate   # Linux/Mac
# .venv\Scripts\activate    # Windows
2

Installer les dépendances

pip install mcp httpx

Le package mcp contient le SDK complet. httpx servira pour les appels HTTP vers notre API interne simulée.

Avec uv, la commande équivalente :

uv pip install mcp httpx
3

Créer la structure

touch server.py

Votre arborescence :

mcp-internal-db/
├── .venv/
└── server.py          # Point d'entrée du serveur MCP

Pas besoin de plus pour démarrer. Un seul fichier suffit pour un MCP simple.

Le serveur MCP minimal

Commencez par le squelette dans server.py :

from mcp.server.fastmcp import FastMCP


# Créer le serveur MCP
mcp = FastMCP("internal-db")

FastMCP est la classe de haut niveau du SDK Python. Elle gère le transport, le protocole JSON-RPC et la découverte des outils. Tout le reste se fait avec des décorateurs.

Définir des Tools avec @mcp.tool

Un tool se crée en décorant une fonction async. Le SDK génère automatiquement le schéma JSON à partir des type hints et des docstrings.

from typing import Any




# Données simulées (remplacez par vos vrais appels API)
USERS_DB: list[dict[str, Any]] = [
    {"id": 1, "name": "Alice Martin", "email": "alice@example.com", "role": "admin", "active": True},
    {"id": 2, "name": "Bob Dupont", "email": "bob@example.com", "role": "dev", "active": True},
    {"id": 3, "name": "Claire Petit", "email": "claire@example.com", "role": "dev", "active": False},
    {"id": 4, "name": "David Moreau", "email": "david@example.com", "role": "manager", "active": True},
]




@mcp.tool()
async def search_users(query: str, active_only: bool = True) -> str:
    """Recherche des utilisateurs dans la base interne par nom ou email.


    Args:
        query: Terme de recherche (nom ou email, insensible à la casse)
        active_only: Si True, ne retourne que les utilisateurs actifs
    """
    query_lower = query.lower()
    results = [
        u for u in USERS_DB
        if query_lower in u["name"].lower() or query_lower in u["email"].lower()
    ]


    if active_only:
        results = [u for u in results if u["active"]]


    if not results:
        return f"Aucun utilisateur trouvé pour '{query}'."


    lines = [f"Résultats pour '{query}' ({len(results)} trouvé(s)) :"]
    for u in results:
        status = "actif" if u["active"] else "inactif"
        lines.append(f"- {u['name']} ({u['email']}) / rôle : {u['role']} / {status}")


    return "\n".join(lines)

Un second tool : statistiques

@mcp.tool()
async def get_user_stats() -> str:
    """Retourne des statistiques sur les utilisateurs de la base interne.


    Nombre total, répartition par rôle et statut d'activité.
    """
    total = len(USERS_DB)
    active = sum(1 for u in USERS_DB if u["active"])
    roles: dict[str, int] = {}
    for u in USERS_DB:
        roles[u["role"]] = roles.get(u["role"], 0) + 1


    role_lines = [f"  - {role} : {count}" for role, count in sorted(roles.items())]


    return "\n".join([
        f"Statistiques utilisateurs :",
        f"- Total : {total}",
        f"- Actifs : {active} / Inactifs : {total - active}",
        f"- Par rôle :",
        *role_lines,
    ])

Tool avec appel HTTP (API REST)

En production, vos tools appelleront de vraies APIs. Voici le pattern avec httpx :

import httpx




@mcp.tool()
async def fetch_user_from_api(user_id: int) -> str:
    """Récupère un utilisateur depuis l'API REST interne.


    Args:
        user_id: Identifiant numérique de l'utilisateur
    """
    # Exemple avec une API publique de test
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://jsonplaceholder.typicode.com/users/{user_id}",
                timeout=10.0,
            )
            response.raise_for_status()
            data = response.json()
            return (
                f"Utilisateur #{data['id']} :\n"
                f"- Nom : {data['name']}\n"
                f"- Email : {data['email']}\n"
                f"- Entreprise : {data['company']['name']}"
            )
    except httpx.HTTPStatusError as e:
        return f"Erreur HTTP {e.response.status_code} pour l'utilisateur #{user_id}."
    except httpx.RequestError as e:
        return f"Erreur de connexion : {e}"

Définir des Resources avec @mcp.resource

Les resources exposent des données en lecture seule, identifiées par des URIs.

import json




@mcp.resource("db://schema")
async def get_schema() -> str:
    """Schéma de la base de données interne."""
    schema = {
        "tables": {
            "users": {
                "columns": {
                    "id": "integer (primary key)",
                    "name": "varchar(255)",
                    "email": "varchar(255) unique",
                    "role": "enum(admin, dev, manager)",
                    "active": "boolean",
                },
                "row_count": len(USERS_DB),
            }
        }
    }
    return json.dumps(schema, indent=2, ensure_ascii=False)




@mcp.resource("db://roles")
async def get_roles() -> str:
    """Liste des rôles disponibles dans la base."""
    roles = sorted(set(u["role"] for u in USERS_DB))
    return json.dumps({"roles": roles}, ensure_ascii=False)

Claude Code peut lire ces resources pour comprendre la structure de vos données avant de formuler des requêtes.

Définir des Prompts avec @mcp.prompt

Les prompts sont des templates d'interaction pré-configurés.

@mcp.prompt()
async def audit_users(role: str = "all") -> str:
    """Génère un audit des utilisateurs, optionnellement filtré par rôle.


    Args:
        role: Rôle à auditer (all, admin, dev, manager)
    """
    if role == "all":
        return (
            "Effectue un audit complet des utilisateurs. "
            "Utilise get_user_stats pour les statistiques globales, "
            "puis search_users pour vérifier chaque rôle. "
            "Identifie les comptes inactifs et les anomalies potentielles."
        )
    return (
        f"Effectue un audit des utilisateurs avec le rôle '{role}'. "
        f"Utilise search_users pour les trouver et vérifie leur statut."
    )

Démarrer le serveur

Ajoutez le point d'entrée à la fin de server.py :

if __name__ == "__main__":
    mcp.run(transport="stdio")

C'est tout. Le SDK gère le reste : parsing JSON-RPC, découverte des outils, sérialisation des réponses.

Tester le MCP

1

Test rapide

# Vérifier que le serveur démarre sans erreur
python server.py

Le serveur attend des messages JSON-RPC sur stdin. Tapez Ctrl+C pour l'arrêter.

2

Test avec l'inspecteur MCP

npx @modelcontextprotocol/inspector python server.py

L'inspecteur ouvre une interface web pour tester interactivement vos tools et resources.

3

Test avec uvx (si uv est installé)

uvx permet d'exécuter un package Python dans un environnement isolé :

# Depuis le dossier du projet
uvx --from . mcp-internal-db

Pour que ça fonctionne, ajoutez un pyproject.toml :

[project]
name = "mcp-internal-db"
version = "1.0.0"
requires-python = ">=3.10"
dependencies = ["mcp", "httpx"]


[project.scripts]
mcp-internal-db = "server:mcp.run"

Intégrer dans Claude Code

Option 1 : via la CLI

claude mcp add internal-db -- python /chemin/absolu/vers/mcp-internal-db/server.py

Option 2 : via .mcp.json

{
  "mcpServers": {
    "internal-db": {
      "command": "python",
      "args": ["/chemin/absolu/vers/mcp-internal-db/server.py"]
    }
  }
}

Option 3 : avec uvx (recommandé pour la distribution)

{
  "mcpServers": {
    "internal-db": {
      "command": "uvx",
      "args": ["mcp-internal-db"]
    }
  }
}

Le fichier server.py complet

import json
from typing import Any


import httpx
from mcp.server.fastmcp import FastMCP


mcp = FastMCP("internal-db")


# --- Données simulées ---


USERS_DB: list[dict[str, Any]] = [
    {"id": 1, "name": "Alice Martin", "email": "alice@example.com", "role": "admin", "active": True},
    {"id": 2, "name": "Bob Dupont", "email": "bob@example.com", "role": "dev", "active": True},
    {"id": 3, "name": "Claire Petit", "email": "claire@example.com", "role": "dev", "active": False},
    {"id": 4, "name": "David Moreau", "email": "david@example.com", "role": "manager", "active": True},
]


# --- Tools ---




@mcp.tool()
async def search_users(query: str, active_only: bool = True) -> str:
    """Recherche des utilisateurs dans la base interne par nom ou email.


    Args:
        query: Terme de recherche (nom ou email, insensible à la casse)
        active_only: Si True, ne retourne que les utilisateurs actifs
    """
    query_lower = query.lower()
    results = [
        u for u in USERS_DB
        if query_lower in u["name"].lower() or query_lower in u["email"].lower()
    ]
    if active_only:
        results = [u for u in results if u["active"]]
    if not results:
        return f"Aucun utilisateur trouvé pour '{query}'."
    lines = [f"Résultats pour '{query}' ({len(results)} trouvé(s)) :"]
    for u in results:
        status = "actif" if u["active"] else "inactif"
        lines.append(f"- {u['name']} ({u['email']}) / {u['role']} / {status}")
    return "\n".join(lines)




@mcp.tool()
async def get_user_stats() -> str:
    """Retourne des statistiques sur les utilisateurs de la base interne."""
    total = len(USERS_DB)
    active = sum(1 for u in USERS_DB if u["active"])
    roles: dict[str, int] = {}
    for u in USERS_DB:
        roles[u["role"]] = roles.get(u["role"], 0) + 1
    role_lines = [f"  - {role} : {count}" for role, count in sorted(roles.items())]
    return "\n".join([
        "Statistiques utilisateurs :",
        f"- Total : {total}",
        f"- Actifs : {active} / Inactifs : {total - active}",
        "- Par rôle :",
        *role_lines,
    ])




@mcp.tool()
async def fetch_user_from_api(user_id: int) -> str:
    """Récupère un utilisateur depuis l'API REST interne.


    Args:
        user_id: Identifiant numérique de l'utilisateur
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://jsonplaceholder.typicode.com/users/{user_id}",
                timeout=10.0,
            )
            response.raise_for_status()
            data = response.json()
            return (
                f"Utilisateur #{data['id']} :\n"
                f"- Nom : {data['name']}\n"
                f"- Email : {data['email']}\n"
                f"- Entreprise : {data['company']['name']}"
            )
    except httpx.HTTPStatusError as e:
        return f"Erreur HTTP {e.response.status_code} pour l'utilisateur #{user_id}."
    except httpx.RequestError as e:
        return f"Erreur de connexion : {e}"




# --- Resources ---




@mcp.resource("db://schema")
async def get_schema() -> str:
    """Schéma de la base de données interne."""
    schema = {
        "tables": {
            "users": {
                "columns": {
                    "id": "integer (primary key)",
                    "name": "varchar(255)",
                    "email": "varchar(255) unique",
                    "role": "enum(admin, dev, manager)",
                    "active": "boolean",
                },
                "row_count": len(USERS_DB),
            }
        }
    }
    return json.dumps(schema, indent=2, ensure_ascii=False)




@mcp.resource("db://roles")
async def get_roles() -> str:
    """Liste des rôles disponibles dans la base."""
    roles = sorted(set(u["role"] for u in USERS_DB))
    return json.dumps({"roles": roles}, ensure_ascii=False)




# --- Prompts ---




@mcp.prompt()
async def audit_users(role: str = "all") -> str:
    """Génère un audit des utilisateurs, optionnellement filtré par rôle.


    Args:
        role: Rôle à auditer (all, admin, dev, manager)
    """
    if role == "all":
        return (
            "Effectue un audit complet des utilisateurs. "
            "Utilise get_user_stats pour les statistiques globales, "
            "puis search_users pour vérifier chaque rôle."
        )
    return (
        f"Effectue un audit des utilisateurs avec le rôle '{role}'. "
        f"Utilise search_users pour les trouver et vérifie leur statut."
    )




# --- Démarrage ---


if __name__ == "__main__":
    mcp.run(transport="stdio")

Comparatif TypeScript vs Python

CritèreTypeScriptPython
Schéma des paramètreszod expliciteType hints + docstring
Syntaxeserver.tool("nom", ...)@mcp.tool() décorateur
Exécution directeNécessite tsx ou compilationpython server.py suffit
Distributionnpm (npx)PyPI (uvx)
PerformanceLéger avantage (V8)Très correct pour du I/O
ÉcosystèmeIdéal si votre stack est JS/TSIdéal pour data, ML, scripts

Le choix dépend de votre stack. Les deux SDK sont maintenus par Anthropic et produisent des MCP parfaitement compatibles.

Prochaines étapes

Précédent

Sécuriser les MCP Claude Code : dangers réels et checklist

Suivant

Protocole MCP avancé : transports et architecture