Python SDK

Brain is a synchronous Python 3.11+ client built entirely on the Python standard library (urllib.request, json). It requires no third-party packages.

InstallCopy Anchor Link

python -m pip install aex-brain==2.0.0

Declare runtime toolsCopy Anchor Link

Keep your Python process alive while the session uses its functions. The SDK owns a reverse SSE connection and sends results over HTTP.

import os
from brain import Brain, RuntimeTool

client = Brain("http://127.0.0.1:8080", os.environ["BRAIN_API_KEY"])

def lookup(arguments, context):
    if context.cancelled.is_set():
        raise RuntimeError("cancelled")
    context.progress({"stage": "reading"})
    return {"value": "application data for " + context.scope["tenant_id"]}

tools = [RuntimeTool("lookup", "Read application data", {"type": "object"}, lookup)]
session = client.create({"harness": "pi", "config": {
    "model": "openai/gpt-4.1-mini", "tools": []
}}, runtime_tools=tools)
client.send(session["id"], {"type": "user.message", "text": "Use runtime_lookup."})
stream = client.observe(session["id"])
try:
    for event in stream:
        print(event)
        if event["type"] == "session.status" and event["data"]["status"] == "idle":
            break
finally:
    stream.close()
    client.close(session["id"])
    client.dispose()

Return a JSON value to populate structured_content; raise an exception for a tool error. Functions run on bounded callback threads and must cooperate with context.cancelled. context.scope is authenticated host identity, separate from arguments. Function source and captured credentials are never registered with Brain.

client.dispose() closes callback connections without closing sessions. Reattach matching descriptors after restart with client.rebind(session_id, tools), then explicitly resume and send a new input. A disconnected callback returns an unavailable or unknown-outcome tool error according to whether dispatch occurred; the SDK never resends an action.

The distribution is named aex-brain; the import package is brain. Wheels and source distributions are published on PyPI and include generated types and the PEP 561 py.typed marker. No checkout or PYTHONPATH setting is needed. Pin the client and matching server commit from the release workflow.

Create a clientCopy Anchor Link

Import Brain from brain and pass the server base URL, your bearer token, and an optional socket timeout in seconds (default 30):

import os
from brain import Brain

brain = Brain("http://127.0.0.1:8080", os.environ["BRAIN_API_KEY"], timeout=30)

Generated TypedDict definitions for request and response payloads (Session, SessionConfig, CreateSession, Input, Event, EventList, SessionList, Tool, ToolCatalog, Call, Content, Result, ToolResult, Checkpoint, Restore, Status, SyncStatus, ErrorDetail, ErrorResponse) live in brain.types.

Create a session and send a messageCopy Anchor Link

Call create() with a harness name and the session configuration: model, tools, and optional instructions and reasoning_effort. Then call send() to queue a user.message:

session = brain.create({
    "harness": "pi",
    "config": {
        "model": "openai/gpt-4.1-mini",
        "instructions": "Answer in two sentences or fewer.",
        "reasoning_effort": "low",
        "tools": [],
    },
})

accepted = brain.send(session["id"], {
    "type": "user.message",
    "text": "Explain append-only storage.",
    "key": "message-1",
})

send() returns the recorded user.message event once Brain writes it to disk (202 Accepted). Keep accepted["id"] to correlate events emitted during that turn.

Stream eventsCopy Anchor Link

Call observe(session_id) to receive a synchronous generator over SSE frames. Always close the generator in a finally block (or with contextlib.closing) when you break out of the loop early so the underlying HTTP socket closes cleanly.

Run this script while Brain is running (see examples/session.py for the CLI version):

import os
from brain import Brain

brain = Brain("http://127.0.0.1:8080", os.environ["BRAIN_API_KEY"])
session = brain.create({
    "harness": "pi",
    "config": {"model": "openai/gpt-4.1-mini", "tools": []},
})
accepted = brain.send(session["id"], {
    "type": "user.message",
    "text": "Explain append-only storage.",
    "key": "message-1",
})

stream = brain.observe(session["id"])
try:
    for event in stream:
        if event["type"] == "assistant.delta":
            print(event["data"]["text"], end="", flush=True)
        if event["type"] == "assistant.message":
            print(event["data"]["text"])
        if event.get("input_id") == accepted["id"]:
            if event["type"] == "session.error":
                raise RuntimeError(event["data"]["message"])
            if (event["type"] == "session.status"
                    and event["data"]["status"] in {"idle", "failed", "interrupted", "closed"}):
                break
finally:
    stream.close()

Reconnect with cursorsCopy Anchor Link

Every retained event carries an opaque cursor string. Save the latest non-empty event.get("cursor") and pass it as after to observe() to resume streaming from that position:

cursor = ""
stream = brain.observe(session["id"], after=cursor)
try:
    for event in stream:
        if event.get("cursor"):
            cursor = event["cursor"]
        if event["type"] == "session.status" and event["data"]["status"] == "idle":
            break
finally:
    stream.close()

Live-only events (assistant.delta and tool.progress) do not include a cursor. To read retained events in batches without opening an SSE stream, call brain.history(session["id"], after=cursor, limit=100).

Handle errors and timeoutsCopy Anchor Link

Non-2xx HTTP responses raise BrainError:

from brain import Brain, BrainError

try:
    brain.send(session["id"], {"type": "user.message", "text": "Continue."})
except BrainError as err:
    print(err.status, err.code, err, err.request_id)
    raise

• BrainError attributes: status (HTTP status int), code (machine-readable error code such as "conflict" or "overloaded", falling back to "http_error"), request_id (str | None), and the error message via str(err).
• Tool failures vs. HTTP errors: Failed or denied tool calls do not raise BrainError; they appear in the event stream as tool.result events with is_error: True. Model provider or harness failures appear as session.error events. See Errors.
• Timeouts: Brain(url, token, timeout=30) sets the socket read/connect timeout in seconds for all calls on that client instance, including flush(), restore(), and individual socket reads in observe() (where keepalives arrive roughly every 15 seconds). Increase timeout (for example, timeout=60) when calling flush() or restore() on large checkpoints. observe() raises ValueError if a single SSE event exceeds 4 MiB or the stream closes mid-frame.
• No automatic retries or redirects: The client installs a custom HTTPRedirectHandler that rejects all HTTP redirects and never retries failed requests automatically.

Methods referenceCopy Anchor Link