# Python SDK — `together-sandbox` ## Installation ```bash pip install together-sandbox ``` Requires Python 3.10+. --- ## Authentication Set your Together AI API key as an environment variable: ```bash export TOGETHER_API_KEY=your_api_key ``` Or pass it directly when constructing the client (see below). --- ## Quick start ```python import asyncio from together_sandbox import TogetherSandbox async def main(): sdk = TogetherSandbox() # reads TOGETHER_API_KEY from env async with await sdk.sandboxes.create(snapshot_alias="my-app@v1") as sandbox: content = await sandbox.files.read("/package.json") print(content) await sandbox.terminate() asyncio.run(main()) ``` > **Note:** The `async with` block closes the HTTP connection on exit. It does **not** automatically terminate the VM — call `await sandbox.terminate()` explicitly when you're done. --- ## `TogetherSandbox` The main entry point for the SDK. ```python sdk = TogetherSandbox(api_key=None, base_url="https://api.bartender.codesandbox.io") ``` | Parameter | Type | Description | | ---------- | --------------------- | -------------------------------------------------------------------------------------------------------------------------- | | `api_key` | `str \| None` | Together AI API key. Falls back to the `TOGETHER_API_KEY` env var. | | `base_url` | `str` | Management API base URL. Defaults to `https://api.bartender.codesandbox.io`. Override via the `TOGETHER_BASE_URL` env var. | | `retry` | `RetryConfig \| None` | Retry configuration for transient failures. See [Retry](#retry) below. | `TogetherSandbox` supports use as an async context manager: ```python async with TogetherSandbox() as sdk: sandbox = await sdk.sandboxes.create(snapshot_alias="my-app@v1") ... # Closes the management API HTTP connection on exit. ``` ### `sdk.sandboxes` Sandbox lifecycle namespace. #### `sdk.sandboxes.create(*, cpu=1.0, memory_bytes=2*1024**3, snapshot_id=None, snapshot_alias=None, ttl=None, tags=None, termination_policy=None) -> Sandbox` Creates a new sandbox from a snapshot, starts the VM, and returns a connected [`Sandbox`](#sandbox) instance. This is the primary way to get a running sandbox — no separate `start()` call is needed. ```python sandbox = await sdk.sandboxes.create(snapshot_alias="my-app@v1") ``` Resource params (`cpu`, `memory_bytes`) default to **1 vCPU / 2 GiB memory** if omitted. | Parameter | Type | Required | Description | | ---------------- | -------------- | -------- | -------------------------------------------------------------------------------------- | | `snapshot_id` | `str \| None` | \* | ID of the snapshot to use. One of `snapshot_id` or `snapshot_alias` is required. | | `snapshot_alias` | `str \| None` | \* | Alias of the snapshot to use. One of `snapshot_id` or `snapshot_alias` is required. | | `cpu` | `float` | No | CPU allocation in cores (0.1–16). Default: `1.0` (1 vCPU). | | `memory_bytes` | `int` | No | Memory allocation in bytes (1–8 GB per CPU). Default: `2 * 1024 ** 3` (2 GiB). | | `ttl` | `int \| None` | No | Seconds after creation before the sandbox is automatically terminated. | | `tags` | `dict \| None` | No | Arbitrary key/value labels to attach to the sandbox. | | `termination_policy` | `dict \| None` | No | Termination policy `{"snapshot": {"memory": bool, "aliases": [...], "ttl": int, "tags": {...}}}`. `"memory": True` also snapshots memory (hibernate). Omit for an ephemeral sandbox (no snapshot, deleted on termination). | Sandboxes start automatically on creation, so there is no separate start step. A terminated sandbox cannot be used again — to continue from its state, create a new sandbox from the snapshot it produced (`snapshot_alias="sandbox:"`). #### `sdk.sandboxes.get(sandbox_id) -> SandboxModel` Fetches a single sandbox's metadata by ID. Returns the same raw `SandboxModel` record that `sdk.sandboxes.list()` yields — not a connected [`Sandbox`](#sandbox) client, so it does not give you `.files`, `.execs`, or the other in-VM namespaces. Use [`connect()`](#sdksandboxesconnectsandbox_id---sandbox) for that. ```python info = await sdk.sandboxes.get("your-sandbox-id") print(info.status, info.status_reason) ``` #### `sdk.sandboxes.connect(sandbox_id) -> Sandbox` Connects to an existing sandbox by ID and returns a connected [`Sandbox`](#sandbox) instance. If the sandbox is still in a transient status (e.g. `starting`), it waits for it to settle. Raises if the sandbox does not end up `running` (e.g. it is `terminated`). ```python sandbox = await sdk.sandboxes.connect("your-sandbox-id") content = await sandbox.files.read("/package.json") ``` #### `sdk.sandboxes.terminate(sandbox_id, *, snapshot=UNSET): Coroutine[None]` Terminates a VM by sandbox ID. `snapshot` (`{"memory": ..., "aliases": [...], "ttl": ..., "tags": {...}}`) overrides what the sandbox's stored termination policy would snapshot for this teardown — omit it to use the stored policy, or pass `None` for an ephemeral teardown (no snapshot). ```python await sdk.sandboxes.terminate("your-sandbox-id", snapshot={"aliases": ["my-app@v2"]}) ``` --- ### `sdk.snapshots` Snapshot creation namespace. Snapshots are images you can pass to `sdk.sandboxes.create()`. #### `sdk.snapshots.create(params): Coroutine[CreateSnapshotResult]` Create a snapshot from either a Docker build context (built remotely by default) or an existing public Docker image. **From a build context (remote build):** Submit a Docker build context to Together's remote image-builder service. The service builds the image, pushes it to the internal registry, and the SDK then registers it as a snapshot. No local Docker installation is required. ```python from together_sandbox import CreateContextSnapshotParams result = await sdk.snapshots.create(CreateContextSnapshotParams( context="./my-app", dockerfile="./my-app/Dockerfile.prod", # optional alias="my-app@v1", # optional ttl=86400, # optional — auto-retire after N seconds on_progress=lambda e: print(e.output), )) # Use the snapshot ID to create a sandbox: sandbox = await sdk.sandboxes.create(snapshot_id=result.snapshot_id) ``` | Parameter | Type | Description | | ------------- | -------------------------------------------- | ----------------------------------------------------------------------------------------------------------- | | `context` | `str` | Path to the Docker build context directory. | | `dockerfile` | `str \| None` | Path to a Dockerfile. Defaults to `Dockerfile` inside `context`. | | `alias` | `str \| None` | Alias for the snapshot. Format: `tag` or `namespace@tag`. Namespace defaults to the context directory name. | | `cache_key` | `str \| None` | Groups remote builds that share a layer cache; builds with the same key reuse each other's layers. Each snapshot builds under a freshly generated image name, so with no key there is nothing to match against and the cache never hits — set a stable key to get reuse. Lowercase path of alphanumerics, `.`, `_`, `-`, `/`; no tag; max 255 chars. Ignored for local builds. | | `ttl` | `int \| None` | Seconds after creation before the snapshot is automatically retired. Omit to keep it indefinitely. | | `on_progress` | `Callable[[SnapshotProgress], None] \| None` | Optional progress callback. Receives a `SnapshotProgress` at each stage. | **From a public Docker image:** The image is pulled and optimized by Together's remote image-builder service (including nydus conversion for fast cold-starts), then registered as a snapshot. No local Docker installation is required. ```python from together_sandbox import CreateImageSnapshotParams result = await sdk.snapshots.create(CreateImageSnapshotParams( image="node:22", alias="my-node@latest", # optional )) print(result.snapshot_id) ``` | Parameter | Type | Description | | ------------- | -------------------------------------------- | ----------------------------------------------------------------------------------------------- | | `image` | `str` | Docker image name or reference (e.g. `node:22`, `registry.example.com/org/app:tag`). | | `alias` | `str \| None` | Alias for the snapshot. Format: `tag` or `namespace@tag`. Namespace defaults to the image name. | | `on_progress` | `Callable[[SnapshotProgress], None] \| None` | Optional progress callback. | #### `CreateSnapshotResult` | Property | Type | Description | | ------------- | ------------- | ----------------------------------------------------- | | `snapshot_id` | `str` | ID of the created snapshot. | | `alias` | `str \| None` | The full alias (`namespace@tag`) if one was assigned. | #### `SnapshotProgress` | Property | Type | Description | | -------- | ----- | ----------------------------------------------------------------------------------------------------------- | | `step` | `str` | Current stage: `"prepare"`, `"build"`, `"auth"`, `"push"`, `"register"`, or `"alias"`. | | `output` | `str` | Human-readable progress message. | #### `sdk.snapshots.get_by_id(id) -> Snapshot` Fetch snapshot metadata by ID. ```python snapshot = await sdk.snapshots.get_by_id("snapshot-id") print(snapshot.id, snapshot.byte_size) ``` #### `sdk.snapshots.get_by_alias(alias) -> Snapshot` Fetch snapshot metadata by alias. ```python snapshot = await sdk.snapshots.get_by_alias("my-app@v1") ``` #### `sdk.snapshots.list(*, limit=None, cursor=None, exclude_retired=None, tags=None) -> Page[Snapshot]` List snapshots. Returns a `Page` that is async-iterable across all pages — iterate it directly to walk every snapshot, or use `get_next_page()` / `next_cursor` for manual page-by-page control. | Parameter | Type | Description | | ----------------- | --------------------- | ----------------------------------------------------------- | | `limit` | `int \| None` | Page size (1–100, default 20). | | `cursor` | `str \| None` | Start from an opaque cursor instead of the first page. | | `exclude_retired` | `bool \| None` | When true, retired snapshots are excluded. Default false. | | `tags` | `dict \| None` | Matches snapshots whose tags contain all the given pairs. | ```python # Iterate every snapshot across all pages async for snapshot in await sdk.snapshots.list(): print(snapshot.id) # Or page-by-page when the cursor matters page = await sdk.snapshots.list(limit=50) while page.has_next_page(): print(page.data, page.next_cursor) page = await page.get_next_page() # Only live snapshots for one service live = await sdk.snapshots.list(exclude_retired=True, tags={"service": "api"}) ``` #### `sdk.sandboxes.list(*, limit=None, cursor=None, statuses=None, snapshot_id=None, tags=None) -> Page[SandboxModel]` List sandboxes. Returns a `Page` (same shape as `snapshots.list()`). | Parameter | Type | Description | | ------------- | ------------------- | --------------------------------------------------------- | | `limit` | `int \| None` | Page size (1–100, default 20). | | `cursor` | `str \| None` | Start from an opaque cursor instead of the first page. | | `statuses` | `list[str] \| None` | Matches sandboxes in any of the given statuses. | | `snapshot_id` | `str \| None` | Matches sandboxes created from the given snapshot. | | `tags` | `dict \| None` | Matches sandboxes whose tags contain all the given pairs. | A status is one of `starting`, `running`, `terminating`, `terminated`, `failed_to_start`, `recovering`, `unrecovered`. ```python async for sandbox in await sdk.sandboxes.list(): print(sandbox.id) # Running sandboxes for one team running = await sdk.sandboxes.list(statuses=["running"], tags={"team": "platform"}) # Every sandbox booted from one snapshot from_snapshot = await sdk.sandboxes.list(snapshot_id="snapshot-id") ``` #### `sdk.snapshots.alias(snapshot_id, alias) -> None` Assign (or update) an alias on an existing snapshot. ```python await sdk.snapshots.alias("snapshot-id", "my-app@v2") ``` #### `sdk.snapshots.retire_by_id(id) -> Snapshot` Retire a snapshot by ID and return the retired snapshot. Once retired, the snapshot can no longer be used to create new sandboxes, and it is eventually deleted, but only once no sandbox still references it. ```python retired = await sdk.snapshots.retire_by_id("snapshot-id") ``` --- ## `Sandbox` A connected, running VM. Returned by `sdk.sandboxes.create()` and `sdk.sandboxes.connect()`. All sub-namespaces are available as properties. ```python async with await sdk.sandboxes.create(snapshot_alias="my-app@v1") as sandbox: ... ``` ### Properties | Property | Type | Description | | --------- | -------------- | ------------------------------------------- | | `id` | `str` | The sandbox/VM ID. | | `vm_info` | `SandboxModel` | Raw sandbox record (id, status, agent, etc.) | --- ### `sandbox.files` File system operations. #### `files.read(path) -> str` Read the content of a file. ```python content = await sandbox.files.read("/src/main.py") ``` #### `files.create(path, content) -> str` Create or overwrite a file. Content can be `str` (encoded as UTF-8) or `bytes`. ```python await sandbox.files.create("/hello.txt", "Hello, world!") await sandbox.files.create("/data.bin", b"\x00\x01\x02") ``` #### `files.delete(path) -> None` Delete a file. ```python await sandbox.files.delete("/old-file.txt") ``` #### `files.move(from_path, to_path) -> None` Move a file. ```python await sandbox.files.move("/src/old.py", "/src/new.py") ``` #### `files.copy(from_path, to_path) -> None` Copy a file. ```python await sandbox.files.copy("/src/template.py", "/src/copy.py") ``` #### `files.stat(path) -> FileInfo` Get file metadata (size, type, modified time, etc.). ```python info = await sandbox.files.stat("/package.json") ``` #### `files.watch(path, *, recursive=None, ignore_patterns=None) -> AsyncIterator[dict]` Watch a directory for file system changes via SSE. Returns an async iterator of event dicts. ```python async for event in sandbox.files.watch("/src", recursive=True, ignore_patterns=["node_modules"]): print(event) ``` | Parameter | Type | Description | | ----------------- | ------------------- | ---------------------------------- | | `recursive` | `bool \| None` | Watch subdirectories recursively. | | `ignore_patterns` | `list[str] \| None` | Glob patterns for paths to ignore. | --- ### `sandbox.directories` Directory operations. #### `directories.list(path) -> list[FileInfo]` List the contents of a directory. ```python files = await sandbox.directories.list("/src") ``` #### `directories.create(path) -> None` Create a directory. ```python await sandbox.directories.create("/src/utils") ``` #### `directories.delete(path) -> None` Delete a directory. ```python await sandbox.directories.delete("/tmp/scratch") ``` --- ### `sandbox.execs` Shell execution operations. #### `execs.list() -> list[Exec]` List all active execs. ```python execs = await sandbox.execs.list() ``` #### `execs.create(command, args, *, autostart=None, pty=None, cwd=None, env=None, user=None) -> Exec` Create a new exec (run a command). ```python exec_ = await sandbox.execs.create( command="npm", args=["install"], cwd="/workspace", env={"NODE_ENV": "production"}, ) ``` | Parameter | Type | Required | Description | | ----------- | ------------------------ | -------- | -------------------------------------------------------------- | | `command` | `str` | Yes | Command to execute (e.g. `"npm"`). | | `args` | `list[str]` | Yes | Command line arguments (e.g. `["install"]`). | | `autostart` | `bool \| None` | No | Whether to automatically start the exec (defaults to true). | | `pty` | `bool \| None` | No | Whether to start a PTY shell session. | | `cwd` | `str \| None` | No | Working directory for the command. | | `env` | `dict[str, str] \| None` | No | Environment variables to set, as a plain dict. | | `user` | `str \| None` | No | $USER:$GROUP ID to run the command as (defaults to 1000:1000). | #### `execs.get(id_) -> Exec` Get an exec by ID. ```python exec_ = await sandbox.execs.get("exec-id") ``` #### `execs.start(id_) -> Exec` Start an exec configured with `autostart=False`. Will succeed if in `CREATED` or `RUNNING` state, or returns `409` error. ```python exec_ = await sandbox.execs.start("exec-id") ``` #### `execs.exec(command, args, *, pty=None, cwd=None, env=None, user=None) -> dict` Run a command to completion and return its result. Creates an exec with `autostart=True`, streams its output via SSE, and waits for the process to exit. Returns a dict with the final `exit_code` and the joined output. Returned dict shape: - `exit_code` (`int`) — the process's exit code. Guaranteed to be present; if the stream ends without an exit code event the call raises `RuntimeError` instead. - `output` (`str`) — the concatenation of all stdout and stderr chunks in arrival order. Use `get_output()` or `stream_output()` if you need per-chunk metadata (type, sequence, timestamp). Raises `RuntimeError` if the stream ends without delivering an exit code (e.g. the process was killed externally or the sandbox was shut down mid-run). ```python result = await sandbox.execs.exec("sh", ["-c", "echo hello && echo oops >&2 && exit 3"]) assert result["exit_code"] == 3 assert "hello" in result["output"] assert "oops" in result["output"] ``` #### `execs.delete(id_) -> None` Delete an exec. ```python await sandbox.execs.delete("exec-id") ``` #### `execs.stream_output(id_, last_sequence=None) -> AsyncIterator[dict]` Stream exec output via SSE. Optionally provide `last_sequence` to resume from a specific point. ```python async for chunk in sandbox.execs.stream_output("exec-id"): print(chunk) ``` #### `execs.get_output(id_, last_sequence=None) -> ExecOutputResult` One-shot poll for the exec output buffer (non-streaming). Returns the same `{exit_code, output}` shape as `execs.exec()` — the only difference is that `exit_code` may be `None` here because the process may still be running when polled. Use `stream_output()` if you want individual events with per-chunk metadata. Returned dict shape (a `TypedDict` named `ExecOutputResult`): - `exit_code` (`int` or `None`) — the process's exit code, or `None` if the process hasn't exited yet. - `output` (`str`) — the concatenation of all stdout and stderr chunks received so far, in arrival order. ```python result = await sandbox.execs.get_output("exec-id") if result["exit_code"] is not None: print(f"Process exited with {result['exit_code']}: {result['output']}") ``` #### `execs.send_stdin(id_, data: str) -> None` Send raw stdin data to a running exec. ```python await sandbox.execs.send_stdin("exec-id", "yes\n") ``` #### `execs.resize(id_, cols: int, rows: int) -> None` Resize the PTY for an interactive exec. ```python await sandbox.execs.resize("exec-id", cols=80, rows=24) ``` #### `execs.stream_list() -> AsyncIterator[dict]` Stream the live list of all active execs via SSE. ```python async for update in sandbox.execs.stream_list(): print(update) ``` --- ### `sandbox.ports` Port discovery. #### `ports.list() -> list[PortInfo]` List all open ports. ```python ports = await sandbox.ports.list() ``` #### `ports.stream_list() -> AsyncIterator[dict]` Stream port changes via SSE. ```python async for event in sandbox.ports.stream_list(): print(event) ``` --- ### Lifecycle methods #### `sandbox.terminate(*, snapshot=UNSET) -> None` Terminate this VM. After this the sandbox is terminal and cannot be used again. `snapshot` (`{"memory": ..., "aliases": [...], "ttl": ..., "tags": {...}}`) overrides what this teardown snapshots — omit it to use the sandbox's stored termination policy, or pass `None` for an ephemeral teardown (no snapshot). ```python # Use the stored termination policy await sandbox.terminate() # Snapshot the filesystem and alias it, so a new sandbox can start from it await sandbox.terminate(snapshot={"aliases": ["my-app@v2"]}) # Snapshot the filesystem and memory, so a new sandbox can resume from it await sandbox.terminate(snapshot={"memory": True}) ``` `memory` defaults to `False`. A sandbox created from a snapshot with memory resumes with its processes intact. #### `sandbox.close() -> None` Close the underlying sandbox HTTP client connection without affecting the VM state. --- ### Async context manager `Sandbox` supports use as an async context manager. Exiting the block closes the HTTP connection but does **not** terminate the VM. ```python async with await sdk.sandboxes.create(snapshot_alias="my-app@v1") as sandbox: content = await sandbox.files.read("/README.md") await sandbox.terminate() ``` --- ## Static factory methods on `Sandbox` Convenience classmethods that create a temporary SDK client internally. ### `Sandbox.create(*, snapshot_id=None, snapshot_alias=None, api_key=None, base_url=..., **kwargs)` Creates a sandbox from a snapshot, starts the VM, and returns a connected [`Sandbox`](#sandbox) instance. ```python sandbox = await Sandbox.create(snapshot_id="your-snapshot-id", api_key="your-key") ``` To terminate a sandbox by ID without a running `Sandbox` instance, use the namespace method: `await sdk.sandboxes.terminate("sandbox-id")`. --- ## Errors ### `HttpError` Raised by SDK operations for **every** failure. Inherits from `RuntimeError`, so existing `except RuntimeError:` clauses keep working. HTTP-level errors (non-success status) and transport-level failures (DNS, timeout, connection refused) both surface as `HttpError`, distinguished by the `status` field. | Attribute | Type | Description | | --------- | ------- | ------------------------------------------------------------------------------------- | | `args` | `tuple` | Standard exception args — first element is the formatted message. | | `status` | `int` | HTTP status code, or `0` as a sentinel for transport failures (no response received). | Because `0` is not a valid HTTP status, `e.status == 0` cleanly identifies "the request never reached the server" without losing the rest of the error-handling shape. The original transport exception (`httpx.TimeoutException`, `httpx.ConnectError`, `httpx.RemoteProtocolError`) is preserved on `__cause__` for debugging tracebacks. ```python from together_sandbox import HttpError try: await sdk.snapshots.get_by_id("...") except HttpError as e: if e.status == 0: # transport-level failure (DNS / timeout / connection refused) raise elif e.status == 404: return None # not found raise ``` --- ## Retry All operations automatically retry on transient failures. By default: - **HTTP status codes** `408`, `429`, `500`, `502`, `503`, `504` trigger a retry. - **Transport-level failures** (`httpx.TimeoutException`, `httpx.ConnectError`, `httpx.RemoteProtocolError`) — these surface as `HttpError` with `status == 0` — trigger a retry. - **3 total attempts** (1 initial + 2 retries). - **Exponential backoff**: starts at 0.5 s, doubles each attempt, plus up to 0.25 s of random jitter. Pass a `RetryConfig` to `TogetherSandbox(retry=...)` to customise this behaviour. ### `RetryConfig` `RetryConfig` is a dataclass — instantiate it with keyword arguments. | Field | Type | Default | Description | | -------------- | ----------------------------------------------------------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `max_attempts` | `int` | `3` | Total number of attempts (including the first). Set to `1` to disable retries. | | `should_retry` | `Callable[[RetryContext], bool \| float] \| Awaitable[...]` | `None` | Override the retry decision. Return `False` to abort immediately, `True` to retry with the default backoff delay, or a `float` (seconds) to retry after a custom delay. May be a coroutine function. | | `on_retry` | `Callable[[RetryContext], None] \| Awaitable[None]` | `None` | Called before each retry. Use for logging, metrics, or UI progress updates. May be a coroutine function. | ### `RetryContext` | Field | Type | Description | | ----------- | ------------- | -------------------------------------------------------------------------------------------- | | `operation` | `str` | The operation that failed, e.g. `'api.terminate_sandbox'`, `'files.read'`. | | `attempt` | `int` | 1-based number of the attempt that just failed. | | `error` | `Exception` | The [`HttpError`](#httperror) that was raised. | | `status` | `int \| None` | HTTP status code, or `0` for transport-level failures. | | `delay` | `float` | **Seconds** to wait before the next attempt (default computed, override via `should_retry`). | ### Example ```python import asyncio from together_sandbox import TogetherSandbox, RetryConfig, RetryContext async def main(): sdk = TogetherSandbox( api_key="...", retry=RetryConfig( max_attempts=4, should_retry=lambda ctx: ( # Never retry snapshot creation — it is not idempotent False if ctx.operation == "snapshots.create" # Give up on auth errors else False if ctx.status in (401, 403) # Retry everything else with default backoff else True ), on_retry=lambda ctx: print( f"[retry] {ctx.operation} attempt {ctx.attempt} failed " f"({'transport' if ctx.status == 0 else f'HTTP {ctx.status}'}) " f"— retrying in {ctx.delay:.2f}s" ), ), ) asyncio.run(main()) ``` > **Note — `snapshots.create` is not idempotent.** Retrying after a transient 500 once the snapshot has already been created will register a duplicate. Exclude it via `should_retry` as shown above, or use the shorthand: > > ```python > RetryConfig(should_retry=lambda ctx: ctx.operation != "snapshots.create") > ``` --- ## Environment variables | Variable | Description | | ------------------- | ----------------------------------------------- | | `TOGETHER_API_KEY` | Required. Your Together AI API key. | | `TOGETHER_BASE_URL` | Optional. Override the management API base URL. |