Python
Run Python workloads in an isolated sandbox.
Install
withruntime is one client for every Runtime Cloud product, sync and async,
method for method. It uses only the standard library, imports in about 15 ms and
keeps its connections open between calls. Python 3.10 or later.
Terminalpip install withruntimeThe client finds its key by itself: RUNTIME_API_KEY when it is set, and
otherwise the connection this machine saved when it was connected (any
npx withruntime command connects it, with one browser approval). On a
server, put a key from https://withruntime.com/account/keys in
RUNTIME_API_KEY from your secret manager. Never put it in
source code, a URL or a command-line argument. With no key anywhere, the first call fails with
missing_api_key and says how to get one.
Hello, sandbox
Pythonfrom withruntime import Sandboxwith Sandbox.create() as sbx: result = sbx.exec("python3 -c 'print(6 * 7)'") print(result.exit_code, result.stdout)Sandbox.create() takes no required arguments and returns once the sandbox is
running. Leaving the with block stops it, even after an exception.
With no arguments you get the free trial while it lasts, the default region, and 2 vCPU, 4 GiB of memory and a 4 GiB disk for up to 30 minutes. Every field is optional and takes snake_case names:
Pythonfrom withruntime import Runtimeruntime = Runtime() # RUNTIME_API_KEY, or this machine's connectionsbx = runtime.sandboxes.create( name="tests-42", labels={"team": "search", "job": "42"}, vcpu=2, memory_mib=4096, disk_mib=8192, timeout_seconds=900, on_lease_end="stop", network={"internet": True, "allow": ["pypi.org", "*.pythonhosted.org"]},)print(sbx.id, sbx.info["funding"], sbx.info["expiresAt"])sbx.stop()The async client is the same with await:
Pythonimport asynciofrom withruntime import AsyncRuntimeasync def main(): async with AsyncRuntime() as runtime: async with await runtime.sandboxes.create() as sbx: results = await asyncio.gather(*(sbx.exec(f"echo {i}") for i in range(10))) print([r.stdout.strip() for r in results])asyncio.run(main())Run commands
A string runs under bash -c. A list runs the program directly, with no shell,
which is what you want for untrusted arguments.
Pythonimport osfrom withruntime import Sandboxwith Sandbox.create() as sbx: sbx.exec("mkdir -p app && echo 'print(1 + 1)' > app/main.py") run = sbx.exec( ["python3", "main.py"], cwd="/workspace/app", env={"API_TOKEN": os.environ.get("API_TOKEN", "")}, timeout_ms=120_000, ) if run.exit_code != 0: print(run.stderr)envis how secrets reach a command. It is never echoed back, and journals record a hash, not the value. Never put a secret in the command line itself.stdingives the command input, then closes it.- The default timeout is 60 seconds; the maximum is 24 hours. A timeout is a
result (
timed_out=True, with the output so far), not an exception. check=TrueraisesCommandErroron a non-zero exit.
Stream output as it happens with callbacks, or iterate the events:
Pythonimport sysfrom withruntime import Sandboxwith Sandbox.create() as sbx: sbx.exec("for i in 1 2 3; do echo line $i; sleep 1; done", on_stdout=sys.stdout.write) for event in sbx.exec_stream("npm --version"): if event["type"] == "stdout": print(event["data"], end="") elif event["type"] == "exit": print("exit", event["exitCode"])Background processes
Pythonfrom withruntime import Sandboxwith Sandbox.create() as sbx: server = sbx.spawn("python3 -m http.server 8000", cwd="/workspace") print(server.id, server.info["state"]) repl = sbx.spawn(["python3", "-i", "-q"], stdin="pipe") repl.write("print(21 * 2)\n") repl.write("exit()\n", eof=True) print(repl.wait().stdout) for process in sbx.processes(): print(process["id"], process["state"], process["command"]) server.kill("SIGTERM")process.output(cursor=0) yields every event from the start until the process
exits. A process outlives your connection; get it back with sbx.process(id).
An interactive terminal
Pythonfrom withruntime import Sandboxwith Sandbox.create() as sbx: term = sbx.terminal(cols=120, rows=40) term.write("echo hello from the terminal\n") term.write("exit\n") while (chunk := term.recv()) is not None: print(chunk.decode(errors="replace"), end="")Files
Pythonfrom withruntime import Sandboxwith Sandbox.create() as sbx: sbx.files.write("/workspace/data/input.csv", "a,b\n1,2\n") text = sbx.files.read_text("/workspace/data/input.csv") data = sbx.files.read("/workspace/data/input.csv") # bytes print(sbx.files.exists("/workspace/data/input.csv"), sbx.files.stat("/workspace/data/input.csv")) for entry in sbx.files.list("/workspace", depth=2): print(entry["type"], entry["size"], entry["path"]) print(sbx.files.glob("**/*.csv")) sbx.files.mkdir("/workspace/out") sbx.files.rename("/workspace/data/input.csv", "/workspace/out/input.csv") sbx.files.remove("/workspace/data", recursive=True)write makes parent directories and replaces the file atomically; large files go
in parallel chunks checked by SHA-256. Whole directories travel as one archive:
Pythonimport pathlibimport tempfilefrom withruntime import Sandboxproject = pathlib.Path(tempfile.mkdtemp())(project / "main.py").write_text("print('hi')\n")with Sandbox.create() as sbx: sbx.files.upload(str(project), "/workspace/project") sbx.exec("cd project && python3 main.py > result.txt") sbx.files.download("/workspace/project", str(project.parent / "project-out"))Pause, wake, extend
Pythonfrom withruntime import Sandboxsbx = Sandbox.create(timeout_seconds=600)sbx.exec("echo state > /workspace/state.txt")sbx.pause() # memory and files are kept; compute billing stopsagain = Sandbox.connect(sbx.id)again.wake(timeout_seconds=1200)again.extend(600)again.stop()Find sandboxes again
Pythonfrom withruntime import Runtimeruntime = Runtime()for sbx in runtime.sandboxes.list(labels={"team": "search"}, state=["running"]): print(sbx.id, sbx.info["name"], sbx.state)Every list returns a page: page.data, page.has_more, page.next_page(),
page.to_list(), and a for loop walks every item on every page.
Errors and retries
Pythonfrom withruntime import NotFoundError, RuntimeError, Sandboxtry: Sandbox.connect("00000000-0000-4000-8000-000000000000")except NotFoundError: print("no such sandbox")except RuntimeError as error: print(error.code, error.hint, error.request_id)The classes match the JavaScript SDK: AuthenticationError,
PermissionDeniedError, NotFoundError, ConflictError, InvalidRequestError,
RateLimitError, ServiceUnavailableError, ConnectionError and CommandError,
all subclasses of withruntime.RuntimeError. Every write carries an idempotency
key, made for you; transport failures, 429 and 503 are retried with the same key,
so a retry never makes two sandboxes or runs a command twice.
Read-only keys and daily limits
An owner can make a read-only key and set a daily spending limit on a key at
API keys. A key reads both and can
change neither. runtime.limits needs withruntime 0.3.1 or later:
Pythonfrom withruntime import Runtimewith Runtime() as runtime: limits = runtime.limits.get() print(limits["access"]) # "full", "read" or "selected" left = limits["daily"]["remainingMicros"] if left is not None and int(left) < 1_000_000: print("less than $1 left in this 24-hour window")Past the limit, a create, wake, extension or renewal fails with a
RuntimeError whose code is spending_limit_reached (HTTP 402), and it is
not retried. A read-only key asking to change anything gets
PermissionDeniedError. See security.
Images, volumes and snapshots
Forks and snapshots are paused while we fix an issue: for now fork, snapshot
and a create naming snapshot answer 503 fork_unavailable. Your sandboxes are
unaffected.
Pythonfrom withruntime import Runtimeruntime = Runtime()image = runtime.images.build(name="data", recipe={"pip": ["pandas"], "apt": ["jq"]}, on_log=lambda line: print(line["text"]))volume = runtime.volumes.create(size_mib=10_240, name="cache")with runtime.sandboxes.create(image=image["id"], volumes=[{"volume_id": volume["id"], "path": "/data"}]) as sbx: print(sbx.exec("python3 -c 'import pandas; print(pandas.__version__)'").stdout) forks = sbx.fork(count=2) # copies as it is now, running for fork in forks: fork.stop() snapshot = sbx.snapshot(name="with-pandas", retention_days=7) with runtime.sandboxes.create(snapshot=snapshot["id"]): pass runtime.snapshots.delete(snapshot["id"])images.build waits until the image is ready; images.create queues it and
returns. A volume lives on one server and is not backed up off it. See
JavaScript for what each does; the Python methods
are the same in snake_case.
Code interpreter and network rules
Pythonfrom withruntime import Runtimeruntime = Runtime()with runtime.sandboxes.create() as sbx: sbx.interpreter.run("import math\nx = math.pi") cell = sbx.interpreter.run("round(x * 2, 3)") print(cell["results"][0]["data"]["text/plain"]) sbx.network.set(internet=True, allow=["pypi.org", "*.pythonhosted.org"]) print(sbx.network.get())Feedback and support are on the client too: runtime.feedback.submit(...) and
runtime.support.message(...); see feedback and support.
Previews and the desktop
Pythonfrom withruntime import Sandboxwith Sandbox.create() as sbx: sbx.spawn("python3 -m http.server 3000") preview = sbx.previews.create(3000, visibility="public") print(preview["url"]) sbx.desktop.start(width=1280, height=800) sbx.desktop.open("https://example.com") sbx.desktop.click(640, 400) with open("screen.png", "wb") as file: file.write(sbx.desktop.screenshot())A preview's address is under runtimehost.com, the domain for everything
sandboxes serve, kept apart from Runtime's own site.
See JavaScript for what each does.
Configuration
Pythonfrom withruntime import Runtimewith Runtime(max_retries=4, timeout=120) as runtime: print(runtime.me()["orgId"])RUNTIME_API_URL points the client at another API origin.
Before 0.3.0 the package was withruntime-cloud, imported as runtime_cloud.
Both still work and give you the same classes, so older code keeps running.