Python SDK

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

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)

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

RelataClient(
    base_url: str,
    *,
    bearer_token: str | None = None,
    purpose: str | None = None,
    timeout: float = 30.0,
    tenant: str | None = None,        # X-Organization-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,
    max_retries: int = 0,
    retry_backoff_secs: float = 0.5,
)

client = RelataClient("http://localhost:9090", purpose="analytics")
result = client.query("SELECT COUNT(*) FROM Person", purpose="audit")  # override

client = RelataClient(
    "https://relata.example.com",
    bearer_token="your-token",   # required when server has RELATA_BEARER_TOKEN set
    purpose="analytics",
)

client = RelataClient(
    url,
    bearer_token=token,
    tenant="org-acme",           # X-Organization-Id
    acting_as="user-bob",        # X-Acting-As
    delegated_by="user-alice",   # X-Delegated-By
)

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

from relata import RelataClient, McpClient, GovernanceClient
 
with RelataClient(url, bearer_token=token, purpose="compliance") as client:
    mcp    = McpClient.from_client(client)
    gov    = GovernanceClient.from_client(client)
 
    profile = mcp.get_entity_profile("person-123", purpose="compliance")
    holds   = gov.list_legal_holds()
    sigma   = gov.import_sigma(open("rules.yml").read())

result = client.query("SELECT * FROM Person LIMIT 10")
for row in result:        # iterate
    print(row["name"])
first = result.rows[0]    # index
if not result:            # truthy when non-empty
    print("empty")
print(f"{len(result)} rows in {result.elapsed_ms} ms")

from relata.exceptions import (
    PurposeError,     # missing or unregistered purpose (HTTP 400)
    QuotaError,       # per-principal cost quota exhausted (HTTP 429)
    AuthError,        # invalid or missing bearer token (HTTP 401)
    ForbiddenError,   # Cedar ACL denied (HTTP 403)
    NotFoundError,    # (HTTP 404)
    ConflictError,    # (HTTP 409)
    ValidationError,  # (HTTP 422)
    RateLimitedError, # (HTTP 429 with retry_after)
    ConnectionError,  # server unreachable
    ServerError,      # 5xx
    RelataError,      # base class — catch-all
)
 
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}")

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")
        async for row in result:
            print(row)
 
asyncio.run(main())

client._sync.post("/types", {
    "name": "AgentTask",
    "fields": [
        {"name": "task_id", "type": "string"},
        {"name": "status", "type": "string"},
    ],
})
client._sync.post("/ingest?object_type=AgentTask",
                  {"task_id": "t-1", "status": "done"})

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