Examples
Four patterns that cover most of what people build: running model-generated code, testing a repository, isolating per-request work in a backend, and never leaking a sandbox.
1. Run LLM-generated Python and feed the result back
The classic loop: a model writes code, the code runs somewhere that cannot hurt you, and the output goes back into the conversation. The one detail worth copying is how the source gets into the sandbox — base64 through a single command, so no amount of quoting, backticks or newlines in the generated code can break out of the shell string.
"""Run untrusted, model-generated Python in a fresh sandbox and return the result."""
import base64
import os
import requests
API = "https://sandbox-as-a-service.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['AAS_API_KEY']}"}
TIMEOUT = 600
def run_generated_python(code: str, packages: list[str] | None = None) -> dict:
sandbox = requests.post(
f"{API}/sandboxes",
headers=HEADERS,
json={"size": "small", "name": "llm-code", "timeout_minutes": 10},
timeout=TIMEOUT,
)
sandbox.raise_for_status()
sandbox_id = sandbox.json()["id"]
def run(command, timeout_ms=60_000):
r = requests.post(
f"{API}/sandboxes/{sandbox_id}/exec",
headers=HEADERS,
json={"command": command, "timeout_ms": timeout_ms},
timeout=timeout_ms / 1000 + 30,
)
r.raise_for_status()
return r.json()
try:
# base64 keeps the generated source opaque to the shell.
encoded = base64.b64encode(code.encode()).decode()
run(f"echo {encoded} | base64 -d > /workspace/main.py")
if packages:
spec = " ".join(f"'{p}'" for p in packages)
install = run(
f"pip install --quiet --break-system-packages --user {spec}",
timeout_ms=300_000,
)
if install["exit_code"] != 0:
return {"ok": False, "stage": "install", "stderr": install["stderr"]}
result = run("python3 /workspace/main.py", timeout_ms=120_000)
return {
"ok": result["exit_code"] == 0,
"stage": "run",
"exit_code": result["exit_code"],
"stdout": result["stdout"],
"stderr": result["stderr"],
"truncated": result["truncated"],
}
finally:
requests.delete(f"{API}/sandboxes/{sandbox_id}", headers=HEADERS, timeout=TIMEOUT)
# --- feeding the result back to the model -------------------------------------
generated = """
import pandas as pd
df = pd.DataFrame({"city": ["Berlin", "Lisbon", "Oslo"], "pop_m": [3.7, 0.55, 0.7]})
print(df.sort_values("pop_m", ascending=False).to_string(index=False))
"""
outcome = run_generated_python(generated, packages=["pandas"])
if outcome["ok"]:
feedback = f"The code ran successfully. Output:\n{outcome['stdout']}"
else:
feedback = (
"The code failed with exit code "
f"{outcome.get('exit_code')}. Fix it and try again.\n\n"
f"stderr:\n{outcome.get('stderr', '')[-4000:]}"
)
print(feedback) # -> next turn of the conversation// Run untrusted, model-generated code in a fresh sandbox and return the result.
const API = "https://sandbox-as-a-service.com/v1";
const headers = {
Authorization: `Bearer ${process.env.AAS_API_KEY}`,
"Content-Type": "application/json",
};
async function api(method, path, body) {
const res = await fetch(`${API}${path}`, {
method,
headers,
body: body === undefined ? undefined : JSON.stringify(body),
});
if (!res.ok) {
const err = await res.json().catch(() => ({}));
throw new Error(`${res.status} ${err?.error?.type}: ${err?.error?.message}`);
}
return res.json();
}
export async function runGeneratedCode(code, { packages = [] } = {}) {
const sandbox = await api("POST", "/sandboxes", {
size: "small",
name: "llm-code",
timeout_minutes: 10,
});
const run = (command, timeoutMs = 60000) =>
api("POST", `/sandboxes/${sandbox.id}/exec`, { command, timeout_ms: timeoutMs });
try {
// base64 keeps the generated source opaque to the shell.
const encoded = Buffer.from(code, "utf8").toString("base64");
await run(`echo ${encoded} | base64 -d > /workspace/main.js`);
if (packages.length) {
const install = await run(
`cd /workspace && npm install --no-fund --no-audit ${packages.join(" ")}`,
300000
);
if (install.exit_code !== 0) {
return { ok: false, stage: "install", stderr: install.stderr };
}
}
const result = await run("node /workspace/main.js", 120000);
return {
ok: result.exit_code === 0,
stage: "run",
exitCode: result.exit_code,
stdout: result.stdout,
stderr: result.stderr,
truncated: result.truncated,
};
} finally {
await api("DELETE", `/sandboxes/${sandbox.id}`).catch(() => {});
}
}
const outcome = await runGeneratedCode(
`const _ = require("lodash");
console.log(_.chunk([1, 2, 3, 4, 5], 2));`,
{ packages: ["lodash"] }
);
console.log(
outcome.ok
? `The code ran successfully. Output:\n${outcome.stdout}`
: `The code failed (exit ${outcome.exitCode}). stderr:\n${outcome.stderr.slice(-4000)}`
);2. Clone a repo, install dependencies, run the tests
A clean machine per test run, with no state carried over from the last one. Note the long
timeout_ms values on install and test steps, the tail -c that keeps the
response under the output cap, and the extend call for suites that outrun the sandbox
lifetime.
"""Check out a repository in a disposable machine and run its test suite."""
import os
import requests
API = "https://sandbox-as-a-service.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['AAS_API_KEY']}"}
class Sandbox:
def __init__(self, size="small", minutes=15, name=None):
r = requests.post(
f"{API}/sandboxes",
headers=HEADERS,
json={"size": size, "name": name, "timeout_minutes": minutes},
timeout=600,
)
r.raise_for_status()
self.data = r.json()
self.id = self.data["id"]
def run(self, command, timeout_ms=60_000, cwd="/workspace", env=None):
r = requests.post(
f"{API}/sandboxes/{self.id}/exec",
headers=HEADERS,
json={"command": command, "timeout_ms": timeout_ms, "cwd": cwd, "env": env or {}},
timeout=timeout_ms / 1000 + 30,
)
r.raise_for_status()
return r.json()
def extend(self, minutes=30):
requests.post(f"{API}/sandboxes/{self.id}/extend", headers=HEADERS,
json={"additional_minutes": minutes}, timeout=60)
def destroy(self):
requests.delete(f"{API}/sandboxes/{self.id}", headers=HEADERS, timeout=600)
def __enter__(self):
return self
def __exit__(self, *exc):
self.destroy()
return False
def test_repo(repo_url, ref="main"):
with Sandbox(size="medium", minutes=60, name="ci") as box:
steps = [
("clone", f"git clone --depth 1 --branch {ref} {repo_url} /workspace/repo", 300_000),
("venv", "python3 -m venv /workspace/repo/.venv", 120_000),
("install", "/workspace/repo/.venv/bin/pip install -q -e '.[test]'", 600_000),
]
for name, command, timeout_ms in steps:
step = box.run(command, timeout_ms=timeout_ms, cwd="/workspace")
if step["exit_code"] != 0:
return {"ok": False, "stage": name, "log": step["stderr"][-8000:]}
box.extend(30) # test suites are the step most likely to outrun the lifetime
# tail keeps the response comfortably under the 1 MiB per-stream cap
tests = box.run(
"/workspace/repo/.venv/bin/pytest -q 2>&1 | tail -c 60000",
timeout_ms=600_000,
cwd="/workspace/repo",
env={"PYTHONUNBUFFERED": "1", "CI": "true"},
)
return {
"ok": tests["exit_code"] == 0,
"stage": "tests",
"exit_code": tests["exit_code"],
"log": tests["stdout"][-8000:],
}
print(test_repo("https://github.com/psf/requests.git"))#!/usr/bin/env bash
# Clone, install and test a repository in a throwaway machine.
set -euo pipefail
REPO="${1:?usage: test-repo.sh <git-url>}"
AUTH="Authorization: Bearer $AAS_API_KEY"
JSON="Content-Type: application/json"
SBX=$(curl -sS -X POST https://sandbox-as-a-service.com/v1/sandboxes -H "$AUTH" -H "$JSON" \
-H "Idempotency-Key: ci-$(date +%s)-$RANDOM" \
-d '{"size":"medium","name":"ci","timeout_minutes":60}' | jq -r .id)
# Destroy the sandbox no matter how the script exits.
trap 'curl -sS -X DELETE https://sandbox-as-a-service.com/v1/sandboxes/$SBX -H "$AUTH" >/dev/null' EXIT
exec_in() { # exec_in <command> <timeout_ms>
jq -n --arg c "$1" --argjson t "${2:-60000}" \
'{command:$c, timeout_ms:$t, cwd:"/workspace"}' \
| curl -sS -X POST https://sandbox-as-a-service.com/v1/sandboxes/$SBX/exec -H "$AUTH" -H "$JSON" -d @-
}
exec_in "git clone --depth 1 $REPO /workspace/repo" 300000 | jq -r .exit_code
exec_in "cd /workspace/repo && python3 -m venv .venv && .venv/bin/pip install -q -e '.[test]'" 600000 | jq -r .exit_code
RESULT=$(exec_in "cd /workspace/repo && .venv/bin/pytest -q 2>&1 | tail -c 60000" 600000)
echo "$RESULT" | jq -r .stdout
exit "$(echo "$RESULT" | jq -r .exit_code)"3. One sandbox per request in a web backend
When each request runs untrusted input, give each request its own machine and destroy it before
responding. The details that matter in production: a semaphore that respects the
20-sandbox concurrency limit, a short timeout_minutes so a crashed worker
cannot leak an expensive machine, and cleanup that runs even when the client disconnects.
"""POST /run — execute a snippet in a dedicated sandbox, one per request."""
import asyncio
import base64
import os
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
API = "https://sandbox-as-a-service.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['AAS_API_KEY']}"}
app = FastAPI()
client = httpx.AsyncClient(base_url=API, headers=HEADERS, timeout=600.0)
# The account-wide concurrency limit is 20; stay under it.
slots = asyncio.Semaphore(20)
class RunRequest(BaseModel):
code: str = Field(max_length=70_000)
timeout_ms: int = Field(default=30_000, ge=1000, le=600_000)
@app.post("/run")
async def run(req: RunRequest):
async with slots:
try:
created = await client.post(
"/sandboxes",
json={"size": "small", "name": "req", "timeout_minutes": 5},
)
except httpx.HTTPError as exc:
raise HTTPException(503, f"could not reach the sandbox API: {exc}") from exc
if created.status_code == 429:
raise HTTPException(429, "sandbox capacity exhausted, retry shortly")
if created.status_code == 402:
raise HTTPException(503, "out of sandbox credit")
created.raise_for_status()
sandbox_id = created.json()["id"]
try:
encoded = base64.b64encode(req.code.encode()).decode()
await client.post(
f"/sandboxes/{sandbox_id}/exec",
json={"command": f"echo {encoded} | base64 -d > /workspace/main.py"},
)
result = (
await client.post(
f"/sandboxes/{sandbox_id}/exec",
json={"command": "python3 /workspace/main.py",
"timeout_ms": req.timeout_ms},
)
).json()
return {
"exit_code": result["exit_code"],
"stdout": result["stdout"],
"stderr": result["stderr"],
"truncated": result["truncated"],
"timed_out": result["exit_code"] == 124,
}
finally:
# Runs on every path, including an exception or a client disconnect.
# Shielded so cancellation of the request cannot skip the teardown.
await asyncio.shield(client.delete(f"/sandboxes/{sandbox_id}"))// POST /run — execute a snippet in a dedicated sandbox, one per request.
import express from "express";
const API = "https://sandbox-as-a-service.com/v1";
const headers = {
Authorization: `Bearer ${process.env.AAS_API_KEY}`,
"Content-Type": "application/json",
};
const app = express();
app.use(express.json({ limit: "1mb" }));
// The account-wide concurrency limit is 20; stay under it.
let inFlight = 0;
const MAX_IN_FLIGHT = 20;
async function api(method, path, body) {
const res = await fetch(`${API}${path}`, {
method,
headers,
body: body === undefined ? undefined : JSON.stringify(body),
});
const json = await res.json().catch(() => ({}));
return { status: res.status, ok: res.ok, json };
}
app.post("/run", async (req, res) => {
if (inFlight >= MAX_IN_FLIGHT) {
return res.status(429).json({ error: "sandbox capacity exhausted, retry shortly" });
}
inFlight++;
let sandboxId = null;
try {
const created = await api("POST", "/sandboxes", {
size: "small",
name: "req",
timeout_minutes: 5,
});
if (!created.ok) {
const status = created.status === 429 ? 429 : 503;
return res.status(status).json({ error: created.json?.error?.type ?? "unavailable" });
}
sandboxId = created.json.id;
const encoded = Buffer.from(String(req.body.code ?? ""), "utf8").toString("base64");
await api("POST", `/sandboxes/${sandboxId}/exec`, {
command: `echo ${encoded} | base64 -d > /workspace/main.js`,
});
const result = await api("POST", `/sandboxes/${sandboxId}/exec`, {
command: "node /workspace/main.js",
timeout_ms: Math.min(Number(req.body.timeout_ms) || 30000, 600000),
});
return res.json({
exit_code: result.json.exit_code,
stdout: result.json.stdout,
stderr: result.json.stderr,
truncated: result.json.truncated,
timed_out: result.json.exit_code === 124,
});
} catch (err) {
return res.status(500).json({ error: String(err) });
} finally {
inFlight--;
// Never await cleanup on the response path, but never skip it either.
if (sandboxId) {
api("DELETE", `/sandboxes/${sandboxId}`).catch(() => {});
}
}
});
app.listen(3000);4. Cleanup patterns
A leaked sandbox is not a correctness bug — it is a bill. Three layers, in order of how much you should rely on them.
Layer 1: destroy it in a finally
This is the one that does the real work. Wrap the sandbox in whatever your language calls a scope guard so that no code path — exception, early return, timeout — can skip the DELETE.
import contextlib
import os
import requests
API = "https://sandbox-as-a-service.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['AAS_API_KEY']}"}
@contextlib.contextmanager
def sandbox(size="small", minutes=15, name=None):
"""Yields a sandbox dict; destroys it on the way out, whatever happened."""
r = requests.post(
f"{API}/sandboxes",
headers=HEADERS,
json={"size": size, "name": name, "timeout_minutes": minutes},
timeout=600,
)
r.raise_for_status()
box = r.json()
try:
yield box
finally:
# DELETE is idempotent, so a failure here is safe to swallow and retry.
try:
requests.delete(f"{API}/sandboxes/{box['id']}", headers=HEADERS, timeout=600)
except requests.RequestException:
pass # expiry is the backstop; the sweeper will collect it
with sandbox(name="report") as box:
... # anything that raises in here still destroys the machineconst API = "https://sandbox-as-a-service.com/v1";
const headers = {
Authorization: `Bearer ${process.env.AAS_API_KEY}`,
"Content-Type": "application/json",
};
/** Runs fn with a fresh sandbox and destroys it afterwards, whatever happened. */
export async function withSandbox(fn, { size = "small", minutes = 15, name } = {}) {
const res = await fetch(`${API}/sandboxes`, {
method: "POST",
headers,
body: JSON.stringify({ size, name, timeout_minutes: minutes }),
});
if (!res.ok) throw new Error(`create failed: ${res.status} ${await res.text()}`);
const sandbox = await res.json();
try {
return await fn(sandbox);
} finally {
// DELETE is idempotent; failing to clean up must not mask the real error.
await fetch(`${API}/sandboxes/${sandbox.id}`, { method: "DELETE", headers })
.catch(() => {});
}
}
const output = await withSandbox(async (box) => {
const r = await fetch(`${API}/sandboxes/${box.id}/exec`, {
method: "POST",
headers,
body: JSON.stringify({ command: "uname -a" }),
}).then((r) => r.json());
return r.stdout;
}, { name: "probe" });
console.log(output);Layer 2: a short timeout_minutes
Set the lifetime to a little more than the job needs. If your process is killed between creating a sandbox and destroying it, the lifetime is exactly how much that mistake costs — ten minutes instead of 1440. You can always extend a sandbox that turns out to need longer.
Layer 3: a sweeper
Belt and braces for anything the first two layers missed — a crashed worker, a bug, an agent that walked away. Run it on a schedule and destroy anything your own records do not recognise.
import os
import requests
from datetime import datetime, timezone, timedelta
API = "https://sandbox-as-a-service.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['AAS_API_KEY']}"}
def sweep(known_ids: set[str], older_than_minutes: int = 30) -> list[str]:
"""Destroy sandboxes this system no longer knows about."""
cutoff = datetime.now(timezone.utc) - timedelta(minutes=older_than_minutes)
destroyed = []
listing = requests.get(f"{API}/sandboxes", headers=HEADERS,
params={"limit": 100}, timeout=60).json()
for box in listing["data"]:
if box["status"] != "running" or box["id"] in known_ids:
continue
created = datetime.fromisoformat(box["created_at"].replace("Z", "+00:00"))
if created > cutoff:
continue # probably still being set up by a live worker
requests.delete(f"{API}/sandboxes/{box['id']}", headers=HEADERS, timeout=600)
destroyed.append(box["id"])
return destroyed
print("swept:", sweep(known_ids=set()))And the backstop under all three: every sandbox has an expires_at, and the platform
destroys it when that passes. Even total failure of your cleanup code has a bounded cost.
Try it with your own code
Sign in, create a key, and run your first sandbox in a few minutes.