Sandboxes

solari-sandbox: classes SandboxClient and Sandbox. A sandbox is a headless microVM for commands, code, files, and git. See the Python SDK hub for install and configuration.

pip install solari-sandbox

Contents

• SandboxClient: SandboxClient(), create(), create_desktop(), connect(), get(), list(), list_all(), kill(), list_snapshots(), get_snapshot(), delete_snapshot(), promote_snapshot(), aclose(), volumes.create(), volumes.list(), volumes.get(), volumes.delete()
• Sandbox: connect(), reconnect(), close(), commands.run(), commands.start(), pty.create(), run_code(), create_code_context(), files.*, git.*, env(), download_url(), upload_url(), volumes.mount(), metrics(), snapshot(), revert(), pause(), resume(), set_timeout(), preview_url(), kill()
• Types: CreateSandboxResponse, SandboxView, CommandResult, _CommandHandle, _PtyHandle, RunCodeResult, CodeResultItem, Chart, FsEntry, FsStat, FsSearchMatch, FsWatchEvent, GitStatus, GitBranch, GitCommit, MetricsResult, SnapshotView, errors

SandboxClient

The sandbox entry point. Talks the gateway HTTP API.

Properties

• volumes VolumeClient: persistent volume CRUD. Attach one at create time via create(volumes=...).

Constructors

SandboxClient()

SandboxClient(
    *,
    api_key: str,
    base_url: str,
    http: httpx.AsyncClient | None = None,
    call_timeout_ms: int | None = None,
    kind: "sandbox" | "desktop" = "sandbox",
) -> SandboxClient

Creates a client. Does not open any connection. All arguments are keyword-only.

Parameters:

• api_key str: required.
• base_url str: required, e.g. https://api.getsolari.com.
• http httpx.AsyncClient | None: reuse an existing client.
• call_timeout_ms int | None: per-RPC timeout for handles this client creates. Default 300_000.
• kind "sandbox" | "desktop": the flavour create() makes. Default "sandbox".

Returns: SandboxClient

Raises: SolariError if api_key or base_url is missing.

Example:

from solari_sandbox import SandboxClient

client = SandboxClient(
    api_key="slr_live_...",
    base_url="https://api.getsolari.com",
)

Methods

create()

async create(
    *,
    template: str | None = None,
    cpu: int | None = None,
    mem_mb: int | None = None,
    disk_gb: int | None = None,
    envs: dict[str, str] | None = None,
    metadata: dict[str, str] | None = None,
    timeout_ms: int | None = None,
    from_snapshot: str | None = None,
    lifecycle: dict | None = None,
    volumes: list[dict[str, str]] | None = None,
) -> Sandbox

Creates a sandbox and returns a live handle. The call sends an idempotency key, so retrying is safe.

Parameters:

• template str | None: e.g. "base", "code", "python-kernel", or a promoted template id.
• cpu, mem_mb, disk_gb int | None: vCPUs, memory in MiB, disk in GiB. The host clamps to slot capacity.
• envs dict[str, str] | None: per-session env.
• metadata dict[str, str] | None: opaque labels, filterable on list().
• timeout_ms int | None: idle window before auto-release.
• from_snapshot str | None: boot from a snapshot instead of the template.
• lifecycle dict | None: idle policy with camelCase wire keys, e.g. {"onTimeout": "pause", "autoResume": True}.
• volumes list[dict[str, str]] | None: each {"volumeId": ..., "path": ...}, mounted before the session starts.

Returns: Sandbox

Raises: AuthError, PlanError, ConcurrencyLimitError, NoCapacityError, or GatewayError.

Example:

sbx = await client.create(
    template="base",
    cpu=2,
    mem_mb=4096,
    envs={"NODE_ENV": "test"},
)

create_desktop()

async create_desktop(
    *,
    template: str | None = None,
    cpu: int | None = None,
    mem_mb: int | None = None,
    disk_gb: int | None = None,
    envs: dict[str, str] | None = None,
    metadata: dict[str, str] | None = None,
    timeout_ms: int | None = None,
    from_snapshot: str | None = None,
    resolution: str | None = None,
    record: bool | None = None,
    lifecycle: dict | None = None,
    volumes: list[dict[str, str]] | None = None,
) -> Desktop

Creates a GUI session over the same /sandboxes route and returns a Desktop handle. Adds resolution and record to the sandbox options.

Parameters:

• All of create(), plus resolution str | None and record bool | None.

Returns: Desktop

Example:

vm = await client.create_desktop(template="office", resolution="1280x720")

connect()

async connect(sandbox_id: str) -> Sandbox

Re-attaches to a running sandbox by id.

Parameters:

• sandbox_id str: the sandbox id.

Returns: Sandbox

Raises: GatewayError if the id is unknown.

Example:

sbx = await client.connect("sbx_abc123")
await sbx.commands.run("uptime")

get()

async get(sandbox_id: str) -> SandboxView

Fetches a sandbox’s current record: state, kind, metadata, resources. Does not return a handle.

Parameters:

• sandbox_id str: the sandbox id.

Returns: SandboxView

Example:

view = await client.get("sbx_abc123")
print(view.state)  # "running" | "paused" | ...

list()

async list(
    *,
    metadata: dict[str, str] | None = None,
    state: SandboxState | None = None,
    kind: SandboxKind | None = None,
    limit: int | None = None,
    cursor: str | None = None,
) -> dict

Lists sandboxes with optional filters. One page. Follow nextCursor, or use list_all().

Parameters:

• metadata dict[str, str] | None: match on caller-set metadata.
• state SandboxState | None: starting, running, paused, archived, releasing, gone.
• kind SandboxKind | None: sandbox or desktop.
• limit int | None, cursor str | None: page size and cursor.

Returns: dict, a plain dict, not a dataclass: {"sandboxes": list[SandboxView], "nextCursor": str | None}.

Example:

page = await client.list(state="running", limit=50)
print(len(page["sandboxes"]), page["nextCursor"])

list_all()

list_all(
    *,
    metadata: dict[str, str] | None = None,
    state: SandboxState | None = None,
    kind: SandboxKind | None = None,
    limit: int | None = None,
) -> AsyncIterator[SandboxView]

Auto-paginates list(), following nextCursor until exhausted. An async generator. Iterate it, do not await it.

Parameters:

• Same filters as list(), minus cursor.

Returns: AsyncIterator[SandboxView]

Example:

async for s in client.list_all(state="running"):
    print(s.sandboxId, s.cpu, s.memMb)

kill()

async kill(sandbox_id: str) -> None

Destroys a sandbox. Idempotent.

Parameters:

• sandbox_id str: the sandbox id.

Returns: None

Example:

await client.kill("sbx_abc123")

list_snapshots()

async list_snapshots(
    *,
    template: str | None = None,
    kind: SandboxKind | None = None,
    limit: int | None = None,
) -> list[SnapshotView]

Lists snapshots, optionally filtered by template or kind.

Parameters:

• template str | None, kind SandboxKind | None, limit int | None.

Returns: list[SnapshotView], the list directly, not wrapped in a dict.

Example:

for snap in await client.list_snapshots(template="base"):
    print(snap.id, snap.sizeBytes)

get_snapshot()

async get_snapshot(snapshot_id: str) -> SnapshotView

Fetches one snapshot record.

Parameters:

• snapshot_id str: e.g. snap_abc123.

Returns: SnapshotView

Example:

snap = await client.get_snapshot("snap_abc123")
print(snap.sizeBytes, snap.parent)

delete_snapshot()

async delete_snapshot(snapshot_id: str) -> None

Deletes a snapshot.

Parameters:

• snapshot_id str: the snapshot id.

Returns: None

Raises: GatewayError if the snapshot has live children.

Example:

await client.delete_snapshot("snap_abc123")

promote_snapshot()

async promote_snapshot(snapshot_id: str, name: str) -> dict

Promotes a snapshot to a reusable template you can pass as template on future creates.

Parameters:

• snapshot_id str: the snapshot id.
• name str: template name.

Returns: dict, the raw body, with templateId and name.

Example:

snap_id = await sbx.snapshot("deps-installed")
tpl = await client.promote_snapshot(snap_id, "my-stack")
fresh = await client.create(template=tpl["templateId"])

aclose()

async aclose() -> None

Closes the underlying HTTP clients, if this client owns them. Does not touch live sandboxes.

Returns: None

Example:

await client.aclose()

volumes.create()

async volumes.create(
    *,
    name: str | None = None,
    size_mb: int | None = None,
    metadata: dict[str, str] | None = None,
) -> dict

Creates a persistent, S3-backed volume your org owns. It outlives any session.

Parameters:

• name str | None: server default "volume".
• size_mb int | None: advisory hint; storage is elastic, not a hard cap.
• metadata dict[str, str] | None: opaque labels.

Returns: dict, the raw volume body, with volumeId and name. The volume calls return plain dicts, not dataclasses.

Example:

vol = await client.volumes.create(name="datasets")
sbx = await client.create(
    volumes=[{"volumeId": vol["volumeId"], "path": "/data"}],
)

volumes.list()

async volumes.list() -> list[dict]

Lists the org’s volumes, newest first.

Returns: list[dict]

Example:

for v in await client.volumes.list():
    print(v["volumeId"], v["name"])

volumes.get()

async volumes.get(volume_id: str) -> dict

Fetches one volume’s metadata.

Parameters:

• volume_id str: e.g. vol_abc123.

Returns: dict

Example:

vol = await client.volumes.get("vol_abc123")

volumes.delete()

async volumes.delete(volume_id: str) -> None

Deletes a volume’s metadata. Idempotent.

Parameters:

• volume_id str: the volume id.

Returns: None

Example:

await client.volumes.delete("vol_abc123")

Sandbox

A live headless session. Construct via SandboxClient.create().

Properties

• id str: the session id.
• sandboxId str: alias of id.
• controlUrl str: wss:// JSON-RPC control channel.
• expiresAt str: ISO 8601 expiry.
• connected bool: whether the control channel is open.
• commands, pty, files, volumes, git: the grouped namespaces below.

Channel

connect()

async connect() -> None

Opens the control WebSocket. Idempotent, but required: call it once after create() (or connect()) before any command, code, file, or git call. Those methods do not open the channel themselves and raise ConnectionError ("Not connected") until it is open.

Returns: None

Example:

await sbx.connect()

reconnect()

async reconnect() -> None

Re-opens the control socket after a drop. commands.connect() is an alias of this.

Returns: None

Example:

if not sbx.connected:
    await sbx.reconnect()

close()

async close() -> None

Closes the control channel locally. Does not release the remote session. Use kill() for that. Unlike the TypeScript SDK this is a coroutine and must be awaited.

Returns: None

Example:

await sbx.close()  # session keeps running until its timeout

Commands

commands.run()

async commands.run(
    cmd: str,
    *,
    args: list[str] | None = None,
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    user: str | None = None,
    timeout_ms: int | None = None,
    background: bool = False,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
) -> CommandResult

Runs a command to completion and returns its exit code and captured output.

Parameters:

• cmd str: the program to run; the only positional argument.
• args list[str] | None: arguments; no shell expansion.
• cwd, user str | None, env dict[str, str] | None: per-command context.
• background bool: return immediately with an empty result; output still reaches the callbacks.
• on_stdout, on_stderr Callable[[str], None] | None: streamed output.

Returns: CommandResult: exitCode, stdout, stderr. A non-zero exit returns normally; it does not raise.

Example:

r = await sbx.commands.run("ls", args=["-la", "/tmp"])
print(r.exitCode, r.stdout)

# stream output as it arrives
import sys
await sbx.commands.run(
    "npm",
    args=["install"],
    cwd="/app",
    on_stdout=sys.stdout.write,
)

commands.start()

async commands.start(
    cmd: str,
    *,
    args: list[str] | None = None,
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    user: str | None = None,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
) -> _CommandHandle

Starts a command and returns a handle immediately, without waiting for exit. Use it for interactive or long-running processes.

Parameters:

• cmd str: the program to run.
• Others as commands.run(); there is no timeout_ms or background.

Returns: _CommandHandle: cmdId, stdin(), on_data(), wait(), kill().

Example:

proc = await sbx.commands.start("python3", args=["-u", "worker.py"])
proc.on_data(lambda stream, data: print(stream, data))
await proc.stdin("input\n")
exit_code = await proc.wait()

pty.create()

async pty.create(
    *,
    cols: int,
    rows: int,
    cmd: str | None = None,
    cwd: str | None = None,
    env: dict[str, str] | None = None,
) -> _PtyHandle

Opens a pseudo-terminal, a real TTY, so interactive programs and curses UIs behave normally.

Parameters:

• cols, rows int: required terminal size; keyword-only.
• cmd str | None: program to run; defaults to the guest shell.
• cwd str | None, env dict[str, str] | None.

Returns: _PtyHandle: ptyId, write(), resize(), on_data(), kill().

Example:

pty = await sbx.pty.create(cols=80, rows=24)
pty.on_data(lambda b: print(b.decode("utf-8", "replace"), end=""))
await pty.write("ls -la\n")
await pty.kill()

Code

run_code()

async run_code(
    code: str,
    *,
    language: CodeLanguage | None = None,
    context_id: str | None = None,
    on_stdout: Callable[[str], None] | None = None,
    on_stderr: Callable[[str], None] | None = None,
) -> RunCodeResult

Runs code in a stateful kernel. Rich outputs (PNG, HTML, JSON, structured charts) come back as results items.

Parameters:

• code str: source to execute.
• language CodeLanguage | None: python, javascript, typescript, bash, r. The guest defaults to python.
• context_id str | None: kernel context to reuse; see create_code_context().
• on_stdout, on_stderr Callable[[str], None] | None: invoked per matching result item.

Returns: RunCodeResult: results, charts, error. A runtime error in the code lands in error; it does not raise.

Example:

r = await sbx.run_code("""
import matplotlib.pyplot as plt
plt.plot([1, 2, 3], [4, 5, 6])
plt.title("demo")
plt.show()
""")

print(r.charts[0].type)   # "line"
print(r.charts[0].title)  # "demo"
png = next((i.png for i in r.results if i.png), None)  # base64

create_code_context()

async create_code_context(language: CodeLanguage = "python") -> str

Creates a fresh kernel context so state persists across run_code() calls, like a REPL.

Parameters:

• language CodeLanguage: default "python".

Returns: str, the context id.

Example:

ctx = await sbx.create_code_context("python")
await sbx.run_code("x = 41", context_id=ctx)
r = await sbx.run_code("print(x + 1)", context_id=ctx)  # 42

Files

files.* moves bytes over the control channel. For large files prefer download_url() / upload_url().

files.read() / files.read_text()

async files.read(path: str) -> bytes
async files.read_text(path: str) -> str

Reads a file’s bytes, or its UTF-8 decoded text.

Parameters:

• path str: absolute in-guest path.

Returns: bytes / str

Example:

text = await sbx.files.read_text("/etc/hostname")
data = await sbx.files.read("/tmp/out.bin")

files.write()

async files.write(path: str, data: bytes | str, mode: int | None = None) -> None

Writes a file, creating or truncating it.

Parameters:

• path str: absolute in-guest path.
• data bytes | str: content; str is UTF-8 encoded.
• mode int | None: permission bits, e.g. 0o755.

Returns: None

Example:

await sbx.files.write("/app/run.sh", "#!/bin/sh\necho hi\n", 0o755)

files.list()

async files.list(path: str) -> list[FsEntry]

Lists a directory’s entries (non-recursive).

Parameters:

• path str: directory path.

Returns: list[FsEntry], each with name, dir, size.

Example:

for e in await sbx.files.list("/app"):
    print("d" if e.dir else "-", e.name, e.size)

files.stat()

async files.stat(path: str) -> FsStat

Returns one entry’s metadata.

Parameters:

• path str: file or directory path.

Returns: FsStat: name, dir, size, mode, modTimeMs.

Example:

st = await sbx.files.stat("/app/run.sh")
print(st.size, oct(st.mode), st.modTimeMs)

files.rename()

async files.rename(frm: str, to: str) -> None

Renames or moves a path. The first parameter is spelled frm, because from is a Python keyword.

Parameters:

• frm, to str: source and destination paths.

Returns: None

Example:

await sbx.files.rename("/tmp/a.txt", "/tmp/b.txt")

files.remove()

async files.remove(path: str, recursive: bool = False) -> None

Deletes a file or directory.

Parameters:

• path str: path to delete.
• recursive bool: default False; required for non-empty directories.

Returns: None

Example:

await sbx.files.remove("/tmp/build", True)

files.mkdir()

async files.mkdir(path: str) -> None

Creates a directory.

Parameters:

• path str: directory to create.

Returns: None

Example:

await sbx.files.mkdir("/app/data")

files.search()

async files.search(path: str, query: str, max_results: int | None = None) -> list[FsSearchMatch]

Recursively greps for query under path.

Parameters:

• path str: root to search.
• query str: the pattern.
• max_results int | None: cap on matches.

Returns: list[FsSearchMatch], each with path, line, text.

Example:

for m in await sbx.files.search("/app", "TODO", 20):
    print(f"{m.path}:{m.line}: {m.text}")

files.watch()

async files.watch(
    path: str,
    cb: Callable[[FsWatchEvent], None],
    recursive: bool = False,
) -> Callable[[], Awaitable[None]]

Watches a path for changes. Returns an async unwatch callable. Await it to stop.

Parameters:

• path str: path to watch.
• cb Callable[[FsWatchEvent], None]: invoked per event with type and path.
• recursive bool: default False.

Returns: Callable[[], Awaitable[None]]: the unwatch function.

Example:

unwatch = await sbx.files.watch(
    "/app",
    lambda ev: print(ev.type, ev.path),
    True,
)

await unwatch()

files.upload() / files.download()

async files.upload(path: str, data: bytes | str) -> None
async files.download(path: str) -> bytes

Transfer a file over the control channel. Same bytes as write() / read().

Parameters:

• path str: in-guest path.
• data bytes | str: upload content.

Returns: None / bytes

Example:

await sbx.files.upload("/app/data.csv", csv_string)
out = await sbx.files.download("/app/result.json")

download_url()

async download_url(path: str) -> dict

Signed, time-limited URL to download an in-guest file directly over HTTP, bypassing the control channel. Use for large files.

Parameters:

• path str: in-guest file path.

Returns: dict, with url and usually expiresAt.

Raises: RuntimeError if the handle was not created by a client.

Example:

import httpx

dl = await sbx.download_url("/app/big.tar.gz")
async with httpx.AsyncClient() as http:
    res = await http.get(dl["url"])

upload_url()

async upload_url(path: str | None = None) -> dict

Signed, time-limited URL to upload a file into the guest directly over HTTP.

Parameters:

• path str | None: destination path; optional, some backends return a prefix.

Returns: dict, with url and usually expiresAt.

Raises: RuntimeError if the handle was not created by a client.

Example:

up = await sbx.upload_url("/app/big.zip")
async with httpx.AsyncClient() as http:
    await http.put(up["url"], content=data)

Git

Every git.* call is a non-shell git invocation in the guest, so there is no injection surface. Requires git on PATH, which the base template ships.

git.clone()

async git.clone(
    url: str,
    *,
    path: str | None = None,
    branch: str | None = None,
    depth: int | None = None,
    username: str | None = None,
    password: str | None = None,
    cwd: str | None = None,
) -> None

Clones a repository into the guest.

Parameters:

• url str: remote URL.
• path str | None: destination dir.
• branch str | None: branch or tag.
• depth int | None: shallow-clone depth.
• username, password str | None: basic auth for private HTTPS remotes; injected into the URL for this invocation only.
• cwd str | None: working directory.

Returns: None

Raises: RuntimeError with git’s stderr on a non-zero exit.

Example:

await sbx.git.clone(
    "https://github.com/org/repo.git",
    path="/app/repo",
    branch="main",
    depth=1,
)

git.status()

async git.status(cwd: str | None = None) -> GitStatus

Parsed working-tree status.

Parameters:

• cwd str | None: repository directory.

Returns: GitStatus

Raises: RuntimeError on a non-zero exit.

Example:

st = await sbx.git.status("/app/repo")
print(st.branch, st.clean, st.ahead, st.modified)

git.add()

async git.add(paths: list[str], cwd: str | None = None) -> None

Stages paths. An empty list is a no-op.

Parameters:

• paths list[str]: use ["."] for everything.
• cwd str | None: repository directory.

Returns: None

Example:

await sbx.git.add(["."], "/app/repo")

git.commit()

async git.commit(
    message: str,
    *,
    cwd: str | None = None,
    author: str | None = None,
    email: str | None = None,
    all: bool = False,
) -> dict

Commits staged changes and returns the new hash. Set author/email, because an ephemeral session has no git identity.

Parameters:

• message str: commit message.
• cwd str | None: repository directory.
• author, email str | None: identity, scoped to this commit only.
• all bool: stage all tracked modifications first.

Returns: dict: {"hash": str}.

Raises: RuntimeError on a non-zero exit.

Example:

c = await sbx.git.commit(
    "add feature",
    cwd="/app/repo",
    author="CI Bot",
    email="ci@example.com",
)
print(c["hash"])

git.push() / git.pull()

async git.push(
    *,
    cwd: str | None = None,
    remote: str | None = None,
    branch: str | None = None,
    username: str | None = None,
    password: str | None = None,
) -> None
async git.pull(...)  # identical signature

Pushes to or pulls from a remote. Credentials are used for this one invocation and never persisted to the repo config.

Parameters:

• cwd str | None: repository directory.
• remote str | None: default "origin".
• branch str | None: defaults to the current branch’s upstream.
• username, password str | None: one-off HTTPS auth.

Returns: None

Raises: RuntimeError on a non-zero exit.

Example:

import os

await sbx.git.push(
    cwd="/app/repo",
    username="x-access-token",
    password=os.environ["GITHUB_TOKEN"],
)

git.checkout()

async git.checkout(ref: str, *, cwd: str | None = None, create: bool = False) -> None

Checks out an existing ref, or creates a branch.

Parameters:

• ref str: branch, tag, or commit.
• cwd str | None: repository directory.
• create bool: create the branch (-b).

Returns: None

Example:

await sbx.git.checkout("feature/x", cwd="/app/repo", create=True)

git.branches()

async git.branches(cwd: str | None = None) -> list[GitBranch]

Lists local branches.

Parameters:

• cwd str | None: repository directory.

Returns: list[GitBranch], each with name, commit, current.

Example:

for b in await sbx.git.branches("/app/repo"):
    print("*" if b.current else " ", b.name, b.commit)

git.log()

async git.log(*, cwd: str | None = None, max_count: int | None = None) -> list[GitCommit]

Recent commits, newest first.

Parameters:

• cwd str | None: repository directory.
• max_count int | None: cap the number returned.

Returns: list[GitCommit], each with hash, author, email, date, message.

Example:

commits = await sbx.git.log(cwd="/app/repo", max_count=10)

Lifecycle & admin

env()

async env(vars: dict[str, str]) -> None

Injects or replaces per-session environment variables in the guest.

Parameters:

• vars dict[str, str]: the variables to set.

Returns: None

Example:

await sbx.env({"API_URL": "https://staging.example.com"})

volumes.mount()

async volumes.mount(vol_id: str, path: str) -> None

Mounts a persistent volume into the running guest. Prefer attaching at create time via create(volumes=...).

Parameters:

• vol_id str: volume id.
• path str: absolute in-guest mount point.

Returns: None

Example:

await sbx.volumes.mount("vol_abc123", "/data")

metrics()

async metrics() -> MetricsResult

Live resource usage for this session.

Returns: MetricsResult: cpuPct, memBytes, memTotalBytes, diskBytes.

Raises: RuntimeError if the handle was not created by a client.

Example:

m = await sbx.metrics()
print(m.cpuPct, m.memBytes / m.memTotalBytes)

snapshot()

async snapshot(name: str | None = None) -> str

Checkpoints this running session and returns the snapshot id. The session keeps running.

Parameters:

• name str | None: optional label.

Returns: str, the snapshot id.

Raises: RuntimeError if the handle was not created by a client.

Example:

await sbx.commands.run("pip", args=["install", "pandas"])
snap_id = await sbx.snapshot("deps-installed")

# boot future sandboxes from it
fast = await client.create(from_snapshot=snap_id)

revert()

async revert(snapshot_id: str) -> None

Restores this session in place from a snapshot. The session id stays stable.

Parameters:

• snapshot_id str: snapshot to restore.

Returns: None

Raises: RuntimeError if the handle was not created by a client.

Example:

await sbx.revert(snap_id)  # back to a known-good state

pause()

async pause() -> None

Saves full RAM + disk state and closes the control channel. Resume later with resume().

Returns: None

Raises: RuntimeError if the handle was not created by a client.

Example:

await sbx.pause()   # stops billing for compute
await sbx.resume()  # full state restored

resume()

async resume() -> None

Resumes a paused session and re-points the control channel at the fresh slot.

Returns: None

Raises: RuntimeError if the handle was not created by a client.

Example:

await sbx.resume()
r = await sbx.commands.run("cat", args=["/tmp/state"])

set_timeout()

async set_timeout(timeout_ms: int) -> dict

Extends the rolling idle window. Every use of the session resets it.

Parameters:

• timeout_ms int: new idle window in ms.

Returns: dict, the raw body, carrying expiresAt.

Raises: RuntimeError if the handle was not created by a client.

Example:

r = await sbx.set_timeout(30 * 60_000)  # 30 min
print(r["expiresAt"])

preview_url()

async preview_url(port: int) -> dict

Resolves a public URL for a port listening inside the guest, e.g. a dev server.

Parameters:

• port int: the in-guest port.

Returns: dict, with url, and token when the gateway signs one.

Raises: RuntimeError if the handle was not created by a client.

Example:

await sbx.commands.start("python3", args=["-m", "http.server", "3000"])
p = await sbx.preview_url(3000)
print(p["url"])

kill()

async kill() -> None

Destroys the remote session and closes the channel. Idempotent.

Returns: None

Example:

try:
    await sbx.commands.run("./build.sh")
finally:
    await sbx.kill()

Types

CreateSandboxResponse

• sandboxId str, kind SandboxKind.
• controlUrl str, expiresAt str.
• streamUrl str | None: present only when kind == "desktop".

SandboxView

• sandboxId str, kind SandboxKind, state SandboxState.
• metadata dict, expiresAt str.
• cpu int, memMb int.

CommandResult

• exitCode int, stdout str, stderr str.

_CommandHandle

Returned by commands.start(); not constructed directly.

• cmdId str: the command id.
• on_data(cb): subscribe to (stream, data) chunks. Synchronous. Chunks that arrive before the first subscriber are buffered and replayed, so early output is never lost.
• await stdin(data): write to the command’s stdin.
• await wait(): int, the exit code. Raises if the control channel drops before the process exits.
• await kill(signal=None): signal the process.

_PtyHandle

• ptyId str: the PTY id.
• on_data(cb): subscribe to raw bytes. Synchronous.
• await write(data), await resize(cols, rows), await kill().

RunCodeResult / CodeResultItem

• results list[CodeResultItem]: each has type (stdout/stderr/result) plus any of text, png, jpeg, svg, html, latex, json, markdown, chart.
• charts list[Chart]: every chart across results, flattened.
• error Any: the raw error object when the code raised.

Chart

• type ChartType: line, scatter, bar, pie, box_and_whisker, composite, unknown.
• title, xLabel, yLabel str | None.
• x, y ChartAxis | None: label, ticks, scale.
• elements list | None: per-type data (points, bars, slices).

FsEntry / FsStat / FsSearchMatch / FsWatchEvent

• FsEntry: name str, dir bool, size int.
• FsStat: FsEntry’s fields plus mode int (permission bits) and modTimeMs int (unix millis).
• FsSearchMatch: path str, line int, text str.
• FsWatchEvent: type str, path str.

GitStatus / GitBranch / GitCommit

• GitStatus.branch str: empty on a detached HEAD.
• GitStatus.detached, clean bool.
• GitStatus.ahead, behind int: relative to the upstream, when set.
• GitStatus.staged, modified, untracked list[str].
• GitBranch: name, commit str, current bool.
• GitCommit: hash, author, email, date, message str.

MetricsResult

• cpuPct float, memBytes int, memTotalBytes int, diskBytes int.

SnapshotView

• id str, parent str | None, name str | None.
• sizeBytes int, createdAt str, kind SandboxKind, template str.

Errors

All extend SolariError. Gateway errors also carry status, code, and body.

from solari_sandbox import ConcurrencyLimitError, NoCapacityError

try:
    sbx = await client.create(template="base")
except NoCapacityError:
    await retry_later()
except ConcurrencyLimitError:
    await wait_for_slot()