Usage#

This is a hands-on guide to the AgensGraph MCP servers. It assumes you have completed Installation: the server commands available (through uvx or an installed package), and, for the Cypher and memory servers, a running AgensGraph and a role that is not a superuser (see Database role).

Two kinds of client use these servers:

Claude Desktop: the LLM picks which tools to call from your natural-language request. This is the primary way to use them (see below).
A FastMCP Client: call tools directly from Python, exactly as an agent would, with no LLM. Good for scripting and testing.

The servers call no LLM and need no API key. The intelligence is the client; the servers are deterministic tools. Tool names use underscores (read_agensgraph_cypher).

Driving a server from Python#

The simplest self-contained client spawns a server over stdio and calls its tools. The examples below run the installed commands, so install the servers into the environment that runs them (pip install "agensgraph-ai[mcp]", which also brings fastmcp). A tool result carries a text block holding JSON, or plain text for the tools that return a string; this data() helper reads either.

import asyncio, json
from fastmcp import Client
from fastmcp.client.transports import StdioTransport

def data(result):
    """A tool result's first text block: parsed JSON, or the text itself."""
    raw = result.content[0].text if getattr(result, "content", None) else ""
    try:
        return json.loads(raw)
    except ValueError:
        return raw

CONN = {
    "AGENSGRAPH_URL": "postgresql://localhost:5432",
    "AGENSGRAPH_USERNAME": "mcp",          # not a superuser: see Installation
    "AGENSGRAPH_PASSWORD": "mcp",
    "AGENSGRAPH_DATABASE": "mydb",
    "AGENSGRAPH_GRAPHNAME": "mygraph",
}

def cypher_client(*args):
    """A client for a Cypher server started with these extra CLI flags."""
    return Client(StdioTransport(command="mcp-agensgraph-cypher", args=list(args), env=CONN))

async def main():
    async with cypher_client() as client:
        print([tool.name for tool in await client.list_tools()])
        print(data(await client.call_tool("get_agensgraph_schema", {})))

asyncio.run(main())

A client for a server already listening over HTTP is just Client(StreamableHttpTransport(url)), with url the server’s endpoint, such as "http://127.0.0.1:8000/mcp/", and StreamableHttpTransport from fastmcp.client.transports.

Cypher server#

mcp-agensgraph-cypher runs Cypher against AgensGraph. The server creates its graph at startup if it is missing, unless it is read-only.

Label and property names are sent as written. A mixed-case label or property name is quoted before the statement is sent, so :Person means the label Person exactly as written. A graph whose labels were created unquoted (for example in psql) holds them in lower case; refer to those in lower case (:city, not :City).

Tool	Input	Returns
`get_agensgraph_schema`	—	Per-label `count`, `properties` and `relationships`.
`read_agensgraph_cypher`	`query`, `params?`, `limit?`, `offset?`, `walk?`	A paginated envelope (below); refuses writes.
`write_agensgraph_cypher`	`query`, `params?`	The five write counters and the `rows` the statement returned.
`explain_agensgraph_cypher`	`query`, `analyze?`	The plan as JSON.
`recommend_property_indexes`	`query`, `params?`	`{findings, verified, note, existing_indexes}`
`agensgraph_health`	—	Health figures for the database and the graph.
`top_cypher_queries`	`limit?` (default 20)	The costliest statements.

get_agensgraph_schema: each property comes with type, declared, indexed and unique.
explain_agensgraph_cypher: with analyze, it runs the statement in a read-only transaction for real timings.
recommend_property_indexes: index and rewrite advice from the plan. It never runs DDL.
agensgraph_health: cache hit ratio, unused indexes, vacuum backlog, connection use, graph metadata, installed extensions.
top_cypher_queries: the statements come from pg_stat_statements; without it, the tool returns a note saying it is not installed.

A parameter is written %(name)s, not $name. Both read_agensgraph_cypher and write_agensgraph_cypher take params as a map and bind it the way the driver does, so the placeholder is the driver’s. $name reaches the server as written and fails with syntax error at or near "$". A list or a map bound this way arrives as JSONB, which is what UNWIND and IN expect.

Writing. The write tool returns insertedvertices, insertededges, deletedvertices, deletededges and updatedproperties (a counter the statement cannot be held to is null), plus rows and row_count. A reply is about work that was committed; a failure means nothing was.

Reading with pagination. read_agensgraph_cypher returns {rows, row_count, offset, limit, has_more, next_offset}. Follow next_offset until has_more is false. limit defaults to 100 and is clamped to 1000. To read a whole result, pass walk: true instead of paging: it reads the result once from a server-side cursor, still returns at most limit rows, and adds the exact total_rows.

async def read_all(client, query, params=None):
    rows, offset = [], 0
    while True:
        page = data(await client.call_tool("read_agensgraph_cypher", {
            "query": query, "limit": 1000, "offset": offset,
            **({"params": params} if params else {}),
        }))
        rows.extend(page["rows"])
        if not page["has_more"]:
            return rows
        offset = page["next_offset"]

Writing a node, then reading it back both ways:

async with cypher_client() as client:
    stats = data(await client.call_tool("write_agensgraph_cypher", {
        "query": 'CREATE (:"Airport" {iata: %(iata)s, city: %(city)s})',
        "params": {"iata": "JFK", "city": "New York"},
    }))
    print(stats)   # {"insertedvertices": 1, "insertededges": 0, ..., "rows": [], "row_count": 0}

    rows = await read_all(client, 'MATCH (a:"Airport") WHERE a.city = %(city)s RETURN a.iata AS iata',
                          {"city": "New York"})
    walked = data(await client.call_tool("read_agensgraph_cypher", {
        "query": 'MATCH (a:"Airport") RETURN a.iata AS iata', "walk": True, "limit": 10,
    }))
    print(len(rows), walked["total_rows"])

A response is also bounded by a token budget (10,000 tokens by default; see Installation). Whole rows are dropped to fit, never part of one, and rows_omitted says how many. A list of 128 items or more in a value is replaced with <omitted: list of N items>.

Read-only safety. Reads run inside a READ ONLY transaction, so a write is refused by the database whatever the text looked like. The read tool also refuses a statement that writes, and the write tool refuses one that only reads. Starting the server with AGENSGRAPH_READ_ONLY=true (or --read-only) goes further and does not register the write tool at all:

async with cypher_client("--read-only") as client:
    names = [tool.name for tool in await client.list_tools()]
    print("write_agensgraph_cypher" in names)   # False

Each tool takes one graph statement. SQL is refused, including SQL behind EXPLAIN. Statements that change a graph’s shape (CREATE VLABEL, CREATE PROPERTY INDEX, DROP GRAPH and the like) are refused by the write tool unless the server was started with --allow-graph-ddl.

Plans, index advice and health. These tools read only:

async with cypher_client() as client:
    plan = data(await client.call_tool("explain_agensgraph_cypher", {
        "query": "MATCH (a:\"Airport\") WHERE a.iata = 'JFK' RETURN a",
    }))
    print(plan[0]["Plan"]["Node Type"])          # e.g. "Seq Scan"

    advice = data(await client.call_tool("recommend_property_indexes", {
        "query": "MATCH (a:\"Airport\") WHERE a.iata = 'JFK' RETURN a",
    }))
    print(advice["findings"], advice["verified"])  # verified is False: advice is reasoned, not measured

    health = data(await client.call_tool("agensgraph_health", {}))
    print(health["cache_hit_ratio"])
    print(data(await client.call_tool("top_cypher_queries", {"limit": 5})))

explain_agensgraph_cypher takes no parameters, so write literal values into the statement. Advice about a query with parameters is given the parameters through params, because a list bound as a parameter plans differently from one written inline. recommend_property_indexes flags two shapes that cannot reach an index at all: STARTS WITH, and an IN list bound as a parameter. It reports each with the rewrite that does use one.

Namespacing and transports. Give a server AGENSGRAPH_NAMESPACE=graph1 (or --namespace graph1) to prefix every tool name (graph1-read_agensgraph_cypher) so several servers can run side by side. The same tools work over stdio and Streamable http; switch with --transport. Read behavior is tunable with --read-timeout and --token-limit (and the AGENSGRAPH_* equivalents in Installation).

Memory server#

mcp-agensgraph-memory is a knowledge-graph memory an agent grows over a conversation. Its model: entities {name, type, observations} and relations {source, target, relationType}. A name identifies one entity, and a relationship type is stored upper-cased. At startup the server creates its graph (default memory) with a Memory label, a unique index on the name and a full-text index; it refuses a graph holding a Memory label it did not make unless started with --adopt-existing-graph.

Tool	Input	Returns
`create_entities`	`entities`	The entities as they now stand.
`create_relations`	`relations`	`{created, skipped}`
`add_observations`	`observations`	Per entity, `{entityName, addedObservations, found}`.
`read_graph`	`limit?`, `order?`	`{entities, relations, truncated}`.
`search_memories`	`query`, `limit?`, `order?`	The matching entities and the relations among them, with `truncated`.
`find_memories_by_name`	`names`, `limit?`	The named entities and their neighbours.
`delete_observations`	`deletions`	Per entity, `{entityName, deletedObservations, found}`.
`delete_relations`	`relations`	`{requested, deletedRelations}`.
`delete_entities`	`entityNames`	`{deleted, notFound, deletedRelations}`.

create_entities: each entity comes back as {name, type, observations, updated}. An existing entity keeps its observations and gains the new ones.
create_relations: a relation whose ends are not in the memory is skipped.
find_memories_by_name: returns the named entities, every relationship touching them, and the entities at the other end.

Here is the full lifecycle through all nine tools:

memory = Client(StdioTransport(command="mcp-agensgraph-memory", args=["--transport", "stdio"],
                               env={**CONN, "AGENSGRAPH_GRAPHNAME": "memory"}))
async with memory as client:
    # create
    await client.call_tool("create_entities", {"entities": [
        {"name": "Jane Roe", "type": "person", "observations": ["Prefers aisle seats"]},
        {"name": "Acme Corp", "type": "company", "observations": []},
    ]})
    await client.call_tool("create_relations", {"relations": [
        {"source": "Jane Roe", "target": "Acme Corp", "relationType": "WORKS_AT"},
    ]})
    await client.call_tool("add_observations", {"observations": [
        {"entityName": "Jane Roe", "observations": ["Planning a trip to Tokyo in March"]},
    ]})

    # read / search
    graph  = data(await client.call_tool("read_graph", {"limit": 1000}))   # {entities, relations, truncated}
    recent = data(await client.call_tool("read_graph", {"limit": 10, "order": "recent"}))
    hits   = data(await client.call_tool("search_memories", {"query": "Tokyo", "limit": 50}))
    named  = data(await client.call_tool("find_memories_by_name", {"names": ["Jane Roe"]}))
    print([e["name"] for e in hits["entities"]])   # ['Jane Roe']

    # delete
    await client.call_tool("delete_observations", {"deletions": [
        {"entityName": "Jane Roe", "observations": ["Prefers aisle seats"]},
    ]})
    await client.call_tool("delete_relations", {"relations": [
        {"source": "Jane Roe", "target": "Acme Corp", "relationType": "WORKS_AT"},
    ]})
    print(data(await client.call_tool("delete_entities", {"entityNames": ["Acme Corp"]})))

Every returned entity also carries updated, when it was last written (UTC, ISO 8601). It is null for an entity written before the memory recorded it, and for the neighbouring entities find_memories_by_name returns, which it does not read the timestamp for. order is "name" (the default) or "recent", most recently written first. search_memories matches every word given, with stemming, so engineers finds engineering but eng finds nothing; "*" asks for everything. read_graph and search_memories return at most limit entities (default AGENSGRAPH_MEMORY_LIMIT, at most 1000) and set truncated when more exists. find_memories_by_name looks up at most limit names.

Data-modeling server#

mcp-agensgraph-data-modeling designs a graph schema and generates the Cypher to build it. It needs no database. The easiest flow starts from a built-in example model, then generates constraints and ingest queries from it:

modeling = Client(StdioTransport(command="mcp-agensgraph-data-modeling", args=["--transport", "stdio"]))
async with modeling as client:
    print(data(await client.call_tool("list_example_data_models", {})))      # browse examples

    example = data(await client.call_tool("get_example_data_model",
                                          {"example_name": "patient_journey"}))
    model = example["data_model"]

    # validate, visualize, and generate build Cypher
    await client.call_tool("validate_data_model", {"data_model": model})
    mermaid = data(await client.call_tool("get_mermaid_config_str", {"data_model": model}))
    constraints = data(await client.call_tool("get_constraints_cypher_queries", {"data_model": model}))
    node_ingest = data(await client.call_tool("get_node_cypher_ingest_query",
                                              {"node": model["nodes"][0]}))
    print(constraints[:2])   # ['create vlabel if not exists "Patient"', 'create unique property index ...']
    print(node_ingest)       # UNWIND %(records)s as record MERGE (n: "Patient" {id: record.id}) ...

get_constraints_cypher_queries returns a list, one whole statement per item. Run them one at a time in order, and do not split them on a semicolon, since a label may contain one. Each can be run again. The ingest queries bind their records to a parameter named records, written %(records)s.

The statements declare labels and indexes, which is graph DDL. To run them through the Cypher server, start it with --allow-graph-ddl; then load records with the ingest query:

async with cypher_client("--allow-graph-ddl") as client:
    for statement in constraints:
        await client.call_tool("write_agensgraph_cypher", {"query": statement})
    stats = data(await client.call_tool("write_agensgraph_cypher", {
        "query": node_ingest,
        "params": {"records": [{"id": "p-001", "first": "Ann", "last": "Lee"}]},
    }))
    print(stats["insertedvertices"])   # 1 the first time; the MERGE matches after that

Other tools: validate_node, validate_relationship, get_relationship_cypher_ingest_query, and load_from_arrows_json / export_to_arrows_json to round-trip with the Arrows app. There are seven examples: patient_journey, supply_chain, software_dependency, oil_gas_monitoring, customer_360, fraud_aml and health_insurance_fraud. The model and example schemas are also exposed as MCP resources (resource://schema/..., resource://examples/...), along with ingest guidance at resource://static/agensgraph_data_ingest_process. The server defines an MCP prompt, create_new_data_model, which Claude Desktop surfaces as a slash command: given a data context and use cases, it guides the agent to design, validate, and visualize a model. You then hand the generated Cypher to the Cypher server’s write_agensgraph_cypher.

Use with Claude Desktop#

Claude Desktop is MCP’s native client. Register the servers in its claude_desktop_config.json and restart; their tools appear automatically. See Installation for the full config. In short, each server is a uvx <server> command with the AGENSGRAPH_* connection in env. Then, for example:

Cypher: “What’s in this graph? Which airports have the most outgoing routes?” The agent calls get_agensgraph_schema, then paginated read_agensgraph_cypher. “Why is that query slow?” leads to explain_agensgraph_cypher and recommend_property_indexes.
Memory: “Remember I prefer aisle seats.” leads to create_entities / add_observations; later “What are my travel preferences?” leads to search_memories / read_graph.
Data modeling: the create_new_data_model prompt designs a schema, renders a Mermaid diagram, and produces ingest Cypher.

You can validate any server without Claude using the MCP Inspector, passing the connection with -e (see Installation):

npx @modelcontextprotocol/inspector \
  -e AGENSGRAPH_USERNAME=mcp -e AGENSGRAPH_PASSWORD=<your-password> \
  -e AGENSGRAPH_DATABASE=mydb -e AGENSGRAPH_GRAPHNAME=mygraph \
  -- uvx mcp-agensgraph-cypher --transport stdio

Run the demos#

The repository ships a runnable demo suite that drives the servers with a FastMCP client (no LLM, fully reproducible), each with a pre-executed notebook.

Demo	Server(s)	What it shows
`01_model_and_load`	data-modeling → cypher	Design a schema and load OpenFlights.
`02_cypher_query`	cypher	Schema, multi-hop reads, pagination, read-only, timeout/token-limit.
`03_cypher_operate`	cypher	Writes, modes and transports.
`04_memory`	memory	All nine memory tools end to end.
`05_cypher_scale`	cypher	Operating a graph it did not build.

01_model_and_load: designs a schema, generates ingest Cypher, and loads OpenFlights through the write tool.
03_cypher_operate: writes + stats, read-only mode, namespacing, stdio and Streamable HTTP transports.
05_cypher_scale: read-only schema, aggregations, plans, index advice, health and top statements on a graph it did not build.

The demos run against a checkout: one virtual environment installs the common core and all three servers in editable mode (see the demos README). Copy .env.example to .env for the connection settings (the demos use the AGENS_* prefix and port 55432 for their own config), then run, e.g.:

cd mcp-agensgraph/examples/demos
.venv/bin/python 01_model_and_load/build.py    # design a schema + load OpenFlights
.venv/bin/python 02_cypher_query/ask.py        # query it through the cypher server

A second track, claude_desktop/, provides a uvx-based Claude Desktop config and guided scenarios for all three servers.