Graph Cypher

Instructions

Implementation Reference

• src/mcp-server/index.ts:537-553 (handler)

  The graph_cypher tool handler function. It checks if the current tenant is an admin (via isAdminTenant), then calls client.executeCypher() to run a read-only Cypher query against the Neo4j graph. Returns results with execution time.

  }, async (args) => {
    const tenantId = currentTenant();
    if (!isAdminTenant(tenantId)) {
      return toolError(
        `graph_cypher is admin-only (your tenant: ${tenantId}). Use graph_query, graph_entities, or graph_search instead — those are tenant-scoped.`,
      );
    }
    try {
      const result = await client.executeCypher(args.cypher, args.params ?? {});
      return toolResult({
        cypher: args.cypher,
        ...result,
      });
    } catch (err) {
      return toolError(`graph_cypher failed: ${err instanceof Error ? err.message : String(err)}`);
    }
  });

• src/mcp-server/index.ts:528-537 (registration)

  Registration of the graph_cypher tool with the MCP server. Defines title, description, input schema (cypher: string, params: optional record), and marks it read-only. The description notes it's admin-only and enforced read-only via Neo4j executeRead().

  server.registerTool("graph_cypher", {
    title: "Graph Cypher",
    description:
      "Execute a read-only Cypher query against the memory graph. You generate the Cypher — this tool just runs it. Enforced read-only via Neo4j executeRead(). Use for custom queries not covered by other tools. Admin-only (must be the bootstrap tenant) — non-admin tenants would otherwise be able to bypass tenant filtering by writing raw Cypher.",
    inputSchema: {
      cypher: z.string().describe("Cypher query to execute (read-only)"),
      params: z.record(z.string(), z.unknown()).optional().describe("Query parameters"),
    },
    annotations: { readOnlyHint: true },
  }, async (args) => {

• src/mcp-server/index.ts:532-536 (schema)

  Input schema for graph_cypher: requires a 'cypher' string (the Cypher query) and optionally a 'params' record for query parameters.

  inputSchema: {
    cypher: z.string().describe("Cypher query to execute (read-only)"),
    params: z.record(z.string(), z.unknown()).optional().describe("Query parameters"),
  },
  annotations: { readOnlyHint: true },

• src/shared/neo4j-client.ts:1090-1098 (helper)

  The executeCypher method on Neo4jClient. Runs the Cypher query via the private run() method (which uses the Neo4j driver session with WRITE access mode), measures execution time, and returns results with row count and timing.

  async executeCypher(
    cypher: string,
    params: Record<string, unknown> = {},
  ): Promise<{ results: Record<string, unknown>[]; result_count: number; execution_time_ms: number }> {
    const start = Date.now();
    const rows = await this.run(cypher, params);
    const elapsed = Date.now() - start;
    return { results: rows, result_count: rows.length, execution_time_ms: elapsed };
  }

• src/shared/auth.ts:219-223 (helper)

  The isAdminTenant helper function. Checks if the given tenantId matches the BOOTSTRAP_TENANT_ID environment variable. Used by graph_cypher's handler to gate access to admin-only users.

  export function isAdminTenant(tenantId: string): boolean {
    const bootstrap = process.env.BOOTSTRAP_TENANT_ID;
    if (!bootstrap) return false;
    return tenantId === bootstrap;
  }