Un serveur MCP sans configuration pour la recherche vectorielle filtrée dans Neo4j

Imaginez un graphe de connaissances rempli d’articles de presse et un agent confronté à cette demande : afficher les reportages positifs sur l’énergie renouvelable publiés par Reuters la semaine dernière. Cette phrase contient trois éléments distincts : une recherche sémantique sur « énergie renouvelable », ainsi que des filtres de métadonnées concernant la source (Reuters), l’opinion exprimée (positive) et la date (la semaine dernière).

La recherche vectorielle classique gère la partie sémantique. Cependant, sans filtres, elle renverra volontiers des articles négatifs de 2019 publiés par des blogs aléatoires, et le LLM n’aura aucun moyen de savoir qu’ils sont irrelevants avant d’en avoir lu chaque un. C’est le problème de la recherche vectorielle avec filtres de métadonnées, et Neo4j propose depuis longtemps deux solutions pour y remédier.

1. Filtrer au préalable à l’aide de Cypher pour réduire l’ensemble des candidats, puis effectuer la recherche vectorielle sur ce qui reste. Les résultats sont précis, mais cela peut devenir lent lorsque l’ensemble des candidats reste important.

Exécutez d’abord la recherche vectorielle sur l’index complet, puis appliquez des filtres ultérieurs. Cela est rapide, mais le taux de rappel peut diminuer à mesure que les filtres se restreignent, car des éléments pertinents pourraient ne jamais figurer dans les premiers k résultats.

Chacune de ces approches présentait des avantages réels, tout en laissant place à des améliorations.

Quelles sont les modifications apportées par Neo4j 2026.02

Neo4j 2026.02 (voir les notes de version) a rendu la recherche vectorielle avec filtres généralement disponible dans le cadre de Cypher 25. Les prédicats peuvent désormais être appliqués à l’intérieur de l’index vectoriel au moment de la requête, permettant ainsi à l’algorithme HNSW d’éliminer les vecteurs non correspondants avant de classer les résultats. La latence reste faible, le taux de rappel élevé, et il n’est pas nécessaire de récupérer plus d’informations que nécessaire. Une nouvelle clause native SEARCH est également introduite, remplaçant l’ancienne procédure db.index.vector.queryNodes() et faisant de la recherche vectorielle une fonctionnalité à part entière du langage de requête.

Cela soulève une question : un serveur MCP pourrait-il découvrir des index vectoriels de lui-même, examiner leurs propriétés de métadonnées et générer des outils de recherche filtrée sans aucune configuration manuelle ?

Un serveur de recherche vectorielle générique avec zéro configuration

Tout ceux qui ont développé des outils MCP pour la recherche vectorielle connaissent la procédure : écrire du Cypher, définir les paramètres, configurer le modèle d’incorporation, documenter les filtres. Cela fonctionne bien pour un index bien connu. Mais un serveur prêt à l’emploi qui fonctionne avec n’importe quelle base de données Neo4j, auquel il suffit de pointer une instance pour qu’il s’occupe du reste, est bien plus attrayant.

C’est exactement ce que fait ce serveur. Lors de son démarrage, il se connecte à Neo4j, trouve tous les index vectoriels, extrait un nœud de chacun pour détecter les types de propriétés, et crée un outil MCP entièrement typé pour chaque index, avec des paramètres de filtre et une documentation que le LLM peut lire.

Une chose ne peut pas être découverte automatiquement : le modèle d’incorporation. Neo4j ne le stocke pas encore dans les métadonnées de l’index, et une clé API serait nécessaire pour appeler ce modèle de toute façon. Par conséquent, le fournisseur d’incorporation et ses identifiants sont la seule configuration à fournir.

Étape 1 : trouver les index vectoriels

Tout d’abord, le serveur demande à Neo4j tous les index vectoriels :

records, _, _ = driver.execute_query(
    """
    SHOW VECTOR INDEXES
    """,
    database_=NEO4J_DATABASE,
)

Le résultat indique le nom de chaque index, l’étiquette du nœud qu’il couvre ainsi que les propriétés indexées. S’il n’y a pas d’index vectoriels, le serveur s’arrête avec une erreur, car il n’y a rien à fournir.

Étape 2 : types d’échantillons et valeurs d’exemple

Pour chaque index, le serveur doit savoir quelle propriété contient l’embedding et quel est le type de chacune des autres propriétés. published_date contient-il une simple date ou un timestamp complet ? source est-il du texte ? Le score d’opinion est-il un nombre décimal ? Un outil dédié _sample_property répond en examinant un seul nœud par propriété :

def _sample_property(driver, label: str, prop: str) -> tuple[str, Any]:
    """Return (type, sample_value) for a single property."""
    cypher = (
        "MATCH (n:$($label)) WHERE n[$prop] IS NOT NULL "
        "WITH n LIMIT 1 RETURN n[$prop] AS val"
    )
    result, _, _ = driver.execute_query(
        cypher, label=label, prop=prop, database_=NEO4J_DATABASE
    )
    if not result:
        return "string", None
    val = result[0]["val"]
    return _detect_type(val), val

Il renvoie le type détecté ainsi que la valeur réelle de l’échantillon. La fonction principale rassemble ces informations en deux dictionnaires, un pour les types et un pour les exemples :

def _sample_property_types(
    driver, label: str, properties: list[str]
) -> tuple[str | None, dict[str, str], dict[str, Any]]:
    embedding_prop: str | None = None
    meta_types: dict[str, str] = {}
    meta_examples: dict[str, Any] = {}

    for p in properties:
        t, sample = _sample_property(driver, label, p)
        if t == "vector":
            embedding_prop = p
        else:
            meta_types[p] = t
            meta_examples[p] = sample

    return embedding_prop, meta_types, meta_examples

La fonction _detect_type vérifie, dans l’ordre, s’il s’agit d’une liste (un vecteur), d’un booléen, d’un entier, d’un flottant, d’une valeur datetime ou d’une date, et recourt sinon à une chaîne de caractères. Les valeurs datetime et date sont considérées comme des types distincts. Une propriété contenant des composantes temporelles est classée comme datetime et fait l’objet du cast datetime() de Neo4j, et non de date(). La liste de flottants représente l’embedding, tandis que toutes les autres propriétés deviennent des champs de métadonnées filtrables. Pour l’exemple relatif aux nouvelles, le résultat est le suivant :

embedding -> vector
published_date -> date (or datetime, if timestamps are present)
source -> string
sentiment -> float

Le dictionnaire meta_examples conserve les valeurs d’échantillon réelles, telles que « Reuters » pour la source ou 0,72 pour l’avis, et elles enrichissent les docstrings générés dans l’étape suivante.

Étape 3 : générer les outils de recherche

Désormais, tout est mis en place. Pour chaque index, une fonction closure capture les métadonnées de cet index et génère une fonction de recherche :

def _make_search_fn(idx: dict[str, Any]):
    """Factory that closes over a single index's metadata."""

    index_name = idx["name"]
    label = idx["label"]
    embed_prop = idx["embed_prop"]
    prop_types = idx["prop_types"]
    prop_examples = idx["prop_examples"]

    # Build the filter-parameter docstring dynamically
    filter_lines = []
    for prop, t in prop_types.items():
        safe = prop.replace(" ", "_").replace("-", "_")
        ex = _format_example(prop_examples.get(prop), t)
        if t in ("float", "int"):
            filter_lines.append(
                f"  {safe}: Optional dict with 'min' and/or 'max' (numeric range). Example value: {ex}"
            )
        elif t == "datetime":
            filter_lines.append(
                f"  {safe}: Optional dict with 'min' and/or 'max' (ISO datetime strings, uses datetime()). Example value: {ex}"
            )
        elif t == "date":
            filter_lines.append(
                f"  {safe}: Optional dict with 'min' and/or 'max' (ISO date strings, uses date()). Example value: {ex}"
            )
        elif t == "bool":
            filter_lines.append(
                f"  {safe}: Optional bool. Example value: {ex}"
            )
        else:
            filter_lines.append(
                f"  {safe}: Optional str (exact match). Example value: {ex}"
            )

    filters_doc = "\n".join(filter_lines) if filter_lines else "  (none)"

    async def search(
        query: str,
        top_k: int = 10,
        filters: Optional[dict[str, Any]] = None,
    ) -> str:
        query_vec = embedder.embed_query(query)
        where_clause, cypher_params = _build_where(prop_types, filters or {})
        where_part = f"\n    WHERE {where_clause}" if where_clause else ""

        cypher = (
            f"CYPHER 25\n"
            f"MATCH (n:$($label))\n"
            f"  SEARCH n IN (\n"
            f"    VECTOR INDEX {index_name}\n"
            f"    FOR $query_vec{where_part}\n"
            f"    LIMIT $top_k\n"
            f"  ) SCORE AS score\n"
            f"RETURN n {{ .*, `{embed_prop}`: null }} AS doc, score\n"
            f"ORDER BY score DESC"
        )

        cypher_params["query_vec"] = query_vec
        cypher_params["top_k"] = top_k

        rows, _, _ = driver.execute_query(
            cypher, cypher_params, database_=NEO4J_DATABASE
        )
        results = [{"doc": dict(r["doc"]), "score": r["score"]} for r in rows]
        return json.dumps(results, indent=2, default=str)

    search.__name__ = f"search_{index_name}"
    search.__qualname__ = f"search_{index_name}"
    search.__doc__ = (
        f"Semantic vector search over :{label} nodes using the "
        f"'{index_name}' index.\n\n"
        f"Args:\n"
        f"  query: Natural-language search text (will be embedded).\n"
        f"  top_k: Number of results to return (default 10).\n"
        f"  filters: Optional dict of metadata filters:\n"
        f"{filters_doc}\n"
    )
    return search

Quelques détails méritent une attention particulière :

• Le préfixe CYPHER 25 active la syntaxe plus récente nécessaire pour la clause SEARCH et le filtrage en temps réel.
• La clause WHERE se trouve à l’intérieur du bloc SEARCH. C’est ainsi que fonctionne le filtrage en temps réel 2026.02. HNSW respecte lui-même les prédicats, de sorte que les résultats d’approximation du voisin le plus proche sont filtrés sans que la capacité de rappel ne diminue après le filtrage.
• La propriété d’incorporation est mise à zéro dans les lignes retournées ({embed_prop}: null), de manière à ce que les grands vecteurs ne soient pas renvoyés au LLM, où ils ne feraient que consommer des tokens inutilement.
• La documentation est générée dynamiquement à partir des types échantillonnés et inclut une valeur d’exemple concrète pour chaque filtre. L’agent peut ainsi savoir quels filtres existent, quels types ils attendent et à quoi ressemblent des valeurs réalistes.

Construction de la clause WHERE

Une fonction _build_where transforme un dictionnaire de filtres Python en prédicat Cypher, en choisissant la forme adaptée au type déclaré de chaque propriété :

• Les nombres flottants et entiers utilisent des filtres de plage via {"min": ..., "max": ...}, ce qui génère des comparaisons >= et <=.
• Les dates et heures emploient la même logique de plage, mais sont encadrées par la fonction datetime().
• Les dates utilisent également cette logique, mais sont encadrées par date().
• Les valeurs booléennes font appel à une égalité stricte.
• Les chaînes de caractères, qui constituent le cas par défaut, utilisent aussi une égalité stricte.

Tous les prédicats sont reliés par AND, ce qui correspond aux simples comparaisons de propriétés prises en charge par Neo4j dans le filtrage par index vectoriel.

def _build_where(
    prop_types: dict[str, str], filters: dict[str, Any]
) -> tuple[str, dict[str, Any]]:
    clauses: list[str] = []
    params: dict[str, Any] = {}

    for prop, t in prop_types.items():
        safe = prop.replace(" ", "_").replace("-", "_")
        val = filters.get(safe)
        if val is None:
            continue

        if t in ("float", "int"):
            if isinstance(val, dict):
                if val.get("min") is not None:
                    k = f"filt_{safe}_min"
                    clauses.append(f"n.`{prop}` >= ${k}")
                    params[k] = val["min"]
                if val.get("max") is not None:
                    k = f"filt_{safe}_max"
                    clauses.append(f"n.`{prop}` <= ${k}")
                    params[k] = val["max"]
        elif t == "datetime":
            if isinstance(val, dict):
                if val.get("min") is not None:
                    k = f"filt_{safe}_min"
                    clauses.append(f"n.`{prop}` >= datetime(${k})")
                    params[k] = val["min"]
                if val.get("max") is not None:
                    k = f"filt_{safe}_max"
                    clauses.append(f"n.`{prop}` <= datetime(${k})")
                    params[k] = val["max"]
        elif t == "date":
            if isinstance(val, dict):
                if val.get("min") is not None:
                    k = f"filt_{safe}_min"
                    clauses.append(f"n.`{prop}` >= date(${k})")
                    params[k] = val["min"]
                if val.get("max") is not None:
                    k = f"filt_{safe}_max"
                    clauses.append(f"n.`{prop}` <= date(${k})")
                    params[k] = val["max"]
        elif t == "bool":
            k = f"filt_{safe}"
            clauses.append(f"n.`{prop}` = ${k}")
            params[k] = val
        else:  # string exact match
            k = f"filt_{safe}"
            clauses.append(f"n.`{prop}` = ${k}")
            params[k] = val

    where = " AND ".join(clauses)
    return where, params

Pourquoi ne pas laisser l’LLM écrire le Cypher ?

Le serveur Cypher MCP de Neo4j pourrait générer des requêtes via la conversion texte en Cypher, mais la recherche vectorielle nécessite des embeddings, et actuellement le seul moyen de transférer un embedding entre outils est via le contexte de l’LLM. Envoyer des vecteurs de 1536 dimensions à travers le modèle gaspille des tokens et du temps.

En regroupant tout le flux de travail — génération des embeddings, construction et exécution du Cypher — dans un seul outil, on évite que les embeddings n’intègrent le contexte du modèle. L’LLM fournit une requête en langage naturel ainsi que des filtres optionnels, et reçoit les résultats classés.

Lancer l’outil

Le serveur peut être exécuté directement depuis GitHub à l’aide de uvx. Ajoutez une entrée similaire à la configuration du client MCP pour Claude Desktop, VS Code, Cursor ou des outils similaires :

{
  "mcpServers": {
    "neo4j-vector": {
      "command": "uvx",
      "args": [
        "--from",
        "git+https://github.com/tomasonjo/neo4j-prefiltering-mcp",
        "--with",
        "langchain-openai",
        "neo4j-prefiltering-mcp"
      ],
      "env": {
        "NEO4J_URI": "neo4j+s://demo.neo4jlabs.com:7687",
        "NEO4J_USER": "companies2",
        "NEO4J_PASSWORD": "companies2",
        "NEO4J_DATABASE": "companies2",
        "EMBEDDING_MODEL": "openai:text-embedding-3-small",
        "OPENAI_API_KEY": "<your-key>"
      }
    }
  }
}

Le flag --with langchain-openai installe le fournisseur d’incrustations OpenAI. Pour utiliser un autre fournisseur, comme Cohere ou un modèle local, remplacez cette dépendance et mettez à jour EMBEDDING_MODEL. La recherche d’incrustations s’effectue via init_embeddings() de LangChain, donc tout fournisseur pris en charge par LangChain fonctionnera.

Lors des tests dans Claude Desktop avec une question d’exemple et l’utilisation de l’extension tool call, on observe des paramètres tels que ceux-ci :

{
  "query": "positive developments in artificial intelligence AI breakthroughs progress",
  "top_k": 10,
  "filters": {
    "language": "en",
    "date": {
      "min": "2026-01-01",
      "max": "2026-12-31"
    },
    "sentiment": {
      "min": 0.5
    }
  }
}

Rien n’a besoin d’être écrit à la main : pas de définitions d’outils, pas de requêtes, pas de schémas de filtre. Créez un autre index vectoriel, redémarrez le serveur et un nouvel outil apparaît.

En résumé

La recherche vectorielle avec filtrage dans Neo4j 2026.02, associée à la syntaxe SEARCH de Cypher 25, offre une solution simple pour des outils de recherche vectorielle filtrée. En combinant cela avec MCP et une légère introspection au démarrage, on obtient un serveur capable de trouver des index vectoriels, de détecter les propriétés pouvant être filtrées et de générer des outils de recherche typés sans écrire une seule ligne de code spécifique à l’outil. En le dirigeant vers une base de données Neo4j contenant des index vectoriels, il assemble son propre ensemble d’outils. La documentation sur le filtrage présente la syntaxe complète.