platinum-sdk (Python)

Python client for Platinum — hardware-isolated sandbox microVMs (one Cloud Hypervisor VM per sandbox, sub-second boots via a warm pool).

pip install platinum-sdk     # import name: platinum

Python ≥ 3.9, typed (PEP 561 — ships py.typed; mypy --strict clean). Sync and async clients. Server-side use only — never ship an API key into client-side code.

Quickstart

from platinum import Platinum

dn = Platinum(token="pt_live_...", api_url="https://api.platinum.dev")
# or: PT_TOKEN / PT_API_URL env vars

# Pick a template that exists on YOUR deployment first:
print(dn.templates.list())

sbx = dn.sandboxes.create(template="pt-base", wait_for_running=True)
r = sbx.exec(["uname", "-a"]).check()     # .check() raises on non-zero exit
print(r.stdout)
sbx.delete()

Async — a full 1:1 mirror of the sync surface:

from platinum import AsyncPlatinum

async with AsyncPlatinum(token="pt_live_...", api_url="https://api.platinum.dev") as dn:
    sbx = await dn.sandboxes.create(template="pt-base", wait_for_running=True)
    r = await sbx.exec("uname -a")        # argv list or plain string
    print(r.stdout)
    await sbx.delete()

Expose a port at create time (URL comes back in the same response):

sbx = dn.sandboxes.create(
    template="pt-base",
    expose=[{"port": 8080, "public": True}],
    wait_for_running=True,
)
print(sbx.exposed_url(8080))

Build a custom image inline (cache-hit on repeat, built on first use):

from platinum import Platinum, Template

dn = Platinum()
image = (Template.from_python_image("3.12-slim")
                 .pip_install(["fastapi", "uvicorn"])
                 .workdir("/app"))
sbx = dn.sandboxes.create(image=image, wait_for_running=True, wait_timeout_ms=600_000)

Configuration

Errors

Everything the SDK raises extends PlatinumError (.status, .body, .code — the API's machine-readable error code, also folded into str(err)):

from platinum import NotFoundError, ConflictError, RateLimitError

try:
    sbx.exec("true")
except NotFoundError:            # sandbox gone
    ...
except ConflictError as e:       # e.g. e.code == "sandbox_not_running"
    ...
except RateLimitError as e:      # back off e.retry_after_seconds — the SDK never retries
    ...

Subclasses: ValidationError (400) · AuthenticationError (401) · ForbiddenError (403) · NotFoundError (404) · ConflictError (409) · RateLimitError (429) · ServerError (5xx) · PlatinumTimeoutError / PlatinumConnectionError (client-side, status == 0). No automatic retries, ever — a 429 or a failed create is surfaced, never silently retried.

Surface (at parity with the TypeScript SDK)

The two SDKs expose the same operations — enforced in CI by verify/sdk-parity.sh (name-level) and the offline unit suites (behaviour). The async client mirrors the sync surface 1:1 (enforced by tests/test_unit.py::test_async_mirrors_sync_surface).

JuiceFS Shared Volumes

With the independent JuiceFS feature enabled:

# Returns the volume named "data", creating it (and waiting for it) if absent.
vol = dn.volumes.get_or_create("data", type="shared", driver="juicefs", size_gib=20, inode_quota=100_000)
dn.volumes.attach(vol["id"], sbx.id, "/mnt/data")

# Choose the key yourself when a retry must be safe across process restarts.
dn.volumes.resize(vol["id"], 40, idempotency_key=f"resize-{vol['id']}-40")

Every lifecycle write (sync and async) sends an Idempotency-Key. The SDK generates one per call when you pass none and never retries on its own. If a call fails with an unknown outcome (lost response, timeout, 5xx), the key it sent is on the error as err.idempotency_key: resend it with idempotency_key=err.idempotency_key to recover the same operation instead of starting a second one. Pass your own idempotency_key when a retry must survive a process restart or comes from your own retry loop. volumes.barrier(id) returns the key it used as idempotency_key on the receipt. get_operation and wait_operation only read the durable operation. Restore is fork-only. See the JuiceFS client parity guide for the cross-client contract. When juicefsGitWorkspacesEnabled is discovered, volumes.git / async_volumes.git expose public project/workspace lifecycle and project usage reconciliation. Git reads return either content or {"pending": True, "error": ...} and never a host command id, path, remote, gitdir, or credential.

Run something that outlives the call — exec kills its process group the moment it returns, process.start does not:

p = sbx.process.start(["python", "-m", "http.server", "8000"], cwd="/workspace")
print(p.handle, p.pid)

for chunk in p.stream():            # resumes at the last byte if the socket drops
    if chunk.type in ("stdout", "stderr"):
        if chunk.dropped:           # the ring buffer evicted bytes — a HOLE, not a stat
            print(f"[{chunk.dropped} bytes lost]")
        print(chunk.data, end="")   # str, decoded here — a split emoji arrives whole
    elif chunk.type == "reconnect":  # incl. the server's 30-min cap: it keeps going
        print(f"[resumed at {chunk.stdout_offset}: {chunk.reason}]")
    elif chunk.type == "exit":      # the only terminal event — the loop ends here
        print("exit", chunk.exit_code)

p.signal("SIGTERM")                 # or p.kill() — the record (logs + exit code) is kept

Reconnect to it later from anywhere — reads are by absolute byte offset, so nothing is replayed and nothing is lost:

p = sbx.process.get(handle)
out = p.logs(stdout_offset=my_last_offset)
print(out.stdout, out.stdout_offset, out.running, out.exit_code)

raw = p.logs(encoding="base64").stdout       # bytes, not str — binary-safe

Interactive PTY (stdin also delivers control characters), and a server-side wait:

sh = sbx.process.start("bash -i", pty=True, cols=120, rows=40)
sh.stdin("ls -la\n")
sh.resize(100, 30)
sh.stdin(b"\x03")                   # ^C

done = sbx.process.start(["make", "build"]).wait(timeout_ms=600_000)
print(done.exit_code)

Async is the same surface with await / async for (await sbx.process.start(...), async for chunk in p.stream()). Managed processes need an in-VM agent that implements them; an older guest answers 409 managed_procs_unsupported and exec still works.

Watch for file changes:

for ev in sbx.files.watch("/workspace", max_seconds=60):
    if ev["event"] == "change":
        print(ev["data"]["type"], ev["data"]["path"])

Not wrapped yet: interactive terminal (WebSocket), cross-host migrate (admin-only), API-key management, audit-log queries, billing.

Limits worth knowing

• files.write bodies must stay ≤ 16 MiB — larger writes currently truncate and fail through the vsock path. Split big files into ≤ 16 MiB chunks, or fetch them directly inside the sandbox (exec with wget).
• exec/run_code are buffered, not streaming — output arrives after the command exits (default timeout 30 s via timeout_ms). run_code source is capped at 1 MiB.
• Template names and minimums are per-deployment. pt-base (busybox — no git/pip/node/httpd inside; nc/wget available) exists on default installs; hosted deployments may only offer other templates with higher cpu/ram minimums. Call templates.list() first; a below-minimum create fails with a clear 400.
• Background processes are reaped when their exec call returns. Use sbx.process.start(...) for anything that has to keep running — it is owned by the guest agent, not by the request, and its output and exit code stay readable. The old workaround (setsid sh -c '<server loop>' >/dev/null 2>&1 < /dev/null &, see examples/02_expose_service.py) still works but loses both.

Testing

pip install -e '.[test]' && pytest runs the offline unit suite (httpx.MockTransport, no network). The live e2e is opt-in: PT_E2E=1 PT_API_URL=… PT_TOKEN=… pytest test_e2e.py.

Examples

Runnable scripts in examples/:

pip install platinum-sdk
PT_API_URL=… PT_TOKEN=… python packages/sdk-py/examples/01_create_exec_delete.py

License

MIT