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):

* 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

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
)

v1.1 typed clients (from_client(client))

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

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)

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

QueryBuilder

Response models

Examples

See the examples/ directory:

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:

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.