boxd 0.1.5.dev29

Python SDK for the boxd cloud VM platform

boxd Python SDK

Python SDK for the boxd cloud VM platform. Sync-first API with full async support.

Requires Python 3.10+.

Install

pip install boxd

Quick Start

from boxd import Compute

with Compute(api_key="bxk_...") as c:
    box = c.box.create(name="my-vm")
    result = box.exec("echo", "hello")
    print(result.stdout)
    box.destroy()

Authentication

Compute(api_key="bxk_...")     # API key (recommended)
Compute(token="eyJ...")        # direct JWT
Compute()                      # reads BOXD_API_KEY or BOXD_TOKEN

Configuration

Pick a named cluster preset:

Compute(api_key="bxk_...")                            # production (default)
Compute(api_key="bxd_...", environment="staging")     # boxd-stg.sh

environment also reads from the BOXD_ENVIRONMENT env var ("production" or "staging").

For custom or self-hosted endpoints, override URLs explicitly — these take precedence over environment:

Compute(
    api_key="bxk_...",
    api_url="http://my-boxd.example.com:9443",
    exchange_url="https://my-boxd.example.com/api/v1/auth/token",
)

Environment variables

All Compute arguments can be supplied via env vars. Constructor args win over env vars; env vars win over the environment preset.

Variable	Sets	Default
BOXD_API_KEY	API key (long-lived, recommended)	—
BOXD_TOKEN	Direct JWT (short-lived)	—
BOXD_ENVIRONMENT	Preset name (production or staging)	production
BOXD_API_URL	gRPC endpoint, overrides preset	http://boxd.sh:9443
BOXD_EXCHANGE_URL	Token-exchange URL, overrides preset	https://boxd.sh/api/v1/auth/token

api_url accepts an optional URL scheme that controls TLS:

api_url value	Transport
http://host:port	plaintext (scheme stripped before connecting)
https://host:port	TLS (scheme stripped before connecting)
bare host:port	TLS, except localhost / 127.* which stay plaintext

The default http://boxd.sh:9443 matches production. Self-hosted clusters can pass api_url="http://my-cluster:9443" to opt into plaintext.

VM Lifecycle

box = c.box.create(name="my-vm")
boxes = c.box.list()
found = c.box.get("my-vm")                             # by name or id
forked = c.box.fork("my-vm", name="f1")

box.start()
box.stop()
box.reboot()
box.destroy()
s = box.suspend()    # SuspendResult
r = box.resume()     # ResumeResult

Box fields

Box always carries server-returned fields, but which ones are populated depends on how it was obtained:

Field	create	fork	list	get
id, name, image, public_ip, status	✓	✓	✓	✓
url, boot_time_ms	✓	✓	None	None
forked_from	None	✓	None	None
restart_policy, disk_bytes, auto_suspend_timeout_secs	None	None	None	✓
ssh_port	None	None	None	✓

If you need the URL or boot time after a list / get round-trip, the https://<name>.boxd.sh form is stable, or call box.proxies() for the full set. If you need the lifecycle fields off a Box from list / create / fork, re-fetch via c.box.get(box.name).

BoxConfig

create, fork, and template.create_vm all take an optional config:

from boxd import BoxConfig, LifecycleConfig

config = BoxConfig(
    vcpu=2,                           # default 2
    memory="4G",                      # default "8G"
    env={"API_KEY": "secret"},        # env vars exposed to the VM
    restart_policy="always",          # "always" | "never"
    lifecycle=LifecycleConfig(
        auto_suspend_timeout=300,     # idle network secs; 0 disables
        auto_destroy_timeout=0,       # total lifetime secs; 0 disables
    ),
)

box = c.box.create(name="my-vm", config=config)

Exec

# Simple — collect all output
r = box.exec("python", "script.py")
r.stdout       # str (subprocess fd 1)
r.stderr       # str (subprocess fd 2)
r.exit_code    # int
r.success      # bool — exit_code == 0

# With env vars and timeout
box.exec("sh", "-c", "echo $FOO", env={"FOO": "bar"}, timeout=30)

# Streaming — proc is an ExecProcess. iter_stdout / iter_stderr are
# separate sync generators. wait() returns the exit code; close() force-
# terminates (idempotent).
proc = box.exec("cargo", "build", stream=True)
for chunk in proc.iter_stdout():
    print(chunk.decode(), end="")
# (read iter_stderr concurrently in another thread if you want warnings live)
exit_code = proc.wait()
proc.close()

# Headless one-shot — close stdin immediately so commands like
# `claude -p`, `jq`, `cat file` see EOF and don't hang waiting for input.
r2 = box.exec(
    "claude", "-p", "summarize this PR",
    stream=True, close_stdin=True,
)

# Interactive (PTY + stdin) — stdin must stay open for user input.
# close_stdin=True is REJECTED with this combination (ValueError).
sh = box.exec("bash", interactive=True)    # interactive implies pty
sh.stdin.write(b"echo hello\n")
sh.stdin.write_eof()
print(sh.wait())

Streams: stdout vs stderr

r.stderr / proc.iter_stderr() is populated for non-PTY execs. The subprocess's fd 2 is delivered separately from fd 1 — useful when a tool's progress lives on stderr while the answer is on stdout (e.g. codex exec, cargo build).

Under pty=True / interactive=True, the kernel TTY layer merges stderr into stdout (that's how terminals work), so everything arrives on stdout and the stderr side stays empty. Set pty=False if you need the split.

close_stdin

Pass close_stdin=True (only valid with stream=True on a non-PTY exec) to have the SDK close the client send half of the bidi stream right after the command starts. Headless one-shots that read stdin (claude -p, jq, cat file) see EOF immediately and proceed; without it they hang for several seconds (or forever) waiting on stdin. Passing it together with pty=True or interactive=True raises ValueError rather than silently dropping the flag — a PTY shell needs stdin open for user input.

PTY size + resize

For PTY/interactive execs, pass cols and rows to set the initial terminal geometry, and call proc.resize(cols, rows) to update it mid-session when the local terminal changes size. This is what makes TUI apps like claude, vim, htop render at the right width.

import os
import signal
import shutil

cols, rows = shutil.get_terminal_size()
proc = box.exec("claude", stream=True, pty=True, cols=cols, rows=rows)

# Forward local SIGWINCH so resize propagates into the VM's PTY.
def _on_resize(_signum, _frame):
    c, r = shutil.get_terminal_size()
    proc.resize(c, r)
signal.signal(signal.SIGWINCH, _on_resize)

# Forward raw bytes; the local terminal renders the ANSI escapes.
for chunk in proc.iter_stdout():
    os.write(1, chunk)

Zero / unset cols and rows fall back to the server default of 80×24. resize() on a non-PTY exec is a harmless no-op.

Files

from pathlib import Path

box.write_file(b"binary content", "/app/file.bin")
box.write_file("text content", "/app/file.txt")
box.write_file(Path("local/file.py"), "/app/file.py")
data = box.read_file("/app/output.json")    # bytes

Proxies

box.proxies()                              # list[Proxy]
proxy = box.create_proxy("api", port=3001) # api.<vm>.boxd.sh -> port 3001
box.set_proxy_port(port=3000)              # change default proxy port
box.set_proxy_port(port=3001, name="api")  # change a named proxy
box.delete_proxy("api")

Logs

# Snapshot of available console output
for chunk in box.stream_logs():
    print(chunk.decode(errors="replace"), end="")

# Follow (keeps the stream open for new chunks)
for chunk in box.stream_logs(follow=True):
    print(chunk.decode(errors="replace"), end="")

Templates

Reusable image + BoxConfig frozen together.

from boxd import BoxConfig

t = c.template.create(
    name="t1",
    image="ghcr.io/org/img:tag",
    config=BoxConfig(vcpu=2, memory="4G"),
)
c.template.list()

# create_vm accepts a Template object OR a template ID string. Pass an
# optional `config` to override the template's defaults (e.g. bump
# memory for one specific VM).
box = c.template.create_vm(template=t, name="from-t")
big = c.template.create_vm(
    template=t.id,
    name="from-t-big",
    config=BoxConfig(memory="16G"),
)
c.template.delete(t.id)

Disks

d = c.disk.create("data", size="10G")
d.id; d.name; d.size_bytes; d.status

# attach / detach take a Box instance OR a name/id string
d.attach(box, mount_path="/mnt/data")
d.attach("my-vm", mount_path="/mnt/data", read_only=True)
d.detach("my-vm")

d.destroy()

# list returns DiskHandle instances — same methods as above
for d in c.disk.list():
    print(d.name, d.status)

Domains

Bind an external domain (DNS must already point at the boxd proxy).

c.domain.bind("app.example.com", box)            # accepts a Box, name, or id
c.domain.bind("app.example.com", "my-vm")
for d in c.domain.list():
    print(d.domain, "->", d.vm_id)
c.domain.unbind("app.example.com")

Networks

n = c.network.create()              # server assigns id
named = c.network.create(name="staging")

# `create` returns the new network's id only — `subnet` and `status` come
# back populated once provisioning settles. Re-fetch via `list` to read them.
for net in c.network.list():
    print(net.id, net.subnet, net.status)

Tokens

Issue scoped JWTs for delegated access. The raw token string is only returned at creation — store it then.

t = c.token.create(expires_in=3600)   # 0 = server default
t.token         # str — "eyJ..."  save this; list() will not return it again
t.expires_at    # int — unix seconds

# list() returns TokenInfo (no raw token; listing-safe metadata).
# The `jti` field here is what revoke() takes — there's no jti on
# the freshly-created Token, so revoke goes through list().
for info in c.token.list():
    info.jti          # str — used by revoke()
    info.created_at   # int — unix seconds
    info.expires_at   # int — unix seconds
    c.token.revoke(info.jti)

# Use the token to authenticate a new client
c2 = Compute(token=t.token)

Identity

me = c.whoami()
me.user_id              # "gh-username"
me.fingerprints         # ["SHA256:..."]
me.default_network_id   # "net-..."

cfg = c.config()
cfg.default_image       # "ubuntu:latest"
cfg.zone                # "boxd.sh"

The package also exposes its installed version:

import boxd
print("on", boxd.__version__)

Errors

from boxd import (
    BoxdError,            # base class
    AuthenticationError,
    NotFoundError,
    QuotaExceededError,
    InvalidArgumentError,
    TimeoutError,
    ConnectionError,
    InternalError,
)

try:
    box = c.box.get("nope")
except NotFoundError:
    ...

Class	gRPC status
AuthenticationError	UNAUTHENTICATED, PERMISSION_DENIED
NotFoundError	NOT_FOUND
QuotaExceededError	RESOURCE_EXHAUSTED
InvalidArgumentError	INVALID_ARGUMENT, ALREADY_EXISTS
TimeoutError	DEADLINE_EXCEEDED
ConnectionError	UNAVAILABLE
InternalError	INTERNAL, UNKNOWN

Each error carries the underlying grpc_code (numeric gRPC status — see grpc.StatusCode) for finer-grained handling:

import grpc

try:
    c.box.create(name="my-vm")
except BoxdError as e:
    if e.grpc_code == grpc.StatusCode.RESOURCE_EXHAUSTED.value[0]:
        ...   # hit per-user quota
    raise

Update notifications

Every gRPC response carries an x-boxd-py-sdk-latest header set by the boxd proxy. The SDK's interceptor compares it to the installed version and prints a one-time sys.stderr line if a newer release is available:

A new version of boxd is available (v0.1.2, you have v0.1.1). Update with:
  pip install --upgrade boxd

The notice fires at most once per process, never causes a request to fail, and is silent if the proxy isn't advertising a newer version. Compares as PEP 440-ish (numeric prefix, then per-component compare on .devN suffixes).

Sync vs Async

The default boxd.Compute is the sync API — fully blocking, safe for scripts, REPLs, notebooks, Django views, anywhere you don't already have an event loop. It wraps the async implementation behind a dedicated background loop, so you don't pay for asyncio setup yourself.

from boxd import Compute             # sync — recommended default

boxd.aio.Compute is the async API — use it from inside an existing event loop (FastAPI, asyncio scripts, Quart, anyio):

from boxd.aio import Compute

async with Compute(api_key="bxk_...") as c:
    box = await c.box.create(name="my-vm")
    result = await box.exec("echo", "hello")

The two are surface-equivalent: same method names, same arguments, same return types. The only differences are with vs async with and the await keyword on every call. Pick boxd unless you already have an event loop.

Development

cd sdk/python
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

pytest tests/                              # unit tests (e2e marker excluded by default)
pytest tests/ -m e2e                       # e2e tests (creates/destroys VMs)
pytest tests/ -m ""                        # everything
bash scripts/compile_proto.sh              # regenerate _generated/ after changing api.proto

Architecture

sdk/python/
├── src/boxd/
│   ├── __init__.py       # public sync API exports (default import)
│   ├── aio.py            # public async API exports
│   ├── _sync.py          # sync wrappers (run_until_complete)
│   ├── client.py         # async Compute (entry point) + auth/transport
│   ├── auth.py           # API key → JWT exchange + refresh
│   ├── boxes.py          # async BoxService (create/list/get/fork)
│   ├── box.py            # async Box (lifecycle/exec/files/proxies/logs)
│   ├── exec.py           # ExecResult, ExecProcess, stream readers/writers
│   ├── templates.py      # async TemplateService
│   ├── disks.py          # async DiskService + DiskHandle
│   ├── domains.py        # async DomainService
│   ├── networks.py       # async NetworkService
│   ├── tokens.py         # async TokenService
│   ├── types.py          # public dataclasses (BoxConfig, Proxy, etc.)
│   ├── errors.py         # BoxdError hierarchy + gRPC mapping
│   ├── _utils.py         # GrpcCaller mixin, parse_size, resolve_endpoint
│   └── _generated/       # protoc-grpc-python output (committed)
└── tests/                # pytest unit + gated e2e