GraphGuru | Text-to-Cypher with Guardrails: Building a Safe Query Agent

"Let the LLM write the Cypher" is the most requested GraphRAG feature and the easiest one to ship dangerously. A model that can generate MATCH can generate DETACH DELETE; a model that does not know your schema will invent labels; and a model with no cost ceiling will happily write a Cartesian product across your largest label. This tutorial builds a text-to-Cypher agent with four guardrails that we consider non-negotiable before one goes near production data:

Schema-constrained generation — the model only sees, and is only allowed to use, labels, relationship types, and properties that exist.
Read-only routing — generated queries execute through a read-only user on a read session, so a bad query cannot write.
EXPLAIN cost gating — every query is planned before it runs and rejected if the plan is too expensive.
Result limits and timeouts — hard caps on rows and transaction time.

Then we expose the same thing to any MCP-capable agent via Neo4j's official MCP server.

Setup

pip install "neo4j>=6" openai pydantic

Create a read-only database user. This is the guardrail that works even when every other one fails:

CREATE USER cypher_agent SET PASSWORD 'agent-only-read-me' CHANGE NOT REQUIRED;
GRANT ROLE reader TO cypher_agent;

On Community edition (no RBAC), run the agent against a read replica or a copy; do not give it the neo4j user.

Guardrail 1 — give the model the real schema, and only the real schema

Pull the schema from the database rather than hand-maintaining a prompt that drifts:

import os, neo4j

driver = neo4j.GraphDatabase.driver(
    os.environ["NEO4J_URI"],
    auth=("cypher_agent", os.environ["AGENT_PASSWORD"]),
)

def load_schema() -> str:
    node_props, _, _ = driver.execute_query(
        "CALL db.schema.nodeTypeProperties() YIELD nodeLabels, propertyName, propertyTypes "
        "RETURN nodeLabels, collect(propertyName + ':' + propertyTypes[0]) AS props"
    )
    rels, _, _ = driver.execute_query(
        "CALL db.schema.relTypeProperties() YIELD relType, propertyName "
        "RETURN relType, collect(propertyName) AS props"
    )
    patterns, _, _ = driver.execute_query(
        "CALL db.schema.visualization() YIELD relationships "
        "UNWIND relationships AS r "
        "RETURN DISTINCT labels(startNode(r))[0] + '-[:' + type(r) + ']->' + labels(endNode(r))[0] AS p"
    )
    lines = ["Node labels and properties:"]
    lines += [f"  {':'.join(r['nodeLabels'])} {{{', '.join(r['props'])}}}" for r in node_props]
    lines += ["Relationship types and properties:"]
    lines += [f"  {r['relType']} {r['props']}" for r in rels]
    lines += ["Valid patterns:"]
    lines += [f"  ({r['p']})" for r in patterns]
    return "\n".join(lines)

The db.schema.* procedures are cheap on any realistically sized graph, but cache the result and refresh it on a timer; you do not want a schema scan per question.

The prompt then states the rule explicitly and asks for structured output:

from pydantic import BaseModel
from openai import OpenAI

client = OpenAI()

class CypherDraft(BaseModel):
    cypher: str
    rationale: str
    uses_only_schema: bool

SYSTEM = """You translate questions into Neo4j Cypher 25 read queries.
Rules:
- Use ONLY the labels, relationship types, properties, and patterns listed in the schema. Never invent any.
- Read-only: MATCH, OPTIONAL MATCH, WITH, WHERE, RETURN, ORDER BY, LIMIT, UNWIND, CALL { } subqueries and built-in functions only.
- Always end with LIMIT 100 or less.
- If the question cannot be answered from the schema, return cypher = "" and explain why in rationale.
"""

def draft_cypher(question: str, schema: str) -> CypherDraft:
    completion = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM + "\n\nSchema:\n" + schema},
            {"role": "user", "content": question},
        ],
        response_format=CypherDraft,
        temperature=0,
    )
    return completion.choices[0].message.parsed

Guardrail 2 — refuse anything that is not a read, before it reaches the database

Never rely on the prompt alone. A small static check rejects write keywords and procedure calls you have not allow-listed:

import re

WRITE_TOKENS = re.compile(
    r"\b(CREATE|MERGE|DELETE|DETACH|SET|REMOVE|DROP|LOAD\s+CSV|FOREACH|"
    r"ALTER|GRANT|DENY|REVOKE|START\s+DATABASE|STOP\s+DATABASE)\b",
    re.IGNORECASE,
)
ALLOWED_PROCS = {"db.index.vector.queryNodes", "db.index.fulltext.queryNodes"}
PROC_CALL = re.compile(r"\bCALL\s+([a-zA-Z0-9_.]+)\s*\(", re.IGNORECASE)

def static_check(cypher: str) -> None:
    if WRITE_TOKENS.search(cypher):
        raise ValueError("write or admin keyword in generated query")
    for proc in PROC_CALL.findall(cypher):
        if proc not in ALLOWED_PROCS:
            raise ValueError(f"procedure not allowed: {proc}")
    if not re.search(r"\bLIMIT\s+\d+\s*;?\s*$", cypher, re.IGNORECASE):
        raise ValueError("missing trailing LIMIT")

This is a belt; the read-only user is the braces. CALL { ... } subqueries are allowed by the regex because the tokens inside them are still checked.

Guardrail 3 — `EXPLAIN` before you `RUN`

EXPLAIN asks the planner for a plan without executing. Two things fall out of it for free: the query is validated against the schema (an invented label fails here), and the plan carries the planner's estimated rows per operator. Walk the plan, sum the estimates, and refuse anything above a budget.

MAX_ESTIMATED_ROWS = 250_000
FORBIDDEN_OPERATORS = {"CartesianProduct", "AllNodesScan"}

def plan_cost(cypher: str) -> tuple[float, set[str]]:
    _, summary, _ = driver.execute_query(
        "EXPLAIN " + cypher,
        routing_="r",
        database_="neo4j",
    )
    total, ops = 0.0, set()
    stack = [summary.plan]
    while stack:
        op = stack.pop()
        ops.add(op["operatorType"])
        total += float(op["args"].get("EstimatedRows", 0))
        stack.extend(op.get("children", []))
    return total, ops

def cost_gate(cypher: str) -> None:
    estimated, ops = plan_cost(cypher)
    bad = ops & FORBIDDEN_OPERATORS
    if bad:
        raise ValueError(f"plan uses forbidden operator(s): {', '.join(sorted(bad))}")
    if estimated > MAX_ESTIMATED_ROWS:
        raise ValueError(f"plan too expensive: ~{estimated:,.0f} estimated rows")

Tune MAX_ESTIMATED_ROWS against your graph: run your twenty most common legitimate questions, note their estimates, and set the cap at a few multiples of the largest. AllNodesScan is almost always a sign the model ignored the schema; CartesianProduct is almost always a missing relationship in the pattern.

Guardrail 4 — run it read-only, bounded, and with a timeout

MAX_ROWS = 100

def run_safely(cypher: str) -> list[dict]:
    records, summary, _ = driver.execute_query(
        cypher,
        routing_="r",                     # read routing in a cluster
        database_="neo4j",
    )
    if len(records) > MAX_ROWS:
        raise ValueError("result too large")
    return [r.data() for r in records]

Set the transaction timeout at the database level so it applies no matter which client connects: db.transaction.timeout=10s in neo4j.conf (or per database via ALTER DATABASE ... SET OPTION). That is a better place for it than driver code, because it also catches the day someone bypasses your wrapper.

Putting it together, with one repair loop

def answer(question: str, schema: str) -> dict:
    draft = draft_cypher(question, schema)
    if not draft.cypher:
        return {"answer": None, "reason": draft.rationale}
    cypher = draft.cypher.strip()
    for attempt in range(2):
        try:
            static_check(cypher)
            cost_gate(cypher)
            rows = run_safely(cypher)
            return {"cypher": cypher, "rows": rows}
        except (ValueError, neo4j.exceptions.Neo4jError) as err:
            if attempt == 1:
                return {"cypher": cypher, "error": str(err)}
            repair = draft_cypher(
                f"The query failed with: {err}\nOriginal question: {question}\n"
                f"Rewrite it to satisfy the rules.", schema,
            )
            cypher = repair.cypher.strip()

schema = load_schema()
print(answer("Which customers placed more than five orders in 2025?", schema))

One repair attempt recovers most schema mistakes; more than one is usually the model looping on a question the graph cannot answer, and you are better off returning the rationale to the user.

Exposing it over MCP

If your agents are built on the Model Context Protocol, you do not need to wrap the driver yourself. Neo4j's official MCP server provides get-schema and read/write Cypher tools over a configured connection. Point it at the same read-only user, and in the client configuration register only the read tool. Roughly, for a Claude Desktop or Cursor-style mcp.json:

{
  "mcpServers": {
    "neo4j": {
      "command": "uvx",
      "args": ["mcp-neo4j-cypher@latest"],
      "env": {
        "NEO4J_URI": "bolt://localhost:7687",
        "NEO4J_USERNAME": "cypher_agent",
        "NEO4J_PASSWORD": "agent-only-read-me",
        "NEO4J_DATABASE": "neo4j"
      }
    }
  }
}

Check the repository README for the current package name and tool list — it moves monthly. What does not change is the principle: the credential the agent holds determines what the agent can do, and the LLM's good intentions are not a security control.

What we still do by hand

Even with all four guardrails, we keep a human-reviewed allow-list of question templates for anything customer-facing, log every generated query with its plan cost, and sample logs weekly. Text-to-Cypher is a productivity tool for analysts and an excellent internal agent capability; as an anonymous public endpoint it still needs a person in the loop. If you want help taking one from prototype to production safely, talk to our team.

Text-to-Cypher with Guardrails: Building a Safe Query Agent