Ecrire son propre serveur MCP en Python#

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

curl -LsSf https://astral.sh/uv/install.sh | sh

uv init entrepot-mcp
cd entrepot-mcp
uv venv
# Windows : .venv\Scripts\activate
# macOS / Linux : source .venv/bin/activate
uv add "mcp[cli]"

from mcp.server import MCPServer

mcp = MCPServer("demo")


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b


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

uv run serveur.py

"""Serveur MCP en lecture seule sur une base SQLite.

Expose trois outils :
  - list_tables    : lister les tables de la base
  - describe_table : donner les colonnes et types d'une table
  - run_select     : executer une requete SELECT et renvoyer les lignes

Regle de conception : ce serveur ne doit JAMAIS pouvoir modifier la base.
"""

import json
import logging
import os
import re
import sqlite3
from contextlib import closing

from mcp.server import MCPServer

# IMPORTANT : en transport stdio, ecrire sur stdout casse le protocole
# JSON-RPC. Le module logging ecrit sur stderr, donc il est sur.
# N'utilise jamais print() dans ce fichier.
logger = logging.getLogger(__name__)

mcp = MCPServer("entrepot")

# Le chemin de la base vient de l'environnement, jamais du code.
DB_PATH = os.environ.get("ENTREPOT_DB_PATH", "entrepot.db")

# Garde-fous
MAX_LIGNES = 200          # on ne renvoie jamais plus que ca au modele
MOTS_INTERDITS = (
    "insert", "update", "delete", "drop", "alter", "create",
    "truncate", "replace", "attach", "detach", "pragma", "vacuum",
)


def _connexion() -> sqlite3.Connection:
    """Ouvre la base en lecture seule via l'URI file: et le mode=ro."""
    uri = f"file:{DB_PATH}?mode=ro"
    conn = sqlite3.connect(uri, uri=True)
    conn.row_factory = sqlite3.Row
    return conn


def _est_un_select(sql: str) -> bool:
    """Verifie que la requete est bien un SELECT unique et sans mot interdit.

    Ce filtre est une ceinture. Les bretelles, c'est le compte de base
    de donnees en lecture seule. Ne compte jamais sur le filtre seul.
    """
    nettoye = re.sub(r"--[^\n]*", " ", sql)          # retire les commentaires
    nettoye = re.sub(r"/\*.*?\*/", " ", nettoye, flags=re.S)
    nettoye = nettoye.strip().rstrip(";").strip()

    if ";" in nettoye:                                # pas de requetes chainees
        return False
    if not nettoye.lower().startswith(("select", "with")):
        return False
    mots = set(re.findall(r"[a-z]+", nettoye.lower()))
    return not (mots & set(MOTS_INTERDITS))


@mcp.tool()
def list_tables() -> str:
    """Liste toutes les tables de l'entrepot avec leur nombre de lignes.

    Utilise cet outil EN PREMIER quand tu ne connais pas encore la structure
    de la base. Il ne prend aucun argument et ne modifie rien.
    """
    with closing(_connexion()) as conn:
        noms = [
            ligne["name"]
            for ligne in conn.execute(
                "SELECT name FROM sqlite_master "
                "WHERE type = 'table' AND name NOT LIKE 'sqlite_%' "
                "ORDER BY name"
            )
        ]
        resultat = []
        for nom in noms:
            nb = conn.execute(f'SELECT count(*) AS n FROM "{nom}"').fetchone()["n"]
            resultat.append({"table": nom, "lignes": nb})

    logger.info("list_tables a renvoye %d tables", len(resultat))
    return json.dumps(resultat, ensure_ascii=False, indent=2)


@mcp.tool()
def describe_table(table: str) -> str:
    """Donne les colonnes, les types et les valeurs d'exemple d'une table.

    Appelle cet outil AVANT d'ecrire une requete SQL sur une table, pour
    connaitre les vrais noms de colonnes et eviter d'en inventer.

    Args:
        table: Nom exact de la table, tel que renvoye par list_tables.
    """
    if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", table):
        return f"Nom de table invalide : {table}. Appelle list_tables d'abord."

    with closing(_connexion()) as conn:
        colonnes = [
            {"nom": c["name"], "type": c["type"], "non_nul": bool(c["notnull"])}
            for c in conn.execute(f'PRAGMA table_info("{table}")')
        ]
        if not colonnes:
            return f"Table inconnue : {table}. Appelle list_tables d'abord."

        exemples = [
            dict(ligne)
            for ligne in conn.execute(f'SELECT * FROM "{table}" LIMIT 3')
        ]

    return json.dumps(
        {"table": table, "colonnes": colonnes, "exemples": exemples},
        ensure_ascii=False,
        indent=2,
        default=str,
    )


@mcp.tool()
def run_select(sql: str) -> str:
    """Execute une requete SELECT en lecture seule et renvoie les lignes en JSON.

    Seules les requetes commencant par SELECT ou WITH sont acceptees. Toute
    tentative d'ecriture est refusee. Le resultat est tronque a 200 lignes :
    si tu as besoin d'un total, calcule-le en SQL avec une agregation plutot
    que de demander toutes les lignes.

    Args:
        sql: La requete SELECT a executer, en dialecte SQLite.
    """
    if not _est_un_select(sql):
        # Erreur metier lisible : le modele peut corriger tout seul.
        return (
            "Requete refusee : ce serveur est en lecture seule. "
            "Seules les requetes SELECT ou WITH, sans point-virgule interne "
            "et sans mot-cle d'ecriture, sont autorisees. "
            "Reformule ta requete en SELECT."
        )

    try:
        with closing(_connexion()) as conn:
            curseur = conn.execute(sql)
            lignes = [dict(l) for l in curseur.fetchmany(MAX_LIGNES)]
            tronque = curseur.fetchone() is not None
    except sqlite3.Error as erreur:
        logger.warning("Echec SQL : %s", erreur)
        return (
            f"La requete a echoue : {erreur}. "
            "Verifie les noms de tables et de colonnes avec describe_table."
        )

    charge = {
        "lignes_renvoyees": len(lignes),
        "tronque": tronque,
        "donnees": lignes,
    }
    return json.dumps(charge, ensure_ascii=False, indent=2, default=str)


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

claude mcp add --scope local --env ENTREPOT_DB_PATH=./entrepot.db \
  --transport stdio entrepot -- uv --directory . run serveur.py

{
  "mcpServers": {
    "entrepot": {
      "type": "stdio",
      "command": "uv",
      "args": ["--directory", "${CLAUDE_PROJECT_DIR}", "run", "serveur.py"],
      "env": {
        "ENTREPOT_DB_PATH": "${ENTREPOT_DB_PATH}"
      }
    }
  }
}

{
  "mcpServers": {
    "entrepot": {
      "type": "stdio",
      "command": "C:\\Users\\moi\\.local\\bin\\uv.exe",
      "args": ["--directory", "C:\\projets\\entrepot-mcp", "run", "serveur.py"],
      "env": {
        "ENTREPOT_DB_PATH": "C:\\projets\\entrepot-mcp\\entrepot.db"
      }
    }
  }
}

claude mcp list
claude mcp get entrepot

npx @modelcontextprotocol/inspector

@mcp.tool()
def query(sql: str) -> str:
    """Runs a query."""
    ...

@mcp.tool()
def run_select(sql: str) -> str:
    """Execute une requete SELECT en lecture seule et renvoie les lignes en JSON.

    Seules les requetes commencant par SELECT ou WITH sont acceptees. Toute
    tentative d'ecriture est refusee. Le resultat est tronque a 200 lignes :
    si tu as besoin d'un total, calcule-le en SQL avec une agregation plutot
    que de demander toutes les lignes.

    Args:
        sql: La requete SELECT a executer, en dialecte SQLite.
    """

Mauvais  : "Error: no such column"
Bon      : "La requete a echoue : no such column: montant_ht.
            Verifie les noms de colonnes avec describe_table."

@mcp.tool()
def [NOM_DE_L_OUTIL_EN_SNAKE_CASE]([PARAM_1]: [TYPE], [PARAM_2]: [TYPE] = [DEFAUT]) -> str:
    """[CE_QUE_FAIT_L_OUTIL_EN_UNE_PHRASE].

    [QUAND_L_UTILISER]. [QUAND_NE_PAS_L_UTILISER].
    [CE_QU_IL_RENVOIE_EXACTEMENT]. [SES_LIMITES].

    Args:
        [PARAM_1]: [DESCRIPTION_ET_FORMAT_ATTENDU].
        [PARAM_2]: [DESCRIPTION_ET_VALEUR_PAR_DEFAUT].
    """
    # 1. Valider les entrees
    if not [CONDITION_DE_VALIDITE]:
        return "[MESSAGE_D_ERREUR_QUI_DIT_QUOI_CORRIGER]"

    # 2. Faire le travail
    try:
        resultat = [APPEL_METIER]
    except [TYPE_D_EXCEPTION] as erreur:
        logger.warning("[CONTEXTE] : %s", erreur)
        return f"[MESSAGE_D_ERREUR] : {erreur}. [ACTION_DE_RATTRAPAGE_SUGGEREE]"

    # 3. Limiter la taille avant de renvoyer
    return json.dumps(resultat, ensure_ascii=False, default=str)

Relis mon serveur MCP dans [CHEMIN_DU_FICHIER].

Pour chaque outil expose, verifie :
1. La description dit-elle quand l'utiliser ET quand ne pas l'utiliser ?
2. La description annonce-t-elle ce qui est renvoye et les limites de taille ?
3. Les messages d'erreur disent-ils quoi corriger, ou juste que ca a echoue ?
4. Une entree malveillante peut-elle contourner le controle [NOM_DU_GARDE_FOU] ?
5. Y a-t-il un print() quelque part ? Le transport est stdio, donc c'est interdit.

Donne-moi les corrections sous forme de diff, sans reecrire tout le fichier.