b3os (Python)

The Python SDK for B3OS beta, generated from packages/spec/dist/openapi.json with the same 14 namespaces and 138 operations as the TypeScript SDK (@b3os/sdk), in snake_case. Sync (B3OS) and async (AsyncB3OS, asyncio or trio) clients on httpx. Every request and response body is a TypedDict; the package is py.typed and passes mypy --strict. Python 3.10+.

from b3os import B3OS

b3 = B3OS()  # B3OS_API_KEY
page = b3.vision.scrape({"url": "https://example.com", "formats": ["markdown"]})
print(page["markdown"])

sbx = b3.sandboxes.create_sandbox({"template": "base"})
out = b3.sandboxes.exec(sbx["id"], {"cmd": "python3 -c 'print(2 + 2)'"})
print(out["stdout"])  # "4\n"
b3.sandboxes.delete_sandbox(sbx["id"])

import anyio
from b3os import AsyncB3OS

async def main() -> None:
    async with AsyncB3OS() as b3:
        crawl = await b3.vision.create_crawl({"url": "https://docs.example.com", "limit": 20}, wait=True)
        page = await b3.vision.list_crawl_pages(crawl["id"])
        async for p in page.auto_paging_iter():
            print(p["url"])

anyio.run(main)

Where requests go

The same rules as the TypeScript SDK:

Constructor arguments: api_key (B3OS_API_KEY), base_url (B3OS_BASE_URL), node_url (B3OS_API_URL), socket_path (B3OS_SOCKET; False disables), transport="http" (B3OS_TRANSPORT), timeout, max_retries (2), max_retry_delay (60 s), headers, on_response, http_client.

Calling operations

Path parameters are positional, the body is a positional dict, query and header parameters are keyword arguments, and every method takes the request options idempotency_key, timeout, max_retries, extra_headers and traceparent:

b3.sandboxes.list_sandboxes(status="running", limit=50)          # Page: a dict, plus auto_paging_iter()
b3.sandboxes.write_file(sbx_id, b"print(1)\n", path="/workspace/main.py", mkdir=1)
data = b3.sandboxes.read_file(sbx_id, path="/workspace/out.bin")   # bytes
listing = b3.sandboxes.read_file(sbx_id, path="/workspace", list=1)
md = b3.vision.read_url("https://example.com/pricing", only_main_content=False)   # str
png = b3.render.create_screenshot({"url": url}, accept="image/png")              # bytes
b3.routines.create_routine(open("digest.yaml").read())                            # YAML or a dict

Results are plain dicts, lists, str or bytes; response_of(result) returns its ResponseInfo (status, headers, request_id, attempts, idempotent_replayed, elapsed_ms, and timing: the parsed Server-Timing header with total_ms, edge_ms and phases). For anything not wrapped: b3.request("GET", "/v1/…").

Errors, retries, idempotency

Non-2xx answers raise B3OSError with code, status, type, title, detail, request_id, errors, problem (reason, limit, required_scope as properties), retry_after, retryable and timing. Client-side failures use B3OSConnectionError (connection_error), B3OSTimeoutError (timeout) and the codes invalid_response, stream_interrupted, unsupported_transport. A job that fails while you wait raises B3OSJobError with .job.

Effectful creates that accept Idempotency-Key get one per call, reused by every retry. Retries follow the TypeScript SDK's rules: always for answers that refused the request before any work (429 rate and concurrency limits, 503 unavailable, 409 in progress) when a problem body says so, or a 429 or Cloudflare 521/522/523/530 without one; for retry-safe calls also retryable 5xx (including a CDN's body-less 502/503/520 page, reported as internal with code_from_status) and dropped connections; never for timeouts or a Retry-After longer than max_retry_delay.

Timeouts: timeout bounds connecting and each read (httpx semantics), default the operation's own (capped at 65 s) plus 5 s; streams use a 30 s idle read timeout and pings arrive every 15 s.

Jobs and streams

job = b3.vision.create_crawl({"url": url})                         # the job as created
done = b3.vision.create_crawl({"url": url}, wait=True)              # until it finishes (B3OSJobError if it fails)
done = b3.vision.create_crawl({"url": url}, wait=30.0)              # at most 30 s
for ev in b3.vision.create_crawl({"url": url}, stream=True):        # its events from the creating request on
    print(ev["type"])
b3.jobs.wait(job, on_event=print)                                  # any job (object or id): wait, events, get, cancel

for ev in b3.sandboxes.exec(sbx_id, {"cmd": "npm test"}, stream=True):
    print(ev.get("data", ""), end="")
for chunk in b3.infer.create_chat_completion({"model": m, "messages": msgs}, stream=True):
    print(chunk["choices"][0]["delta"].get("content", ""), end="")

wait=True sends the create with ?wait (up to 60 s) and then follows the job's event stream to its terminal event, polling when the deployment has no events operation. Event streams resume after a dropped connection with Last-Event-ID; stream.last_event_id is where to resume later, JobEventStream.result the finished job.

Terminals, CDP, webhooks

open_pty(sbx_id, cols=…, rows=…) mints a stream ticket and returns a connected PtySession (AsyncPtySession on AsyncB3OS): write() keystrokes, resize(cols, rows), iterate (or read()) the output, wait() for the PtyExit (code, or the WebSocket close code). It needs the optional WebSocket client: pip install 'b3os[pty]' (the websockets package; everything else needs only httpx). pty_url(...) returns the TicketedUrl (url, protocols) instead, for another WebSocket client. connect_cdp(session_id) returns a CdpEndpoint whose URL carries the ticket (playwright.chromium.connect_over_cdp(ep.url)). verify_webhook(raw_body, headers["b3os-signature"], secret) checks a delivery and returns the event.

Development

node packages/sdk/scripts/generate.mjs          # regenerates src/b3os/_generated with the TypeScript SDK
cd packages/sdk-python
uv venv .venv && uv pip install -e '.[dev]'
.venv/bin/mypy                                   # strict
.venv/bin/pytest -q                              # unit tests + the Prism contract test (B3OS_PRISM=0 skips it)

The live test runs on dev-h200 against a dev b3osd: ops/box.sh sdk -- bash packages/sdk/test/live/run.sh py.