SDK reference

Python + Node, batteries included.

Idiomatic clients for both languages: Cypher, search, indexes, cursor streaming, capability + Cognito token providers, agent-memory helpers. Same API surface, language-native types.

First, create an API key. Every example below assumes a gq_... key in your environment. Get yours → · Then export it: export INVARIANTDB_API_KEY=gq_...

Install

Python

pip install invariantdb

Node

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

Connect

Python

from invariantdb import Client

db = Client(
    "https://invariantdb.com",
    api_key="gq_paste_your_api_key_here",
)
# Or via env: INVARIANTDB_URL + INVARIANTDB_API_KEY
db = Client.from_env()

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 shape: { columns, rows, stats, mutations_applied, last_seq, warnings }

// Node
const { rows } = await db.cypher({
  graph: 'my-graph',
  query: 'MATCH (p:Person) WHERE p.role = $role RETURN p.name AS name',
  parameters: { role: 'engineer' },
});

Graphs

Method	Effect
`db.graphs.ensure(name)`	Create if missing; idempotent.
`db.graphs.list()`	List every graph the caller can see.
`db.graphs.info(name)`	Current version identifier, label/edge type counts, schema version.
`db.graphs.drop(name)`	Delete the graph + every snapshot. Irreversible.

Indexes

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

// Vector index — HNSW
db.indexes.ensure(graph="my-graph", label="Doc", property="embedding",
                  type="vector", metric="cosine")

// Composite — multi-property
db.indexes.ensure(graph="my-graph", label="Claim",
                  properties=["state", "year"], type="composite")

Search

Fulltext

hits = db.search.fulltext(
    graph="my-graph",
    label="Document", property="body",
    query="pricing strategy", k=10,
)
# [{nodeId, score, snippet}]

Vector

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

Pre-build (warm cache)

db.search.prebuild(graph="my-graph", label="Episode",
                   property="embedding", metric="cosine")
# Returns vectorsIndexed, dim, backend ("flat" | "hnsw")

Cursor streaming

// For queries with millions of result rows, stream them.
for batch in db.cursor.stream(
    graph="huge",
    query="MATCH (n) RETURN n.id, n.score",
    batch_size=10_000,
):
    process_batch(batch.rows)
    # Each batch yields .rows + .cursor for resumption

Authentication helpers

Capability tokens (per-conversation / per-session)

// Issue a short-lived, scoped capability
cap = db.capabilities.issue(
    graph="memory",
    principal="agent-deepthought",
    scope=["cypher_query"],
    session="conv-abc-123",        # restricts to one conversation
    ttl_seconds=3600,
)
# Pass `cap.token` to the agent process.

// Use it
agent_db = Client("https://invariantdb.com", api_key=cap.token)

Cognito JWT

db = Client("https://invariantdb.com",
            auth=CognitoTokenProvider(
                pool_id="us-east-1_XXXXXXX",
                client_id="your-app-client-id",
                username="alice@example.com",
                password=os.environ["ALICE_PASSWORD"],
            ))
# Provider refreshes JWT before expiry automatically.

Agent memory helpers

session = "conv-abc-123"

# Record one event
db.memory.record_episode(graph="memory", session=session,
                         kind="observation",
                         text="User asked about Q3 forecast",
                         payload={"topic": "forecast"})

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

# Record + revise beliefs
belief_id = db.memory.record_belief(graph="memory",
    agent="agent-deepthought",
    key="churn-risk-acme",
    content="High churn risk (0.85)",
    evidence=["event-99", "event-104"])

db.memory.revise_belief(graph="memory",
    key="churn-risk-acme",
    new_content="Moderate (0.55) — upsell saved them",
    evidence=["event-181"])

# Walk lineage
lineage = db.memory.belief_lineage(graph="memory",
                                    key="churn-risk-acme")

Bitemporal queries

// Pure Cypher — the SDK doesn't wrap, just delegates
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
""")

// New 2026-06-27 — ad-hoc time-travel via `atTime` body field
// (no Cypher rewrite needed; the engine broadcasts AT VALID 'ts'
// to every MATCH clause server-side):
db.cypher(graph="my-graph",
          query="MATCH (e:Event) RETURN count(e)",
          atTime="2025-01-01T00:00:00Z")
// Rejected with HTTP 400 if `query` already contains AT VALID/AT RECORDED.

One-call ranked recall (new 2026-06-27)

// POST /graphs/{name}/memory/recall — BM25 + graph-expand + salience
// rerank server-side. Replaces 3 client-side cypher round-trips.
hits = db.memory.recall(graph="memory",
                        query="what did the user ask about pricing yesterday",
                        k=5,
                        hops=1,
                        salience_weight=0.3)
for snippet in hits.snippets:
    print(snippet.score, snippet.content)
// Defaults: k=10, hops=1, salience_weight=0 (pure BM25).
// Caps: k≤50, hops≤3.

Cypher YIELD AS aliasing (new 2026-06-27)

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

Compliance helpers

// GDPR Article 20 export
export_data = db.compliance.subject_export(graph="my-graph",
                                            subject="user_42")
write_json("user_42_export.json", export_data)

// GDPR Article 17 crypto-shred (irreversible)
receipt = db.compliance.destroy_subject_key(graph="my-graph",
                                             subject="user_42")
write_json("user_42_erasure_receipt.json", receipt)

// Audit chain verify
status = db.compliance.verify_acl_chain(graph="my-graph")
assert status.status == "clean"

Differential privacy

// Issue a private-mode API key (admin only)
dp_key = db.api_keys.create(
    name="research-partner-mit",
    role="reader",
    graphs=["clinical_2026"],
    ttl_days=90,
    privacy_mode={
        "epsilon_budget": 10.0,
        "period": "Daily",
        "max_per_query": 1.0,
    },
)
# Send dp_key.key to the partner ONCE.

// The partner uses it like any other key, but the planner refuses
// non-aggregate returns and Laplace noise is added automatically.

// Check budget remaining
budget = db.cypher(graph="clinical_2026", query="""
    CALL db.dpBudget() YIELD principal, epsilonSpent
    RETURN principal, epsilonSpent
""")

Error handling

from invariantdb.errors import (
    AuthError,         # 401 / 403
    RateLimitError,    # 429
    PrivateModeError,  # DP gate refusals
    UnclampedProperty, # sum/avg without dpClamp
    BudgetExhausted,   # ε budget hit
    CypherParseError,
    TimeoutError,
)

try:
    db.cypher(...)
except BudgetExhausted as e:
    # Wait until the period resets; the engine doesn't lie about when.
    print(f"Need to wait until {e.period_resets_at}")

Where the SDK code lives

Python: PyPI · invariantdb
Node: sdk.invariantdb.com · invariantdb-0.7.0
Source: proprietary; contact sales@invariantdb.com for source-license terms.

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