Python SDK

The b3os package has the same namespaces and methods as the TypeScript SDK, in snake_case, with a sync client (B3OS) and an async one (AsyncB3OS, for asyncio or trio). It is built on httpx, needs Python 3.10 or later, and types every request and response as a TypedDict, so mypy and your editor check field names.

bash
pip install b3os

Create a client

python
from b3os import B3OSb3os = B3OS()  # reads B3OS_API_KEY

Constructor arguments: api_key (B3OS_API_KEY), base_url (B3OS_BASE_URL), node_url (B3OS_API_URL), timeout, max_retries, max_retry_delay, headers, on_response and http_client. Inside a B3OS sandbox, B3OS_API_URL is set to http://127.0.0.1:7700 and the client calls the node with no key; account operations (keys, usage, limits, organizations, projects, the pack registry) still go to the API host. The Unix-socket transport (socket_path, B3OS_SOCKET) is for B3OS's own tools on the node; your code never needs it.

Call a product

Path parameters are positional, the request body is a positional dict, and query and header parameters are keyword arguments. Results are plain dicts, lists, str or bytes:

python
page = b3os.vision.scrape({"url": "https://example.com", "formats": ["markdown"]})print(page["markdown"])sandbox = b3os.sandboxes.create_sandbox({"template": "base"})b3os.sandboxes.write_file(sandbox["id"], b"print(2 + 2)\n", path="/workspace/main.py", mkdir=1)result = b3os.sandboxes.exec(sandbox["id"], {"cmd": "python3 /workspace/main.py"})if result["object"] == "exec_result":  # a background command returns its process instead    print(result["stdout"])b3os.sandboxes.delete_sandbox(sandbox["id"])

Every method also takes the request options idempotency_key, timeout, max_retries, extra_headers and traceparent. The method index lists every method under its TypeScript name; the Python name is the same in snake_case (createSandbox is create_sandbox). For an operation the client does not wrap, b3os.request("GET", "/v1/...") sends it with the same key, retries and errors.

Lists

A list method returns one page, a dict with data and next_cursor; auto_paging_iter() walks every item across pages:

python
for s in b3os.sandboxes.list_sandboxes(status="running").auto_paging_iter():    print(s["id"])

Async

python
import asynciofrom b3os import AsyncB3OSasync def main() -> None:    b3os = AsyncB3OS()    try:        pages = await asyncio.gather(*[            b3os.vision.scrape({"url": url, "formats": ["markdown"]})            for url in ["https://example.com/a", "https://example.com/b"]        ])        print([p["tier"] for p in pages])    finally:        await b3os.aclose()asyncio.run(main())

Jobs and streams

Methods that start a job take wait and stream:

python
job = b3os.vision.create_crawl({"url": "https://docs.example.com", "limit": 50})  # the job as createddone = b3os.vision.create_crawl({"url": "https://docs.example.com", "limit": 50}, wait=True)  # until it endsprint(done["status"], done.get("result"))for event in b3os.vision.create_crawl({"url": "https://docs.example.com"}, stream=True):    print(event["type"])b3os.jobs.wait(job["id"], on_event=print)  # any job ID; jobs also has get, events and cancel

wait=True sends ?wait on the create and then follows the job's event stream to its end; wait=30.0 waits at most that many seconds. A job that fails or is canceled raises B3OSJobError, with the job on .job. Command output streams the same way, with stream=True on sandboxes.exec. Job and log streams resume after a dropped connection with Last-Event-ID.

Errors

A failed call raises B3OSError with the problem's code, status, type, title, detail, request_id, errors, problem, retry_after, retryable and timing:

python
from b3os import B3OSErrortry:    b3os.sandboxes.get_sandbox("sbx_01m5t68q1b590rj801nbzsqzkk")except B3OSError as error:    if error.code != "sandbox_not_found":        raise    print("gone", error.request_id)

Failures no server reported raise B3OSConnectionError (connection_error) or B3OSTimeoutError (timeout), or carry the codes invalid_response, stream_interrupted and unsupported_transport; none of them is on the errors page.

Retries, idempotency and timing

Retries follow the same rules as the TypeScript SDK: always when the server refused the request before doing any work, for calls that are safe to repeat also retryable 5xx answers, never after a timeout. Operations that accept an Idempotency-Key get one per call, reused by every retry; pass idempotency_key= to choose it.

response_of(result) returns the response behind any result: status, headers, request_id, attempts, idempotent_replayed, elapsed_ms, and timing, the parsed Server-Timing header:

python
from b3os import response_ofpage = b3os.vision.scrape({"url": "https://example.com"})info = response_of(page)if info is not None:    print(info.request_id, info.timing)

Terminals, CDP and webhooks

The Python SDK opens no WebSockets itself. open_pty(sandbox_id, cols=120, rows=40) mints a stream ticket and returns its url and protocols for any WebSocket library, and connect_cdp(session_id) returns a CDP URL with a ticket in it for Playwright's connect_over_cdp. verify_webhook(raw_body, header, secret) checks a delivery and returns the event; see verify the signature.