Relata Python SDK

Python SDK for the Relata enterprise-grade data engine.

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

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.

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)
    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 none imports its framework (so they load with or without it installed):

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

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*).

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 is kept as a silent legacy alias for free (ADR-204).

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.