SDK reference

Python and Node SDKs

Clients for Cypher, search, indexes, cursor streaming, capability tokens, and procedure helpers. This page describes SDK version 0.9.0.

First, create an API key. The examples below assume a gq_... key in your environment. Create a key →  ·  Then export it: export INVARIANTDB_API_KEY=gq_...
Graph names in the examples (my-graph, memory) stand for your full graph name as the dashboard shows it, for example a1b2c3_my-graph.

Install

Python

pip install "https://sdk.invariantdb.com/python/invariantdb-0.9.0-py3-none-any.whl#sha256=ae451496e49e3b15a2654e41535203560531922f14600fdbc390d6bfaf455f7f"

The Python SDK is not published on PyPI, so do not run pip install invariantdb without the URL. pip checks the downloaded file against the SHA-256 in the URL. In a requirements file, use:

invariantdb @ https://sdk.invariantdb.com/python/invariantdb-0.9.0-py3-none-any.whl#sha256=ae451496e49e3b15a2654e41535203560531922f14600fdbc390d6bfaf455f7f

Node

npm install https://sdk.invariantdb.com/node/invariantdb-0.9.0.tgz

SHA-256 of the 0.9.0 tarball: 318f40daf3579cfd3af5ffd842154cddd77651e8ab55cc63a419d0053d1eedf5.

Connect

Python

import os
from invariantdb import Client

db = Client("https://invariantdb.com", api_key=os.environ["INVARIANTDB_API_KEY"])

Node

import { Client } from 'invariantdb';

const db = new Client({
  url: 'https://invariantdb.com',
  apiKey: process.env.INVARIANTDB_API_KEY,
});

Cypher

# Python
result = db.cypher(
    graph="my-graph",
    query="MATCH (p:Person) WHERE p.role = $role RETURN p.name, p.startDate",
    parameters={"role": "engineer"},
)
for name, started in result.rows:
    print(name, started)

# Result fields: columns, rows, stats, mutations_applied, last_seq, warnings
// Node: cypher(graph, query, options)
const result = await db.cypher(
  'my-graph',
  'MATCH (p:Person) WHERE p.role = $role RETURN p.name AS name',
  { parameters: { role: 'engineer' } },
);
const rows = result.asDicts();

Provenance

To record who made a change, send X-Provenance-* headers with a Cypher write. In Node, pass them with the headers option:

await db.cypher(
  'my-graph',
  'MATCH (p:Person {name: $n}) SET p.role = $r',
  {
    parameters: { n: 'Alice', r: 'staff engineer' },
    headers: { 'X-Provenance-Actor': 'hr-sync', 'X-Provenance-Reason': 'promotion' },
  },
);

In Python, from SDK 0.9.0, pass the same headers with headers= on db.cypher (0.8.0 could not send them). Over the HTTP API you can also send a provenance body field. Read it back with CALL db.propertyProvenance(id(p), "role"). See Record who made a change for all fields.

Graphs

MethodEffect
db.graphs.ensure(name)Create the graph if it does not exist. Requires an admin or tenant-admin key; graphs created from the dashboard already exist.
db.graphs.list()List graphs. Requires an admin or tenant-admin key.
db.graphs.info(name)Returns name, current_snapshot_seq, and next_seq.
db.graphs.drop(name)Delete the graph and all of its data. Requires an admin or tenant-admin key. Cannot be undone.

Indexes

# Property index: speeds up MATCH (n:Person {email: $v}) and WHERE filters
db.indexes.create("my-graph", label="Person", property="email")

# Vector index
db.indexes.create("my-graph", label="Doc", property="embedding",
                  value_type="vector", metric="cosine")

# Several property indexes at once; existing ones are reported with created=False
db.indexes.ensure("my-graph", [{"label": "Person", "property": "email"}])

For a composite index, run Cypher: CREATE INDEX ON :Claim(year, code). Composite keys must be integer, float, or boolean properties.

Constraints

Create constraints with Cypher through db.cypher:

db.cypher("my-graph", "CREATE CONSTRAINT person_email IF NOT EXISTS FOR (p:Person) REQUIRE p.email IS UNIQUE")
db.cypher("my-graph", "CREATE CONSTRAINT person_name IF NOT EXISTS FOR (p:Person) REQUIRE p.name IS NOT NULL")

# Check stored data against every constraint (read-only)
db.cypher("my-graph", "CALL db.validateConstraints()").as_dicts()

# List constraints
db.cypher("my-graph", "CALL db.constraints() YIELD name, label, property, type "
                      "RETURN name, label, property, type").as_dicts()

# Remove one
db.cypher("my-graph", "DROP CONSTRAINT person_email")

Search

Search indexes are built on the first search against a label and property.

Full-text

hits = db.search.fulltext(
    "my-graph",
    label="Document", property="body",
    query="pricing strategy", k=10,
)
# list of SearchHit objects with node_id and score

Vector

hits = db.search.vector(
    "my-graph",
    label="Episode", property="embedding",
    vector=[0.1, 0.2, ...],
    k=10, metric="cosine",
)

Build a vector index ahead of time

db.indexes.create("my-graph", label="Episode", property="embedding",
                  value_type="vector", metric="cosine")
# returns vectors_indexed, dim, backend

Cursor streaming

# Iterate a large result without loading it all at once.
for row in db.cursor.stream("huge", "MATCH (n) RETURN n.id AS id, n.score AS score",
                            batch_size=10_000):
    process(row)  # {'id': {'string': 'n-1'}, 'score': {'float': 0.5}}

batch_size can be up to 10,000. Cursors are read-only and expire after 300 seconds idle. Row values from cursor.stream use the tagged form shown in the comment.

Authentication helpers

Capability tokens (short-lived, scoped)

# Issue a short-lived, scoped capability (requires an admin key)
cap = db.capabilities.issue(graph="memory", principal="agent-1",
                            scope=["read"], ttl_seconds=300)

# Use it
agent_db = Client("https://invariantdb.com", capability=cap["token"])

Requires an admin key. Scopes: read, write, admin. ttl_seconds can be at most 300.

Agent memory helpers

session = "conv-abc-123"

# Record one event. The payload is a JSON string.
db.procedures.record_episode("memory", session, "observation",
                             "User asked about Q3 forecast",
                             payload='{"topic": "forecast"}')

# Replay
events = db.procedures.replay_episodes("memory", session, limit=50)

# Record and revise beliefs (fictional IDs; evidence IDs must exist in the graph)
db.procedures.record_belief("memory", "churn-risk-acme", "High churn risk", 0.85,
                            evidence_str_ids=["event-99", "event-104"])

db.procedures.revise_belief("memory", "churn-risk-acme", "Moderate churn risk", 0.55,
                            "upsell completed")

# Walk the revisions
lineage = db.procedures.belief_lineage("memory", "churn-risk-acme")

Bitemporal queries

# Cypher does the work; the SDK passes the query through.
db.cypher(graph="my-graph", query="""
    MATCH (p:Policy)-[:HAS_COVERAGE]->(c)
      AT VALID '2024-10-15'
      AT RECORDED '2024-10-15'
    RETURN p.id, c.amount
""")

Over HTTP, POST /graphs/{name}/cypher accepts "atTime": "2025-01-01T00:00:00Z", which applies AT VALID to each top-level MATCH. It is refused when a MATCH already has a temporal clause or when the query has no top-level MATCH. The Python SDK 0.9.0 does not expose atTime.

Ranked recall over HTTP

# POST /graphs/{name}/memory/recall: ranked search with graph expansion and optional salience weighting
curl -fsS -X POST "https://invariantdb.com/graphs/memory/memory/recall" \
  -H "Authorization: Bearer $INVARIANTDB_API_KEY" \
  -H "Content-Type: application/json" \
  --data '{"query": "pricing", "label": "Event", "property": "text", "k": 5}'

Defaults: label Memory, property content, k 10 (max 50), hops 1 (max 3). To search episodes recorded with db.recordEpisode, set label to Event and property to text, as above. The Python SDK 0.9.0 does not wrap this endpoint.

Cypher YIELD AS aliasing

# CALL ... YIELD col AS alias RETURN alias
db.cypher(graph="memory", query="""
    CALL db.recordDecision('alice', 'claim-review', ['claim:a'], 'approve', 'sound risk')
    YIELD decisionId AS did
    RETURN did
""")

Compliance helpers

# GDPR Article 20 export
export_data = db.procedures.subject_export("my-graph", "user_42")

# Destroy a subject's encryption key (irreversible)
receipt = db.procedures.destroy_subject_key("my-graph", "user_42")

# History chain
status = db.audit.verify("my-graph")
assert status.status == "clean"

# Field-access-control event log
acl = db.procedures.verify_acl_chain("my-graph")
assert acl["status"] == "clean"

Crypto-shred (per-subject encryption keys and key destruction) is in development. On the current release, declaring an encrypted property is refused by default, and destroy_subject_key is refused without recording a receipt. Contact us before relying on it; see Crypto-shred.

Private mode

Create a private-mode key by sending a privacyMode object ({"epsilon_budget":10.0,"period":"Daily","max_per_query":1.0}) to POST /cloud/me/api-keys, or to POST /admin/api-keys as an admin. The Python SDK does not expose this.

# Create a private-mode key with your dashboard sign-in token (fictional names)
curl -fsS -X POST https://invariantdb.com/cloud/me/api-keys \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  --data '{"name": "research-partner", "graphs": ["a1b2c3_clinical"], "ttlDays": 90,
           "privacyMode": {"epsilon_budget": 10.0, "period": "Daily", "max_per_query": 1.0}}'
# The response field "key" holds the new key. It is shown once.

A private-mode key is used like any other key, but only expressions built on count, sum or avg may be returned; noise is added and each aggregate spends from the key’s ε budget. This release does not give a differential-privacy guarantee; do not rely on private mode alone to protect individual records.

# Check the budget used
budget = db.cypher(graph="a1b2c3_clinical", query="""
    CALL db.dpBudget() YIELD principal, consumed_epsilon, period_started_ms
    RETURN principal, consumed_epsilon, period_started_ms
""")

Error handling

from invariantdb.errors import (
    AuthError, NotFoundError, ConflictError, ValidationError,
    RateLimitError, ServerError, APIError,
)

try:
    db.cypher(graph, query)
except ValidationError as e:
    print(e.body.get("subcode"))   # e.g. PRIVATE_MODE_BUDGET_EXHAUSTED, SYNTAX_ERROR
except RateLimitError:
    ...  # wait for Retry-After, then retry

Errors map to status codes: 400/422 ValidationError, 401/403 AuthError, 404 NotFoundError, 409 ConflictError, 429 RateLimitError, 5xx ServerError. Check e.body["subcode"] for the specific cause. From server release 1.4.2 an error body also has retryable and retry (retryable, terminal or ambiguous); from SDK 0.9.0 errors also carry them as fields (subcode, code, retry_after_ms / retryAfterMs, retryable, retry), and the error text is the server’s message. The SDKs resend only requests that carry an idempotency key; see Idempotency keys and retries. See the API reference.

Idempotency keys and retries

From SDK 0.9.0, with server release 1.4.2 or later. An idempotency key makes a write safe to send again: the server runs a write at most once per key, and a resend with the same key and the same request gets the first answer back without writing again. Keys are 1 to 255 visible ASCII characters, are kept for 24 hours, and belong to one graph and one credential. Use a new key for each logical write and the same key when you resend it.

# Python
r = db.cypher(
    "my-graph",
    "CREATE (:Order {id: $id})",
    {"id": "o-42"},
    idempotency_key="order-o-42",
    headers={"x-request-id": request_id},
)
r.replayed   # True when the server answered from the first request

# Or let the client choose a new key for every call:
db = Client(url, api_key=key, auto_idempotency_key=True)
// Node
const r = await db.cypher('my-graph', 'CREATE (:Order {id: $id})', {
  parameters: { id: 'o-42' },
  idempotencyKey: 'order-o-42',
  headers: { 'x-request-id': requestId },
});
r.replayed; // true when the server answered from the first request

// Or let the client choose a new key for every call:
const db2 = new Client({ url, apiKey, autoIdempotencyKey: true });

Where the SDK code lives

Quickstart →  ·  MCP →  ·  Cypher reference →  ·  All features →