brain-db-sdk (Python)

The Python client for the Brain memory database. Speaks Brain's BRN0 wire protocol directly; no dependency on Brain's internal code.

Status: typed-graph verbs (Phase 6). The wire codec (Phase 1) is verified byte-for-byte against the shared conformance corpus (all 86 .bin/.json cases re-encode to identical bytes). On top of it, a synchronous connection layer (a transport over a socket, a Connection running the handshake HELLO → WELCOME → AUTH → AUTH_OK and request/response, a BrainClient holding the negotiated session), the three v1 verbs with ergonomic builders (encode(), recall() streaming to EOS / recall_frames(), forget()), a with_retry helper + RetryPolicy (exponential backoff, server retry_after_ms), and the typed-graph verbs: create_entity(), create_statement(), create_relation(), upload_schema(), and materialize_procedural(); and the space/session registry verbs: create_space() / list_spaces() / delete_space() and create_session() / list_sessions() / delete_session() for managing the per-request isolation unit (space) and conversation groupings (session) that a trusted principal targets via act_as. BrainClient is built on a MuxConnection: a background reader thread demultiplexes responses by stream_id, so every verb is concurrency-safe and many requests run in flight at once over one connection from multiple threads. A Pool opens a fixed set of such connections and hands them out round-robin for socket-level parallelism. Transparent reconnect is a later phase.

The full build plan is in the repo-root ../PLAN.md; the layered architecture is in ../ARCHITECTURE.md.

from brain_db_sdk import (
    BrainClient, EncodeBuilder, RecallBuilder, ForgetBuilder, RetryPolicy, with_retry,
)

with BrainClient.connect("127.0.0.1", 7878) as client:
    stored = client.encode(EncodeBuilder("the user prefers dark mode").build())
    answer = client.recall(RecallBuilder("ui preferences").limit(5).build())
    # answer.answer_kind is Single/Set (fact answer.values) or Episodic
    # (memory answer.results); NoSubject/NoMemory means "don't know".
    # Ride out transient ResourceExhausted/Unavailable; the stable request_id
    # makes the resend idempotent.
    req = ForgetBuilder(stored.memory_id).build()
    with_retry(lambda: client.forget(req), RetryPolicy())

Develop + verify against the corpus:

python3 -m venv .venv && .venv/bin/pip install cbor2 pytest
PYTHONPATH=src .venv/bin/python -m pytest tests/ -q

The integration suites run against a real server when BRAIN_SDK_IT_DATA is set, and skip otherwise; scripts/it-server.sh up boots one and prints the vars. Set BRAIN_SDK_IT_REQUIRED=1 wherever a server is meant to be reachable so a misconfigured one fails instead of reading as green.

Layout (folder-per-concern, mirroring the reference Rust SDK):

src/brain_db_sdk/
  wire/        frame + CBOR codec, opcodes, typed payloads (corpus-verified)
  errors.py    client error taxonomy (BrainError + subclasses)
  transport.py sync read/write of whole frames over a socket
  connection.py handshake + one-at-a-time request/response
  client.py    high-level BrainClient: connect, handshake, encode/recall/forget
  verbs.py     ergonomic EncodeBuilder / RecallBuilder / ForgetBuilder
  retry.py     RetryPolicy + with_retry (exponential backoff, server retry_after)
  aio/         the asyncio tier: AsyncBrainClient, AsyncPool, async transport+mux

asyncio tier

brain_db_sdk.aio.AsyncBrainClient is the asyncio twin of BrainClient: the same verbs with the same signatures, every one a coroutine. It speaks the same BRN0 frames over the same multiplexed connection — a request in flight suspends its task instead of blocking a thread, so one connection carries thousands of concurrent requests on a single OS thread. No new dependency: it is built on asyncio from the standard library.

import asyncio
from brain_db_sdk import Auth, EncodeBuilder, RecallBuilder
from brain_db_sdk.aio import AsyncBrainClient

async def main() -> None:
    async with await AsyncBrainClient.connect(
        "127.0.0.1", 18080, Auth.token(b"brain_...")
    ) as client:
        # Independent writes cost one round trip, not three.
        await asyncio.gather(
            client.encode(EncodeBuilder("Ada works at Arc Labs").build()),
            client.encode(EncodeBuilder("Arc Labs builds Brain").build()),
            client.encode(EncodeBuilder("Brain is a memory database").build()),
        )
        answer = await client.recall(RecallBuilder("where does Ada work?").build())
        for memory in answer.memories:
            print(memory.text)

asyncio.run(main())

AsyncPool, AsyncSubscription (drained with async for) and an awaiting with_retry come with it. Everything below the transport is shared rather than duplicated: the frame codec, the CBOR payload types, the request builders, the error taxonomy, the retry policy and its backoff curve, and the Auth / ClientConfig / ConnectionInfo configuration types are the same objects both tiers use. tests/aio_parity compares the two surfaces by reflection and fails the build if a verb, a signature or a default ever drifts.

Two things worth knowing:

• Pick one tier per program, not per call. The sync client blocks the event loop from inside a coroutine; wrapping the async one in asyncio.run from sync code opens and tears down a loop per call and raises outright if a loop is already running (FastAPI, Jupyter).
• A connection belongs to the loop it was opened on. That binding is checked on every call rather than left to chance, because a connection driven from a second loop does not raise — it waits forever on a future nothing will complete.

HTTP tier

For hosted Brain (the Arc cloud gateway) or a self-hosted brain-edge edge, the package also ships BrainHttpClient — a JSON-over-HTTP client with the same verb surface, field names, and error shape as the Rust and TypeScript SDKs (the canonical contract is ../HTTP_CONTRACT.md). Use it when you talk to Brain through an HTTP edge and authenticate with an API key; use the wire BrainClient above for a direct socket, streaming, transactions, and typed-graph management. It has no third-party dependency — pure urllib.

from brain_db_sdk import BrainHttpClient

brain = BrainHttpClient(api_key, base_url="https://api.arc-labs.ai")
stored = brain.encode(text="the kettle whistled")
answer = brain.recall(query="what whistled?", max_results=3)
who = brain.whoami()  # namespace + space_id + permissions

Dependencies: cbor2 (runtime), pytest + pytest-asyncio (dev). CRC32C is pure Python; the asyncio tier adds no runtime dependency.

The live HTTP edge smoke tests cover identity, capabilities, and a memory encode/recall/list/forget lifecycle across all SDKs. Run them from the repo root with BRAIN_SDK_IT_HTTP=https://edge.example BRAIN_SDK_IT_HTTP_KEY=brain_… scripts/edge-it.sh. License: Apache-2.0.