Python SDK

For: application developers.

oz0-client lets a Python application start agents and flows, wait for answers, watch jobs, answer questions, decide approvals, read the catalog and inspect cost.

The SDK requires Python 3.11 or newer. Its messages are dataclasses generated from runtime.proto, and it sends Connect JSON through httpx2. It needs no protobuf runtime. See the runtime API for the interface it calls.

Connect

from oz0 import Client

oz0 = Client("https://edge.example.com:7443", api_key="oz0_...", tenant="acme", ca_file="ca.crt")

• The key is a tenant API key; an operator key starts jobs and decides approvals, a reader key follows them. An operator can pass cert=("operator.crt", "operator.key") instead.
• The CA is the cluster's own ca.crt, which signed the edge's certificate; ca_pem= takes it as text.
• Without arguments the client reads OZ0_URL, OZ0_API_KEY, OZ0_TENANT and OZ0_CA_FILE.
• timeout= is each call's, in seconds (30 by default); streams have none. Client is a context manager, and AsyncClient takes the same arguments for asyncio.

Run an agent

job = oz0.run("acme-reviewer", {"repo": "acme/api", "pr": 42}, timeout=900, budget_usd=0.50, tags=["ci"])
print(job.output["verdict"], f"${job.cost.usd:.4f}")

run starts the job and waits for it, and raises JobFailed (with the job in .job) when it failed, was canceled or timed out; check=False returns it instead. The steps on their own are start, which returns the running job, wait(id, timeout=...), which follows the job's stream and tries again when it breaks, and get(id). A text agent takes a string. An input that does not fit the agent's schema raises ConnectError with code == "invalid_argument" before any job starts.

Conversations

continue_from=<job> makes the job the next turn of that job's conversation, and conversation=<id> the next turn of the conversation with that ID (its first, when there is none yet): the agent gets the earlier turns (Conversations). The job says which turn it is.

first = oz0.run("acme-assistant", "Our release is on Friday.")
second = oz0.run("acme-assistant", "When is the release?", continue_from=first.id)
print(second.turn, second.conversation, second.output)

start and run retry a start that meets unavailable twice, with the same key: without id=, each start sends an Idempotency-Key of its own, so the server gives the retries the same job, and with an ID the retries repeat it. A start again under an ID with the same request returns that job, running or ended; another request under it raises ConnectError with code already_exists (job IDs and retries).

candidate=<id> starts a job on a private candidate instead of the installed version. Every RPC of the API, GetConversation and ListConversations included, is typed on oz0.runtime.

Watch, questions and approvals

import asyncio
from oz0 import AsyncClient


async def main() -> None:
    async with AsyncClient() as oz0:
        job = await oz0.start("acme-asker", "Pick a colour for the report")
        async for ev in oz0.watch(job.id):
            if ev.approval_requested and ev.approval_requested.kind == "question":
                await oz0.answer(ev.approval_requested, "blue")
            elif ev.text:
                print(ev.text, end="")
            elif ev.finished:
                print(ev.finished.status, ev.finished.output)


asyncio.run(main())

Each event has one of text, tool_call, tool_result, child, approval_requested, approval_decided, finished and the others of WatchJobResponse; ev.event_case names it. Outside a stream:

Every other method of the API is on oz0.runtime, typed: oz0.runtime.get_job_trace(GetJobTraceRequest(id=job.id)), with the messages in oz0.models.

A catalog entry does not establish that the caller may start it now. Use the runtime API's ListStartable for a launch picker; it also checks start permissions, kill switches and retirement.

watch resumes a broken stream by itself after the last event (since_seq and the resume token), so no event comes twice; an event whose resync is set says a gap could not be resumed over, and the job is fetched again. watch(id, since_seq=..., resume_token=...) resumes a watch an earlier process made.

Feeds

oz0.feed(feed.conversation(id), feed.job(id), feed.my_activity()) (with from oz0 import feed) follows many chats and jobs on one stream; iterate its events(), which resume by itself on any edge, change what it follows with subscribe and unsubscribe, and resume it in another process with state() and resume_from(...). AsyncClient.feed is the same for asyncio. See Live feeds.

Stream tickets and apps that act as their people

client.stream_ticket(job=id) (or feed=) asks for a one-time ticket that lets a browser's EventSource open a job's or a feed's Server-Sent Events without the key. A client made with the app's key exchanges a person's ID token for an app token with client.exchange_token(id_token) (apps that act as their people); pass its access_token as another client's api_key to act as the person until expires_at. A refused exchange raises TokenExchangeError with OAuth's code. AsyncClient has both.

Lists, page after page

The runtime API's lists that can grow come a page at a time. iter_jobs(), iter_approvals(), iter_conversations(), iter_connections() and iter_routine_fires() yield every row and read the next page only when the loop reaches it, so a loop that stops early reads no further; jobs(limit) is the first limit rows of iter_jobs() (20 when it is 0 or less), read in pages of that size up to 100, and approvals() reads every page. AsyncClient has the same as async iterators.

for job in client.iter_jobs(mine=True):  # only the jobs the caller started
    print(job.id, job.status)

A person's chats

my_chats() yields the caller's own chats and those shared with them, page after page; update_chat() pins, archives, marks read, renames (the owner) or removes one from the caller's list, which erases nothing. Both need a person's credentials: an app's key or an operator's certificate has no chats. AsyncClient has the same.

from oz0.models import ChatAction

for chat in client.my_chats(agent="helper"):
    print(chat.id, chat.title, chat.unread)
client.update_chat("conv-1", ChatAction.PIN)

Typed agents

oz0-client generate writes a module for one tenant: a TypedDict per object of its agents' input and output schemas, and classes Agents and AsyncAgents with two methods per agent and flow.

OZ0_URL=https://edge.example.com:7443 OZ0_API_KEY=oz0_... oz0-client generate --tenant acme --ca-file ca.crt --out agents.py

from agents import Agents

agents = Agents(oz0)
job = agents.acme_reviewer({"repo": "acme/api", "pr": 42})  # starts it: a Job
review = agents.run_acme_reviewer({"repo": "acme/api", "pr": 42})  # runs it: an AcmeReviewerOutput

The module records the catalog's ETag and the generator's version. Run the command again in your app's CI: it asks the edge with that ETag and leaves the file alone until a plugin changes an agent or a schema, or a newer oz0-client writes it again. --from openapi.json reads a saved document instead, and oz0 client generate --lang python writes the same file.

Fitting a tool's answer to a budget

oz0.fit is for plugin authors, not for the client: it fits a tool's long answer (a page, a file, a diff) to a budget in tokens, as stub, outline summary, section or full (results that fit a budget). It uses only the standard library, so a plugin may depend on oz0-client for it or copy the file.

from oz0 import fit

fit.count("How many tokens is this?")  # 6, the platform's estimate (pkg/toolcost's Count)
fit.weigh(" " * 1600)  # 101: what fitting charges, with long runs of white space charged by length
f = fit.fit_document(page, budget=2000, query="Install", head=url, chars=20_000)  # chars: a bound of your own
f.view, f.text, f.meta()  # "section", the section under "Install", the fit note for _meta["oz0/fit"]

Versions

The SDK's version is the server release it was made with, and every 1.x SDK works with every 1.x server: from v1.0 the API only grows. Fields a newer server adds are ignored, and an enum value the SDK does not know yet reads as a member of its own, such as JobStatus("JOB_STATUS_PAUSED").