boxd Python SDK

Python SDK for the boxd cloud machine platform. Create machines, run commands in them, move files, and manage everything around them.

Requires Python 3.10+.

Install

pip install boxd

Quick start

from boxd import Boxd

boxd = Boxd(api_key="bxd_...")

machine = boxd.machines.create("my-machine")
boxd.machines.wait_until_ready(machine.id)

result = boxd.machines.exec(machine.id, "uname -a")
print(result.stdout)

boxd.machines.delete(machine.id)
boxd.close()

Everything follows the same shape: boxd.<resource>.<verb>(id, ...). Resources return plain data — a Machine has fields, not methods.

Client

Boxd()                                     # production
Boxd(api_key="bxd_...")
Boxd(base_url="https://boxd.example.com:9443")   # any other cluster

Every argument is keyword-only.

base_url accepts an optional scheme that controls TLS:

boxd.base_url reports the cluster the client settled on.

The client holds a connection, so keep one around rather than making a new one per call. Close it when you're done — or use it as a context manager:

with Boxd(api_key="bxd_...") as boxd:
    ...

Authentication

The first of these that is present wins:

1. token= — used as given
2. api_key= — exchanged for a short-lived credential and kept fresh for you
3. BOXD_TOKEN, then BOXD_API_KEY
4. running inside a boxd machine — see below
5. otherwise AuthenticationError

If your key is revoked mid-session, the SDK fails fast with AuthenticationError rather than retrying.

Inside a machine

Inside a boxd machine, Boxd() authenticates automatically — no API key needed, and it talks to that machine's own cluster unless you pass base_url.

from boxd import Boxd

boxd = Boxd()
for machine in boxd.machines.list():
    print(machine.name, machine.status)

One limit: inside a shared machine the automatic credential can manage the organization's shared machines, but cannot read environment variables or secrets, and cannot reach private machines. Pass an API key for those.

Sync and async

Boxd and AsyncBoxd are the same surface — same namespaces, same method names, same arguments, same return types. Switching is await and an import, not a rewrite.

from boxd import AsyncBoxd

boxd = AsyncBoxd(api_key="bxd_...")

machine = await boxd.machines.create("my-machine")
result = await boxd.machines.exec(machine.id, "echo hello")
await boxd.close()

AsyncBoxd is an async context manager too:

async with AsyncBoxd(api_key="bxd_...") as boxd:
    ...

Two methods stay un-awaited, because they hand back something to iterate rather than a result: stream_exec returns the session object directly, and logs is an async generator you drive with async for.

Use AsyncBoxd when you already have an event loop (FastAPI, asyncio scripts, anyio). Use Boxd everywhere else — scripts, notebooks, Django views.

Machines

machine = boxd.machines.create(
    "my-machine",
    vcpu=4,
    memory="16G",
    env={"MODE": "production"},
)
boxd.machines.get("my-machine")     # by name or id
boxd.machines.list()                     # a plain list
boxd.machines.list(org="acme")           # one organization's machines
boxd.machines.list(all_contexts=True)    # every org you belong to
boxd.machines.delete("my-machine")

Only name is positional; everything else is keyword-only. The full set of create options:

boxd.machines.create(
    "builder",
    image="ubuntu:24.04",
    org="acme",                   # create inside an organization
    shared=True,                  # and make it visible to every member
    env={"API_URL": "https://example.com"},
    cmd=["/usr/local/bin/start"],
    restart_policy="always",      # "always" | "never"
    vcpu=2,
    memory="8G",                  # "8G", "512M", or a byte count
    disk="100G",
    auto_suspend_timeout=300,     # seconds; 0 disables
    auto_destroy_timeout=0,
    ssh=True,                     # give the machine an SSH port
    proxies=[ProxyEntry(name="api", port=3000)],
    volumes=[VolumeMount(disk_id="d_...", mount_path="/data", read_only=False)],
)

boxd.machines.create()            # every option is optional — cluster default image

ProxyEntry and VolumeMount are importable from boxd. A ProxyEntry with port=0 has its port detected inside the machine.

State:

boxd.machines.start(id)
boxd.machines.stop(id)
boxd.machines.reboot(id)
boxd.machines.pause(id)        # suspend to RAM — fast to resume; PauseResult(suspend_us)
boxd.machines.resume(id)       # ResumeResult(resume_us)
boxd.machines.hibernate(id)    # suspend to disk — cheaper, slower to wake
boxd.machines.wake(id)

Everything else:

boxd.machines.fork("my-machine", "my-copy")     # live clone
boxd.machines.fork(id, shared=True, vcpu=8)     # same sizing options as `create`
boxd.machines.rename(id, "new-name")            # returns the new name; reboots the machine
boxd.machines.share(id)                         # visible to your whole org
boxd.machines.unshare(id)
boxd.machines.set_auto_suspend_timeout(id, 300) # seconds idle; 0 disables
boxd.machines.set_auto_hibernate_timeout(id, 0)
boxd.machines.wait_until_ready(id)
boxd.machines.wait_until_ready(id, timeout=180.0, poll_interval=1.0)   # seconds
boxd.machines.suggest_name()

A fork inherits the source's sizing for anything you leave unset, and is private to you unless you pass shared=True.

create and fork return once the machine is scheduled, not once it is usable. Call wait_until_ready before doing anything that depends on it running — especially before forking it again.

The Machine record

machine.id, machine.name, machine.image_ref
machine.status                  # "pending" | "starting" | "running" | "suspended" |
                                # "hibernated" | "stopped" | "failed" | "destroyed" |
                                # "migrating"
machine.restart_policy          # str | None
machine.created_at              # datetime | None — None when none is on record

machine.resources.vcpu          # what the machine actually got, not what you
machine.resources.memory_bytes  # asked for — always concrete
machine.resources.disk_bytes

machine.org                     # OrgRef(id, name) | None — None = personal quota
machine.shared                  # shared with that org, or private to you

machine.access.ssh_port         # int | None — None until allocated
machine.access.domain
machine.access.url              # https://<name>.<domain>

machine.idle.suspend_after      # seconds; 0 = that timer is disabled
machine.idle.hibernate_after
machine.idle.destroy_after

machine.source                  # MachineSource | None — None = booted from an image
machine.source.kind             # "fork" | "snapshot"
machine.source.name             # source machine, or snapshot name
machine.source.version          # int | None — snapshots only; a fork has none
machine.source.id               # str | None — provenance; may not resolve

machine.hibernated_at           # datetime | None — None = not hibernated
machine.last_connected_at       # datetime | None — None = never connected
machine.boot_time_ms            # int | None — last boot; None = never booted

None always means "not set": a port that was never allocated, a boot that never happened, an org you do not have. Where 0 is a real answer — a disabled idle timer — it stays 0.

org is the org the machine belongs to and is billed to; shared says whether your teammates can see it. A private machine can still be org-billed, so org set with shared=False is normal, not a contradiction.

source.id points at the machine or snapshot this one came from. It is a record of where the machine came from, not a live link — it may not resolve, and a lookup that finds nothing is normal.

MachineStatus is importable from boxd when you want the literal type; a status a newer server introduces is passed through as a plain string.

Creating from a snapshot

boxd.snapshots.create(machine_id, "golden")
machine = boxd.machines.create("from-golden", from_snapshot="golden")

Restoring a snapshot replays the machine as it was captured, so from_snapshot goes with name, org and the sizing options. Combining it with image, env, cmd, restart_policy or shared raises ValueError.

Exec

result = boxd.machines.exec(id, "cargo build")
result.stdout      # str
result.stderr      # str — populated for non-PTY execs
result.exit_code   # int
result.success     # bool

boxd.machines.exec(id, ["echo", "a b"])              # a list is quoted for you
boxd.machines.exec(id, "env", env={"FOO": "bar"})
boxd.machines.exec(id, "cargo build", timeout=30)    # seconds

# Under a PTY, stderr merges into stdout and `stderr` comes back empty.
boxd.machines.exec(id, "top -b -n1", tty=True, cols=120, rows=40)

command takes a list of argv — shell-quoted for you — or a ready-made command line as a string. timeout gives up on the call; whatever it started inside the machine may well still be running.

For anything interactive, stream_exec gives you a live session — the one handle in the SDK, because a bidirectional stream really is stateful:

with boxd.machines.stream_exec(id, command="bash", tty=True) as stream:
    stream.write(b"ls\n")       # bytes or str
    stream.write_eof()          # half-close stdin; the process sees EOF
    for chunk in stream:        # bytes — merged output, what a terminal would show
        print(chunk.decode(errors="replace"), end="")
    print("exited", stream.exit_code)

stream_exec takes command and the rest as keywords, and hands back the session without a round trip. exit_code is None until the stream is exhausted. Leaving the with block — or calling close() — ends the session.

iter_chunks() yields OutputChunk(data, is_stderr) when you need the two streams apart. Under tty=True the terminal merges them, so everything arrives as stdout — set tty=False if you need the split.

For a headless one-shot that reads stdin (jq, cat, claude -p), pass close_stdin=True so it sees end-of-input immediately instead of hanging. Combining it with tty=True raises ValueError — a shell needs stdin open.

Set the terminal size with cols/rows, and call stream.resize(cols, rows) when the local terminal changes size:

import shutil, signal

cols, rows = shutil.get_terminal_size()
stream = boxd.machines.stream_exec(id, command="htop", tty=True, cols=cols, rows=rows)
signal.signal(signal.SIGWINCH, lambda *_: stream.resize(*shutil.get_terminal_size()))

Logs

for chunk in boxd.machines.logs(id):
    print(chunk.decode(errors="replace"), end="")

for chunk in boxd.machines.logs(id, follow=True):   # stays open
    ...

Files

from pathlib import Path

written = boxd.machines.files.upload(id, "/app/config.json", '{"debug": true}')
boxd.machines.files.upload(id, "/app/data.bin", Path("local.bin").read_bytes())
data = boxd.machines.files.download(id, "/app/output.json")   # bytes

upload takes str or bytes, streams it in chunks so large files are fine, and returns the number of bytes the machine confirmed it wrote.

Ports and proxies

fwd = boxd.machines.ports.expose(id, 8080)            # public TCP forward
boxd.machines.ports.expose(id, 5353, protocol="udp")  # "tcp" | "udp" | "both"
fwd.dns, fwd.public_port, fwd.machine_port, fwd.protocol
fwd.machine_id, fwd.machine_name
boxd.machines.ports.unexpose(id, 8080)                # echoes back what it removed
boxd.machines.ports.list(id)                          # one machine's forwards
boxd.machines.ports.list()                            # every forward you own

Connect on dns:public_port. Max 3 forwards per machine. Re-exposing a machine port keeps its public port and just updates the protocol set; "both" shares one public port across TCP and UDP.

ports.list() is account-wide — pass a machine to narrow it, or filter on .machine_id / .machine_name.

route = boxd.machines.proxies.create("my-machine", "api", 3001)  # api.<machine>...
route.name, route.port
routes = boxd.machines.proxies.list("my-machine")
routes[0].name          # str | None — None on the machine's default route
routes[0].domain        # the hostname this route answers on
routes[0].port          # int — where traffic actually goes
routes[0].port_mode     # "locked" (you pinned it) | "auto" (detected for you)
routes[0].is_default
routes[0].machine_id, routes[0].machine_name
boxd.machines.proxies.set_port("my-machine", 3000, name="api")
boxd.machines.proxies.set_port("my-machine", "auto")  # default route, auto-detected
boxd.machines.proxies.delete("my-machine", "api")

These take an id or a name, like everything else on machines. name is a subdomain label: lowercase letters, digits and hyphens, not starting or ending with one. create answers as soon as the route is accepted, so it confirms the subdomain and the port it was pointed at; list() reports the full domain and the resolved port.

Checkpoints

Per-machine captures, restored in place. They are deleted with the machine.

cp = boxd.machines.checkpoints.create(id, "before-upgrade")
cp.id, cp.name, cp.status
saved = boxd.machines.checkpoints.list(id)
saved[0].size_bytes
saved[0].created_at     # datetime
saved[0].created_by     # str | None
saved[0].available      # restorable right now
boxd.machines.checkpoints.restore(id, "before-upgrade")
boxd.machines.checkpoints.delete(id, "before-upgrade")

The machine must be running to take a checkpoint. status is "pending" until the artifact lands, then "ready" (or "failed"); restore wants one that is "ready" and available.

Environment variables and secrets

Two namespaces with identical methods. The difference is that a secret's value is write-only — the server never returns it, and the Secret model has no value field at all.

boxd.env.set("MODE", "production", scope="all")
boxd.env.list()                    # EnvVar(name, scope, value)
boxd.env.list(org="acme")
boxd.env.delete("MODE", scope="all")

boxd.secrets.set("API_TOKEN", "s3cr3t", scope="shared")
boxd.secrets.list()                # Secret(name, scope) — no value
boxd.secrets.delete("API_TOKEN", scope="shared")

scope defaults to "shared" on set and delete; list takes org only and reports every scope. Pass org="acme" to any of these to work in an organization instead of your personal scope.

set, delete and move each return the server's human-readable confirmation of what it did.

Scope decides which machines a name applies to:

Scope is part of a name's identity — the same name can exist in several scopes at once — so changing it is a move between two addresses, and both ends are required:

boxd.secrets.move("API_TOKEN", from_scope="private", to_scope="shared")

Calling it twice fails the second time. Environment variables and secrets share one namespace within a scope, so an environment variable can block a secret of the same name moving in, and vice versa.

Snapshots and disks

snap = boxd.snapshots.create(machine_id, "golden")   # re-saving bumps the version
snap.id, snap.name, snap.version, snap.status
boxd.snapshots.get("golden")                         # by name or id
boxd.snapshots.list()
boxd.snapshots.delete("golden")
boxd.snapshots.list(org="acme")      # `org` works on get/list/delete too

disk = boxd.disks.create("data", "10G")   # bytes or a human string
disk.id, disk.name, disk.size_bytes, disk.status
boxd.disks.attach(disk.id, machine_id, "/mnt/data")
boxd.disks.attach(disk.id, machine_id, "/mnt/data", read_only=True)
boxd.disks.detach(disk.id, machine_id)
boxd.disks.list()
boxd.disks.delete(disk.id)

The machine must be running to snapshot it, and create answers before the artifact lands — status is "pending" until it does. Snapshots stay inside one organization.

A disk is always created writable; read-only is chosen per attachment. A disk can be attached to only one machine at a time.

create confirms only what the server can answer immediately; the full records come back from get and list:

snapshot = boxd.snapshots.get("golden")
snapshot.id, snapshot.name
snapshot.version        # int | None — latest ready version; None = nothing captured yet
snapshot.status         # "pending" | "ready" | "failed"
snapshot.size_bytes
snapshot.created_at     # datetime | None — the first capture; it stays put
snapshot.updated_at     # datetime | None — the most recent capture
snapshot.vcpu           # the sizing the machine was captured at
snapshot.memory_bytes
snapshot.use_count      # machines restored from it so far

volume = boxd.disks.list()[0]
volume.id, volume.name, volume.size_bytes
volume.status           # "creating" | "ready" | "destroyed" — attach once "ready"
volume.created_at       # datetime | None
volume.attachments      # [DiskAttachment(machine_id, machine_name, mount_path, mount_mode)]
                        # mount_mode is "ro" or "rw"

Organizations, credentials, account

orgs = boxd.orgs.list()              # a plain list
orgs[0].id
orgs[0].name                         # display label — it can repeat across organizations
orgs[0].slug                         # the organization's unique key
orgs[0].is_admin                     # you administer it
orgs[0].is_default                   # where your personal machines are billed

Anywhere a call takes org, it accepts an organization's name or id.

key = boxd.api_keys.create(
    "ci",
    org="acme",                      # the organization the key is fenced to
    kind="member",                   # "member" (default) acts as you within that org;
                                     # "org" is a userless service credential, limited
                                     # to the org's shared fleet, org admin only
    expires_in=60 * 60 * 24 * 30,    # seconds; 0 for no expiry
)
key.id
key.api_key                          # the raw key — shown once, store it now
key.expires_at                       # datetime | None

keys = boxd.api_keys.list()
keys[0].name, keys[0].key_prefix, keys[0].created_at
keys[0].last_used_at                 # datetime | None — None = never used
keys[0].expires_at                   # datetime | None — None = no expiry
keys[0].org, keys[0].kind            # "member" | "org"
boxd.api_keys.delete(key.id)

Every key is fenced to exactly one organization. Deleting one takes effect immediately.

from pathlib import Path

me = boxd.account.get()
me.user_id
me.display_name                      # str | None — falls back to `user_id`
me.pubkey_fingerprints               # list[str]
me.billing.subscription_status       # "active", "trialing", … | None
me.billing.past_due_since            # datetime | None
me.billing.max_vms                   # effective quota
me.billing.vcpu, me.billing.memory_bytes

pubkey = (Path.home() / ".ssh/id_ed25519.pub").read_text()
boxd.account.link_ssh_key(pubkey)
boxd.account.link_ssh_key(
    pubkey,
    device_id="laptop",              # one key kept per device — re-linking replaces it
    label="MacBook Pro",             # shown wherever the device is listed
)

cfg = boxd.account.config()
cfg.default_image, cfg.zone

Errors

from boxd import (
    BoxdError,               # base class — catch this to catch everything
    AuthenticationError,     # no usable credential, or it was rejected
    PermissionDeniedError,   # authenticated, but not allowed
    NotFoundError,
    ConflictError,           # already exists, or fights the current state
    RateLimitError,          # rate limit or quota
    APIStatusError,          # any other error from the server
    APIConnectionError,      # could not reach the server
)

try:
    boxd.machines.get("nope")
except NotFoundError:
    ...

Every error carries .message, .code (the canonical status name, e.g. "not_found") and .grpc_code, the numeric status code.

Connection failures are retried with exponential backoff, max_retries times. Timeouts are never retried — the server may already have applied the request — and neither is AuthenticationError.

Update notices

The SDK prints a one-time note to stderr if the server reports a newer release:

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

It fires at most once per process and never causes a request to fail.

The installed version is available as boxd.__version__.

Development

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

pytest                                     # unit tests
bash scripts/compile_proto.sh              # regenerate stubs after an API change