# Python SDK

The `taskcloud` package is written by hand against the [OpenAPI spec](../api-reference.md) (fields and paths follow the spec exactly), with sandbox handles, streaming, retries and webhook verification on top. Python 3.9+, one dependency (`httpx`). Sync (`Client`) and async (`AsyncClient`) clients share the same methods.

```bash
pip install taskcloud            # core
pip install "taskcloud[pty]"     # adds websockets for sandbox.pty.connect()
```

## Client

```python
import os
from taskcloud import Client

client = Client(
    api_key=os.environ["TASKCLOUD_API_KEY"],  # default: this env var
    base_url="https://api.task.cloud/v1",          # default
    max_retries=2,                                 # retries errors with retryable=True
    timeout=120.0,                                 # seconds
)
```

Fields use the API's **snake_case names** (`auto_pause`, `timeout_seconds`, `exit_code`). Responses are typed objects with attributes (`result.exit_code`) and `.to_dict()`.

## Sandboxes

```python
sandbox = client.sandboxes.create(
    template="base",                      # or snapshot_id=..., or image={"ref": "ghcr.io/acme/env:1"}
    size="default",                       # small | default | large
    name="user-42",
    labels={"user": "42"},
    env={"NODE_ENV": "production"},
    auto_pause={"enabled": True, "idle_timeout_seconds": 300},
    always_on=False,
    priority="standard",                  # or "batch" (−50%, may queue)
    ports=[{"port": 3000, "visibility": "private"}],
    wait=True,
    idempotency_key="user-42-create",
)
print(sandbox.id, sandbox.state)

same = client.sandboxes.get(sandbox.id)
for s in client.sandboxes.list(state=["running"], label=["user=42"]):   # auto-paginates
    print(s.id, s.state)

sandbox.update(auto_pause={"enabled": True, "idle_timeout_seconds": 900})
sandbox.update(always_on=True)
sandbox.pause()
sandbox.resume()                      # usually unnecessary: calls wake it
fork = sandbox.fork(name="user-42-try-2")
forks = sandbox.fork(count=10)        # list of 10 Sandbox objects
sandbox.refresh()
sandbox.wait_for("running", timeout=30)
sandbox.delete()
```

### Bulk fork

`fork(count=N)` creates up to 50 copies of a sandbox (memory included) in one call. Without `count`, or with `count=1`, it returns one `Sandbox`, as before. With `count > 1` the API returns `{"data": [...]}` and the SDK returns a `list` of `Sandbox`. `fork_many(count, ...)` always returns a list. `AsyncClient` has the same methods (`await sandbox.fork(count=N)`).

Admission is all-or-nothing: if the source host cannot fit every fork, nothing is created and the call raises `ApiError` with `capacity_unavailable` (or `concurrency_limit_reached` if the plan's limit would be exceeded). With `priority="batch"` the forks are queued instead and come back in state `queued`.

```python
rollouts = sandbox.fork_many(32, labels={"run": "eval-7"})
results = [s.exec(command="python3 rollout.py") for s in rollouts]

queued = sandbox.fork(count=50, priority="batch", wait=False)   # -50%, may queue
```

## Exec

```python
r = sandbox.exec(command="pytest -q", cwd="/home/user/app", timeout_seconds=300)
print(r.exit_code, r.stdout, r.stderr, r.truncated)

sandbox.exec(args=["python3", "-c", "print(42)"])     # no shell

for ev in sandbox.exec_stream(command="pip install -r requirements.txt"):
    if ev.type in ("stdout", "stderr"):
        print(ev.data, end="")
    elif ev.type == "exit":
        print("exit", ev.exit_code)

cmd = sandbox.exec(command="python3 -m http.server 8080", background=True)
for ev in cmd.stream():          # attach; replays earlier output
    ...
cmd.write_stdin("q\n", eof=True)
cmd.kill(signal="SIGTERM")
done = cmd.wait()

sandbox.commands.list()
sandbox.commands.get(cmd.id)
```

### PTY

`pty.connect()` needs the `pty` extra (`pip install "taskcloud[pty]"`) and is available on the sync `Client` only. `AsyncClient` supports `pty.create()`, which returns the `websocket_url` for your own WebSocket client.

```python
with sandbox.pty.connect(cols=120, rows=40) as term:
    term.write(b"ls -la\n")
    print(term.read(timeout=1.0).decode())
    term.resize(160, 48)
```

## Files

```python
sandbox.files.write("/home/user/app/.env", "PORT=3000\n", mode="0600")
text = sandbox.files.read_text("/home/user/app/package.json")
data = sandbox.files.read("/home/user/out.png")            # bytes
entries = sandbox.files.list("/home/user/app", recursive=False)
info = sandbox.files.stat("/home/user/app")
sandbox.files.mkdir("/home/user/data")
sandbox.files.move("/home/user/a.txt", "/home/user/b.txt")
sandbox.files.remove("/home/user/tmp", recursive=True)
sandbox.files.upload("/home/user/app", {"src/hello.txt": b"hi"})
archive = sandbox.files.download_archive("/home/user/app", format="tar.gz")
url = sandbox.files.download_url(path="/home/user/big.bin").url
up = sandbox.files.upload_url(path="/home/user/dataset.parquet", expires_in_seconds=900)

for ev in sandbox.files.watch("/home/user/app/src", recursive=True):
    print(ev.type, ev.path)
```

## Networking

```python
port = sandbox.ports.expose(port=3000, visibility="private")
print(port.url)                         # https://<short_id>-3000.taskhost.cloud
token = sandbox.access_tokens.create(ttl_seconds=3600, ports=[3000]).token
sandbox.ports.list()
sandbox.ports.remove(3000)

ssh = sandbox.ssh(public_key="ssh-ed25519 AAAA...", ttl_seconds=3600)
print(ssh.command)
```

## Desktop

```python
stream = sandbox.desktop.stream(ttl_seconds=600, read_only=True)
print(stream.url)
sandbox.desktop.revoke(stream.id)

shot = sandbox.desktop.screenshot(format="png")    # .data (base64), .media_type, .width, .height
sandbox.desktop.input(
    actions=[
        {"action": "left_click", "coordinate": [640, 412]},
        {"action": "type", "text": "hello"},
        {"action": "key", "text": "Return"},
    ],
    return_screenshot=True,
)
pos = sandbox.desktop.cursor()   # .x, .y
```

Full agent loop: [Desktop and computer use](../guides/desktop-and-computer-use.md).

## Snapshots and templates

```python
snap = sandbox.snapshots.create(name="after-pip-install")
copy = client.sandboxes.create(snapshot_id=snap.id)
client.snapshots.list(sandbox_id=sandbox.id)
client.snapshots.delete(snap.id)

tpl = client.templates.create(
    name="agent-env",
    image={"ref": "ghcr.io/acme/agent-env:1.4"},
    ready_check={"command": "curl -sf localhost:8080/health", "timeout_seconds": 120},
)
print(client.templates.logs(tpl.id))
client.sandboxes.create(template="agent-env")
```

## Webhooks, usage, keys, account

```python
hook = client.webhooks.create(
    url="https://example.com/hooks/taskcloud",
    events=["sandbox.ready", "sandbox.paused", "sandbox.resumed", "sandbox.error", "sandbox.deleted"],
)
print(hook.secret)   # whsec_... (shown once)

from taskcloud.webhooks import verify_webhook
event = verify_webhook(payload=raw_body, headers=request.headers, secret=os.environ["WEBHOOK_SECRET"])

usage = client.usage.get(group_by="sandbox", from_="2026-11-01T00:00:00Z")   # `from` is a Python keyword: use from_=
limits = client.limits.get()
key = client.api_keys.create(name="user-42", preset="full-sandbox", restrict={"labels": {"user": "42"}})
client.account.set_spend_limit(monthly_limit_usd=500, on_limit="pause_running")
```

## Async

```python
import asyncio
from taskcloud import AsyncClient

async def main():
    client = AsyncClient()
    sandboxes = await asyncio.gather(*[client.sandboxes.create(template="base", priority="batch") for _ in range(20)])
    results = await asyncio.gather(*[s.exec(command="python3 eval.py") for s in sandboxes])
    await asyncio.gather(*[s.delete() for s in sandboxes])

asyncio.run(main())
```

Batch sandboxes may wait in `queued`; `create` with `priority="batch"` returns immediately, and `exec` waits for the sandbox to start.

## Errors

```python
from taskcloud import ApiError

try:
    client.sandboxes.create(size="large")
except ApiError as e:
    print(e.status, e.code, e.retryable, e.request_id, e.details)
```

Errors with `retryable=True` are retried automatically up to `max_retries`, honouring `Retry-After`.