Relata Python SDK

Python SDK for the Relata DB enterprise-grade data engine.

Repo: github.com/relatadb/sdk-python
Docs: relatadb.dev · Issues
By: ZySec AI — Frontier Sovereign Intelligence · hello@zysec.ai

Relata DB is a Rust-native, ontology-driven engine built for bi-temporal, governed knowledge workloads: link analysis, identity resolution, full-text and vector search, access-scoped restricted data, and provenance-tracked analytics.

Requirements

Python 3.11+
httpx >= 0.27
pydantic >= 2.6

Installation

pip install relata-sdk
# or with uv
uv add relata-sdk

Quick start

from relata import RelataClient

with RelataClient("http://localhost:9090", purpose="analytics") as client:
    result = client.query("SELECT * FROM Person WHERE name LIKE 'Ahmed%' LIMIT 10")
    for row in result:
        print(row)

Flagship retrieval surface — `namespace().query()` (#1991)

For the search / RAG persona, the SDK ships a typed JSON door that compiles to a governed SQL plan — no SQL on the client side. text compiles to a BM25 rank_by directive; filters narrow the set server-side; and .write() is schemaless (auto-creates the type with inferred fields via POST /ingest/auto — no DDL).

from relata import RelataClient

with RelataClient("http://localhost:9090", purpose="rag") as client:
    docs = client.namespace("Document")
    docs.write([{"id": "d1", "title": "Knowledge graphs 101", "body": "..."}])
    res = docs.query(
        text="graph retrieval",
        match_column="title",
        filters=[{"field": "status", "op": "eq", "value": "published"}],
        limit=10,
    )
    for row in res:
        print(row["title"])

rank_by (explicit) accepts the server's array forms (["bm25","title","..."], ["vector","ann","..."]); filters take {field, op, value} (ops eq/ne/gt/gte/lt/lte/like/ilike/in/between) or an equality-dict shorthand. Async: AsyncNamespace / await docs.query(...). HTTP compression is off by default (pass compress=True to re-enable for slow links); one connection pool is reused across every namespace.

Cypher

RelataDB auto-detects Cypher: any query starting with MATCH is translated to SQL before execution. The SDK is language-agnostic — send the string as you would SQL:

result = client.query("MATCH (n:Person {id: 'p1'}) RETURN *")
# → SELECT * FROM Person WHERE id = 'p1'

GraphQL

graphql() (ADR-220) translates a GraphQL SELECT subset to SQL and runs it through the governed query path. Returns data; raises RelataError on a non-empty errors envelope.

data = client.graphql(
    "{ Person(where: { age: { _gt: 30 } }, limit: 10) { id name age } }",
)
# data == [{"id": "p1", "name": "Ada", "age": 36}, ...]

# With variables + an explicit operation name:
data = client.graphql(
    "query Q($n: Int!) { Person(where: { age: { _gt: $n } }) { name } }",
    variables={"n": 30},
    operation_name="Q",
)

# Async variant:
# data = await client.agraphql("{ __schema { types { name } } }")

Supported: MATCH / OPTIONAL MATCH, WHERE, RETURN, UNION / UNION ALL (#378), and CALL traverse.* / CALL gds.* procedures (#377). CREATE / MERGE writes route through the governed write door. See the SQL reference.

Hybrid search — `VectorClient.hybrid_search()`

VectorClient wraps the server's HYBRID_SEARCH operator (ADR-175): supply query_text (BM25 leg), query_embedding + embedding_slot (vector leg), or both — when both are present the server fuses the two rankings via reciprocal rank fusion. This is Relata's genuine edge over a plain vector DB or a plain full-text engine.

from relata import RelataClient
from relata.vectors import VectorClient

with RelataClient("http://localhost:9090", purpose="rag") as client:
    vectors = VectorClient.from_client(client)

    query_embedding = vectors.embed("graph retrieval")["embedding"]

    hits = vectors.hybrid_search(
        "Document",
        query_text="graph retrieval",
        query_embedding=query_embedding,
        embedding_slot="_emb_text",
        k=10,
    )
    for row in hits:
        print(row["title"], row.get("_score"))

See examples/hybrid_search.py for a full runnable walkthrough (embed → ingest → BM25-only vs. vector-only vs. fused search). The higher-level namespace().query(rank_by=["vector", "ann", "..."]) door (above) reaches the same fused path when both a text and a vector rank_by are supplied; VectorClient.hybrid_search() is the typed entry point when you want direct control over embedding_slot and both legs at once.

Agent memory in three lines

For agent-memory workloads, Memory is a Mem0-style one-liner surface over the governed /memory/* verbs — governance (purpose + ACL) stays on by default:

from relata import Memory

with Memory("http://localhost:9090", purpose="agent-notes") as m:
    mem_id = m.add("Alice prefers dark mode")     # store
    hits = m.search("ui preferences", top_k=5)    # recall (confidence × recency × relevance)

    # ADR-145 retrieval-quality operators (#2674): bound recall by token
    # budget and confidence so one call can't blow an agent's context window.
    # `search_detailed` also surfaces the read-only `recall_cost_tokens`/
    # `cancelled` response fields.
    detailed = m.search_detailed("ui preferences", min_confidence=0.5, budget_tokens=200)

    m.forget(mem_id)                              # governed retract, not a hard delete

See examples/memory_quickstart.py. For async apps, AsyncMemory has the same surface: async with AsyncMemory(...) as m: await m.add(...); await m.search(...).

Agent-framework adapters

Drop-in, governed memory backends for the major agent frameworks — each maps the framework's memory/storage interface onto Relata, and (with one exception) none imports its framework (so they load with or without it installed):

Framework	Import	Shape
LangChain	`relata_adapters.langchain.RelataMemory`	`BaseMemory`
LlamaIndex	`relata_adapters.llamaindex.RelataMemory`	`BaseMemory`
CrewAI	`relata_adapters.crewai.RelataStorage`	`Storage`
AutoGen	`relata_adapters.autogen.RelataMemory`	`Memory` (async)
LangGraph*	`relata_langgraph.RelataCheckpointer` / `AsyncRelataCheckpointer`	`BaseCheckpointSaver`

* LangGraph is the exception: it gates checkpoint behavior behind isinstance(checkpointer, BaseCheckpointSaver), so relata_langgraph is a real subclass with a real (optional-extra) dependency on langgraph — see pip install "relata-sdk[langgraph]" and the LangGraph checkpointer guide.

from relata_adapters.langchain import RelataMemory

mem = RelataMemory(base_url="http://localhost:9090", purpose="agent")
# chain = ConversationChain(llm=..., memory=mem)

The adapters use the semantic-memory model (store + relevance recall), the same paradigm as Mem0. They wrap the Memory client, so governance stays on.

Core concepts

Every query must declare a purpose

Relata enforces a mandatory purpose on every query. The purpose must be a string registered in the tenant's PurposeRegistry (e.g. "analytics", "audit", "analysis"). Queries without a purpose are rejected at the wire.

Set a default on the client:

client = RelataClient("http://localhost:9090", purpose="analytics")

Or override per-call:

result = client.query("SELECT COUNT(*) FROM Person", purpose="audit")

Bearer token authentication

When the server is configured with RELATA_BEARER_TOKEN, pass the token to the client:

client = RelataClient(
    "https://relata.example.com",
    bearer_token="your-token",
    purpose="analytics",
)

What identity does my SDK client use by default?

Without a bearer token, every request runs as the api-user principal — a built-in identity with read access to all domain types and write access in open (lite) mode. Audit entries, quota charges, and ACL decisions all attribute to api-user.

MCP requests (McpClient) run as mcp-client — a different principal with separate audit and quota accounting.

For multi-tenant production: set a bearer token + tenant= on every request. See docs/src/end-users/defaults.md for the full default-behavior reference.

Fluent query builder

result = (
    client.select("Person")
    .where("nationality = 'IN'")
    .where("age > 25")
    .as_of("2025-01-01")          # bi-temporal point-in-time
    .with_provenance()             # include PROV-O columns
    .order_by("name ASC")
    .limit(20)
    .execute()
)

Relata SQL extensions

Extension	Example
Bi-temporal	`SELECT * FROM Person AS OF '2024-06-01'`
Provenance	`SELECT * FROM Person WITH PROVENANCE`
Graph traversal	`FROM PATHS_BETWEEN('person-123', 'org-456', max_hops => 4)`
Face search	`WHERE MATCH_FACE(probe_embedding, stored_embedding, 0.92)`
Identity lookup	`FROM LOOKUP_IDENTITY('+919876543210')`
Hybrid search	`SELECT *, HYBRID_SCORE('terror finance') AS score FROM Document`

QueryResult is iterable

result = client.query("SELECT * FROM Person LIMIT 10")

# Iterate over rows
for row in result:
    print(row["name"])

# Access by index
first = result.rows[0]

# Check if empty
if not result:
    print("No results")

# Row count
print(f"{len(result)} rows in {result.processing_time_ms} ms")

Async support

import asyncio
from relata import RelataClient

async def main():
    async with RelataClient("http://localhost:9090", purpose="analytics") as client:
        result = await client.aquery("SELECT * FROM Person LIMIT 5")
        for row in result:
            print(row)

asyncio.run(main())

Plain gRPC query with Arrow-IPC opt-in (#4090)

RelataClient.query_grpc runs a SQL query over the server's plain gRPC RelataQuery/Execute RPC (default port 50051, RELATA_GRPC_PORT) and opts into the negotiated Arrow-IPC row encoding (QueryRequest.wants_arrow_ipc_rows — server side landed in PR #4086). The server answers with whichever encoding it supports (arrow_ipc_rows or a rows_json fallback); query_grpc decodes either wire shape into the same QueryResult shape query() returns, so callers never need to know which one came back.

result = client.query_grpc("SELECT * FROM Person LIMIT 1000", purpose="analytics")
print(result.row_count, result.rows[0]["name"])

Requires the optional grpcio dependency: pip install relata-sdk[grpc] (or pip install grpcio directly). Unlike query_flight/query_arrow (which only need the undeclared-optional pyarrow), the rest of this SDK has zero gRPC dependency — Python previously had none at all, so grpcio is declared as a real, versioned extra (mirrors the langgraph extra) rather than an undeclared lazy import. arrow_ipc_rows decoding still needs pyarrow (already optional elsewhere in this SDK) — only reached when the server actually answers with that encoding. TLS is selected automatically for grpcs:///tls:// endpoints (grpc_endpoint="grpcs://host:port").

query_grpc_stream is the same opt-in over the server-streaming RelataQuery/ExecuteStream RPC: the server answers with a stream of RowBatch frames instead of one buffered QueryResponse, and each frame independently negotiates arrow_ipc_rows vs rows_json (same empty-means-fall-back contract, applied per frame). Every frame is collected into the same QueryResult shape, so a large result set needs no different consumption model than a small one.

result = client.query_grpc_stream("SELECT * FROM Person", purpose="analytics")
print(result.row_count)

Because RowBatch carries no row_count field, the returned row_count is the number of rows actually received (whereas query_grpc reports the server's own QueryResponse.row_count). aquery_grpc / aquery_grpc_stream are the async variants of both.

Error handling

from relata.exceptions import (
    PurposeError,   # missing or unregistered purpose
    QuotaError,     # per-principal cost quota exhausted
    AuthError,      # invalid or missing bearer token
    ConnectionError, # server unreachable
    RelataError,    # base class — catch all SDK errors
)

try:
    result = client.query("SELECT * FROM Person")
except PurposeError as exc:
    print(f"Fix: set purpose= on client or per query. {exc}")
except QuotaError as exc:
    print(f"Quota exhausted — reduce query scope. {exc}")
except AuthError as exc:
    print(f"Auth failed — check bearer_token. {exc}")
except ConnectionError as exc:
    print(f"Cannot reach server. {exc}")

API reference

`RelataClient`

RelataClient(
    base_url: str,
    *,
    bearer_token: str | None = None,
    purpose: str | None = None,
    timeout: float = 30.0,
    tenant: str | None = None,        # X-Relata-Tenant-Id (multi-tenant)
    acting_as: str | None = None,     # X-Acting-As (delegation)
    delegated_by: str | None = None,  # X-Delegated-By
    headers: dict[str, str] | None = None,  # arbitrary header overlay
    max_retries: int = 0,             # retry on connection errors
    retry_backoff_secs: float = 0.5,  # exponential backoff base
)

Method	Returns	Description
`query(sql, *, purpose)`	`QueryResult`	Execute a SQL query
`health()`	`HealthResponse`	Check node health
`status()`	`StatusResponse`	Node status + quota
`stats()`	`Stats`	Engine-wide counts (records/states/snapshot_rows/log_leaves/tokens)
`version()`	`VersionInfo`	Build info (version, commit, profile, schema_version)
`ready()`	`ReadyReport`	9-condition readiness report
`audit_count()`	`AuditCountResponse`	Audit log entry count + chain validity
`cluster_nodes()`	`list[ClusterNode]`	List cluster nodes
`select(*cols)`	`QueryBuilder`	Start a fluent query
`close()`	—	Close sync connections
`aquery(...)`	`QueryResult`	Async query
`ahealth()` ... `astats()` ... `aready()` etc.	—	Async mirrors of every method above
`aclose()`	—	Close async connections

v1.1 typed clients (`from_client(client)`)

Each module inherits the parent client's auth, tenant, and purpose context:

Module	Class	Surface
`relata.governance`	`GovernanceClient`	Rules, retention (holds + WORM), breakglass, alerts, DSAR
`relata.mcp`	`McpClient`	22 typed MCP tool wrappers + generic `call_tool`
`relata.a2a`	`A2AClient`	A2A tasks + LangGraph checkpoints + agent card
`relata.audit`	`AuditClient`	Audit entries (filtered/paginated) + signed receipts + PDF export
`relata.identity`	`IdentityClient`	Identity label/uncertainty + lookup tables + ERASE SUBJECT
`relata.objects`	`ObjectClient`	Typed upsert + batch + point-lookup (`get`) + delete via `/ingest?object_type=`
`relata.ingest`	`IngestClient`	Bulk NDJSON + CSV + media status
`relata.vectors`	`VectorClient`	KNN + hybrid search + similar-to (SQL-backed)
`relata.s3`	`S3Client`	boto3 / httpx / aiobotocore wrapper for the S3 protocol door
`relata.system`	`SystemClient`	LLM config + test + jobs status
`relata.streaming`	`StreamingClient`	NDJSON row streams + SSE consumers (watch/alerts) + Arrow IPC
`relata.tenants`	`TenantAdminClient`	Tenant CRUD + quota + sharing agreements + platform admin

Each has an async mirror (Async*).

Admin-listener reachability (admin_base_url, #2321): on a hardened server/cluster deployment, /admin/* and /platform/* are mounted only on the loopbound admin control-plane listener (RELATA_ADMIN_BIND, default 127.0.0.1:9091 — ADR-0261), never the main data-plane base_url. Pass admin_base_url= to RelataClient (inherited automatically by BackupClient/TenantAdminClient.from_client(...)) or directly to BackupClient/TenantAdminClient to reach those routes. Leave it unset (the default) on the free profile, where the admin listener isn't split from the data plane. A bare 404 from an admin/platform-only route with no error detail is rewritten into a hint pointing at this option instead of a confusing plain 404.

v1.1 transport hardening (#79)

RFC 7807 problem+json parsing — every error carries code, type_url, retryable, request_id.
Typed exceptions — ForbiddenError (403), NotFoundError (404), ConflictError (409), ValidationError (422), RateLimitedError (429, with retry_after).
Retry — RelataClient(..., max_retries=3, retry_backoff_secs=0.5).
X-Request-ID — auto-generated per request; caller-supplied IDs respected; server's response ID stamped on exceptions.

Custom object types

RelataDB is ontology-governed — unknown types are rejected on both ingest and read. Register custom types at runtime via POST /types (persisted across restart):

# Register a custom type
client._sync.post("/types", {
    "name": "AgentTask",
    "fields": [
        {"name": "task_id", "type": "string"},
        {"name": "status", "type": "string"},
    ]
})

# Now ingest + query work
client._sync.post("/ingest?object_type=AgentTask", {"task_id": "t-1", "status": "done"})

For ACL access in strict mode, grant via env var:

RELATA_ACL_GRANT=AgentTask:read+write

Unknown types return 400 on both ingest and read — fail-closed by design.

v1.1 QueryBuilder extensions (#76)

Method	Description
`.limit(n, after="cursor")`	Keyset pagination (`LIMIT n AFTER 'cursor'`)
`.since(cursor)`	Incremental reads (`WHERE system_from > cursor`)

The builder also validates identifiers (from_(), select()) and refuses dangerous tokens in where() (SQL-injection stopgap). For user-supplied values use .where_param("id = ?", value) — ? placeholders are replaced with safely-typed literals (str → single-quoted + escaped, int/float → numeric).

Agent-framework adapters

Framework	Import	Shape
LangChain	`relata_adapters.langchain.RelataMemory`	`BaseMemory`
LlamaIndex	`relata_adapters.llamaindex.RelataMemory`	`BaseMemory`
CrewAI	`relata_adapters.crewai.RelataStorage`	`Storage`
AutoGen v0.2	`relata_adapters.autogen.RelataMemory`	`Memory` (async)
LangGraph	`relata_langgraph.RelataCheckpointer` / `AsyncRelataCheckpointer`	`BaseCheckpointSaver`
AG2 (v0.4+)	`relata_adapters.ag2.RelataAG2Memory`	`MemoryProtocol`
Pydantic-AI	`relata_adapters.pydantic_ai.RelataMemoryBackend`	memory backend
smolagents	`relata_adapters.smolagents.RelataTool`	tool callable
Auto-detect	`relata_adapters.registry.get_memory_adapter()`	picks the right class

`QueryBuilder`

Method	Description
`.select(*cols)`	Columns or table name shorthand
`.from_(table)`	FROM table
`.where(condition)`	Add WHERE predicate (multiple calls → AND)
`.where_param(condition, *values)`	Safe WHERE with `?` placeholder interpolation (str/int/float)
`.as_of(timestamp)`	Bi-temporal AS OF
`.with_provenance()`	Append WITH PROVENANCE
`.purpose(p)`	Override purpose for this query
`.limit(n)`	LIMIT
`.offset(n)`	OFFSET
`.order_by(*cols)`	ORDER BY
`.paths_between(src, dst, max_hops)`	PATHS_BETWEEN graph operator
`.match_face(probe, candidate, threshold)`	MATCH_FACE operator
`.lookup_identity(value)`	LOOKUP_IDENTITY operator
`.hybrid_score(text)`	HYBRID_SCORE ranking
`.raw(sql)`	Raw SQL escape hatch
`.sql()`	Return assembled SQL without executing
`.execute()`	Execute (sync)
`.aexecute()`	Execute (async)

Response models

Model	Fields
`QueryResult`	`rows`, `query_id`, `processing_time_ms`, `elapsed_ms` (alias, removed v1.6.0), `row_count`
`HealthResponse`	`status`, `profile`, `node_id`
`StatusResponse`	`profile`, `role`, `query_quota`
`AuditCountResponse`	`entries`, `chain_valid`
`ClusterNode`	`node_id`, `role`, `url`
`VersionInfo`	`version`, `commit`, `profile`, `schema_version`, `features`
`Stats`	`records`, `states`, `snapshot_rows`, `log_leaves`, `tokens`
`ReadyReport`	`is_ready`, `status`, `reason`, `detail`

Examples

See the examples/ directory:

File	Demonstrates
`basic_query.py`	Getting started, health check, quota check
`analytics.py`	Full analytics workflow: identity → cases → graph → provenance
`face_search.py`	MATCH_FACE operator, threshold comparison, CCTV temporal search
`hybrid_search.py`	`VectorClient.hybrid_search()` — embed, ingest, BM25-only vs. vector-only vs. fused (RRF) search
`graph_traversal.py`	PATHS_BETWEEN, temporal network shift, hub detection
`audit.py`	Chain verification, audit summary, anomaly detection

Deployment profiles

lite was a silent legacy alias for free (ADR-204); it is now rejected outright — startup fails (FATAL) if RELATA_PROFILE=lite is set. Use free instead.

Relata ships three deployment profiles. The SDK works identically across all three:

Profile	Use case	Binary
`free`	Embedded / single-process	`RELATA_PROFILE=free relata serve`
`server`	Single-node production	`RELATA_PROFILE=server relata serve`
`cluster`	Multi-node distributed	`RELATA_PROFILE=cluster relata serve`

Testing with an ephemeral server

tests/conftest_ephemeral.py provides a session-scoped pytest fixture that starts a real relata serve process on a random port and tears it down at the end of the test session.

# In any test file:
from tests.conftest_ephemeral import relata_server

def test_health(relata_server):
    import httpx
    r = httpx.get(f"{relata_server['base_url']}/health",
                  headers={"Authorization": f"Bearer {relata_server['token']}"})
    assert r.status_code == 200

Set RELATA_BIN to the binary path (default: relata on $PATH) and RELATA_TEST_TOKEN to override the bearer token (default: relata-test).

License

AGPL-3.0-only. See the root LICENSE file.

Navigation Menu

Uh oh!

Relata Python SDK

Requirements

Installation

Quick start

Flagship retrieval surface — `namespace().query()` (#1991)

Cypher

GraphQL

Hybrid search — `VectorClient.hybrid_search()`

Agent memory in three lines

Agent-framework adapters

Core concepts

Every query must declare a purpose

Bearer token authentication

Fluent query builder

Relata SQL extensions

QueryResult is iterable

Async support

Plain gRPC query with Arrow-IPC opt-in (#4090)

Error handling

API reference

`RelataClient`

v1.1 typed clients (`from_client(client)`)

v1.1 transport hardening (#79)

Custom object types

v1.1 QueryBuilder extensions (#76)

Agent-framework adapters

`QueryBuilder`

Response models

Examples

Deployment profiles

Testing with an ephemeral server

License

Uh oh!

FilesExpand file tree

README.md

Latest commit

History

README.md

File metadata and controls

Relata Python SDK

Requirements

Installation

Quick start

Flagship retrieval surface — namespace().query() (#1991)

Cypher

GraphQL

Hybrid search — VectorClient.hybrid_search()

Agent memory in three lines

Agent-framework adapters

Core concepts

Every query must declare a purpose

Bearer token authentication

Fluent query builder

Relata SQL extensions

QueryResult is iterable

Async support

Plain gRPC query with Arrow-IPC opt-in (#4090)

Error handling

API reference

RelataClient

v1.1 typed clients (from_client(client))

v1.1 transport hardening (#79)

Custom object types

v1.1 QueryBuilder extensions (#76)

Agent-framework adapters

QueryBuilder

Response models

Examples

Deployment profiles

Testing with an ephemeral server

License

Flagship retrieval surface — `namespace().query()` (#1991)

Hybrid search — `VectorClient.hybrid_search()`

`RelataClient`

v1.1 typed clients (`from_client(client)`)

`QueryBuilder`