Schéma de serveur MCP pour catalogue de données

Un serveur MCP de catalogue de données expose les données d’un catalogue interne comme outils accessibles par l’IA, permettant à un assistant de répondre à des questions telles que « quelles tables avons-nous sur les clients ? » ou « qu’est-ce qui alimente le tableau de bord des revenus ? » sans ouvrir un navigateur ou naviguer dans une interface de catalogue. Si le catalogue est développé en interne ou tourne sur une plateforme sans serveur MCP officiel, ce schéma fournit un point de départ.

Le schéma ci-dessous utilise un catalogue simulé en mémoire. En production, remplacez les accès au dictionnaire par des appels à l’API de votre catalogue réel — DataHub, OpenMetadata, Atlan, ou ce que vous utilisez.

Le serveur

from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("DataCatalogMCP")

# Données de catalogue simulées - remplacez par votre API de catalogue réelle
CATALOG = {
    "sales.orders": {
        "description": "Order transactions from all channels",
        "columns": ["order_id", "customer_id", "total_amount", "created_at"],
        "owner": "sales-team",
        "tags": ["pii", "financial"],
        "upstream": ["raw.shopify_orders", "raw.pos_transactions"],
        "downstream": ["analytics.revenue_daily", "ml.churn_features"]
    },
    "sales.customers": {
        "description": "Customer master data with demographics",
        "columns": ["customer_id", "email", "segment", "lifetime_value"],
        "owner": "marketing-team",
        "tags": ["pii"],
        "upstream": ["raw.crm_contacts"],
        "downstream": ["sales.orders", "analytics.cohorts"]
    }
}

Trois outils essentiels

Recherche de tables

L’outil le plus utilisé. Les utilisateurs posent des questions vagues — « quelque chose sur les commandes ? » — et l’IA doit trouver les tables pertinentes :

@mcp.tool()
def search_tables(
    query: str,
    tags: list[str] | None = None,
    limit: int = 10
) -> str:
    """Search for tables in the data catalog.

    Args:
        query: Search term to match against table names and descriptions
        tags: Optional list of tags to filter by (e.g., ['pii', 'financial'])
        limit: Maximum number of results to return

    Returns:
        Matching tables with descriptions
    """
    results = []
    for table_name, metadata in CATALOG.items():
        if query.lower() in table_name.lower() or query.lower() in metadata["description"].lower():
            if tags is None or any(t in metadata["tags"] for t in tags):
                results.append({
                    "name": table_name,
                    "description": metadata["description"],
                    "owner": metadata["owner"],
                    "tags": metadata["tags"]
                })
    return json.dumps(results[:limit], indent=2)

En production, cette recherche interrogerait l’API de recherche de votre catalogue — la plupart des catalogues proposent une recherche en texte libre, un filtrage par tags et un classement par pertinence nativement. L’outil MCP est un wrapper fin qui traduit les paramètres de l’IA dans le format de requête de votre catalogue.

Détails d’une table

Une fois que l’IA a trouvé une table, elle a besoin d’une vue complète :

@mcp.tool()
def get_table_details(table_name: str) -> str:
    """Get detailed metadata for a specific table.

    Args:
        table_name: Fully qualified table name (e.g., 'sales.orders')

    Returns:
        Complete table metadata including columns, owner, and tags
    """
    if table_name not in CATALOG:
        return json.dumps({"error": f"Table '{table_name}' not found in catalog"})

    metadata = CATALOG[table_name]
    return json.dumps({
        "table": table_name,
        **metadata
    }, indent=2)

Cet outil retourne tout : colonnes, propriétaire, tags, dépendances amont et aval. L’IA l’utilise pour comprendre la structure d’une table avant d’écrire des requêtes ou de formuler des recommandations.

Lignage des données

Le traçage du lignage est une capacité clé du serveur de catalogue. « Qu’est-ce qui alimente le tableau de bord des revenus ? » est une question que la plupart des interfaces de catalogue nécessitent plusieurs étapes de navigation pour répondre ; l’outil MCP la résout en un seul appel :

@mcp.tool()
def get_data_lineage(
    table_name: str,
    direction: str = "both",
    depth: int = 2
) -> str:
    """Trace data lineage for a table.

    Args:
        table_name: Table to trace lineage for
        direction: 'upstream', 'downstream', or 'both'
        depth: How many levels to traverse (1-5)

    Returns:
        Lineage graph showing data flow
    """
    if table_name not in CATALOG:
        return json.dumps({"error": f"Table '{table_name}' not found"})

    metadata = CATALOG[table_name]
    lineage = {"table": table_name}

    if direction in ("upstream", "both"):
        lineage["upstream"] = metadata.get("upstream", [])
    if direction in ("downstream", "both"):
        lineage["downstream"] = metadata.get("downstream", [])

    return json.dumps(lineage, indent=2)

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

Le paramètre depth importe en production. Une requête de lignage à profondeur 1 est rapide. Une requête à profondeur 5 peut parcourir des centaines de nœuds dans un entrepôt de données complexe. Définissez des valeurs par défaut et des limites raisonnables pour éviter que l’IA ne demande des traversées de graphe coûteuses.

Ce que vous pouvez demander

Avec ce serveur connecté, les conversations deviennent réellement utiles pour la découverte de données :

« Trouvez toutes les tables liées aux clients »
« Quelles tables contiennent des données PII ? »
« Montrez-moi le lignage de la table orders »
« Qui possède la table revenue_daily ? »
« Quelles sources brutes alimentent le modèle de churn ? »

Considérations pour la production

Mise en cache. Les métadonnées d’un catalogue évoluent lentement. Mettez en cache les résultats de recherche et les détails de table pendant 5 à 15 minutes pour éviter de surcharger votre API de catalogue à chaque question.

Contrôle d’accès. Votre catalogue dispose probablement de son propre contrôle d’accès. Le serveur MCP doit le respecter — si un utilisateur ne peut pas voir une table dans l’interface du catalogue, il ne devrait pas la voir via MCP non plus. Transmettez l’identité de l’utilisateur à vos appels d’API de catalogue.

Qualité de la recherche. La recherche simple par sous-chaîne de l’exemple est un placeholder. Les catalogues réels proposent des correspondances floues, une recherche sémantique et un filtrage par tags. Utilisez la recherche native de votre catalogue — elle a été optimisée exactement pour ce cas d’usage.

Implémentations de référence. Avant de construire de zéro, étudiez le serveur MCP DataHub et l’intégration MCP OpenMetadata. Les deux sont open source et démontrent des patterns de production pour l’intégration de catalogues, l’implémentation de la recherche et les API de lignage.