# Python SDK (/brain/docs/sdks/python)

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

## Install [#install]

```sh
python -m pip install aex-brain==2.0.0
```

## Declare runtime tools [#declare-runtime-tools]

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

```python
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 client [#create-a-client]

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

```python
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 message [#create-a-session-and-send-a-message]

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

```python
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 events [#stream-events]

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

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

<Callout type="info" title="Closing the observer stream">
Calling `stream.close()` only closes the client's HTTP connection. The session remains open on the server, and any running turn continues unless you call `brain.interrupt(session["id"])`.
</Callout>

## Reconnect with cursors [#reconnect-with-cursors]

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:

```python
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 timeouts [#handle-errors-and-timeouts]

Non-2xx HTTP responses raise `BrainError`:

```python
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](/brain/docs/reference/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 reference [#methods-reference]

| Method | Signature | Returns | Description |
| --- | --- | --- | --- |
| `create` | `create(config: CreateSession)` | `Session` | Creates a session (`POST /sessions`). |
| `get` | `get(session_id: str)` | `Session` | Fetches session metadata and status (`GET /sessions/{id}`). |
| `list` | `list(after: str = "", limit: int = 100)` | `dict[str, Any]` | Returns `{"sessions": [...]}` in ascending ID order (`GET /sessions`). |
| `send` | `send(session_id: str, event: Input)` | `Event` | Queues a `user.message` (or sends `user.interrupt`) (`POST /sessions/{id}/events`). |
| `interrupt` | `interrupt(session_id: str)` | `None` | Sends `{"type": "user.interrupt"}` to stop the active turn (`POST /sessions/{id}/events`). |
| `history` | `history(session_id: str, after: str = "", limit: int = 100)` | `dict[str, Any]` | Returns `{"events": [...], "next_cursor": "..."}` (`GET /sessions/{id}/events`). |
| `observe` | `observe(session_id: str, after: str = "")` | `Iterator[Event]` | Yields retained and live events over SSE (`GET /sessions/{id}/events`). |
| `resume` | `resume(session_id: str)` | `Session` | Moves an `interrupted` or `failed` session back to `idle` (`POST /sessions/{id}/resume`). |
| `close` | `close(session_id: str)` | `None` | Permanently closes a session (`POST /sessions/{id}/close`). |
| `flush` | `flush(session_id: str)` | `SyncStatus` | Uploads local state to external storage immediately (`POST /sessions/{id}/flush`). |
| `sync` | `sync(session_id: str)` | `SyncStatus` | Reads background checkpoint sync status (`GET /sessions/{id}/sync`). |
| `restore` | `restore(session_id: str, checkpoint: str)` | `Session` | Replaces local session state with a specific external checkpoint (`POST /sessions/{id}/restore`). |
| `catalog` | `catalog()` | `dict[str, Any]` | Returns `{"tools": [...]}` available to the caller's tenant (`GET /catalog`). |