Serveur MCP sur PostgreSQL en Python : agent IA sécurisé pas-à-pas

📍 Guide principal : Créer un serveur MCP : architecture, primitives, premier déploiement. Ce tutoriel suppose acquis le squelette d’un serveur MCP en Python ; en cas de doute, lisez le guide principal d’abord.

Brancher un agent IA sur une base PostgreSQL est l’un des cas d’usage les plus puissants de MCP : on transforme une base relationnelle en source de vérité interrogeable en langage naturel, sans réécrire de couche d’API. Mais c’est aussi l’un des plus dangereux. Une mauvaise séparation lecture/écriture, un pool de connexions mal dimensionné, une requête générée à la volée et c’est la fuite de données ou le déni de service. Ce tutoriel construit, étape par étape, un serveur MCP en Python qui expose un catalogue produits PostgreSQL avec des outils de lecture sécurisés, un outil d’écriture contrôlé, et le pattern de pool qui tient en production.

L’exemple choisi est un catalogue de formations en ligne (titre, durée, prix, niveau) — assez simple pour rester lisible, assez réaliste pour exposer les pièges typiques. À la fin, vous aurez un serveur MCP qui répond à des questions du type « combien de formations Python avancées avons-nous » et qui sait ajouter une formation sur demande, le tout sur asyncpg avec un pool partagé. Comptez 50 minutes pour parcourir le tutoriel intégralement.

Prérequis

• Python 3.11 ou plus récent
• PostgreSQL 14+ accessible localement (instance Docker, installation native, ou cloud comme Supabase, Neon, Railway)
• Le client psql ou un GUI comme pgAdmin pour vérifier les données
• L’inspecteur MCP installé (pip install "mcp[cli]" dans l’étape 1)
• Niveau attendu : intermédiaire (à l’aise avec SQL et async Python)
• Temps estimé : 50 minutes

Étape 1 — Créer le projet et installer les dépendances

On part d’un environnement Python isolé pour éviter tout mélange avec d’autres projets. Le choix entre asyncpg et psycopg3 mérite une note : asyncpg implémente le protocole binaire PostgreSQL nativement et reste la voie la plus rapide pour des charges asynchrones (les benchmarks officiels le donnent en moyenne plusieurs fois plus rapide que psycopg (le ratio exact dépend du workload) sur du throughput pur), tandis que psycopg3 embarque plus de fonctionnalités côté typage avancé et notifications. Pour MCP, asyncpg est le bon défaut : on est en async, on veut des temps de réponse courts, et on n’a pas besoin des extensions exotiques de Postgres.

mkdir mcp-postgres-server && cd mcp-postgres-server
python -m venv .venv
source .venv/bin/activate  # Windows : .venv\Scripts\activate
pip install "mcp[cli]" asyncpg python-dotenv

Trois paquets : mcp[cli] pour le SDK MCP et l’inspecteur de développement, asyncpg pour le driver Postgres, python-dotenv pour lire la connection string depuis un fichier .env. Si asyncpg échoue à compiler, vérifiez que vous avez les en-têtes de développement Python (python3-dev sur Debian/Ubuntu) ; sur macOS, Xcode Command Line Tools suffisent.

Étape 2 — Préparer la base et le schéma

Si vous n’avez pas déjà une instance Postgres, lancez-en une rapidement avec Docker. La commande ci-dessous démarre Postgres 16 sur le port 5432 avec un mot de passe simple — adaptez à votre convenance pour un usage non local.

docker run --name mcp-pg -e POSTGRES_PASSWORD=motdepasse \
  -e POSTGRES_DB=catalogue -p 5432:5432 -d postgres:16

Au bout de quelques secondes, le conteneur tourne et Postgres écoute sur localhost:5432. Vérifiez avec docker ps que le statut est « healthy ». Si le port 5432 est déjà pris (autre Postgres en local), changez le mapping en -p 5433:5432 et reportez ce port dans la connection string plus loin.

Connectez-vous et créez le schéma. Le but est d’avoir une seule table formations avec quelques lignes pour pouvoir tester immédiatement.

docker exec -it mcp-pg psql -U postgres -d catalogue

Au prompt catalogue=#, collez le bloc SQL suivant. Il crée la table avec une clé primaire autoincrémentée, des contraintes de validation simples (durée et prix positifs, niveau dans une liste fermée), un index sur le titre pour les recherches, et insère cinq lignes de démonstration.

CREATE TABLE formations (
  id SERIAL PRIMARY KEY,
  titre TEXT NOT NULL,
  duree_heures INT NOT NULL CHECK (duree_heures > 0),
  prix_eur NUMERIC(10,2) NOT NULL CHECK (prix_eur >= 0),
  niveau TEXT NOT NULL CHECK (niveau IN ('debutant','intermediaire','avance')),
  cree_le TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_formations_titre ON formations USING gin (to_tsvector('french', titre));

INSERT INTO formations (titre, duree_heures, prix_eur, niveau) VALUES
  ('Python pour debutants', 30, 99.00, 'debutant'),
  ('Linux administration', 50, 199.00, 'intermediaire'),
  ('Cybersecurite offensive', 80, 399.00, 'avance'),
  ('Docker et Kubernetes', 40, 249.00, 'intermediaire'),
  ('Python avance et async', 35, 179.00, 'avance');

Quittez psql avec \q. Vous avez maintenant une base prête à requêter. L’index GIN sur le vecteur de tsvector français permettra plus tard d’implémenter une recherche plein texte rapide ; pour l’instant, on s’en sert juste pour valider que tout est en place.

Étape 3 — Configurer la connexion

Le mot de passe Postgres ne doit jamais finir dans le code source. Créez à la racine du projet un fichier .env avec la connection string, puis ajoutez-le immédiatement à .gitignore.

echo 'DATABASE_URL=postgresql://postgres:motdepasse@localhost:5432/catalogue' > .env
echo '.env' >> .gitignore
echo '.venv/' >> .gitignore

L’ordre de ces commandes compte : créer le .env, puis l’exclure de git. Si vous inversez ou si vous oubliez le .gitignore, votre mot de passe finira dans l’historique git, et même un git rm ultérieur ne suffira pas à l’effacer (il faudra réécrire l’historique). En production, remplacez le fichier .env par les variables d’environnement de votre plateforme d’hébergement.

Étape 4 — Écrire le serveur MCP avec pool partagé

Le pool de connexions est l’élément structurant de tout serveur qui parle à une base. Sans pool, chaque appel d’outil ouvrirait une nouvelle connexion TCP+TLS+auth Postgres — l’overhead tue les performances dès la deuxième invocation parallèle. Avec pool, on maintient un petit nombre de connexions chaudes et chaque outil emprunte une connexion le temps de sa requête. Le SDK MCP de Python permet d’attacher un cycle de vie au serveur via le paramètre lifespan, exactement le bon endroit pour créer le pool au démarrage et le fermer proprement à l’arrêt.

Créez le fichier server.py avec ce contenu :

import os
from contextlib import asynccontextmanager
from dataclasses import dataclass

import asyncpg
from dotenv import load_dotenv
from mcp.server.fastmcp import Context, FastMCP

load_dotenv()
DATABASE_URL = os.environ["DATABASE_URL"]


@dataclass
class AppContext:
    pool: asyncpg.Pool


@asynccontextmanager
async def lifespan(_server: FastMCP):
    pool = await asyncpg.create_pool(
        DATABASE_URL, min_size=1, max_size=5, command_timeout=10,
    )
    try:
        yield AppContext(pool=pool)
    finally:
        await pool.close()


mcp = FastMCP("Catalogue formations", lifespan=lifespan)

Ce que ce code met en place : un pool de 1 à 5 connexions, un timeout par requête de 10 secondes (toute requête plus longue est annulée — protection contre les SELECT mal écrits qui partent en boucle), et l’injection du pool dans le contexte des outils via le pattern lifespan. Le décorateur asynccontextmanager garantit que le pool est fermé proprement même si le serveur se termine sur une exception.

Étape 5 — Ajouter l’outil de comptage

Premier outil, le plus simple : compter les formations par niveau. C’est une requête en lecture pure, sans paramètre arbitraire, idéale pour valider que le pool fonctionne.

@mcp.tool()
async def count_formations(ctx: Context) -> dict:
    """Compte les formations du catalogue, regroupées par niveau."""
    pool: asyncpg.Pool = ctx.request_context.lifespan_context.pool
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT niveau, COUNT(*) AS n FROM formations GROUP BY niveau ORDER BY niveau"
        )
    return {"par_niveau": {r["niveau"]: r["n"] for r in rows}}

Notez le pattern async with pool.acquire() as conn : il emprunte une connexion au pool et la rend automatiquement à la sortie du bloc, même en cas d’exception. C’est la formulation idiomatique d’asyncpg, et c’est elle qui empêche les fuites de connexions qui finissent par épuiser le pool en production. Quand vous lancez l’inspecteur et appelez l’outil, vous devriez voir {"par_niveau": {"avance": 2, "debutant": 1, "intermediaire": 2}}.

Étape 6 — Ajouter l’outil de recherche paramétrée

Deuxième outil : rechercher par mot-clé et filtrer par niveau. C’est ici que le risque d’injection SQL apparaît si on construit la requête à la main. La règle absolue avec asyncpg : on n’utilise jamais la concaténation de chaînes. Tous les paramètres passent par les placeholders $1, $2, etc., et asyncpg gère l’échappement et le typage.

from typing import Literal, Optional

@mcp.tool()
async def search_formations(
    ctx: Context,
    keyword: str,
    niveau: Optional[Literal["debutant", "intermediaire", "avance"]] = None,
    limit: int = 10,
) -> list[dict]:
    """Recherche dans le catalogue par mot-cle, filtre optionnel par niveau."""
    pool: asyncpg.Pool = ctx.request_context.lifespan_context.pool
    limit = max(1, min(limit, 50))  # garde-fou : 1 a 50 max

    sql = """
        SELECT id, titre, duree_heures, prix_eur, niveau
        FROM formations
        WHERE titre ILIKE '%' || $1 || '%'
          AND ($2::text IS NULL OR niveau = $2)
        ORDER BY id
        LIMIT $3
    """
    async with pool.acquire() as conn:
        rows = await conn.fetch(sql, keyword, niveau, limit)
    return [dict(r) for r in rows]

Trois sécurités importantes ici. La limit est bornée côté code (1 à 50) pour empêcher un agent un peu trop enthousiaste de demander 100000 lignes d’un coup. Le filtre niveau utilise une clause $2::text IS NULL OR ... qui rend le paramètre véritablement optionnel sans devoir construire dynamiquement la requête. Et le Literal du typage Python se traduit côté JSON Schema en enum — le client refusera tout niveau hors de la liste autorisée avant même que la requête atteigne Postgres.

Étape 7 — Ajouter l’outil d’écriture contrôlé

L’écriture mérite une vigilance redoublée. Un outil MCP qui peut INSERT, UPDATE ou DELETE est une porte d’entrée vers la corruption de données — surtout quand l’invocation est décidée par un LLM. La bonne pratique consiste à exposer des outils d’écriture spécifiques (par exemple « ajouter une formation »), plutôt qu’un générique execute_sql, et à les soumettre à des validations métier strictes côté serveur.

@mcp.tool()
async def add_formation(
    ctx: Context,
    titre: str,
    duree_heures: int,
    prix_eur: float,
    niveau: Literal["debutant", "intermediaire", "avance"],
) -> dict:
    """Ajoute une nouvelle formation au catalogue."""
    if not titre.strip() or len(titre) > 200:
        return {"error": "Titre invalide (1 a 200 caracteres)"}
    if duree_heures <= 0 or duree_heures > 1000:
        return {"error": "Duree invalide (1 a 1000 heures)"}
    if prix_eur < 0 or prix_eur > 10000:
        return {"error": "Prix invalide (0 a 10000 EUR)"}

    pool: asyncpg.Pool = ctx.request_context.lifespan_context.pool
    async with pool.acquire() as conn:
        new_id = await conn.fetchval(
            """INSERT INTO formations (titre, duree_heures, prix_eur, niveau)
               VALUES ($1, $2, $3, $4) RETURNING id""",
            titre.strip(), duree_heures, prix_eur, niveau,
        )
    return {"id": new_id, "message": f"Formation '{titre.strip()}' ajoutee."}

Observez la triple protection : validations Python en amont (longueur, bornes), contraintes SQL en aval (les CHECK qu’on a posés à l’étape 2), et utilisation des placeholders. Si une seule de ces couches saute, les deux autres tiennent encore. C’est la discipline du « defense in depth » appliquée aux outils MCP — le LLM peut être convaincu d’envoyer n’importe quoi, mais la base reste cohérente.

Étape 8 — Lancer le serveur

Ajoutez la ligne d’exécution à la fin de server.py et lancez le serveur dans l’inspecteur pour valider l’ensemble du chaînage.

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

mcp dev server.py

L’inspecteur web s’ouvre, vous voyez les trois outils listés. Testez chacun : count_formations sans paramètre doit retourner les comptes par niveau, search_formations avec keyword= »Python » doit ramener deux lignes, et add_formation avec un titre valide doit retourner un ID puis vous pouvez vérifier dans psql que la ligne est bien présente. Si une erreur apparaît, lisez attentivement le terminal où tourne mcp dev — la stack trace y est complète.

Étape 9 — Brancher à Claude Desktop

Dernière étape : enregistrer le serveur dans claude_desktop_config.json. Comme on est en stdio, la config est directe — pas besoin de proxy mcp-remote comme dans le tutoriel Next.js.

{
  "mcpServers": {
    "catalogue-formations": {
      "command": "/chemin/absolu/vers/.venv/bin/python",
      "args": ["/chemin/absolu/vers/server.py"],
      "env": {
        "DATABASE_URL": "postgresql://postgres:motdepasse@localhost:5432/catalogue"
      }
    }
  }
}

Le bloc env est essentiel : Claude Desktop ne lit pas votre .env automatiquement, il faut passer la connection string explicitement. Et le chemin Python doit pointer sur le binaire de l’environnement virtuel, pas sur le Python global, sinon les imports d’asyncpg échoueront. Redémarrez Claude Desktop, ouvrez un chat et demandez « Combien de formations avancées avons-nous au catalogue ? » — Claude doit appeler count_formations et vous répondre 3 (les deux d’origine plus celle ajoutée pendant les tests).

Erreurs fréquentes

Erreur	Cause	Solution
connection refused	Postgres non démarré ou port différent	Vérifier docker ps et la connection string
password authentication failed	Mot de passe incorrect dans .env	Rectifier DATABASE_URL et redémarrer le serveur
Pool épuisé après quelques appels	Connexions non rendues (oubli du async with)	Toujours utiliser le context manager async with pool.acquire()
module asyncpg not found	Mauvais Python dans la config Claude	Pointer sur le binaire du venv, pas le Python système
Outil d’écriture refusé silencieusement	Validation Python qui retourne un dict error	Logger côté serveur pour voir quelle validation échoue
Timeout sur requête longue	command_timeout=10 trop court	Optimiser la requête ou augmenter le timeout pour les outils analytiques

Pour aller plus loin

• 🔝 Retour au guide principal : Créer un serveur MCP : architecture, primitives, premier déploiement
• 📚 Tutoriel précédent — Déployer un serveur MCP sur Next.js (Vercel) : tutoriel pas-à-pas
• 🔐 Tutoriel suivant — Sécuriser un serveur MCP avec OAuth 2.1 : tutoriel pas-à-pas
• Documentation asyncpg — usage et pools
• PostgreSQL — CREATE TABLE et contraintes CHECK
• postgres-mcp (Crystal DBA) — serveur Postgres MCP de référence

FAQ

Pourquoi pas un outil execute_sql générique ?

Parce qu’il revient à donner un accès postgres root à un LLM. Quelqu’un finira par demander à l’agent de « nettoyer la base » et l’agent exécutera DELETE FROM formations. Toujours préférer des outils spécifiques avec validations métier.

Quel rôle Postgres utiliser pour MCP ?

Jamais le superuser. Créez un rôle dédié avec uniquement les privilèges nécessaires : SELECT sur les tables à exposer, plus INSERT/UPDATE sur les tables où l’agent doit pouvoir écrire. Si l’agent ne doit jamais supprimer, ne lui accordez pas DELETE — la défense la plus simple.

Comment scaler le pool en production ?

La taille max du pool doit rester bien inférieure à max_connections de Postgres. Sur une base partagée, max_size=5 à max_size=20 est raisonnable pour un serveur MCP unique. Si vous lancez plusieurs instances du serveur, multipliez par le nombre d’instances et vérifiez que la somme reste sous le quota.

asyncpg supporte-t-il les transactions ?

Oui, via async with conn.transaction():. Pour les outils MCP qui modifient plusieurs lignes corrélées, enveloppez tout dans une transaction — soit tout passe, soit rien. C’est le seul moyen d’éviter les états incohérents en cas d’erreur en cours d’écriture.

Et pour Supabase, Neon ou un Postgres managé ?

La connection string change (avec sslmode=require), tout le reste est identique. Pensez juste à exposer la chaîne complète dans DATABASE_URL, en incluant ?sslmode=require à la fin pour les hébergeurs qui imposent TLS.

Le LLM peut-il voir le mot de passe Postgres ?

Non, jamais. Le mot de passe ne vit que dans la variable d’environnement du processus serveur. Le LLM n’a accès qu’aux résultats des outils — il ne voit ni le code, ni les variables d’environnement, ni les requêtes SQL générées. C’est l’isolation host/server qui le garantit.