Async-first via httpx. One required dep. Typed exceptions.
⚠️ Async-only. All public methods are coroutines. Wrap withasyncio.run()from sync code.
pip install rustbox # or: uv pip install rustbox / poetry add rustboximport 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["verdict"], result["stdout"]) # AC hello
asyncio.run(main())run() submits, waits for sync completion, polls if needed, returns the verdict.
# 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")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 -qrespx mocks httpx. No network.
- 📦 PyPI: rustbox
- 🪝 Webhooks
- 🛣️ Roadmap
- 🦀 SDK index