> ## Documentation Index
> Fetch the complete documentation index at: https://docs.beta.b3os.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

Install the b3os package and call every B3OS product from Python with a typed sync or async client, with jobs, streams, typed errors and timing.

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 B3OS

b3os = 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](/sdks#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 asyncio
from b3os import AsyncB3OS


async 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 created
done = b3os.vision.create_crawl({"url": "https://docs.example.com", "limit": 50}, wait=True)  # until it ends
print(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 B3OSError

try:
    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](/sdks/typescript#retries-and-idempotency): 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_of

page = 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](/concepts/jobs#verify-the-signature).