--- title: "Python SDK" description: "oz0-client, the typed Python client of the runtime API, sync and asyncio, with typed agents per tenant." canonical_url: "https://docs.oz0.ai/build/apps/sdk-python" --- # Python SDK > oz0-client, the typed Python client of the runtime API, sync and asyncio, with typed agents per tenant. 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](https://pypi.org/project/httpx2/). It needs no protobuf runtime. See the [runtime API](https://docs.oz0.ai/build/apps/runtime-api) for the interface it calls. The package is not on PyPI yet: the name is reserved for the first public release. Until then, install it from a checkout of the repository with `uv add ./sdk/python` or `pip install ./sdk/python`. ## Connect ```python [app.py] 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](https://docs.oz0.ai/build/apps/runtime-api/authentication#authentication); 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 ```python [app.py] 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=` makes the job the next turn of that job's conversation, and `conversation=` the next turn of the conversation with that ID (its first, when there is none yet): the agent gets the earlier turns ([Conversations](https://docs.oz0.ai/agents/conversations)). The job says which turn it is. ```python [app.py] 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](https://docs.oz0.ai/build/apps/runtime-api#job-ids-and-retries)). `candidate=` starts a job on a [private candidate](https://docs.oz0.ai/build/plugins/develop-and-test/candidate-checks#try-it-on-a-real-cluster) instead of the installed version. Every RPC of the API, `GetConversation` and `ListConversations` included, is typed on `oz0.runtime`. ## Watch, questions and approvals ```python [watch.py] 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:
Method Does
approvals(job="", tree=False) What waits for a person, oldest first: tool calls, hooks and questions
questions(job="", tree=False) The agents' questions only
approve(approval, comment=...) , deny(...) Decide one, given as listed or as an ID with job=
answer(question, "blue") Answer a question; with choices, one of them
cancel(id, reason) Stop a job and its children
jobs(limit=20) Recent jobs, newest first
catalog() The tenant's agents and flows, filtered by the key's scopes, with input_schema and output_schema parsed
cost(id) What a job and every job below it cost, with a branch per job
openapi(etag="") The tenant's OpenAPI document ; with the last ETag, document is None unless it changed
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](https://docs.oz0.ai/build/apps/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](https://docs.oz0.ai/build/apps/live-feeds/browser-and-access#stream-tickets) 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](https://docs.oz0.ai/build/apps/runtime-api/authentication#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](https://docs.oz0.ai/reference/list-pages). `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. ```python 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](https://docs.oz0.ai/agents/conversations/access-and-history#chats-yours-private-by-default) 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. ```python 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. ```bash [Terminal] OZ0_URL=https://edge.example.com:7443 OZ0_API_KEY=oz0_... oz0-client generate --tenant acme --ca-file ca.crt --out agents.py ``` ```python [app.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](https://docs.oz0.ai/build/parts/tools/results-and-files#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. ```python 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")`. ## Sitemap See the full [sitemap](https://docs.oz0.ai/sitemap.md) for all pages.