Python SDK.

olbrain-python-sdk has the same two clients as the JavaScript SDK: AgentClient chats with conversational agents, and Olbrain triggers workflow agents, controls their runs, resolves approvals, and drives research agents. Both are synchronous and built on requests.

Install

pip install -U olbrain-python-sdk

This page describes version 0.5.0. Olbrain needs 0.5.0 or later; versions before 0.4.0 called a host that no longer exists, so upgrade. It needs Python 3.7+, and you import it as olbrain. The package ships a complete offline API reference at site-packages/olbrain/AGENTS.md, written for coding assistants.

Send a message

import os
from olbrain import AgentClient

with AgentClient(agent_id="your-agent-id", api_key=os.environ["OLBRAIN_API_KEY"]) as client:
    session_id = client.create_session(title="Support chat", user_id="u_42")
    reply = client.send_and_wait(session_id, "Hello!")
    if reply.success:
        print(reply.text)
    else:
        print("agent error:", reply.error)

Chat calls are POST /api/agents/webhook/{agent_id}; session reads are GET /api/agents/{agent_id}/sessions/{session_id} and …/messages. Every call sends Authorization: Bearer <api_key>. The client is a context manager; leaving the with block calls close(). Check reply.success as well as catching errors — a 200 can still carry success=False and an error.

Methods

Listen for messages

def on_message(msg: dict):
    print(f"{msg.get('role')}: {msg['content']}")

client = AgentClient(agent_id="…", api_key=os.environ["OLBRAIN_API_KEY"])
sid = client.create_session(on_message=on_message, title="Listening chat")
client.send(sid, "Tell me a story")
client.run()  # blocks; Ctrl+C to stop

on_message gets each new message as a raw dict (role, content, …). It runs on the listener thread, so keep it fast and don't raise from it.

Responses

class ChatResponse:
    text: str
    session_id: str
    success: bool = True
    # prompt_tokens, completion_tokens, total_tokens, cost
    token_usage: Optional[TokenUsage] = None
    model_used: Optional[str] = None
    response_time_ms: Optional[int] = None
    metadata: Optional[dict] = None
    error: Optional[str] = None

token_usage is None when the server sends none, so check it before you read it.

AgentClient errors

Every SDK error subclasses OlbrainError, which has message, error_code, and details.

Workflow runs and approvals

import os
from olbrain import Olbrain

olbrain = Olbrain(api_key=os.environ["OLBRAIN_API_KEY"])
run = olbrain.workflows.run("your-agent-id", payload={"url": "https://example.com"})
print(olbrain.runs.get("your-agent-id", run["run_id"]))

Olbrain(api_key=None, get_id_token=None, base_url=None, timeout=30). Give exactly one of api_key (must start ak_) or get_id_token (a callable that returns a signed-in user's token, called before every request), or it raises ValidationError. timeout is in seconds. It is the same API as the JavaScript Olbrain client with Python names; every method takes the agent id first and returns the API's JSON as dicts and lists.

There is no helper for waiting on a run: poll runs.get until its status is final, resolving approvals as they appear. Each run is a real, billed run.

Research agents

olbrain.research has the same calls as the JavaScript research API, in snake_case (sessions.stream_message, plans.approve_by_id, reports.get, review.stream_apply, …). It authenticates as a signed-in user, so create the client with get_id_token; an Access Key gets NotFoundError. Streams are plain generators — stop iterating to cancel. The JavaScript live.* subscriptions are not in Python; poll sessions.get or reports.get instead.

olbrain = Olbrain(get_id_token=lambda: user_id_token)
for ev in olbrain.research.sessions.stream_message(agent_id, sid, "Size the Indian EV charger market"):
    if ev.get("type") == "text_delta":
        print(ev["text"], end="", flush=True)

Olbrain errors

BillingError and NotFoundError subclass ApiError, so catch them first. AgentClient logs through logging.getLogger("olbrain"); set it to DEBUG to see requests.