Un servidor MCP de configuración cero para búsquedas vectoriales filtradas en Neo4j
Descubra automáticamente los índices vectoriales de Neo4j, pruebe tipos de propiedades y genere herramientas de búsqueda MCP tipadas que utilicen el filtrado in-index de Cypher 25 sin necesidad de configuración manual alguna.
Imagínese un grafo de conocimiento repleto de artículos de noticias y un agente que recibe esta solicitud: mostrar la cobertura positiva sobre energías renovables de Reuters de la semana pasada. Esa oración contiene tres elementos distintos: una búsqueda semántica sobre “energías renovables”, filtros de metadatos por fuente (Reuters), sentimiento (positivo) y fecha (semana pasada).
La búsqueda vectorial simple maneja la parte semántica. Sin embargo, sin filtros, devolverá amablemente artículos negativos de 2019 publicados por blogs aleatorios, y el modelo de lenguaje grande no tendrá forma de saber que son irrelevantes hasta que los haya leído todos. Ese es el problema de la búsqueda vectorial con filtrado de metadatos, y Neo4j ha ofrecido desde hace tiempo dos formas de abordarlo.
- Filtrar previamente con Cypher para reducir el conjunto de candidatos y luego realizar la búsqueda vectorial sobre lo que queda. Los resultados son precisos, pero pueden volverse lentos cuando el conjunto de candidatos sigue siendo grande.
Ambos métodos tenían ventajas reales, y ambos dejaban espacio para mejoras.
Qué cambia en Neo4j 2026.02
Neo4j 2026.02 (consulte las notas de lanzamiento) hizo que la búsqueda vectorial con filtros estuviera disponible de forma general como parte de Cypher 25. Ahora se pueden aplicar predicados dentro del índice vectorial en el momento de la consulta, de modo que el algoritmo HNSW omite los vectores que no coinciden antes de realizar el ranking. La latencia se mantiene baja, el recuerdo se mantiene alto y no es necesario recuperar datos en exceso. También llega una nueva cláusula nativa SEARCH, que reemplaza al procedimiento anterior db.index.vector.queryNodes() y convierte a la búsqueda vectorial en una funcionalidad de primer nivel del lenguaje de consultas.
Eso plantea una pregunta: ¿podría un servidor MCP descubrir índices vectoriales por sí mismo, inspeccionar sus propiedades de metadatos y generar herramientas de búsqueda filtrada sin necesidad de configuración manual?
Un servidor de búsqueda vectorial genérico con cero configuración
Cualquiera que haya creado herramientas MCP para búsqueda vectorial conoce el procedimiento habitual: escribir el Cypher, definir los parámetros, configurar el modelo de embedding y documentar los filtros. Para un índice bien conocido eso funciona bien. Pero un servidor listo para usar que funcione con cualquier base de datos Neo4j, al cual simplemente se le indica la instancia y que se encarga del resto, resulta mucho más atractivo.
Eso es exactamente lo que hace este servidor. Al iniciar, se conecta a Neo4j, encuentra todos los índices vectoriales, toma una muestra de cada uno para detectar los tipos de propiedades y crea una herramienta MCP completamente tipada por cada índice, con parámetros de filtro y una documentación que el LLM puede leer.
Hay algo que no se puede descubrir automáticamente: el modelo de embedding. Neo4j aún no almacena eso en los metadatos del índice, y de todas formas se necesitaría una clave API para llamar al modelo. Por lo tanto, el proveedor de embedding y sus credenciales son la única configuración que debe proporcionarse.
Paso 1: encontrar los índices vectoriales
Primero, el servidor solicita a Neo4j todos los índices vectoriales:
records, _, _ = driver.execute_query(
"""
SHOW VECTOR INDEXES
""",
database_=NEO4J_DATABASE,
)
El resultado muestra el nombre de cada índice, la etiqueta del nodo que abarca y las propiedades indexadas. Si no existen índices vectoriales, el servidor finaliza con un error, ya que no hay nada para servir.
Paso 2: tipos de muestra y valores de ejemplo
Para cada índice, el servidor necesita saber qué propiedad contiene la representación vectorial y qué tipo tiene cada una de las demás. ¿Contiene published_date una fecha simple o un timestamp completo? ¿Es source texto? ¿El puntaje de sentimiento es un número decimal? Una función auxiliar dedicada _sample_property responde al examinar un único nodo por propiedad:
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
Devuelve el tipo detectado y el valor real de la muestra. La función principal reúne estos datos en dos diccionarios, uno para los tipos y otro para los ejemplos:
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 función _detect_type verifica, en orden, si se trata de una lista (un vector), bool, int, float, datetime y date, recayendo en string si no coincide con ninguno de ellos. Datetime y date se consideran tipos diferentes. Una propiedad que contenga componentes de tiempo tiene el tipo datetime y recibe la conversión datetime() de Neo4j, no date(). La lista de floats corresponde al embedding, y todas las demás propiedades se convierten en campos de metadatos filtrables. En el ejemplo de noticias, el resultado es el siguiente:
embedding -> vector
published_date -> date (or datetime, if timestamps are present)
source -> string
sentiment -> float
El diccionario meta_examples almacena los valores reales de las muestras, como “Reuters” para la fuente o 0.72 para el sentimiento, y estos enriquecen las cadenas de documentación generadas en el paso siguiente.
Paso 3: generar las herramientas de búsqueda
Ahora todo se combina. Para cada índice, una función closure captura los metadatos de ese índice y genera una función de búsqueda:
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
Algunos detalles merecen atención:
- El prefijo
CYPHER 25activa la sintaxis más reciente necesaria para la cláusula SEARCH y el filtrado en tiempo de indexación. - La cláusula
WHEREse encuentra dentro del bloque SEARCH. Eso es el filtrado en tiempo de indexación 2026.02 en acción. HNSW respeta directamente los predicados, por lo que los resultados aproximados de vecino más cercano se devuelven ya filtrados sin perder precisión tras el filtrado posterior. - La propiedad de embedding se establece en null en las filas devueltas (
{embed_prop}: null), de modo que los vectores grandes no vuelven al LLM, donde solo consumirían tokens innecesariamente. - La documentación se genera dinámicamente a partir de los tipos muestreados e incluye un valor de ejemplo concreto para cada filtro. Así, el agente puede ver qué filtros existen, qué tipos se esperan y cómo son los valores realistas.
Construcción de la cláusula WHERE
La función _build_where convierte un diccionario de filtros en Python en un predicado Cypher, eligiendo la forma según el tipo declarado de cada propiedad:
- Los números flotantes e enteros utilizan filtros de rango a través de
{"min": ..., "max": ...}, lo que genera comparaciones>=y<=. - Las fechas utilizan la misma lógica de rango, envuelta en el cast
datetime(). - Las fechas simples también la usan, envueltas en
date(). - Los valores booleanos emplean la igualdad exacta.
- Las cadenas de texto, que son el valor por defecto, también utilizan la igualdad exacta.
Todos los predicados se unen con AND, lo cual corresponde a las simples comparaciones de propiedades que Neo4j admite dentro del filtrado por índices vectoriales.
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
¿Por qué no dejar que el LLM escriba el Cypher?
El servidor Neo4j MCP Cypher podría generar consultas mediante conversión de texto a Cypher, pero la búsqueda vectorial necesita embeddings, y actualmente la única forma de transferir un embedding entre herramientas es a través del contexto del LLM. Transmitir vectores de 1536 dimensiones por el modelo desperdicia tokens y tiempo.
Al encapsular todo el flujo de trabajo —generación de embeddings, construcción y ejecución de Cypher— en una sola herramienta, los embeddings quedan fuera del contexto del modelo. El LLM proporciona una consulta en lenguaje natural y filtros opcionales, y recibe los resultados con puntuación.
Ejecutarlo
El servidor se puede ejecutar directamente desde GitHub con uvx. Añada una entrada como esta a la configuración del cliente MCP para Claude Desktop, VS Code, Cursor o herramientas similares:
{
"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>"
}
}
}
}
La bandera --with langchain-openai instala el proveedor de embeddings de OpenAI. Con otro proveedor, como Cohere o un modelo local, sustituya esa dependencia y actualice EMBEDDING_MODEL. La búsqueda de embeddings se realiza a través de init_embeddings() de LangChain, por lo que cualquier proveedor que soporte LangChain funcionará.
Al probar en Claude Desktop con una pregunta de ejemplo y expandir la llamada a la herramienta, se muestran parámetros como estos:
{
"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
}
}
}
Nada necesita ser escrito a mano: no hay definiciones de herramientas, ni consultas, ni esquemas de filtros. Crea otro índice vectorial, reinicia el servidor y aparecerá otra herramienta.
Conclusión
La búsqueda vectorial con filtrado en Neo4j 2026.02, junto con la sintaxis SEARCH de Cypher 25, ofrece una vía sencilla para herramientas de búsqueda vectorial filtrada. Combinada con MCP y un poco de introspección al iniciar el servidor, el resultado es un servidor que encuentra índices vectoriales, detecta propiedades filtrables y genera herramientas de búsqueda tipadas sin necesidad de escribir ni una sola línea de código específico para la herramienta. Dirígetela a una base de datos Neo4j que tenga índices vectoriales y esta creará su propio conjunto de herramientas. La documentación sobre filtrado incluye toda la sintaxis.
Leer más
- Receta de Spring AI: Filtrar resultados RAG con metadatos por título del juego — Una guía de Spring AI que etiqueta los documentos con un título, utiliza un LLM para seleccionar el valor adecuado a partir de la pregunta y filtra la búsqueda vectorial en función de él.
- MCP en detalle: mensajes JSON-RPC, servidores FastMCP y un seguimiento ReAct — Sigue una solicitud del Protocolo de Contexto de Modelo desde las herramientas/list hasta la respuesta final, con servidores FastMCP, un agente LangGraph ReAct y un rastreador que registra cada paso.