boxd 0.1.1.dev5

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",
)

Env vars BOXD_API_URL / BOXD_EXCHANGE_URL override the preset too.

Equivalent env vars: BOXD_API_KEY, BOXD_TOKEN, BOXD_API_URL, BOXD_EXCHANGE_URL.

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 exposes the server-returned fields: id, name, image, public_ip, status, url, boot_time_ms. Forked VMs additionally carry forked_from. VMs returned by c.box.get(...) also expose restart_policy, disk_bytes, and auto_suspend_timeout_secs.

Exec

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

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

# Streaming
proc = box.exec("tail", "-f", "/var/log/syslog", stream=True)
for chunk in proc.iter_stdout():
    print(chunk.decode(), end="")
exit_code = proc.wait()

# Interactive (PTY + stdin)
sh = box.exec("bash", interactive=True)   # interactive implies pty

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

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()
box = c.template.create_vm(template=t, name="from-t")
c.template.delete(t.id)

Disks

d = c.disk.create("data", size="10G")
d.attach(box, mount_path="/mnt/data")
d.attach(box, mount_path="/mnt/data", read_only=True)
d.detach(box)
d.destroy()

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")
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         # "eyJ..." — save this; list() will not return it
t.expires_at    # unix seconds

for info in c.token.list():
    print(info.jti, info.created_at, info.expires_at)
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"

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 for finer-grained handling.

Sync vs Async

The default import is the sync API, which wraps the async implementation using a dedicated event loop:

from boxd import Compute             # sync — recommended for scripts and notebooks

For async code, import from boxd.aio:

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 APIs are surface-equivalent — only the call style (sync vs await) differs.

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