boxd 0.1.2.dev13

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	✓

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
r.stderr       # str
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. Use iter_stdout / iter_stderr
# (sync generators that block until the next chunk arrives), then wait()
# for the exit code. close() force-terminates the stream.
proc = box.exec("tail", "-f", "/var/log/syslog", stream=True)
for chunk in proc.iter_stdout():
    print(chunk.decode(), end="")
exit_code = proc.wait()
proc.close()                               # idempotent

# Interactive (PTY + stdin) — write to proc.stdin, end with write_eof
sh = box.exec("bash", interactive=True)    # interactive implies pty
sh.stdin.write(b"echo hello\n")
sh.stdin.write_eof()
print(sh.wait())

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