The RunloopSDK builds on top of the underlying REST client and provides a Pythonic, object-oriented API for managing devboxes, blueprints, snapshots, and storage objects. The SDK exposes synchronous and asynchronous variants to match your runtime requirements.
- Installation
- Quickstart (synchronous)
- Quickstart (asynchronous)
- Core Concepts
- Accessing the Underlying REST Client
- Error Handling
- Advanced Configuration
- Async Usage
- Polling Configuration
- Complete API Reference
- Feedback
The SDK ships with the runloop_api_client package—no extra dependencies are required.
pip install runloop_api_clientfrom runloop_api_client import RunloopSDK
runloop = RunloopSDK()
# Create a ready-to-use devbox
with runloop.devbox.create(name="my-devbox") as devbox:
result = devbox.cmd.exec("echo 'Hello from Runloop!'")
print(result.stdout())
# Stream stdout in real time
devbox.cmd.exec(
"ls -la",
stdout=lambda line: print("stdout:", line),
)
# Blueprints
blueprint = runloop.blueprint.create(
name="my-blueprint",
dockerfile="FROM ubuntu:22.04\nRUN echo 'Hello' > /hello.txt\n",
)
devbox = blueprint.create_devbox(name="dev-from-blueprint")
# Storage objects
obj = runloop.storage_object.upload_from_text("Hello world!", name="greeting.txt")
print(obj.download_as_text())import asyncio
from runloop_api_client import AsyncRunloopSDK
async def main():
runloop = AsyncRunloopSDK()
async with await runloop.devbox.create(name="async-devbox") as devbox:
result = await devbox.cmd.exec("pwd")
print(await result.stdout())
def capture(line: str) -> None:
print(">>", line)
await devbox.cmd.exec("ls", stdout=capture)
asyncio.run(main())The main SDK class that provides access to all Runloop functionality:
from runloop_api_client import RunloopSDK
runloop = RunloopSDK(
bearer_token="your-api-key", # defaults to RUNLOOP_API_KEY env var
# ... other options
)The SDK provides object-oriented interfaces for all major Runloop resources:
runloop.devbox- Devbox management (create, list, execute commands, file operations)runloop.blueprint- Blueprint management (create, list, build blueprints)runloop.snapshot- Snapshot management (list disk snapshots)runloop.storage_object- Storage object management (upload, download, list objects)runloop.agent- Agent management (create, list agents from npm/pip/git)runloop.axon- [Beta] Axon management (create, publish events, subscribe to SSE streams, SQL queries)runloop.network_policy- Network policy management (create, list, update egress rules)runloop.gateway_config- Gateway config management (create, list API proxy configurations)runloop.mcp_config- MCP config management (create, list MCP server configurations)runloop.secret- Secret management (create, update, list, delete encrypted key-value pairs)runloop.api- Direct access to the underlying REST API client
Secret / AsyncSecret objects from create(), list(), and from_name() always operate by name, even when they carry a captured .id. Use a separate SecretById / AsyncSecretById handle for one exact row:
created = runloop.secret.create(name="API_TOKEN", value="synthetic-example")
assert created.id is not None
exact = runloop.secret.from_id(created.id) # Lazy; never falls back to a name.
exact.update("synthetic-rotated") # Returns SecretView, as do get_info() and delete().
exact.delete() # Deletes only this row.For AsyncRunloopSDK, await the operations but not from_id(). The handle types are exported from runloop_api_client.sdk. ID handles have .id, not .name; use their instance methods rather than name-only manager helpers. Environment-secret maps still take names; pass exact.id explicitly for an ID-based gateway/MCP binding. Name reads/updates select the greatest ID; name deletion removes selected matches nonatomically. IDs identify mutable rows, not value versions, and existing retries can replay writes. Rotation does not rewrite running devbox environments or issued tokens. Avoid debug request logging with real credentials.
Object-oriented interface for working with devboxes. Created via runloop.devbox.create(), runloop.devbox.create_from_blueprint_id(), runloop.devbox.create_from_blueprint_name(), runloop.devbox.create_from_snapshot(), or runloop.devbox.from_id():
# Create a new devbox
devbox = runloop.devbox.create(name="my-devbox")
# Create a devbox from a blueprint ID
devbox_from_blueprint = runloop.devbox.create_from_blueprint_id(
blueprint_id="bpt_123",
name="my-devbox-from-blueprint",
)
# Create a devbox from a blueprint name
devbox_from_name = runloop.devbox.create_from_blueprint_name(
blueprint_name="my-blueprint-name",
name="my-devbox-from-blueprint",
)
# Create a devbox from a snapshot
devbox_from_snapshot = runloop.devbox.create_from_snapshot(
snapshot_id="snp_123",
name="my-devbox-from-snapshot",
)
# Or get an existing one (waits for it to be running)
existing_devbox = runloop.devbox.from_id(devbox_id="dbx_123")
# List all devboxes
devboxes = runloop.devbox.list(limit=10)
# Get devbox information
info = devbox.get_info()
print(f"Devbox {info.name} is {info.status}")Execute commands synchronously or asynchronously:
# exec blocks until completion - use for commands that return immediately
result = devbox.cmd.exec("ls -la")
print("Output:", result.stdout())
print("Exit code:", result.exit_code)
print("Success:", result.success)
# exec_async returns immediately - use for long-running processes
execution = devbox.cmd.exec_async("npm run dev")
# Check execution status
state = execution.get_state()
print("Status:", state.status)
# Wait for completion and get result
result = execution.result()
print("Final output:", result.stdout())
# Kill the process
execution.kill()The Execution object provides fine-grained control over asynchronous command execution:
# Start a long-running process
execution = devbox.cmd.exec_async("python train_model.py")
# Get the execution ID
print("Execution ID:", execution.execution_id)
print("Devbox ID:", execution.devbox_id)
# Poll for current state
state = execution.get_state()
print("Status:", state.status) # "running", "completed", etc.
print("Exit code:", state.exit_status) # only set when execution has completed
# Wait for completion and get results
result = execution.result()
print("Exit code:", result.exit_code)
print("Output:", result.stdout())
print("Errors:", result.stderr())
# Or kill the process early
execution.kill()Key methods:
execution.get_state()- Get current execution state (status, exit_code, etc.)execution.result()- Wait for completion and returnExecutionResultexecution.kill()- Terminate the running processexecution.execution_id- Get the execution ID (property)execution.devbox_id- Get the devbox ID (property)
The ExecutionResult object contains the output and exit status of a completed command:
# From synchronous execution
result = devbox.cmd.exec("ls -la /tmp")
# Or from asynchronous execution
execution = devbox.cmd.exec_async("echo 'test'")
result = execution.result()
# Access execution results
print("Exit code:", result.exit_code)
print("Success:", result.success) # True if exit code is 0
print("Failed:", result.failed) # True if exit code is non-zero
# Get output streams
stdout = result.stdout()
stderr = result.stderr()
print("Standard output:", stdout)
print("Standard error:", stderr)
# Access raw result data
raw_result = result.raw
print("Raw result:", raw_result)Key methods and properties:
result.exit_code- The process exit code (property)result.success- Boolean indicating success (exit code 0) (property)result.failed- Boolean indicating failure (non-zero exit code) (property)result.stdout()- Get standard output as stringresult.stderr()- Get standard error as stringresult.raw- Get the raw result data (property)
Callback requirement: All callbacks (
stdout,stderr,output) must be synchronous functions. Even when usingAsyncDevbox, callbacks cannot be async. Use thread-safe queues or other coordination primitives if you need to bridge into async code.
Pass callbacks into cmd.exec / cmd.exec_async to process logs in real time:
def handle_output(line: str) -> None:
print("LOG:", line)
result = devbox.cmd.exec(
"python train.py",
stdout=handle_output,
stderr=lambda line: print("ERR:", line),
output=lambda line: print("ANY:", line),
)
print("exit code:", result.exit_code)Async example (note that the callback itself is still synchronous):
def capture(line: str) -> None:
# Callbacks must be synchronous
# Use thread-safe data structures if needed
log_queue.put_nowait(line)
await devbox.cmd.exec(
"tail -f /var/log/app.log",
stdout=capture,
)# Write files
devbox.file.write(
path="/home/user/app.js",
contents='console.log("Hello from devbox!");',
)
# Read files
content = devbox.file.read(path="/home/user/app.js")
print(content)
# Upload files
from pathlib import Path
devbox.file.upload(
path="/home/user/upload.txt",
file=Path("local_file.txt"),
)
# Download files
data = devbox.file.download(path="/home/user/download.txt")
with open("local_download.txt", "wb") as f:
f.write(data)# Create SSH key for remote access
ssh_key = devbox.net.create_ssh_key()
print("SSH URL:", ssh_key.url)
# Create tunnel to expose port
tunnel = devbox.net.create_tunnel(port=8080)
print("Public URL:", tunnel.url)
# Remove tunnel when done
devbox.net.remove_tunnel(port=8080)# Create a snapshot (waits for completion)
snapshot = devbox.snapshot_disk(
name="my-snapshot",
commit_message="Added new features",
)
# Create a snapshot asynchronously (returns immediately)
snapshot = devbox.snapshot_disk_async(
name="my-snapshot",
commit_message="Added new features",
)
# Wait for it to complete later
snapshot.await_completed()
# Create new devbox from snapshot
new_devbox = snapshot.create_devbox(name="devbox-from-snapshot")# Suspend devbox (pause without losing state)
devbox.suspend()
# Resume suspended devbox
devbox.resume()
# Keep devbox alive (extend timeout)
devbox.keep_alive()
# Wait for devbox to reach running state
devbox.await_running()
# Wait for devbox to be suspended
devbox.await_suspended()
# Shutdown devbox
devbox.shutdown()State-waiting behavior: In the synchronous SDK,
runloop.devbox.from_id()anddevbox.resume()block until the devbox reaches therunningstate, anddevbox.suspend()blocks until the devbox issuspended. The async counterparts return immediately; callawait_running()/await_suspended()explicitly if you need to wait for the state transition.
Devboxes support context managers for automatic cleanup:
# Synchronous
with runloop.devbox.create(name="temp-devbox") as devbox:
result = devbox.cmd.exec("echo 'Hello'")
print(result.stdout())
# devbox is automatically shutdown when exiting the context
# Asynchronous
async with await runloop.devbox.create(name="temp-devbox") as devbox:
result = await devbox.cmd.exec("echo 'Hello'")
print(await result.stdout())
# devbox is automatically shutdown when exiting the contextRetrieve logs from a devbox, optionally filtered by execution ID or shell name:
# Get all devbox logs
logs = devbox.logs()
for log in logs.logs:
print(f"[{log.level}] {log.message}")
# Filter logs by execution ID
result = devbox.cmd.exec('echo "hello"')
exec_logs = devbox.logs(execution_id=result.execution_id)
# Filter logs by shell name
shell_logs = devbox.logs(shell_name="my-shell")Key methods:
devbox.get_info()- Get devbox details and statusdevbox.cmd.exec()- Execute commands synchronouslydevbox.cmd.exec_async()- Execute commands asynchronouslydevbox.logs()- Retrieve devbox logs (optionally filter by execution_id or shell_name)devbox.file.read()- Read file contentsdevbox.file.write()- Write file contentsdevbox.file.upload()- Upload filesdevbox.file.download()- Download filesdevbox.net.create_ssh_key()- Create SSH key for remote accessdevbox.net.create_tunnel()- Create network tunneldevbox.net.remove_tunnel()- Remove network tunneldevbox.snapshot_disk()- Create disk snapshot (waits for completion)devbox.snapshot_disk_async()- Create disk snapshot (async)devbox.suspend()- Suspend devboxdevbox.resume()- Resume suspended devboxdevbox.keep_alive()- Extend devbox timeoutdevbox.await_running()- Wait for devbox to be runningdevbox.await_suspended()- Wait for devbox to be suspendeddevbox.shutdown()- Shutdown the devbox
Object-oriented interface for working with blueprints. Created via runloop.blueprint.create() or runloop.blueprint.from_id():
# Create a new blueprint
blueprint = runloop.blueprint.create(
name="my-blueprint",
dockerfile="FROM ubuntu:22.04\nRUN apt-get update && apt-get install -y python3\n",
system_setup_commands=["pip install numpy pandas"],
)
# Or create a blueprint with a Docker build context from a local directory
from pathlib import Path
from runloop_api_client.lib.context_loader import build_docker_context_tar
context_root = Path("./my-app")
tar_bytes = build_docker_context_tar(context_root)
build_ctx_obj = runloop.storage_object.upload_from_bytes(
data=tar_bytes,
name="my-app-context.tar.gz",
content_type="tgz",
)
blueprint_with_context = runloop.blueprint.create(
name="my-blueprint-with-context",
dockerfile="""\
FROM node:22
WORKDIR /usr/src/app
# copy using the build context from the object
COPY package.json package.json
COPY src src
RUN npm install --only=production
CMD ["node", "src/app.js"]
""",
# Build context
build_context=build_ctx_obj.as_build_context(),
)
# Or get an existing one
blueprint = runloop.blueprint.from_id(blueprint_id="bpt_123")
# List all blueprints
blueprints = runloop.blueprint.list()
# Get blueprint details and build logs
info = blueprint.get_info()
logs = blueprint.logs()
# Create a devbox from this blueprint
devbox = blueprint.create_devbox(name="devbox-from-blueprint")
# Delete the blueprint when done
blueprint.delete()Key methods:
blueprint.get_info()- Get blueprint detailsblueprint.logs()- Get build logs for the blueprintblueprint.delete()- Delete the blueprintblueprint.create_devbox()- Create a devbox from this blueprint
Object-oriented interface for working with disk snapshots. Created via runloop.snapshot.from_id():
# Get an existing snapshot
snapshot = runloop.snapshot.from_id(snapshot_id="snp_123")
# List all snapshots
snapshots = runloop.snapshot.list()
# List snapshots for a specific devbox
devbox_snapshots = runloop.snapshot.list(devbox_id="dbx_123")
# Get snapshot details and check status
info = snapshot.get_info()
print(f"Snapshot status: {info.status}")
# Update snapshot metadata
snapshot.update(
name="updated-snapshot-name",
metadata={"version": "v2.0"},
)
# Wait for async snapshot to complete
snapshot.await_completed()
# Create a devbox from this snapshot
devbox = snapshot.create_devbox(name="devbox-from-snapshot")
# Delete the snapshot when done
snapshot.delete()Key methods:
snapshot.get_info()- Get snapshot details and statussnapshot.update()- Update snapshot name and metadatasnapshot.delete()- Delete the snapshotsnapshot.await_completed()- Wait for snapshot completionsnapshot.create_devbox()- Create a devbox from this snapshot
Object-oriented interface for working with storage objects. Created via runloop.storage_object.create() or runloop.storage_object.from_id():
# Create a new storage object
storage_object = runloop.storage_object.create(
name="my-file.txt",
content_type="text",
metadata={"project": "demo"},
)
# Upload content to the object
storage_object.upload_content("Hello, World!")
storage_object.complete()
# Upload from file
from pathlib import Path
uploaded = runloop.storage_object.upload_from_file(
Path("/path/to/file.txt"),
name="my-file.txt",
)
# Upload text content directly
uploaded = runloop.storage_object.upload_from_text(
"Hello, World!",
name="my-text.txt",
metadata={"source": "text"},
)
# Upload from bytes
uploaded = runloop.storage_object.upload_from_bytes(
b"binary content",
name="my-file.bin",
content_type="binary",
)
# Get object details and download
info = storage_object.refresh()
download_url = storage_object.get_download_url(duration_seconds=3600)
# Download content
text_content = storage_object.download_as_text()
binary_content = storage_object.download_as_bytes()
# List all storage objects
objects = runloop.storage_object.list()
# Delete when done
storage_object.delete()The storage helpers manage the multi-step upload flow (create → PUT to presigned URL → complete):
from pathlib import Path
# Upload local file with content-type detection
obj = runloop.storage_object.upload_from_file(file_path=Path("./report.csv"))
# Manual control
obj = runloop.storage_object.create(
name="data.bin",
content_type="binary",
)
obj.upload_content(b"\xde\xad\xbe\xef")
obj.complete()Key methods:
storage_object.refresh()- Get updated object detailsstorage_object.upload_content()- Upload content to the objectstorage_object.complete()- Mark upload as completestorage_object.get_download_url()- Get presigned download URLstorage_object.download_as_text()- Download content as textstorage_object.download_as_bytes()- Download content as bytesstorage_object.delete()- Delete the object
Static upload methods:
runloop.storage_object.upload_from_file()- Upload from filesystemrunloop.storage_object.upload_from_text()- Upload text content directlyrunloop.storage_object.upload_from_bytes()- Upload from bytes
You can mount storage objects to devboxes to access their contents:
# Create a storage object first
storage_object = runloop.storage_object.upload_from_text(
"Hello, World!",
name="my-data.txt",
)
# Create a devbox and mount the storage object
devbox = runloop.devbox.create(
name="my-devbox",
mounts=[
{
"type": "object_mount",
"object_id": storage_object.id,
"object_path": "/home/user/data.txt",
},
],
)
# The storage object is now accessible at /home/user/data.txt in the devbox
result = devbox.cmd.exec("cat /home/user/data.txt")
print(result.stdout()) # "Hello, World!"
# Mount archived objects (tar, tgz, gzip) - they get extracted to a directory
archive_object = runloop.storage_object.upload_from_file(
Path("./project.tar.gz"),
name="project.tar.gz",
)
devbox_with_archive = runloop.devbox.create(
name="archive-devbox",
mounts=[
{
"type": "object_mount",
"object_id": archive_object.id,
"object_path": "/home/user/project", # Archive gets extracted here
},
],
)
# Access extracted archive contents
result = devbox_with_archive.cmd.exec("ls -la /home/user/project/")
print(result.stdout())Beta: Axon APIs are in beta and may change.
Object-oriented interface for working with axons — event communication channels that support
publishing events, subscribing to event streams via SSE, and executing SQL queries against an
embedded SQLite database. Created via runloop.axon.create() or runloop.axon.from_id():
# Create a new axon
axon = runloop.axon.create(name="my-axon")
# Or get an existing one
axon = runloop.axon.from_id(axon_id="axn_123")
# List all axons
axons = runloop.axon.list()
# Get axon details
info = axon.get_info()
print(f"Axon {info.name} created at {info.created_at_ms}")Publish structured events to an axon:
result = axon.publish(
event_type="task_completed",
origin="AGENT_EVENT", # EXTERNAL_EVENT | AGENT_EVENT | USER_EVENT
payload='{"task_id": 42, "status": "done"}',
source="my-agent",
)
print(f"Sequence: {result.sequence}, Timestamp: {result.timestamp_ms}")Subscribe to real-time events via server-sent events:
with axon.subscribe_sse() as stream:
for event in stream:
print(f"[{event.source}] {event.event_type}: {event.payload}")
# event.axon_id, event.origin, event.sequence, event.timestamp_ms also availableAsync:
async with await axon.subscribe_sse() as stream:
async for event in stream:
print(f"[{event.source}] {event.event_type}: {event.payload}")Each axon has an embedded SQLite database accessible via axon.sql:
# Create a table
axon.sql.query(sql="CREATE TABLE tasks (id INTEGER PRIMARY KEY, name TEXT, done INTEGER)")
# Insert with parameterized queries (? placeholders)
axon.sql.query(
sql="INSERT INTO tasks (id, name, done) VALUES (?, ?, ?)",
params=[1, "Write docs", 0],
)
# Query
result = axon.sql.query(sql="SELECT * FROM tasks WHERE done = ?", params=[0])
for row in result.rows:
print(row) # e.g. [1, "Write docs", 0]
print(result.columns) # [SqlColumnMetaView(name="id", type="INTEGER"), ...]
print(result.meta) # SqlResultMetaView(changes=0, duration_ms=0.5, rows_read_limit_reached=False)Execute multiple statements atomically with batch:
result = axon.sql.batch(
statements=[
{"sql": "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)"},
{"sql": "INSERT INTO users (id, name) VALUES (?, ?)", "params": [1, "Alice"]},
{"sql": "INSERT INTO users (id, name) VALUES (?, ?)", "params": [2, "Bob"]},
{"sql": "SELECT * FROM users ORDER BY id"},
],
)
# Each statement gets a result
for step in result.results:
if step.success:
print(f"Rows: {step.success.rows}, Changes: {step.success.meta.changes}")
elif step.error:
print(f"Error: {step.error}")Key methods:
axon.get_info()– Get axon details (name, created_at_ms)axon.publish()– Publish an event (event_type, origin, payload, source)axon.subscribe_sse()– Subscribe to event stream (returnsStream[AxonEventView])axon.sql.query()– Execute a single SQL statement with optional paramsaxon.sql.batch()– Execute multiple SQL statements atomically
The SDK always exposes the underlying client through the .api attribute:
runloop = RunloopSDK()
raw_devbox = runloop.api.devboxes.create()This makes it straightforward to mix high-level helpers with low-level calls whenever you need advanced control.
The SDK provides comprehensive error handling with typed exceptions:
from runloop_api_client import RunloopSDK
import runloop_api_client
runloop = RunloopSDK()
try:
devbox = runloop.devbox.create(name="example-devbox")
result = devbox.cmd.exec("invalid-command")
except runloop_api_client.APIConnectionError as e:
print("The server could not be reached")
print(e.__cause__) # an underlying Exception, likely raised within httpx.
except runloop_api_client.RateLimitError as e:
print("A 429 status code was received; we should back off a bit.")
except runloop_api_client.APIStatusError as e:
print("Another non-200-range status code was received")
print(e.status_code)
print(e.response)Error codes are as follows:
| Status Code | Error Type |
|---|---|
| 400 | BadRequestError |
| 401 | AuthenticationError |
| 403 | PermissionDeniedError |
| 404 | NotFoundError |
| 422 | UnprocessableEntityError |
| 429 | RateLimitError |
| >=500 | InternalServerError |
| N/A | APIConnectionError |
import httpx
from runloop_api_client import RunloopSDK, DefaultHttpxClient
runloop = RunloopSDK(
bearer_token="your-api-key", # defaults to RUNLOOP_API_KEY env var
base_url="https://api.runloop.ai", # or use RUNLOOP_BASE_URL env var
timeout=60.0, # 60 second timeout (default is 30)
max_retries=3, # Retry failed requests (default is 5)
default_headers={
"X-Custom-Header": "value",
},
# Custom HTTP client with proxy
http_client=DefaultHttpxClient(
proxy="http://my.test.proxy.example.com",
transport=httpx.HTTPTransport(local_address="0.0.0.0"),
),
)The async SDK has the same interface as the synchronous version, but all I/O operations are async:
import asyncio
from runloop_api_client import AsyncRunloopSDK
async def main():
runloop = AsyncRunloopSDK()
# All the same operations, but with await
async with await runloop.devbox.create(name="async-devbox") as devbox:
result = await devbox.cmd.exec("pwd")
print(await result.stdout())
# Streaming (note: callbacks must be synchronous)
def capture(line: str) -> None:
print(">>", line)
await devbox.cmd.exec("ls", stdout=capture)
# Async file operations
await devbox.file.write(path="/tmp/test.txt", contents="Hello")
content = await devbox.file.read(path="/tmp/test.txt")
# Async network operations
tunnel = await devbox.net.create_tunnel(port=8080)
print("Tunnel URL:", tunnel.url)
asyncio.run(main())Many operations that wait for state changes accept a polling_config parameter:
from runloop_api_client.lib.polling import PollingConfig
# Create devbox with custom polling
devbox = runloop.devbox.create(
name="my-devbox",
polling_config=PollingConfig(
timeout_seconds=300.0, # Wait up to 5 minutes
interval_seconds=2.0, # Poll every 2 seconds
),
)
# Wait for snapshot completion with custom polling
snapshot.await_completed(
polling_config=PollingConfig(
timeout_seconds=600.0, # Wait up to 10 minutes
interval_seconds=5.0, # Poll every 5 seconds
),
)For the full REST API documentation and all available parameters, see:
- api.md - Complete REST API documentation
- README.md - Advanced topics (retries, timeouts, error handling, pagination)
The object-oriented SDK is new for Python—feedback and ideas are welcome! Please open an issue or pull request on GitHub if you spot gaps, bugs, or ergonomic improvements.