Skip to content

python-sdk.md

Latest commit

 

History

History
697 lines (479 loc) · 28.2 KB

python-sdk.md

File metadata and controls

697 lines (479 loc) · 28.2 KB
Raw

Python SDK — together-sandbox

Installation

pip install together-sandbox

Requires Python 3.10+.

Authentication

Set your Together AI API key as an environment variable:

export TOGETHER_API_KEY=your_api_key

Or pass it directly when constructing the client (see below).

Quick start

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.

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 below.

TogetherSandbox supports use as an async context manager:

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 instance. This is the primary way to get a running sandbox — no separate start() call is needed.

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": {"aliases": [...], "ttl": int, "tags": {...}}}. 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:<id>").

sdk.sandboxes.terminate(sandbox_id, *, snapshot=UNSET): Coroutine[None]

Terminates a VM by sandbox ID. snapshot ({"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).

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.

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)

Local build opt-in. Set TOGETHER_LOCAL_BUILD=1 in the environment to build the image with your own Docker daemon and push it to the registry from your machine instead of using the remote image-builder. This requires Docker to be installed and running. Useful for debugging build issues locally or when working in restricted network environments.

export TOGETHER_LOCAL_BUILD=1
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.

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.

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.

snapshot = await sdk.snapshots.get_by_alias("my-app@v1")

sdk.snapshots.list(*, limit=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).
exclude_retired bool | None When true, retired snapshots are excluded. Default false.
tags dict | None Matches snapshots whose tags contain all the given pairs. 
# 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, statuses=None, tags=None) -> Page[Sandbox]

List sandboxes. Returns a Page (same shape as snapshots.list()).

Parameter Type Description
limit int | None Page size (1–100, default 20).
statuses list[str] | None Matches sandboxes in any of the given statuses.
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.

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"})

sdk.snapshots.alias(snapshot_id, alias) -> None

Assign (or update) an alias on an existing snapshot.

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.

retired = await sdk.snapshots.retire_by_id("snapshot-id")

Sandbox

A connected, running VM. Returned by sdk.sandboxes.create(). All sub-namespaces are available as properties.

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.

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.

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.

await sandbox.files.delete("/old-file.txt")

files.move(from_path, to_path) -> None

Move a file.

await sandbox.files.move("/src/old.py", "/src/new.py")

files.copy(from_path, to_path) -> None

Copy a file.

await sandbox.files.copy("/src/template.py", "/src/copy.py")

files.stat(path) -> FileInfo

Get file metadata (size, type, modified time, etc.).

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.

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.

files = await sandbox.directories.list("/src")

directories.create(path) -> None

Create a directory.

await sandbox.directories.create("/src/utils")

directories.delete(path) -> None

Delete a directory.

await sandbox.directories.delete("/tmp/scratch")

sandbox.execs

Shell execution operations.

execs.list() -> list[Exec]

List all active execs.

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).

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.

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.

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).

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.

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.

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.
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.

await sandbox.execs.send_stdin("exec-id", "yes\n")

execs.resize(id_, cols: int, rows: int) -> None

Resize the PTY for an interactive exec.

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.

async for update in sandbox.execs.stream_list():
    print(update)

sandbox.ports

Port discovery.

ports.list() -> list[PortInfo]

List all open ports.

ports = await sandbox.ports.list()

ports.stream_list() -> AsyncIterator[dict]

Stream port changes via SSE.

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 ({"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).

# 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"]})

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.

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 instance.

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.

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 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

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:

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.
TOGETHER_LOCAL_BUILD Optional. Set to 1 to build context-based snapshots with your local Docker daemon instead of Together's remote image-builder. See above.