forkd Python SDK — spawn, branch, exec, and eval microVMs

pip install forkd

Controller(base_url=None, token=None, timeout=60.0)

list_snapshots() -> list[dict]

delete_snapshot(tag: str) -> None

spawn_sandboxes(
    snapshot_tag: str,
    n: int = 1,
    per_child_netns: bool = False,
    memory_limit_mib: int | None = None,
    prewarm: bool = False,
    live_fork: bool = False,
    hugepages: bool = False,
) -> list[dict]

list_sandboxes() -> list[dict]

get_sandbox(sandbox_id: str) -> dict

kill_sandbox(sandbox_id: str) -> None

branch_sandbox(
    sandbox_id: str,
    tag: str | None = None,
    diff: bool = False,
    measure_diff: bool = False,
    mode: BranchMode | None = None,
    wait: bool = True,
) -> dict

exec_command(sandbox_id: str, args: list[str], timeout_secs: int = 30) -> dict

eval_code(sandbox_id: str, code: str) -> dict

ping_sandbox(sandbox_id: str) -> dict

Sandbox(
    tag: str | None = None,
    target: str | None = None,
    timeout: int = 30,
    *,
    spawn: bool = True,
)

with Sandbox() as sb:
    print(sb.commands.run("echo hi").stdout)  # hi

Sandbox.create(*args, **kwargs) -> Sandbox

sb.commands.run(cmd: str | list[str], timeout: int = 30) -> CommandResult

sb.eval(code: str) -> object

sb.ping() -> dict

sb.kill() -> None

BranchMode = Literal["full", "diff", "live"]

@dataclass
class CommandResult:
    stdout: str
    stderr: str
    exit_code: int

class ControllerError(RuntimeError):
    status: int    # HTTP status code
    body: Any      # Parsed JSON body, or raw string if not JSON
    url: str       # Full URL of the failed request

from forkd import Controller, ControllerError

c = Controller()
try:
    c.kill_sandbox("sb-missing")
except ControllerError as e:
    if e.status == 404:
        print("sandbox not found")
    else:
        raise

import time
from forkd import Controller, Sandbox

# ── Sandbox: E2B-compatible in-guest execution ───────────────────────
print("=== with-block (auto-kill on exit) ===")
with Sandbox() as sb:
    print("agent:", sb.ping())

    t = time.perf_counter()
    r = sb.commands.run("python3 -c 'import numpy; print(numpy.eye(3))'")
    print(f"exec (fresh subprocess) [{(time.perf_counter()-t)*1000:.0f} ms]")
    print(r.stdout.rstrip())

    t = time.perf_counter()
    r = sb.commands.run("echo hello from sandbox")
    print(f"exec echo [{(time.perf_counter()-t)*1000:.0f} ms]: {r.stdout.rstrip()}")

    t = time.perf_counter()
    result = sb.eval("numpy.zeros(5).tolist()")
    print(f"eval (warm PID-1 numpy) [{(time.perf_counter()-t)*1000:.0f} ms]: {result}")

print()

# ── Controller: lifecycle + live BRANCH fan-out ──────────────────────
c = Controller()  # reads FORKD_URL, FORKD_TOKEN from env

# Spawn a live-fork-capable parent. live_fork=True backs guest RAM with
# a memfd so later BRANCHes can use mode="live" (UFFD_WP).
children = c.spawn_sandboxes("pyagent", n=1, per_child_netns=True, live_fork=True)
sb_id = children[0]["id"]
print(f"spawned {sb_id} at {children[0]['guest_addr']}")

# Drive the sandbox via Sandbox (attach, don't spawn a new VM)
with Sandbox(target=children[0]["guest_addr"], spawn=False) as sb:
    print(sb.commands.run("uname -a").stdout.rstrip())

# BRANCH: mode="live" drops source pause to sub-50 ms.
# wait=False returns after ~10 ms; background copy completes asynchronously.
# Poll c.list_snapshots() until status="ready" before spawning grandchildren.
branch = c.branch_sandbox(sb_id, tag="checkpoint-1", mode="live", wait=False)
print(f"branch tag={branch['tag']} status={branch.get('status')}")

# Fan out 5 grandchildren from the branch.
grandchildren = c.spawn_sandboxes(branch["tag"], n=5, per_child_netns=True)
print(f"spawned {len(grandchildren)} grandchildren")

# Cleanup
c.kill_sandbox(sb_id)
for gc in grandchildren:
    c.kill_sandbox(gc["id"])