Python SDK reference

pip install orkestr

import os
from orkestr import Sandbox

# 1) Implicit: SDK reads ORKESTR_API_KEY from env
sbx = Sandbox.create(template="python-3.12")

# 2) Explicit: per-call api_key
sbx = Sandbox.create(template="python-3.12", api_key="ork_...")

# 3) Reusable client (good for long-running agents)
from orkestr import OrkestrClient
client = OrkestrClient(api_key="ork_...", base_url="https://api.orkestr.eu")
sbx = Sandbox.create(template="python-3.12", client=client)

from orkestr import Sandbox

sbx = Sandbox.create(
    template="python-3.12",
    size="small",
    network="off",
    timeout_seconds=600,
    env={"OPENAI_API_KEY": "sk-..."},
    metadata={"agent_run": "r_42"},
    region="fsn1",
)
print(sbx.id, sbx.status)

# Reattach from another process
sbx = Sandbox.get("sbx_01HXYZ...")

# List your sandboxes (newest first)
for sbx in Sandbox.list(status="running"):
    print(sbx.id, sbx.template, sbx.created_at)

# Find the sandbox for a session by its metadata tags (AND across keys).
# list() auto-pages; pass max_results to cap.
for sbx in Sandbox.list(metadata={"user_id": "42", "session": "abc"}):
    print(sbx.id)

limits = Sandbox.limits()
limits.plan                        # "free"
limits.allowed_sizes               # ["small"]
"medium" in limits.allowed_sizes   # False
limits.max_concurrent              # 2    (RAM budget as a count of small boxes)
limits.max_snapshots               # 1    (paused sandboxes you may hold)
limits.trial_credit_eur            # 10.0 (one-time trial credit, EUR)
limits.trial_credit_used_eur       # 2.5  (spent so far; the trial never resets)
limits.default_egress_domains      # the restricted-mode default allowlist
for r in limits.regions:           # regions a create may pin right now
    print(r.id, r.label)
for s in limits.sizes:             # full menu, each with .allowed
    print(s.size, s.cpu, s.memory_mb, s.allowed)

result = sbx.exec("python /workspace/main.py", timeout_seconds=60)
result.stdout       # str
result.stderr       # str
result.exit_code    # int
result.duration_ms  # int

for chunk in sbx.exec_stream("python long_task.py"):
    if chunk.stream == "stdout":
        print(chunk.data, end="", flush=True)
    elif chunk.stream == "stderr":
        print(chunk.data, end="", flush=True, file=sys.stderr)
    if chunk.is_final:
        print(f"exit={chunk.exit_code} duration={chunk.duration_ms}ms")

# Start a dev server without nohup gymnastics:
handle = sbx.exec("python -m http.server 8000", background=True)
handle.pid, handle.status        # 41, "running"

# Reattach from another process:
for cmd in sbx.commands():
    print(cmd.id, cmd.command, cmd.status)

print(handle.logs().content)     # buffered output so far
for chunk in handle.stream():    # ...or follow it live
    print(chunk.data, end="")

handle.wait(timeout=120)         # block until it stops; returns exit code
handle.kill()                    # SIGTERM the process group, then SIGKILL

sbx = Sandbox.create(template="code-interpreter")

sbx.run_code("import pandas as pd; df = pd.read_csv('sales.csv')")
execution = sbx.run_code("df.describe()")   # df persisted from the last call

execution.text                # text rendering of the final expression
execution.results[0].html     # ...and the HTML table
execution.error               # structured CodeError on a raising cell, else None

# Isolated contexts: separate interpreters inside one sandbox
ctx = sbx.create_code_context()
ctx.run_code("x = 1")         # invisible to the default context

# Text I/O
sbx.files.write("/workspace/script.py", "print('hello')")
content = sbx.files.read("/workspace/output.txt")

# Binary I/O
sbx.files.write_bytes("/workspace/blob.bin", b"\x00\x01\x02")
raw = sbx.files.read_bytes("/workspace/blob.bin")

# Directory listing
for entry in sbx.files.list("/workspace"):
    print(entry.name, entry.is_dir, entry.size, entry.mode)

# Directory + move + stat/exists (no shelling out)
sbx.files.mkdir("/workspace/data")           # mkdir -p by default
sbx.files.move("/workspace/a", "/workspace/data/a")
st = sbx.files.stat("/workspace/data/a")     # .exists is a normal check
if st.exists and not st.is_dir:
    print(st.size, oct(st.mode))
sbx.files.exists("/workspace/nope")          # False, not an error

# Remove
sbx.files.delete("/workspace/output.txt")

# Many files in one call (scaffolding a project without N round-trips).
# Per-file results: a bad path fails its own row, the rest still land.
results = sbx.files.write_batch({
    "/workspace/src/app.py": "print('hi')",
    "/workspace/README.md": "# my app",
})
for r in results:
    print(r.path, r.bytes_written, r.error)   # error is None on success

# Large files: mint a pre-signed URL and move bytes directly - plain
# HTTP, no auth header, no base64, no 16 MiB cap. Reusable until expiry.
url = sbx.files.presign_upload("/workspace/dataset.csv", expires_in=600)
# requests.put(url.url, data=open("dataset.csv", "rb"))

url = sbx.files.presign_download("/workspace/out/report.pdf")
print(url.url)   # hand to a browser, curl, or another service

from orkestr import Sandbox, Volume

# Attach a volume by name - created on first use (default 10 GB).
# Anything under /persist survives terminate.
sbx = Sandbox.create(template="python-3.12", volume="my-cache")
sbx.exec("echo trained-weights-v1 > /persist/model.txt")
sbx.terminate()

# A brand-new sandbox re-attaches the same volume and sees the data:
sbx = Sandbox.create(template="python-3.12", volume="my-cache")
print(sbx.exec("cat /persist/model.txt").stdout)   # trained-weights-v1

# Manage volumes explicitly (custom size, a region pin, listing, cleanup):
vol = Volume.create("datasets", size_gb=50, region="rbx")
for v in Volume.list():
    print(v.name, v.size_mb, v.status, v.region)
Volume.move(vol.id, "hel1")   # re-home a detached volume to another region
Volume.delete(vol.id)   # rejected while attached to a running sandbox

from orkestr import Webhook, verify_webhook_signature

# Register once; store the secret - it is shown ONLY in this response.
whk = Webhook.create(
    "https://example.com/hooks/orkestr",
    event_types=["sandbox.paused", "sandbox.expired"],  # omit for all events
)
save_somewhere(whk.secret)   # whsec_...

# In your receiver (FastAPI shown), verify the RAW body before parsing:
@app.post("/hooks/orkestr")
async def hook(request: Request):
    raw = await request.body()
    sig = request.headers.get("X-Orkestr-Signature")
    if not verify_webhook_signature(secret, raw, sig):
        raise HTTPException(status_code=401)
    delivery = json.loads(raw)   # {"event": "sandbox.paused", "data": {...}}

Webhook.list()             # your registrations; secret never included
Webhook.enable(whk.id)     # re-arm one disabled after repeated failures
Webhook.delete(whk.id)

m = sbx.metrics()
m.sandbox_status        # "running" (usage is null when paused / terminated)
m.cpu.cores             # 1.0
m.cpu.usage_percent     # 94.0  (% of allocated cores; 1-core pegged = 100)
m.cpu.usage_cores       # 0.94  (cores in use)
m.memory.usage_bytes    # 188743680  (working set, excludes reclaimable cache)
m.memory.usage_percent  # 17.6  (% of m.memory.limit_bytes)
m.lifetime.cpu_seconds  # 12.84 (on-CPU seconds since the sandbox started)
for s in m.samples:     # rolling ~60s window, oldest first
    print(s.t, s.cpu_percent, s.mem_bytes)

sbx = Sandbox.create(template="node-22", network="open")

# Start a server on a port inside the sandbox (in the background):
sbx.exec("nohup python3 -m http.server 3000 >/tmp/srv.log 2>&1 &")

host = sbx.get_host(3000)     # "3000-01hxyz....sbx.orkestr.run"
url = f"https://{host}"       # open in a browser or embed it in your app

sbx.set_timeout(3600)   # live for another hour from now
print(sbx.expires_at)   # the new deadline

# Pause: returns the sandbox id, stops compute meter
sandbox_id = sbx.pause()

# ... later ...
sbx = Sandbox.resume(sandbox_id)

with Sandbox.create(template="python-3.12") as sbx:
    result = sbx.exec("python -c 'print(2+2)'")
    print(result.stdout)
# sbx.terminate() is called automatically on exit (including on exceptions)

from orkestr import (
    OrkestrError,
    AuthError,
    SandboxNotFound,
    ExecTimeout,
    SnapshotCapReached,
)

try:
    sbx = Sandbox.create(template="python-3.12")
    result = sbx.exec("sleep 9999", timeout_seconds=5)
except ExecTimeout:
    print("Command exceeded timeout - the sandbox is still running.")
except AuthError as e:
    print(f"Auth failed ({e.status_code}): {e.detail}")
except OrkestrError as e:
    print(f"Something else broke: {e}, request_id={e.request_id}")