# Python SDK

The Python SDK wraps the [REST API](/api) and the [WebSocket](/websocket) behind typed
resources. It ships sync and async clients with the same methods on both.

## Install

```bash
pip install 0bull
# or
uv add 0bull
```

The package is `0bull` on [PyPI](https://pypi.org/project/0bull/) and imports as
`zerobull`. It needs Python 3.10 or newer. Source and changelog:
[github.com/0bull/0bull-python-sdk](https://github.com/0bull/0bull-python-sdk).

This page documents **0.3.0**.

Upgrading from 0.2.0: a phone-controller session phone is keyed `slot`, not `id`, and
`ice_servers` is gone from the session. Live video is not part of the API.

## Authenticate

Create a token at
[0bull.net/settings/api-tokens](https://0bull.net/settings/api-tokens), scoped to the
abilities your code needs. The [ability table](/introduction#token-abilities) lists what each
one grants.

Pass the token, or set `ZEROBULL_API_TOKEN` and pass nothing:

```python
from zerobull import ZeroBull

client = ZeroBull(api_token="...")
# or, with ZEROBULL_API_TOKEN set:
client = ZeroBull()
```

A call made with a token that lacks the ability raises `PermissionDeniedError`.

| Option | Default | Environment |
|---|---|---|
| `api_token` | required | `ZEROBULL_API_TOKEN` |
| `base_url` | `https://0bull.net/api` | `ZEROBULL_BASE_URL` |
| `timeout` | 30 seconds | |
| `max_retries` | 2 | |
| `http_client` | none; inject your own `httpx.Client` | |

## First call

```python
from zerobull import ZeroBull

with ZeroBull() as client:
    for phone in client.phones.list():
        print(phone.slot, phone.name, phone.can_control)
```

`phones.list()` returns `Phone` objects with `slot`, `name`, `video_live`, `input_present`,
`can_control`, `model` and `os_version`. The `slot` is the UUID every other phone method
takes. The `name` is a label, and passing it where a slot is wanted matches no phone.

The same code with the async client:

```python
import asyncio
from zerobull import AsyncZeroBull

async def main() -> None:
    async with AsyncZeroBull() as client:
        phones = await client.phones.list()
        print(phones[0].slot)

asyncio.run(main())
```

Every method below exists on both clients with the same name and arguments. Await it on
`AsyncZeroBull`.

## Phones

Coordinates are fractions of the screen, 0 to 1. `snapshot` and `ocr` take an optional
`width` between 120 and 2000.

```python
jpeg: bytes = client.phones.snapshot(slot, width=600)
text: str = client.phones.ocr(slot)

client.phones.tap(slot, fx=0.5, fy=0.9)
client.phones.swipe(slot, fx1=0.5, fy1=0.8, fx2=0.5, fy2=0.2, steps=30)
client.phones.type(slot, "hello")
client.phones.hotkey(slot, "enter")
```

Input runs on the phone before the call returns, so there is nothing to poll.

`hotkey` accepts `home`, `app_switcher`, `control_center`, `notifications`, `back`,
`run_shortcut`, `enter`, `backspace`, `copy`, `cut`, `paste` and `select_all`. An unknown key
raises `ValueError` before any request is sent.

## Runs

Commands, macros and agent tasks each queue a `Run` and return it right away:

```python
run = client.phones.run_command(slot, "brightness", level=0.5)
run = client.phones.run_macro(slot, workflow="post-to-story", params={"caption": "hi"})
run = client.phones.run_agent(slot, "Open Settings and turn on Wi-Fi")
```

`run_command` ops are `clipboard_set` (needs `text`), `clipboard_get`, `open_url` (needs
`url`), `reboot`, `clear_photos`, `get_ip`, `brightness` (needs `level`, 0 to 1), and
`wifi`, `airplane`, `cellular` and `flashlight` (each needs `on`).

`run_macro` takes exactly one of `workflow` (with optional scalar `params`) or `steps`, up to
200 steps. `run_agent` takes a task of 1 to 2000 characters.

### Waiting on a run

The run comes back before the phone has done the work. Poll it with `runs.wait()`, which
returns once the status is `succeeded`, `failed` or `cancelled`:

```python
run = client.phones.run_command(slot, "get_ip")
run = client.runs.wait(run, timeout=300, interval=2)

if run.succeeded:
    print(run.result["value"])
else:
    print(run.error)
```

`get_ip` and `clipboard_get` put what the phone reported in `result["value"]`. Every other
command finishes with `result` set to `None`, so check `run.succeeded` rather than the
result.

`wait()` raises `WaitTimeoutError` if `timeout` seconds pass first. Defaults are 300 seconds
with a 2 second interval. `run.is_terminal` tells you whether a run you already hold is
finished.

Read the history with `runs.list()`, which returns a `Page`. Call `iter_all()` to walk every
page:

```python
for run in client.runs.list(slot).iter_all():
    print(run.id, run.kind, run.status)
```

## Accounts

```python
account = client.accounts.create(platform="tiktok", handle="@me", slot=slot)
client.accounts.update(account.id, notes="test")
client.accounts.delete(account.id)

for account in client.accounts.list(platform="tiktok").iter_all():
    print(account.id, account.handle)
```

`platform` defaults to `tiktok`. TikTok accounts need a `slot`, YouTube accounts need a
`google_email`, and passing `google_email` on any other platform raises `ValueError`.
`update` changes only the fields you pass.

## Submissions and uploads

```python
from pathlib import Path

submission = client.submissions.create(
    account_id=account.id,
    video=Path("clip.mp4"),
    caption="hello from 0bull",
)
```

Pass exactly one of `video` (a path, bytes, or an open file), `video_url`, or `upload_id`.
On YouTube the caption is the Short's title, so it is required there and capped at 100
characters instead of 2200.

To post the same video to several accounts, upload it once and reuse the id:

```python
upload_id = client.uploads.upload("clip.mp4")
client.submissions.create(account_id=account.id, upload_id=upload_id, caption="hi")
```

Publishing happens on a real phone after the call returns, so wait for it:

```python
submission = client.submissions.wait(submission, timeout=900, interval=5)
print(submission.status, submission.failure)
```

`wait()` returns once the status is `published`, `drafted`, `failed` or `cancelled`, and
raises `WaitTimeoutError` otherwise. `cancel(id)` stops one that is still running, and
`delete(id)` removes it and its stored video.

## Billing

```python
summary = client.billing.summary()
rental = client.billing.start_rental(accept_terms=True, phones=5, country="US")
change = client.billing.request_phone_count(accept_terms=True, add=2)
request = client.billing.get_request(change.request_id)
```

`start_rental` returns a `checkout_url` for the user to open. `request_phone_count` applies
straight away when it fits your standing allowance, otherwise it returns an `approval_url`
and stays pending until someone acts on it. Poll `get_request` for the answer.

`accept_terms=True` confirms you showed the user the [terms](https://0bull.net/terms) and
[privacy policy](https://0bull.net/privacy), including that a rental renews monthly, before
the charge. `False` raises `ValueError`.

## Events over WebSocket

`client.socket()` gives you the same resource methods over one connection that also pushes
status events, so you can stop polling:

```python
from zerobull import RunEvent

with client.socket() as socket:
    run = socket.phones.run_macro(slot, workflow="post-to-story")
    for event in socket.events():
        if isinstance(event, RunEvent) and event.run.id == run.id and event.run.is_terminal:
            print(event.run.status)
            break
```

`events()` yields `RunEvent`, `SubmissionEvent` and `BillingRequestEvent`. The socket
reconnects on its own after an unexpected drop; calls that were in flight raise
`SocketClosedError` rather than being resent. `user`, `session` and any call carrying a local
file raise `NotImplementedError` on the socket. Use `socket.call(fun, data)` for a
[fun](/websocket) the SDK does not model yet.

## Errors

Everything the SDK raises on purpose subclasses `ZeroBullError`.

| Condition | Exception |
|---|---|
| 400 | `BadRequestError` |
| 401 | `AuthenticationError` |
| 403 | `PermissionDeniedError` |
| 404 | `NotFoundError` |
| 409 | `ConflictError` |
| 422 | `ValidationError`, with `.errors` |
| 429 | `RateLimitError`, with `.retry_after` |
| 502, 503 | `UnavailableError` |
| Any other 5xx | `InternalServerError` |
| Any other status | `APIStatusError` |
| Network failure | `APIConnectionError` |
| Timeout | `APITimeoutError` |
| Socket closed mid-call | `SocketClosedError` |
| `wait()` timed out | `WaitTimeoutError` |

```python
from zerobull import APIStatusError, ValidationError

try:
    client.submissions.create(
        account_id=account.id, video_url="https://example.com/missing.mp4", caption="hi"
    )
except ValidationError as e:
    print(e.errors)
except APIStatusError as e:
    print(e.status, e.message)
```

Arguments the SDK can check itself raise `ValueError` before any request goes out. Only a
request that reached the server can raise `ValidationError`.

## Rate limits

The API allows 250 requests a minute, and 60 a minute for `ocr`. A `429` is retried
automatically up to `max_retries`, honoring `Retry-After`, before `RateLimitError` is raised.
Nothing else is retried.