subako

The async Python client for the Subako API: one client with a namespace per resource, typed errors, cursor pagination, a session connection that reconnects on its own, and a tool client that serves this process's functions to a session.

Requires Python 3.12 or newer. The package is async only and in beta.

uv add subako

pip install subako

Getting started

The backend holds the API key, creates the session, and hands the frontend the session token that comes back with it:

# backend
import os
from uuid import UUID

from subako import (
    CreatedSessionBody,
    CreatedSessionReceiptBody,
    CreateSessionBody,
    SessionTokenBody,
    SubakoClient,
)

subako = SubakoClient(base_url="https://api.kikuvi.com", api_key=os.environ["SUBAKO_API_KEY"])


async def open_session(agent_id: UUID) -> None:
    created = await subako.sessions.create(CreateSessionBody(agent_id=agent_id, display_name="support"))
    hand_to_frontend(session_id=created.id, token=await token_for(created))


# A create the SDK had to retry answers with a receipt instead: the session is
# there, its token is never sent twice, so mint a fresh one. See "Receipts".
async def token_for(created: CreatedSessionBody | CreatedSessionReceiptBody) -> str:
    if isinstance(created, CreatedSessionBody):
        return created.session_token
    minted = await subako.sessions.mint_token(created.id)
    if isinstance(minted, SessionTokenBody):
        return minted.session_token
    raise RuntimeError(f"no token in hand for session {created.id}")

The frontend holds only session tokens, one per session, and drives the session from there. A session token goes to SubakoClient as api_key:

# frontend
from uuid import UUID

from pydantic import BaseModel

from subako import MessageAssistantEvent, RunCompletedEvent, SubakoClient, TextContentBlock, Tool, ToolCall


class ContactFormArgs(BaseModel):
    text: str


async def open_contact_form(args: ContactFormArgs, call: ToolCall) -> str:
    ...
    return f"Navigated to /contact with {len(args.text)} characters."


async def talk(session_id: UUID, token: str) -> None:
    async with SubakoClient(base_url="https://api.kikuvi.com", api_key=token) as subako:
        # One connection follows the log; the client on it serves the tools this process offers.
        async with subako.sessions.connect(session_id) as session:
            web = session.create_tool_client("web")
            web.add_tool(
                "open_contact_form",
                Tool(
                    description="Navigate to the contact page and prefill the form.",
                    parameters=ContactFormArgs,
                    execute=open_contact_form,
                ),
            )
            await web.ready()

            await session.send("I want to send an inquiry.")

            async for event in session:
                match event:
                    case MessageAssistantEvent():
                        for block in event.message.content:
                            if isinstance(block, TextContentBlock):
                                print(block.data.text)
                    case RunCompletedEvent():
                        break
                    case _:
                        pass

A backend that serves the tools itself needs no token at all: the client that created the session connects to it, and connect and create_tool_client work there exactly as they do above.

Authentication

Exactly one credential per SubakoClient, sent as Authorization: Bearer.

Option	Secret	Reaches
api_key	sbk_ak_...	One workspace, with the permissions the key was minted with.
access_token	sbk_at_...	Whatever the signed-in user may reach; workspace routes need a workspace.
api_key	sbk_st_...	A session token: one session's own routes and nothing else.

A session token implies its session and its workspace, so a client built on one needs no workspace_id; one passed anyway is sent, and the server refuses a header naming another workspace. Your backend hands it out from session_token on sessions.create, or mints a fresh one with sessions.mint_token; tokens already issued stay live. Both answer with a receipt rather than a secret when the attempt was a replay, which "Receipts" covers.

A browser reaches the session routes cross-origin, and an agent answers no origin until it is told to. Name the page's origin once, from the backend:

await subako.agents.set_security(agent_id, PutAgentSecurityBody(allowed_origins=["https://app.example.com"]))

An entry is exactly what a browser puts in Origin, scheme, host, and an optional port, with no trailing slash and no path. set_security replaces the whole document, so an empty body returns the agent to refusing every browser.

Either option takes a string or a function, sync or async:

subako = SubakoClient(
    base_url="https://api.kikuvi.com", access_token=auth.current_access_token, workspace_id=workspace_id
)

A function is called when a token is first needed, not on every request. What it answers with is cached and carried by every request after it, the event stream included, and requests that start together share one call rather than each making their own. It is called again only when the server answers 401: the refused request then goes once more with the fresh token, under the same Idempotency-Key and without spending a retry, and a second 401 raises UnauthorizedError. So a caller can fetch a token from wherever it lives and leave it to the SDK to decide when that is worth doing again. A string credential has nothing to refresh, so its 401 raises at once, as does an upload whose archive is an AsyncIterable[bytes], which cannot be sent twice; a function that raises surfaces its own error, with the next request calling it again.

A missing, empty, or duplicated credential, and a missing base_url, raise TypeError from the constructor. Close the client with async with or await subako.aclose(); an httpx.AsyncClient passed in as http_client stays open.

Workspaces

workspace_id rides every request as Kikuvi-Workspace. A user credential needs it for workspace-scoped routes; an API key implies its own workspace and refuses a header naming another, and so does a session token. Any call may name a different workspace:

await subako.agents.list(ListAgentsQuery(limit=50), RequestOptions(workspace_id=other))

Calling convention

Path ids are positional, in url order, as a str or the UUID a response carries. The request body is the generated model for the operation, and listing parameters are the generated <Operation>Query model, both under the field names the API itself uses. A trailing RequestOptions carries timeout, max_retries, headers, workspace_id, and idempotency_key.

await subako.agents.get(agent_id)
await subako.agents.publish_version(agent_id, config)
await subako.sessions.resolve_approval(session_id, call_id, decision, RequestOptions(timeout=5.0))

A body is sent with the fields you set: an unset field stays out, and an explicit None travels as null, which the PATCH bodies tell apart. Methods return the parsed response model, or None where the API answers 204. Failures raise.

Pagination

Listings return a Page. Iterate it to walk every item from that page onward, or follow the cursor by hand.

async for agent in await subako.agents.list(ListAgentsQuery(limit=100)):
    print(agent.name)

page = await subako.sessions.list_events(session_id, ListSessionEventsQuery(after_seq=41))
print(page.items, page.next_cursor)
following = await page.next_page()  # Page | None

Every listing answers newest first, and order=OrderParam.asc walks the other way. The cursor follows the direction of travel, so a walk continues past its last item whichever way it runs.

oldest = await subako.agents.list(ListAgentsQuery(order=OrderParam.asc, limit=100))

Cursors are plain values: a UUID for most listings, an int seq for the session event log. The event log is the one listing whose cursor changes name with the direction, before_seq descending and after_seq ascending, and the Page follows whichever the order names. Both bounds are exclusive, so a window is ListSessionEventsQuery(after_seq=..., before_seq=...), and the bound you passed stays put while the walk moves the other one.

The session connection

connect opens one connection to a session and keeps it: every event that has arrived, the readings derived from them, and the clients attached to it.

session = subako.sessions.connect(session_id, history=True)

await session.ready()  # the history is read and the stream is open

session.status  # ConnectionStatus: connecting | open | reconnecting | closed | failed
session.error  # the failure, set when the status is failed
session.last_seq  # int | None
session.events  # every event seen, in seq order
session.transcript  # the messages of the log, each call joined to its result and approval
session.approvals  # the approvals no decision has settled
session.is_running  # a run queued or under way

async for event in session:
    ...  # the events so far, then live ones; `break` stops iterating, not the connection

session.close()  # ends the stream; the clients on it are deleted
session.reconnect()  # only after a failure: opens the stream again on this same connection

async with does the first and the last of these: it waits for ready() on entry and on exit closes the connection and waits for the stream and the clients' deletes to finish.

The connection is also how the session is driven, so a program holds nothing else:

outcome = await session.send("book me a room")  # posts a user turn; `outcome.run_id` names the run
await session.cancel()  # stops the active run at its next checkpoint
await session.resolve_approval(call_id, ApprovalDecisionBody.deny, "not this one")  # allow | deny

transcript, approvals, and is_running are derived from events when they are first read, and again only when new events have arrived since.

events is this client's copy of the session's event log, which is the source of truth: the wire bodies in seq order, parsed into their Pydantic models. SessionEvent is the union of every event, discriminated by type, and each carries seq. Match on the classes; keep a case _: arm so a type this version has not heard of does not fail the match under pyright's strict mode.

Event	Meaning
MessageUserEvent, MessageAssistantEvent	A turn; message.content holds text, tool call, and thinking blocks.
ToolCallEvent, ToolResultEvent	A tool the model called, and what it answered.
RunQueuedEvent, RunStartedEvent, RunCompletedEvent, RunFailedEvent, RunCancelledEvent	The life of one run, each carrying its run_id.
ApprovalRequestedEvent, ApprovalResolvedEvent	A held tool call and the decision on it.

The client events — ClientRegisteredEvent, ClientToolsUpdatedEvent, ClientLeftEvent, ClientToolDispatchedEvent, ClientToolAckedEvent, and ClientToolFailedEvent — and SessionCreatedEvent, ErrorEvent, and CustomEvent are in the union too.

The three readings above are these pure functions, which are exported too, for a log that is not a connection's. transcript answers with the user and assistant messages in seq order, each assistant one carrying its tool_call blocks joined to the tool_result and the approval that followed them; pending_approvals answers with the approval_requested events no decision has settled; is_running says whether a run is queued or under way. All three take a Sequence[SessionEvent] and keep no state of their own:

from subako import is_running, pending_approvals, transcript

messages = transcript(events)
waiting = pending_approvals(events)
busy = is_running(events)

Option	What it does
history	Reads the log to its end first, oldest first; the stream then starts at the last seq.
after_seq	Where the log is picked up: the stream's start, or the history read's when history is set. Omit for the live head, or pass 0 for the whole log.
max_retries	Retries of the first request that opens the stream, as for any call; a reconnect gets one try.
max_reconnects	Consecutive reconnects that carry no event, tolerated; defaults to 5.
headers, workspace_id	Carried by the stream and by every call the connection makes.

A connection with neither history nor after_seq starts at the live head: it carries what happens from now on, and nothing that happened before. history=True is what a chat that draws the conversation so far wants.

The server tails the log forever, so the connection ends only when you end it:

• Opening the stream is a request like any other: it is retried under max_retries on the same terms, and a refusal such as a 404 fails the connection right away, with ready() raising and session.error set.
• Once open, every close is followed by a wait and a fresh connection carrying Last-Event-ID, whether the server closed cleanly or the connection dropped. The server opens every stream with a cursor frame naming where it starts, so a reconnect resumes from there even when the connection carried no event at all. The wait is the server's own retry: value when it sends one, else 3 seconds, doubling up to 30 seconds and resetting as soon as an event arrives. The budget is spent the same way: after max_reconnects reconnects in a row have carried no event — whether they were refused or opened and went quiet — the status is failed and session.error holds the last failure. A reconnect the server refuses outright fails at once: a 404, or a 401 that a freshly resolved token did not settle.
• Iterating a failed connection raises session.error; iterating a closed one ends.
• A connection that failed can be opened again: reconnect() starts the stream over from last_seq on the same connection object, the same log and the same clients, with the reconnect budget reset and ready() settling again when the stream is open. The tool clients the failure sent away register again once it is, as they would on a connection opened from scratch. In any other status, a closed connection included, it does nothing. Nothing calls it for you: it is what a "reconnect" button in a chat is wired to.

Client tools

A client is this process offering tools to one session. The connection creates it and the connection deletes it; in between it registers, keeps itself alive, and answers every dispatch.

from pydantic import BaseModel
from subako import ClientCallResultBody, Tool, ToolCall, TtlClientLifetimeForm, TtlClientLifetimeFormData


class LookupArgs(BaseModel):
    order_id: str


async def lookup(args: LookupArgs, call: ToolCall) -> str | ClientCallResultBody:
    order = await orders.find(args.order_id)
    if order is None:
        return ClientCallResultBody(text=f"no order {args.order_id}", is_error=True)
    return order.status


worker = session.create_tool_client(
    "worker",  # the model sees `client__worker__lookup`
    lifetime=TtlClientLifetimeForm(type="ttl", data=TtlClientLifetimeFormData(seconds=60)),  # the default
    on_error=lambda error: logger.error(error),  # a ping, an ack, an answer, an update, a delete
)

worker.add_tool("lookup", Tool(description="Look up an order.", parameters=LookupArgs, execute=lookup))

await worker.ready()  # registered; `worker.id` and `worker.name` are set

worker.remove_tool("lookup")  # stops offering it; a call already running still answers
await session.delete_tool_client(worker)  # also run when the connection closes

The Pydantic model is the whole declaration: its JSON Schema is the parameters the model reads, and a call's arguments are validated against it before execute runs. Arguments it refuses never reach execute. The model is answered with every issue Pydantic raised, each under the path it was at, which is what lets it correct the call rather than only learn that it failed.

A client starts empty, and the tools added to it are the whole declaration. A tool added before the registration goes out with it; after it, every add_tool and remove_tool made in one tick reaches the server as a single update. A definition whose description and schema are unchanged tells the server nothing and only swaps the handler behind the name, so a program that rebuilds its tools on every turn costs no requests.

worker.status  # ToolClientStatus: joining | serving | leaving | left | failed
worker.error  # the registration's failure, set when the status is failed
worker.id  # and `name`: what the server called this client
worker.tools  # the tool names served right now

Await ready(): a registration the server refused is raised there, and only there, with the status at failed and error set. The status walks joining, serving, leaving, left. Once a client has left, add_tool and remove_tool reach the server no more. delete_tool_client also ends the ping task; it cancels the handlers in flight, sends the DELETE, and raises if the server refuses, so a delete that failed is worth asking for again.

Registration waits for the stream, so no dispatch can arrive before the client is listening. Per dispatch, in order: the call is acked at once, the handler is looked up, the arguments are validated, then execute(args, call). Calls run concurrently.

execute answers with a str, or with a ClientCallResultBody carrying text and is_error. Every dispatch the client picked up is answered: an unknown tool, arguments the schema refused, an execute that raised, and one that outlived the deadline all settle the call with is_error=True and a short reason, so handlers need no try of their own. Asking a person first is execute's own business: await the answer there, and return what they decided.

The handler is cancelled when the client is deleted, and shortly before the server's five-minute result deadline, so a long tool can stop rather than time out. lifetime=NeverClientLifetimeForm(type="never") sends no ping, for a process whose supervisor guarantees the delete; the ttl is the default everywhere else, because the ping is the server's only way to learn that a client is gone. Background failures are logged at WARNING on the subako logger and passed to on_error when given.

Errors

Every refusal the API answers with is a SubakoError: status, code, body, and a message. A failure with no error envelope behind it — a dropped connection, an unparseable body — still arrives as one, with code unset. The codes the API declares have a class each, and a code this version has not heard of still arrives with its code and message intact.

try:
    await subako.sessions.post_event(session_id, PostSessionEventBody(type="input", text=text))
except NotFoundError:
    ...
except ConflictError:
    ...
except SubakoError as error:
    print(error.status, error.code, error)

UnauthorizedError, ForbiddenError, NotFoundError, InvalidRequestError, ConflictError, ClientOutdatedError, PreconditionFailedError, TooManyRequestsError, and InternalError cover the codes. SubakoConnectionError means no response arrived, and its subclass SubakoTimeoutError is the attempt that outlived its timeout. A cancellation of your own passes through untouched, never wrapped.

Timeouts and retries

One attempt may take timeout seconds (default 60). A failed attempt is retried up to max_retries times (default 2, so three attempts), on the same terms for every call. Every operation that creates or consumes something carries an Idempotency-Key the SDK generates for it, so a repeated attempt is deduped by the server rather than acting twice; the rest are idempotent already.

Retried:

• A connection failure, or a timeout, including while reading a response body.
• A 5xx, a 429, or a 408.

Not retried:

• Any other 4xx, a 409 among them: the request itself needs changing. A repeat the server is still executing under the same key waits on its own lock rather than answering, so there is nothing here for a retry to pick up.
• An upload whose archive is an AsyncIterable[bytes], which cannot be read twice.

A 401 is the one refusal handled outside this budget: with a function credential the token is resolved again and the request goes once more, spending no retry, as "Authentication" describes.

The wait honors Retry-After when the server sends one, and is otherwise exponential with jitter. Both settings sit on the client and on each call.

subako = SubakoClient(base_url=base_url, api_key=api_key, timeout=10.0, max_retries=4)
await subako.agents.list(options=RequestOptions(max_retries=0))

Idempotency keys

The generated key covers the SDK's own retries. Pass idempotency_key to dedupe across your retries too, or across processes: the same key with the same request returns the first attempt's receipt instead of acting again, for at least seven days after it completed.

await subako.sessions.create(CreateSessionBody(agent_id=agent_id), RequestOptions(idempotency_key=f"nightly-{date}"))

A key is sent only where the operation takes one, and is ignored elsewhere. Reusing one key for a different request is a 409, so a ConflictError no retry will settle.

Receipts

Three operations answer with a secret shown once: sessions.create, sessions.mint_token, and api_keys.mint. A replay never carries that secret again, so each of them returns either the fresh body or a receipt, and the type says so:

Operation	Fresh (201)	Replayed (200)
sessions.create	CreatedSessionBody	CreatedSessionReceiptBody
sessions.mint_token	SessionTokenBody	SessionTokenReceiptBody
api_keys.mint	MintedApiKeyBody	ApiKeyReceiptBody

The receipt confirms what the first attempt committed, the session and its id, the key and its prefix, and drops the field carrying the secret, which is what isinstance narrows on:

created = await subako.sessions.create(CreateSessionBody(agent_id=agent_id))
if isinstance(created, CreatedSessionBody):
    use(created.session_token)
else:
    # The first attempt committed and its answer was lost: the session is there
    # under `created.id`, and a fresh mint gives it a token.
    minted = await subako.sessions.mint_token(created.id)
    if isinstance(minted, SessionTokenBody):
        use(minted.session_token)

A receipt reaches you only where a key was spent twice: the SDK's own retry of a lost response, or an idempotency_key you repeated yourself. A first attempt under a fresh key always answers with the secret.

Uploading skills

A skill archive is a gzipped tar of the bundle directory, with SKILL.md at its root. Pass bytes, a binary file object, or an AsyncIterable[bytes]; a file object is read off the event loop.

from pathlib import Path

skill = await subako.skills.create(Path("skill.tar.gz").read_bytes())
with Path("skill-v2.tar.gz").open("rb") as archive:
    await subako.skills.push_version(skill.skill_id, archive)