isb for Python

An asyncio SDK for isb: declarative incus sandboxes (containers and VMs). It is a thin client of isb rpc, a line-delimited JSON protocol over the isb binary's stdin and stdout (docs/rpc.md). All the work (planning, reconciling, readiness, exec) happens in isb; this package starts it, sends requests and maps the answers to Python types.

• Python 3.10 or later, Linux.
• No runtime dependencies (standard library only).
• Typed (py.typed), with TypedDicts for the spec generated from isb schema.

Install

pip install isb-sdk

Platform wheels (x86_64 and aarch64 Linux) bundle a static isb binary at isb/_bin/isb, so nothing else is needed. The pure wheel and the sdist do not; they use an isb binary from elsewhere.

The binary is looked up in this order:

1. Client(isb_bin="/path/to/isb")
2. the ISB_BIN environment variable
3. the bundled binary
4. isb on PATH

It must be a version with the rpc command (protocol 1). The client refuses a server that announces any other protocol version.

isb needs access to the incus socket ($INCUS_SOCKET, else $INCUS_DIR/unix.socket, else /var/lib/incus/unix.socket), which usually means membership in incus-admin. That access is root-equivalent on the host.

Quickstart

import asyncio
import isb
from isb import PortBinding, Sandbox, Volume


async def main() -> None:
    async with isb.Client() as client:
        sb = await Sandbox.connect_or_create(
            "dev-web",
            image="dev-base",
            client=client,
            cpus=4,
            memory="4GiB",
            idmap="auto",
            labels={"app": "web"},
            volumes={
                "/home/dev/src": Volume.bind("./src", device="src"),
                "/home/dev/.cache": Volume.named("dev-cache", owner="dev"),
            },
            ports=[
                PortBinding.host("5173", "5173", name="vite", search=20),
            ],
            ready=["running", "default_route", {"user_exists": "dev"}],
            exec={"user": "dev", "cwd": "/home/dev/src"},
            on_progress=print,
        )
        print(sb.last_report)

        # Captured output. argv is never joined into a shell string.
        out = await sb.exec("printf", ["[%s]", "a b", "$HOME"])
        assert out.stdout_text == "[a b][$HOME]"

        # Streaming output, as it is produced.
        script = "for i in 1 2 3; do echo $i; sleep 1; done"
        proc = await sb.exec_stream(["sh", "-c", script])
        async with proc:
            async for event in proc:
                print(event.kind, event.text, end="")
            print("exit code:", await proc.wait())

        await sb.remove(force=True)


asyncio.run(main())

Compose files

project = await isb.Project.load("isb.yaml", vars={"WORKTREE": "/srv/wt"})

for plan in await project.plan():
    print(plan.name, plan.status, plan.actions)

# Like `isb up -d`: returns once the sandboxes are ready. A service's
# `command` is for the foreground CLI `isb up` and is not run here.
for service, report in await project.up(on_progress=print):
    print(service, report.created, report.ports)

# A sandbox from the project carries its service's exec defaults.
web = project.sandbox("web")
await web.exec(["bun", "install"])

await project.down(volumes=True)

API overview

Client

Client(isb_bin=None, socket=None, project=None, create_timeout=None) owns one isb rpc subprocess, started on first use (or await client.start()) and stopped by await client.close() or async with. socket, project and create_timeout are passed as the global flags --socket, --project and --create-timeout. Requests run concurrently over the one process.

Every function takes an optional client=. Without one it uses isb.default_client(), created lazily with default settings (one per event loop). client.call(method, params) sends any protocol method directly.

A client belongs to the event loop it started on. If the subprocess exits, every pending and later request fails with ProcessError, whose message includes the tail of isb's stderr.

The server's working directory is fixed when it starts, so the SDK resolves relative paths itself: base_dir for bind mounts defaults to the current directory at the time of the call, and compose file paths are made absolute.

Sandbox

await Sandbox.create(name, *, image, **spec)	Create. AlreadyExistsError if the name is taken.
await Sandbox.connect_or_create(name, *, image, prune_devices=False, **spec)	Create or reconcile (only what differs changes). The report is on sb.last_report. Also Sandbox.ensure.
await Sandbox.plan(name, *, image, **spec)	What connect_or_create would do, as a Plan.
await Sandbox.get(name)	Handle on an existing sandbox. NotFoundError if missing.
await Sandbox.list_with(labels) / Sandbox.list()	list[SandboxInfo]. labels is {"k": "v", "k2": None} or ["k=v", "k2"].
await Sandbox.remove(name, force=False)	Delete. A running sandbox needs force.

**spec are the SandboxSpec fields of the compose format (docs/spec.md): cpus, memory, storage, type ("container", "virtual-machine" or "vm"), privileged, idmap, profiles, labels, env, volumes, ports, ready, ready_timeout, exec, raw_config, raw_devices. A full spec dict can be passed as spec= instead. The server rejects unknown fields (InvalidError). The create-style methods also take base_dir, wait_ready (default true), named_volumes (top-level named volume definitions, as in a compose file's volumes:) and on_progress (called with each progress line).

On an instance: info(), labels(), start(), stop(force=False, timeout="30s"), restart(), wait_ready(ready=None, ready_timeout=None), remove(force=False), add_port(port) (returns the listen address in use), remove_device(name) (returns whether it existed).

A handle from create, connect_or_create or Project.sandbox carries the spec's exec defaults (user, cwd, env, login) and sends them with every exec; per-call arguments override them.

Exec

out = await sb.exec(
    cmd,
    args=None,
    *,
    cwd=None,
    user=None,
    env=None,
    login=None,
    timeout=None,
    stdin=None,
    tty=False,
)

cmd is a program plus args, or a full argv list. Returns ExecOutput(exit_code, stdout: bytes, stderr: bytes) with stdout_text, stderr_text and success. A non-zero exit is not an exception. stdin is bytes or str, sent followed by EOF. timeout is seconds or a duration string ("90s"); when it passes, the command is killed and IsbTimeoutError is raised. With tty=True, all output arrives as stdout.

p = await sb.exec_stream(cmd, args=None, *, ..., stdin=None | "piped" | bytes)

returns an ExecProcess: iterate it for ExecEvent(kind, data) chunks in order, and drive it with await p.write(b), await p.close_stdin(), await p.signal(15), await p.resize(w, h). await p.wait() returns the exit code, await p.collect() gathers the rest into an ExecOutput. Used as async with, it kills the command on exit if it is still running.

Builders

These return plain dicts for the spec:

• Volume.bind(host_path, *, readonly=False, device=None, options=None)
• Volume.named(name, *, mode=NamedVolumeMode.ENSURE_EXISTS, owner=None, readonly=False, pool=None, device=None); NamedVolumeMode.EXISTING means the volume must exist (external: true).
• PortBinding.host(listen, connect, *, name=None, search=None): listen on the host, connect in the guest.
• PortBinding.guest(listen, connect, *, name=None): listen in the guest, connect on the host.

Project

await Project.load(files=None, *, env_files=None, project_name=None, vars=None) loads and resolves compose files (default ./isb.yaml, else ./isb.yml). vars win over the environment for ${VAR}. The result has name, base_dir, files, file (the resolved file, every sandbox named) and services. Then up(services=None, *, prune_devices=False, wait_ready=True) returns (service, ApplyReport) pairs, plan(...) returns list[Plan], down(services=None, *, volumes=False) deletes, and sandbox("web") returns a handle with that service's exec defaults.

Volumes and prune

isb.volumes.list(pool=None), get(name, pool=None), create(name, pool=None, *, config=None) (returns {"created", "pool"}, a no-op if it exists), remove(name, pool=None). isb.Volumes(client) has the same methods bound to one client. await isb.prune(label, dry_run=True) lists (or, with dry_run=False, deletes) sandboxes whose label value is a host path that no longer exists.

Types

SandboxInfo, ApplyReport, Plan, VolumeInfo, PruneResult, ExecOutput and ExecEvent are dataclasses. Plan actions are dicts tagged by action. The spec types (SandboxSpec, VolumeSpec, PortSpec, ReadyCheck, ExecDefaults, IdmapSpec, NamedVolumeSpec, ComposeFile) are TypedDicts in isb._spec, generated from the JSON Schema by scripts/gen_types.py:

ISB_BIN=/path/to/isb python3 scripts/gen_types.py          # regenerate
ISB_BIN=/path/to/isb python3 scripts/gen_types.py --check  # fail if stale

Errors

Every error is an IsbError with code (the protocol's stable code), message and data:

class	codes
NotFoundError	not_found
AlreadyExistsError	already_exists
NotReadyError	not_ready
IsbTimeoutError	request_timeout, operation_timeout, exec_timeout
InvalidError	invalid, interpolation, parse
ConnectError	connect (isb cannot reach incusd)
ApiError	api, operation_failed
ProtocolError	protocol (also unknown method), bad_request, a bad hello
ProcessError	process: the subprocess could not start or exited
BinaryNotFoundError	binary_not_found (a ProcessError)

Other codes (websocket, io, json) raise IsbError itself. IsbTimeoutError does not derive from the built-in TimeoutError.

Development

Unit tests need only an isb binary with rpc (they use a socket that does not exist, and fake servers for failure paths); integration tests need incusd and a local image with a dev user at uid 1000 and python3 (ISB_TEST_IMAGE, default dev-base). Both use the standard library's unittest, so no install is needed:

# from the repository root
cargo build
ISB_BIN=target/debug/isb PYTHONPATH=sdk/python/src python3 -m unittest discover -s sdk/python/tests -v
ISB_INTEGRATION=1 ISB_BIN=target/debug/isb PYTHONPATH=sdk/python/src python3 -m unittest discover -s sdk/python/tests -v

Integration tests name everything isb-test-py-<pid>-..., label it isb-test=py, and remove it afterwards.

Lint, type check and build (in sdk/python):

uv sync --group dev
uv run ruff check . && uv run ruff format --check .
uv run mypy
uv build                                      # pure wheel and sdist
ISB_WHEEL_BINARY=../../target/x86_64-unknown-linux-musl/release/isb \
ISB_WHEEL_PLAT=manylinux_2_17_x86_64.musllinux_1_2_x86_64 \
uv build --wheel                              # platform wheel with the binary

License

MIT