Python SDK

Reference for upstm-py, the Python SDK. It has a synchronous client, Upstream, and an asynchronous one, AsyncUpstream, with the same methods.

$ pip install upstm-py

Clients#

Upstream#

Upstream(api_key, *, base_url="https://api.upstream.build", timeout=30.0)

Synchronous client. api_key is your ak_… key; timeout is in seconds. Call close() when you are done with it.

AsyncUpstream#

AsyncUpstream(api_key, *, base_url="https://api.upstream.build", timeout=30.0)

Asynchronous client with the same methods, each awaited. Use it as an async context manager:

async with AsyncUpstream(api_key=os.environ["UPSTREAM_API_KEY"]) as client:
    sb = await client.sandbox(template="python")
    print((await sb.exec(["echo", "hello"])).stdout)
    await sb.destroy()

Creating sandboxes#

client.sandbox()#

client.sandbox(template="python", *, ready_timeout=30.0, metadata=None,
               volume_mounts=None, mcp=None, on_timeout=None) -> Sandbox

Create a sandbox from a template and wait until it is ready. Use it as a context manager to destroy the sandbox on exit, or call destroy() yourself.

CPU and memory come from the template. The method also accepts vcpu and memory_mib for compatibility, but they have no effect.

Sandbox.connect()#

Sandbox.connect(client, sandbox_id) -> Sandbox

Attach to an existing sandbox by ID.

Running commands#

sb.exec()#

sb.exec(command, *, stream=False, background=False, timeout=None)
    -> ExecResult | StreamingExecResult | dict

Run command, a list of program and arguments, and wait for it to finish. Returns an ExecResult. Use a shell for pipes and &&: ["sh", "-c", "…"].

• timeout: seconds before the command is stopped.
• stream=True: return a StreamingExecResult that yields output as it arrives.
• background=True: start the command and return immediately with its process ID.

sb.exec_stream()#

sb.exec_stream(command) -> StreamingExecResult

Run a command and stream its stdout, stderr and exit status as they happen.

for chunk in sb.exec_stream(["sh", "-c", "for i in 1 2 3; do echo $i; sleep 1; done"]):
    if chunk.kind == "stdout":
        print(chunk.data.decode(), end="")

sb.list_processes()#

sb.list_processes() -> list[dict]

List processes started with background=True.

sb.signal()#

sb.signal(pid, sig) -> None

Send a signal, such as 15 for SIGTERM, to a process in the sandbox.

Files#

sb.write_file()#

sb.write_file(path, content) -> None

Write str or bytes to a file, creating it if needed.

sb.write_files()#

sb.write_files(files) -> list[dict]

Write several files in one request. files is a list of (path, content) pairs.

sb.read_file() and sb.read_file_text()#

sb.read_file(path) -> bytes
sb.read_file_text(path, encoding="utf-8") -> str

Read a file as raw bytes, or decoded as text.

sb.list_files()#

sb.list_files(path="/workspace") -> list[dict]

List a directory.

sb.stat_file() and sb.file_exists()#

sb.stat_file(path) -> dict
sb.file_exists(path) -> bool

Get a file's metadata, or check whether a path exists.

sb.mkdir(), sb.remove() and sb.rename()#

sb.mkdir(path, *, recursive=True) -> None
sb.remove(path, *, recursive=False) -> None
sb.rename(old_path, new_path) -> None

Create directories, delete files or directories, and move or rename them.

sb.watch_files()#

sb.watch_files(path, callback, *, recursive=True) -> WatchHandle

Call callback with a dict for each change under path. Stop watching with handle.close().

sb.upload_url() and sb.download_url()#

sb.upload_url(path) -> str
sb.download_url(path) -> str

Get a pre-signed URL for moving a large file in or out of the sandbox directly, without sending its contents through the API.

Ports#

sb.expose_port()#

sb.expose_port(port) -> dict

Expose a port from inside the sandbox through the gateway. The result's url opens whatever is listening on that port.

sb.exec(["sh", "-c", "python3 -m http.server 8000"], background=True)
print(sb.expose_port(8000)["url"])

sb.unexpose_port(), sb.list_ports() and sb.get_host()#

sb.unexpose_port(port) -> None
sb.list_ports() -> list[dict]
sb.get_host(port) -> str

Stop exposing a port, list exposed ports, or get the proxy path for one.

Forks, clones and snapshots#

See the forking guide for how these fit together.

sb.fork()#

sb.fork(workload_id=None) -> Sandbox

Fork this running sandbox. The new sandbox resumes with the same memory, files and processes, and shares this sandbox's pages until it writes. It has two extra attributes: parent_sandbox_id and fork_latency_ms. Requires a forkable template.

sb.clone()#

sb.clone(label=None) -> Sandbox

Create a new, independent copy of this sandbox.

sb.create_snapshot()#

sb.create_snapshot(label=None) -> dict

Save this sandbox's state. Returns the snapshot's metadata, including its id.

sb.list_snapshots()#

sb.list_snapshots() -> list[dict]

List this sandbox's snapshots.

client.list_snapshots(), client.restore_snapshot() and client.delete_snapshot()#

client.list_snapshots() -> list[dict]
client.restore_snapshot(snapshot_id) -> Sandbox
client.delete_snapshot(snapshot_id) -> None

List all your snapshots, start a new sandbox from one, or delete one.

sb.pause() and sb.resume()#

sb.pause() -> None
sb.resume() -> None

Suspend the sandbox in place and resume it later with its state intact.

Lifetime#

sb.keep_alive() and sb.set_timeout()#

sb.keep_alive(timeout_seconds=1800) -> None
sb.set_timeout(timeout_seconds) -> None

Extend the sandbox's lifetime, or set it outright.

sb.get_info()#

sb.get_info() -> dict

Get the sandbox's status, template, resources and metadata.

sb.destroy()#

sb.destroy() -> None

Destroy the sandbox and discard its private state. Leaving a with client.sandbox(…) block does this for you.

client.list_sandboxes()#

client.list_sandboxes() -> list[dict]

List your sandboxes.

Observability#

sb.get_metrics() and sb.get_logs()#

sb.get_metrics() -> dict
sb.get_logs(since_ms=0) -> list[dict]

Get CPU, memory and disk usage, or the sandbox's logs since a timestamp in milliseconds.

client.list_events()#

client.list_events(*, offset=0, limit=50, event_type=None, sandbox_id=None) -> list[dict]

List lifecycle events, optionally filtered by type or sandbox.

client.register_webhook()#

client.register_webhook(name, url, secret, *, event_types=None) -> dict
client.list_webhooks() -> list[dict]
client.delete_webhook(webhook_id) -> None

Deliver lifecycle events to your own URL. Each delivery is signed with secret in the X-Webhook-Signature header.

client.usage()#

client.usage() -> Usage

Your account's usage: total_sessions and total_vcpu_seconds.

Templates and volumes#

client.create_template()#

client.create_template(name, dockerfile="") -> dict
client.get_template(name) -> dict
client.list_templates() -> list[dict]
client.delete_template(name) -> None

Build a custom template from a Dockerfile, then use it by name in client.sandbox(). See custom templates.

client.create_volume()#

client.create_volume(name, size_gib=1) -> dict
client.get_volume(volume_id) -> dict
client.list_volumes() -> list[dict]
client.delete_volume(volume_id) -> None

Manage persistent volumes that outlive any one sandbox. Mount them with volume_mounts.

MCP#

sb.get_mcp_url()#

sb.get_mcp_url() -> str

Get the URL of the sandbox's MCP proxy, for MCP servers passed in client.sandbox(mcp=…).

Results#

ExecResult#

ExecResult(exit_code: int, stdout: str, stderr: str)

The outcome of a finished command.

StreamingExecResult and ExecChunk#

ExecChunk(kind: str, data: bytes, code: int | None)

Iterating a StreamingExecResult yields ExecChunks, where kind is "stdout", "stderr", "exit" or "pid". Its stdout() method yields just the standard output. Call close() to stop early.

Errors#

Every error derives from UpstreamError, which carries a status_code when there is one.

from upstream import AuthError, SandboxError

try:
    with client.sandbox(template="python") as sb:
        sb.exec(["python3", "main.py"])
except AuthError:
    print("check UPSTREAM_API_KEY")
except SandboxError as e:
    print("sandbox failed:", e)