Ein Zero-Config MCP-Server für gefilterte Vektorabfragen in Neo4j

Stellen Sie sich ein Wissensgraph mit vielen Nachrichtenartikeln vor sowie einen Agenten, der folgende Anfrage erhält: Zeigen Sie die positiven Berichte über erneuerbare Energien von Reuters aus der letzten Woche. In diesem Satz sind drei verschiedene Aspekte enthalten. Einer ist eine semantische Suche nach „erneuerbare Energien“. Die anderen sind Metadatenfilter für die Quelle (Reuters), die Stimmung (positiv) und das Datum (letzte Woche).

Eine einfache Vektorsuche kümmert sich um den semantischen Teil. Ohne Filter liefert sie jedoch gerne negative Artikel aus dem Jahr 2019 von zufälligen Blogs – und das große Sprachmodell hat keine Möglichkeit, zu erkennen, dass diese irrelevant sind, bevor es sie alle gelesen hat. Das ist das Problem der Vektorsuche mit Metadatenfiltern, und Neo4j bietet seit Langem zwei Ansätze, damit umzugehen.

1. Filtern Sie zunächst mit Cypher, um die Anzahl der möglichen Ergebnisse zu verringern, und führen Sie anschließend die Vektorsuche auf dem verbleibenden Datensatz durch. Die Ergebnisse sind genau, können aber langsam werden, wenn die Anzahl der möglichen Ergebnisse weiterhin groß ist.

Führen Sie zunächst die Vektorabfrage über den gesamten Index durch und filtern Sie anschließend. Das ist schnell, doch die Trefferquote kann sinken, je enger die Filter werden, da relevante Elemente möglicherweise nie zu den ersten k Ergebnissen gehören.

Sowohl Ansätze hatten echte Stärken, und bei beiden gab es Raum für Verbesserungen.

Was hat Neo4j 2026.02 geändert

Neo4j 2026.02 (siehe die Veröffentlichungshinweise) machte die Vektorabfrage mit Filtern allgemein verfügbar als Teil von Cypher 25. Prädikate können nun zur Abfragezeit innerhalb des Vektorindex angewendet werden, sodass das HNSW-Algorithmus nicht passende Vektoren vor der Rangliste überspringt. Die Latenz bleibt niedrig, die Trefferquote hoch und es muss nichts übermäßig abgerufen werden. Zudem gibt es eine neue native SEARCH-Klausel, die den älteren Prozess db.index.vector.queryNodes() ersetzt und die Vektorabfrage zu einem vollwertigen Bestandteil der Abfragesprache macht.

Das wirft eine Frage auf: Kann ein MCP-Server von selbst Vektorindizes entdecken, ihre Metadaten-Eigenschaften überprüfen und ohne manuelle Konfiguration gefilterte Suchwerkzeuge erstellen?

Ein generischer Vektor-Suchserver ohne Konfiguration

Jeder, der bereits MCP-Tools für die Vektorsuche entwickelt hat, kennt den Ablauf: Man schreibt den Cypher-Code, definiert die Parameter, stellt das Embedding-Modell ein und dokumentiert die Filter. Für einen bekannten Index ist das in Ordnung. Doch ein Server, der mit jeder Neo4j-Datenbank funktioniert, bei dem einfach auf eine Instanz gezeigt wird und der den Rest selbst erledigt, ist attraktiver.

Das ist genau das, was dieser Server tut. Beim Start verbindet er sich mit Neo4j, findet alle Vektorindizes, entnimmt jedem Index einen Knoten, um die Eigenschaftstypen zu erkennen, und erstellt für jeden Index ein vollständig typisiertes MCP-Tool mit Filterparametern sowie einer Dokumentation, die das LLM lesen kann.

Eines kann nicht automatisch entdeckt werden: das Embedding-Modell. Neo4j speichert dieses noch nicht in den Index-Metadaten, und es wäre ohnehin eine API-Schlüssel erforderlich, um das Modell aufrufen zu können. Daher sind der Embedding-Anbieter sowie seine Anmeldeinformationen die einzige Konfiguration, die bereitgestellt werden muss.

Schritt 1: Die Vektor-Indizes finden

Zuerst bittet der Server Neo4j um alle Vektor-Indizes:

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

Das Ergebnis zeigt den Namen jedes Index, die von ihm abgedeckte Knotenbezeichnung sowie die indizierten Eigenschaften an. Wenn es keine Vektorindizes gibt, beendet der Server die Ausführung mit einem Fehler, da es nichts zu liefern gibt.

Schritt 2: Beispieltypen und Beispielwerte

Für jeden Index muss der Server wissen, welche Eigenschaft die Embedding-Daten enthält und welchen Typ alle anderen Eigenschaften haben. Enthält published_date ein einfaches Datum oder einen vollständigen Zeitstempel? Ist source Text? Ist die Sentiment-Wertung eine Dezimalzahl? Ein spezieller _sample_property-Hilfsfunktion beantwortet diese Fragen, indem er für jede Eigenschaft einen einzelnen Knoten überprüft:

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

Es gibt den erkannten Typ sowie den tatsächlichen Wert des Beispiels zurück. Die Hauptfunktion sammelt diese in zwei Wörterbücher zusammen, eines für die Typen und eines für die Beispiele:

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

Funktion _detect_type prüft nacheinander auf eine Liste (einen Vektor), bool, int, float, datetime und date und greift bei Bedarf auf string zurück. Datetime und date gelten als unterschiedliche Typen. Eine Eigenschaft mit Zeitkomponenten wird als datetime klassifiziert und erhält die Neo4j-Funktion datetime() zur Umwandlung, nicht date(). Die Liste der Float-Werte stellt die Embedding-Daten dar, während alle anderen Eigenschaften zu filterbaren Metadatenfeldern werden. Im Beispiel mit Nachrichten sieht das Ergebnis wie folgt aus:

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

Das Dictionary meta_examples enthält die tatsächlichen Beispielwerte, wie zum Beispiel „Reuters“ für die Quelle oder 0,72 für die Sentiment-Einstufung, und diese bereichern die in der nächsten Schritt erzeugten Docstrings.

Schritt 3: Erstellung der Suchwerkzeuge

Nun kommt alles zusammen. Für jeden Index erfasst eine Closure die Metadaten dieses Index und erstellt eine Suchfunktion:

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

Einige Details verdienen Aufmerksamkeit:

• Das Präfix CYPHER 25 aktiviert die neuere Syntax, die für die SEARCH-Klausel sowie das In-Index-Filtern erforderlich ist.
• Die WHERE-Klausel befindet sich innerhalb des SEARCH-Blocks. Das ist das In-Index-Filtern von 2026.02 im Einsatz. HNSW berücksichtigt selbst die Prädikate, sodass annähernde Ergebnisse des nächsten Nachbarn gefiltert zurückgegeben werden, ohne dass die Trefferquote durch nachträgliches Filtern beeinträchtigt wird.
• Die Embedding-Eigenschaft wird in den zurückgegebenen Zeilen auf null gesetzt ({embed_prop}: null), damit große Vektoren nicht an das LLM zurückgesendet werden, wo sie nur Token verbrauchen würden.
• Die Dokumentation wird dynamisch anhand der ausgewählten Typen erstellt und enthält einen konkreten Beispielwert für jedes Filter. Dadurch kann der Agent erkennen, welche Filter vorhanden sind, welche Type erwartet werden und wie realistische Werte aussehen.

Aufbau der WHERE-Klausel

Die Funktion _build_where wandelt ein Python-Filter-Dictionary in einen Cypher-Predikat um und wählt dabei die passende Form je nach vom jeweiligen Attribut deklariertem Typ aus:

• Fließkommazahlen und Ganzzahlen erhalten Bereichsfilter über {"min": ..., "max": ...}, wodurch Vergleiche mit >= und <= entstehen.
• Datumszeiten nutzen dieselbe Bereichslogik, allerdings eingebettet in den datetime()-Umwandlungsfunktion.
• Daten verwenden ebenfalls diese Logik, allerdings eingebettet in die date()-Funktion.
• Boolesche Werte nutzen die exakte Gleichheit.
• Zeichenketten, der Standardfall, verwenden ebenfalls die exakte Gleichheit.

Sämtliche Predikate werden mit AND verbunden, was den einfachen Attributvergleichen entspricht, die Neo4j im Rahmen des Vektor-Index-Filterns unterstützt.

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

Warum nicht dem LLM erlauben, den Cypher zu schreiben?

Der Neo4j MCP Cypher-Server könnte Abfragen über Text-zu-Cypher generieren, doch die Vektorsuche benötigt Embeddings – und derzeit ist die einzige Möglichkeit, Embeddings zwischen Tools zu übertragen, der Kontext des LLMs. Das Durchgeben von 1536-dimensionalen Vektoren durch das Modell verschwendet Tokens und Zeit.

Durch die Zusammenfassung des gesamten Workflows – Erstellung von Embeddings, Konstruktion und Ausführung von Cypher-Abfragen – in einem einzigen Tool bleiben die Embeddings außerhalb des Modellkontexts. Das LLM liefert eine Abfrage in natürlicher Sprache sowie optional Filter und erhält die bewerteten Ergebnisse.

Ausführung

Der Server kann direkt über GitHub mit uvx ausgeführt werden. Fügen Sie eine Eintragung wie diese zur MCP-Klientkonfiguration für Claude Desktop, VS Code, Cursor oder ähnliche Tools hinzu:

{
  "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>"
      }
    }
  }
}

Das Flag --with langchain-openai installiert den OpenAI-Embedding-Anbieter. Bei einem anderen Anbieter, wie z. B. Cohere oder einem lokalen Modell, muss diese Abhängigkeit ausgetauscht und EMBEDDING_MODEL aktualisiert werden. Die Abfrage der Embeddings erfolgt über LangChains init_embeddings(), sodass jeder von LangChain unterstützte Anbieter funktioniert.

Beim Testen in Claude Desktop mit einer Beispielfrage und der Erweiterung des Tool-Calls werden Parameter wie diese angezeigt:

{
  "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
    }
  }
}

Es muss nichts per Hand geschrieben werden: keine Tool-Definitionen, keine Abfragen, keine Filter-Schemata. Erstellen Sie einen weiteren Vektorindex, starten Sie den Server neu – und ein weiteres Tool erscheint.

Die wichtigsten Erkenntnisse

Die Vektorsuche mit Filterfunktionen in Neo4j 2026.02, in Kombination mit der Cypher 25 SEARCH-Syntax, bietet einen einfachen Weg zu Tools für gefilterte Vektorsuchen. In Kombination mit MCP sowie etwas Startzeit-Introspektion entsteht so ein Server, der Vektorindizes findet, filterbare Eigenschaften erkennt und typisierte Suchtools ohne eine einzige Zeile spezifischen Codes generiert. Richten Sie dies auf eine Neo4j-Datenbank mit Vektorindizes aus – sie erstellt anschließend ihr eigenes Toolset. In der Dokumentation zur Filterung finden Sie die vollständige Syntax.