rustbox 0.4.0

Official Python SDK for Rustbox - run untrusted code in a kernel-enforced sandbox.

🐍 Rustbox Python SDK

Async-first via httpx. One required dep. Typed exceptions.

⚠️ Async-only. All public methods are coroutines. Wrap with asyncio.run() from sync code.

🚀 Install

pip install rustbox        # or: uv pip install rustbox / poetry add rustbox

⚡ Quickstart

import asyncio, os
from rustbox import Rustbox

async def main():
    client = Rustbox(os.environ["RUSTBOX_API_KEY"])
    result = await client.run(language="python", code="print('hello')")
    print(result["result"]["verdict"], result["output"]["stdout"])  # AC hello

asyncio.run(main())

run() submits, waits for sync completion, polls if needed, returns the verdict.

Response shape (schema v2)

{
  "id": "uuid",
  "schema_version": "2.0",
  "language": "python",
  "profile": "judge",
  "status": "completed",
  "timestamps": { "created_at": "...", "started_at": "...", "completed_at": "..." },
  "result": {
    "verdict": "AC",
    "cause": "normal_exit",
    "actor": "runtime",
    "exit_code": 0,
    "signal": null,
    "error_message": null
  },
  "output": { "stdout": "42\n", "stderr": "", "integrity": "complete" },
  "metrics": {
    "cpu_time_secs": 0.013,
    "wall_time_secs": 0.02,
    "memory_peak_bytes": 3715072,
    "cpu_wall_ratio": 0.65,
    "divergence": "cpu_bound"
  },
  "evidence": {
    "isolation": { "mode": "strict", "controls_applied": [], "controls_missing": [] },
    "limits": { "cgroup_backend": "cgroup_v2", "cgroup": { "memory_peak_bytes": 3715072, "oom_kill_events": 0 } },
    "teardown": { "reap_status": "clean", "zombie_count": 0, "cleanup_verified": true }
  }
}

verdict is one of AC | RE | TLE | MLE | PLE | FSE | SIG | IE. status is one of pending | running | completed | error. Read the verdict at result["result"]["verdict"], output at result["output"]["stdout"], and timing at result["metrics"]["cpu_time_secs"].

Profiles

# Judge profile (default) - short evaluation runs, no egress proxy.
await client.run("python", "print(1)")

# Agent profile - longer jobs, egress proxy on, per-key byte budget.
# Requires a non-trial API key.
await client.run("python", "...", profile="agent")

🔒 Errors

from rustbox import (
    RustboxAuthError, RustboxRateLimitError, RustboxServerError, RustboxError,
)

try:
    await client.run("python", "...")
except RustboxAuthError:       pass  # 401/403
except RustboxRateLimitError:  pass  # 429 - back off
except RustboxServerError:     pass  # 5xx - retry
except RustboxError:           pass  # other

🧰 Full API

Rustbox(
    api_key: str,
    base_url: str = DEFAULT_BASE_URL,
    *,
    timeout_secs: float = 65.0,
    max_retries: int = 2,
)

await client.run(
    language: str,
    code: str,
    stdin: str = "",
    profile: Literal["judge", "agent"] | None = None,
)  # -> dict

await client.submit(
    language, code, stdin="",
    profile=None, wait=False,
    idempotency_key=None,
)  # -> dict

await client.get_result(job_id)    # -> dict
await client.get_languages()       # -> list[str]
await client.get_health()          # -> dict
await client.get_ready()           # -> dict
await client.aclose()              # close httpx connection pool

🧪 Tests

cd python
uv pip install pytest pytest-asyncio respx httpx
python -m pytest -q

respx mocks httpx. No network.

🔗

• 📦 PyPI: rustbox
• 🪝 Webhooks
• 🛣️ Roadmap
• 🦀 SDK index