Python SDK#

The dormo package is written by hand against the OpenAPI spec (fields and paths follow the spec exactly), with sandbox handles, streaming, retries and webhook verification on top. Python 3.9+, one dependency (httpx). Sync (Client) and async (AsyncClient) clients share the same methods.

pip install dormo            # core
pip install "dormo[pty]"     # adds websockets for sandbox.pty.connect()

Client#

import os
from dormo import Client

client = Client(
    api_key=os.environ["DORMO_API_KEY"],  # default: this env var
    base_url="https://api.dormo.dev/v1",          # default
    max_retries=2,                                 # retries errors with retryable=True
    timeout=120.0,                                 # seconds
)

Fields use the API's snake_case names (auto_pause, timeout_seconds, exit_code). Responses are typed objects with attributes (result.exit_code) and .to_dict().

Sandboxes#

sandbox = client.sandboxes.create(
    template="base",                      # or snapshot_id=..., or image={"ref": "ghcr.io/acme/env:1"}
    size="default",                       # small | default | large
    name="user-42",
    labels={"user": "42"},
    env={"NODE_ENV": "production"},
    auto_pause={"enabled": True, "idle_timeout_seconds": 300},
    always_on=False,
    priority="standard",                  # or "batch" (−50%, may queue)
    ports=[{"port": 3000, "visibility": "private"}],
    wait=True,
    idempotency_key="user-42-create",
)
print(sandbox.id, sandbox.state)

same = client.sandboxes.get(sandbox.id)
for s in client.sandboxes.list(state=["running"], label=["user=42"]):   # auto-paginates
    print(s.id, s.state)

sandbox.update(auto_pause={"enabled": True, "idle_timeout_seconds": 900})
sandbox.update(always_on=True)
sandbox.pause()
sandbox.resume()                      # usually unnecessary: calls wake it
fork = sandbox.fork(name="user-42-try-2")
forks = sandbox.fork(count=10)        # list of 10 Sandbox objects
sandbox.refresh()
sandbox.wait_for("running", timeout=30)
sandbox.delete()

Bulk fork#

fork(count=N) creates up to 50 copies of a sandbox (memory included) in one call. Without count, or with count=1, it returns one Sandbox, as before. With count > 1 the API returns {"data": [...]} and the SDK returns a list of Sandbox. fork_many(count, ...) always returns a list. AsyncClient has the same methods (await sandbox.fork(count=N)).

Admission is all-or-nothing: if the source host cannot fit every fork, nothing is created and the call raises ApiError with capacity_unavailable (or concurrency_limit_reached if the plan's limit would be exceeded). With priority="batch" the forks are queued instead and come back in state queued.

rollouts = sandbox.fork_many(32, labels={"run": "eval-7"})
results = [s.exec(command="python3 rollout.py") for s in rollouts]

queued = sandbox.fork(count=50, priority="batch", wait=False)   # -50%, may queue

Exec#

r = sandbox.exec(command="pytest -q", cwd="/home/user/app", timeout_seconds=300)
print(r.exit_code, r.stdout, r.stderr, r.truncated)

sandbox.exec(args=["python3", "-c", "print(42)"])     # no shell

for ev in sandbox.exec_stream(command="pip install -r requirements.txt"):
    if ev.type in ("stdout", "stderr"):
        print(ev.data, end="")
    elif ev.type == "exit":
        print("exit", ev.exit_code)

cmd = sandbox.exec(command="python3 -m http.server 8080", background=True)
for ev in cmd.stream():          # attach; replays earlier output
    ...
cmd.write_stdin("q\n", eof=True)
cmd.kill(signal="SIGTERM")
done = cmd.wait()

sandbox.commands.list()
sandbox.commands.get(cmd.id)

PTY#

pty.connect() needs the pty extra (pip install "dormo[pty]") and is available on the sync Client only. AsyncClient supports pty.create(), which returns the websocket_url for your own WebSocket client.

with sandbox.pty.connect(cols=120, rows=40) as term:
    term.write(b"ls -la\n")
    print(term.read(timeout=1.0).decode())
    term.resize(160, 48)

Files#

sandbox.files.write("/home/user/app/.env", "PORT=3000\n", mode="0600")
text = sandbox.files.read_text("/home/user/app/package.json")
data = sandbox.files.read("/home/user/out.png")            # bytes
entries = sandbox.files.list("/home/user/app", recursive=False)
info = sandbox.files.stat("/home/user/app")
sandbox.files.mkdir("/home/user/data")
sandbox.files.move("/home/user/a.txt", "/home/user/b.txt")
sandbox.files.remove("/home/user/tmp", recursive=True)
sandbox.files.upload("/home/user/app", {"src/hello.txt": b"hi"})
archive = sandbox.files.download_archive("/home/user/app", format="tar.gz")
url = sandbox.files.download_url(path="/home/user/big.bin").url
up = sandbox.files.upload_url(path="/home/user/dataset.parquet", expires_in_seconds=900)

for ev in sandbox.files.watch("/home/user/app/src", recursive=True):
    print(ev.type, ev.path)

Networking#

port = sandbox.ports.expose(port=3000, visibility="private")
print(port.url)                         # https://<short_id>-3000.dormo.host
token = sandbox.access_tokens.create(ttl_seconds=3600, ports=[3000]).token
sandbox.ports.list()
sandbox.ports.remove(3000)

ssh = sandbox.ssh(public_key="ssh-ed25519 AAAA...", ttl_seconds=3600)
print(ssh.command)

Desktop#

stream = sandbox.desktop.stream(ttl_seconds=600, read_only=True)
print(stream.url)
sandbox.desktop.revoke(stream.id)

shot = sandbox.desktop.screenshot(format="png")    # .data (base64), .media_type, .width, .height
sandbox.desktop.input(
    actions=[
        {"action": "left_click", "coordinate": [640, 412]},
        {"action": "type", "text": "hello"},
        {"action": "key", "text": "Return"},
    ],
    return_screenshot=True,
)
pos = sandbox.desktop.cursor()   # .x, .y

Full agent loop: Desktop and computer use.

Snapshots and templates#

snap = sandbox.snapshots.create(name="after-pip-install")
copy = client.sandboxes.create(snapshot_id=snap.id)
client.snapshots.list(sandbox_id=sandbox.id)
client.snapshots.delete(snap.id)

tpl = client.templates.create(
    name="agent-env",
    image={"ref": "ghcr.io/acme/agent-env:1.4"},
    ready_check={"command": "curl -sf localhost:8080/health", "timeout_seconds": 120},
)
print(client.templates.logs(tpl.id))
client.sandboxes.create(template="agent-env")

Webhooks, usage, keys, account#

hook = client.webhooks.create(
    url="https://example.com/hooks/dormo",
    events=["sandbox.ready", "sandbox.paused", "sandbox.resumed", "sandbox.error", "sandbox.deleted"],
)
print(hook.secret)   # whsec_... (shown once)

from dormo.webhooks import verify_webhook
event = verify_webhook(payload=raw_body, headers=request.headers, secret=os.environ["WEBHOOK_SECRET"])

usage = client.usage.get(group_by="sandbox", from_="2026-11-01T00:00:00Z")   # `from` is a Python keyword: use from_=
limits = client.limits.get()
key = client.api_keys.create(name="user-42", preset="full-sandbox", restrict={"labels": {"user": "42"}})
client.account.set_spend_limit(monthly_limit_usd=500, on_limit="pause_running")

Async#

import asyncio
from dormo import AsyncClient

async def main():
    client = AsyncClient()
    sandboxes = await asyncio.gather(*[client.sandboxes.create(template="base", priority="batch") for _ in range(20)])
    results = await asyncio.gather(*[s.exec(command="python3 eval.py") for s in sandboxes])
    await asyncio.gather(*[s.delete() for s in sandboxes])

asyncio.run(main())

Batch sandboxes may wait in queued; create with priority="batch" returns immediately, and exec waits for the sandbox to start.

Errors#

from dormo import ApiError

try:
    client.sandboxes.create(size="large")
except ApiError as e:
    print(e.status, e.code, e.retryable, e.request_id, e.details)

Errors with retryable=True are retried automatically up to max_retries, honouring Retry-After.