Python: Sandbox

​

Sandbox

​

Create, run, clean up — the basic loop

from sandkiln import Sandbox, SandkilnApiError

sandbox = Sandbox.create(tags={"env": "ci", "owner": "pipeline"}, vcpu_count=2, mem_size_mib=1024)

result = sandbox.run_command("python3", ["analyze.py", "--input", "data.csv"])
if result.exit_code != 0:
    raise RuntimeError(f"analyze.py failed: {result.stderr}")
print(result.stdout)

sandbox.write_file("/tmp/report.json", '{"ok": true}')
report = sandbox.read_file("/tmp/report.json")

# One-shot run, never needed again -- skip the snapshot entirely.
sandbox.stop(keep=False)

• Sandbox.create(name=None, tags=None, base_url=None, auth_token=None, vcpu_count=None, mem_size_mib=None, image_id=None, rate_limit=None, drives=None) — boots a sandbox. name is a caller-given identity, unique among live sandboxes and held snapshots (409 if already taken) — see by_name/get_or_create below. vcpu_count/mem_size_mib override the daemon’s configured defaults, subject to its ceiling. image_id boots from a registered image instead of the default rootfs (see Image). rate_limit caps host I/O via Firecracker’s own token-bucket limiter (unlimited if omitted). drives is a list of DriveAttachment to attach existing persistent drives (see Drive). A call with no drives/rate_limit matching a configured pool’s image/resources transparently resumes a warm snapshot instead of cold-booting — see Pool.
• sandbox.run_command(command, args=None) — returns an ExecResult (stdout, stderr, exit_code).
• sandbox.read_file(path) — returns file contents as bytes.
• sandbox.write_file(path, content) — content is str or bytes.
• sandbox.chmod(path, mode), sandbox.chown(path, uid, gid), sandbox.mkdir(path, parents=False), sandbox.rename(from_path, to_path), sandbox.copy(from_path, to_path), sandbox.symlink(target, link_path), sandbox.readlink(path), sandbox.truncate(path, size), sandbox.list_dir(path) (returns list[DirEntry] with real is_dir/is_symlink/size/mode/mtime metadata) — the full filesystem operation set.
• sandbox.stop(keep=None) — stops the sandbox. Default (keep omitted or True) preserves state as a resumable snapshot, returns a StopResult(kept, snapshot_id). keep=False fully destroys instead.

​

Attaching to an id you already have

import os
from sandkiln import Sandbox

# A worker process that only has an id passed to it -- no create() round-trip needed.
sandbox = Sandbox.attach(os.environ["SANDBOX_ID"], auth_token=os.environ["SANDKILN_TOKEN"])
sandbox.run_command("echo", ["still here"])

• Sandbox.attach(id, base_url=None, auth_token=None) — wraps an existing sandbox id without a network round-trip.

​

Named sandboxes — find the same environment again later

from sandkiln import Sandbox, SandkilnApiError

# First call today: creates a fresh sandbox named "agent-session-42".
# Every later call, any day: resumes it if it was stopped, returns it as-is if still live.
sandbox, created = Sandbox.get_or_create(name="agent-session-42")
if created:
    sandbox.run_command("pip", ["install", "-r", "requirements.txt"])  # one-time setup
sandbox.run_command("pytest")
sandbox.stop()  # preserved by default -- nothing to do here

# Elsewhere: resolve the same name to a *live* sandbox without creating anything.
try:
    live = Sandbox.by_name("agent-session-42")
except SandkilnApiError as e:
    pass  # 409 if the name currently belongs to a stopped sandbox -- use get_or_create for that

• Sandbox.by_name(name, base_url=None, auth_token=None) — resolves a name to a live sandbox. Raises SandkilnApiError (409) if the name currently belongs to a stopped (snapshotted) sandbox instead — use get_or_create for that.
• Sandbox.get_or_create(name, tags=None, base_url=None, auth_token=None, vcpu_count=None, mem_size_mib=None) — resolves name to a sandbox in one race-safe call: live as-is, resumed if stopped, created fresh otherwise. Returns (sandbox, created).

​

Listing and filtering

ci_sandboxes = Sandbox.list(tags={"env": "ci"})
for info in ci_sandboxes:
    print(info.id, info.name, info.created_at)

• Sandbox.list(tags=None, base_url=None, auth_token=None) — lists sandboxes; tags filters by exact match on every given key.

​

Dev-server preview

sandbox = Sandbox.create()
sandbox.run_command("sh", ["-c", "npm run dev &"])  # background the dev server
url = sandbox.preview_url(3000, path="/health")
print(url)  # open this in a browser, or request it yourself

• sandbox.preview_url(port, path="/") — the URL a browser can open to reach a server listening on port inside the sandbox.

​

Local tunnel

sandbox = Sandbox.create()
tunnel = sandbox.tunnel(8080, local_port=3000)
# Code inside the sandbox connecting to 127.0.0.1:8080 now reaches
# whatever's listening on this machine's 127.0.0.1:3000.
sandbox.run_command("curl", ["http://127.0.0.1:8080/"])
tunnel.close()

• sandbox.tunnel(guest_port, local_port, local_host="127.0.0.1") — returns a TunnelHandle with .tunnel_id and .close(). The relay runs on a background thread (AsyncSandbox.tunnel(...) runs it as an asyncio.Task instead). See Local tunnel.

​

Snapshot, resume, fork

snapshot_id = sandbox.snapshot()  # save state, stop the VM

# ...later, possibly in a different process...
resumed = Sandbox.resume(snapshot_id)  # consumes the snapshot
resumed.run_command("echo", ["back from a snapshot"])

# Or keep the snapshot reusable by forking instead of resuming:
fork1 = Sandbox.fork(snapshot_id)
fork1.stop()  # frees the fork; the snapshot is still there
fork2 = Sandbox.fork(snapshot_id)  # fork it again

• sandbox.snapshot() — saves full state to disk and stops; returns a snapshot id.
• Sandbox.resume(snapshot_id, base_url=None, auth_token=None) — boots from a snapshot, consuming it.
• Sandbox.fork(snapshot_id, base_url=None, auth_token=None) — boots from a snapshot without consuming it. Raises SandkilnApiError (409) while an earlier fork is still live.
• Sandbox.list_snapshots(source_sandbox_id=None, base_url=None, auth_token=None) — lists snapshots. source_sandbox_id narrows to the one taken from that original sandbox id.

​

Remote storage mounts

sandbox = Sandbox.create()
mount = sandbox.mount("build-cache", "https://s3.example.com", access_key="...", secret_key="...", mount_path="/mnt/cache")
# Ordinary read_file()/write_file() against mount_path now read/write through to the bucket.
sandbox.unmount(mount.id)

• sandbox.mount(bucket, endpoint, access_key, secret_key, mount_path, read_only=False) — mounts an S3-compatible bucket at mount_path via rclone mount. endpoint is the full URL of the S3-compatible service, never defaulted; access_key/secret_key are written into a 0600 rclone config file inside the guest, never passed as a command-line argument. Needs the optional FUSE-capable rootfs setup in SELF_HOSTING.md; raises SandkilnApiError on a guest-side mount failure. Not carried across resume()/fork() — it’s a live guest FUSE process, already captured by Firecracker’s own snapshot.
• sandbox.list_mounts() — lists every remote-storage mount currently active in this sandbox.
• sandbox.unmount(mount_id) — tears down one mount. Best-effort on the guest side — a failed umount is a daemon-log warning, not a failed request.