Sandbox

class Sandbox(*, command: str | None = None, args: list[str] | None = None, defaults: SandboxDefaults | None = None, container_image: str | None = None, tags: list[str] | None = None, base_url: str | None = None, request_timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None, max_lifetime_seconds: float | None = None, profile_ids: list[str] | None = None, profile_names: list[str] | None = None, runner_ids: list[str] | None = None, resources: ResourceOptions | dict[str, Any] | None = None, mounted_files: list[dict[str, Any]] | None = None, s3_mount: dict[str, Any] | None = None, ports: list[dict[str, Any]] | None = None, network: NetworkOptions | dict[str, Any] | None = None, max_timeout_seconds: int | None = None, environment_variables: dict[str, str] | None = None, annotations: dict[str, str] | None = None, secrets: Sequence[Secret | dict[str, Any]] | None = None)

​

Properties

​

sandbox_id

@property
def sandbox_id(self) -> str | None

​

returncode

@property
def returncode(self) -> int | None

​

runner_id

@property
def runner_id(self) -> str | None

​

profile_id

@property
def profile_id(self) -> str | None

​

status

@property
def status(self) -> SandboxStatus | None

​

status_updated_at

@property
def status_updated_at(self) -> datetime | None

​

started_at

@property
def started_at(self) -> datetime | None

​

runner_group_id

@property
def runner_group_id(self) -> str | None

​

service_address

@property
def service_address(self) -> str | None

​

exposed_ports

@property
def exposed_ports(self) -> tuple[tuple[int, str], ...] | None

​

applied_ingress_mode

@property
def applied_ingress_mode(self) -> str | None

​

applied_egress_mode

@property
def applied_egress_mode(self) -> str | None

​

resource_limits

@property
def resource_limits(self) -> dict[str, str] | None

​

resource_requests

@property
def resource_requests(self) -> dict[str, str] | None

​

resource_gpu

@property
def resource_gpu(self) -> dict[str, Any] | None

​

exec_stats

@property
def exec_stats(self) -> dict[str, int]

​

Methods

​

run

run(*args: str = (), container_image: str | None = None, defaults: SandboxDefaults | None = None, request_timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None, max_lifetime_seconds: float | None = None, tags: list[str] | None = None, profile_ids: list[str] | None = None, profile_names: list[str] | None = None, runner_ids: list[str] | None = None, resources: ResourceOptions | dict[str, Any] | None = None, mounted_files: list[dict[str, Any]] | None = None, s3_mount: dict[str, Any] | None = None, ports: list[dict[str, Any]] | None = None, network: NetworkOptions | dict[str, Any] | None = None, max_timeout_seconds: int | None = None, environment_variables: dict[str, str] | None = None, annotations: dict[str, str] | None = None, secrets: Sequence[Secret | dict[str, Any]] | None = None) -> Sandbox

• *args (str): Optional command and arguments (e.g., “echo”, “hello”, “world”). If omitted, uses default command from SandboxDefaults.
• container_image (str | None): Container image to use
• defaults (SandboxDefaults | None): Optional SandboxDefaults to apply
• request_timeout_seconds (float | None): Timeout for API requests (client-side)
• poll_retry_budget_seconds (float | None): Wall-clock budget for retrying transient errors on the sandbox-status poll loop (default: 30s). Set to 0 to disable retry.
• poll_rpc_timeout_seconds (float | None): Per-call timeout for poll Get RPCs (default: 15s). Separate from request_timeout_seconds.
• max_lifetime_seconds (float | None): Max sandbox lifetime (server-side)
• tags (list[str] | None): Optional tags for the sandbox
• profile_ids (list[str] | None): Optional list of profile IDs for infrastructure selection. See SandboxDefaults.profile_ids for semantics. Prefer profile_names when selecting by name.
• profile_names (list[str] | None): Optional list of profile names for infrastructure selection (preferred over profile_ids). See SandboxDefaults.profile_names for semantics.
• runner_ids (list[str] | None): Optional list of runner IDs
• resources (ResourceOptions | dict[str, Any] | None): Resource configuration. Accepts ResourceOptions for separate requests/limits, or a flat dict for backward-compatible Guaranteed QoS.
• mounted_files (list[dict[str, Any]] | None): Files to mount into the sandbox
• s3_mount (dict[str, Any] | None): S3 bucket mount configuration
• ports (list[dict[str, Any]] | None): Port mappings for the sandbox
• network (NetworkOptions | dict[str, Any] | None): Network configuration (NetworkOptions dataclass)
• max_timeout_seconds (int | None): Maximum timeout for sandbox operations
• environment_variables (dict[str, str] | None): Environment variables to inject into the sandbox. Merges with and overrides matching keys from the session defaults. Use for non-sensitive config only.
• annotations (dict[str, str] | None): Kubernetes pod annotations for the sandbox. Merges with and overrides matching keys from the session defaults. Use for non-sensitive metadata only.
• secrets (Sequence[Secret | dict[str, Any]] | None): Secrets to inject as environment variables. Merged with defaults (defaults first, then this list).

# Using defaults (shell-trapped keep-alive)
sb = Sandbox.run()

# Fire and forget style
sb = Sandbox.run("echo", "hello")
# sb.sandbox_id is set, but sandbox may still be starting

# Wait for ready if needed
sb = Sandbox.run("sleep", "infinity").wait()
result = sb.exec(["echo", "hello"]).result()

# Or use context manager for automatic cleanup
with Sandbox.run("sleep", "infinity") as sb:
    result = sb.exec(["echo", "hello"]).result()

​

session

session(defaults: SandboxDefaults | Mapping[str, Any] | None = None) -> Session

• Shared configuration via defaults
• Automatic cleanup of orphaned sandboxes
• Function execution via @session.function() decorator Parameters
• defaults (SandboxDefaults | Mapping[str, Any] | None): Optional defaults to apply to sandboxes created via session

• Session: A Session instance

session = Sandbox.session(defaults)
sb = session.create(command="sleep", args=["infinity"])

@session.function()
def compute(x, y):
    return x + y

await session.close()

​

list

list(*, tags: list[str] | None = None, status: str | None = None, profile_ids: list[str] | None = None, profile_names: list[str] | None = None, runner_ids: list[str] | None = None, include_stopped: bool = False, base_url: str | None = None, timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None) -> OperationRef[list[Sandbox]]

• tags (list[str] | None): Filter by tags (sandboxes must have ALL specified tags)
• status (str | None): Filter by status (“running”, “completed”, “failed”, etc.)
• profile_ids (list[str] | None): Optional list of profile IDs for infrastructure selection. See SandboxDefaults.profile_ids for semantics. Prefer profile_names when selecting by name.
• profile_names (list[str] | None): Optional list of profile names for infrastructure selection (preferred over profile_ids). See SandboxDefaults.profile_names for semantics.
• runner_ids (list[str] | None): Filter by runner IDs
• include_stopped (bool): If True, include terminal sandboxes (completed, failed, terminated). Defaults to False.
• base_url (str | None): Override API URL (default: CWSANDBOX_BASE_URL env or default)
• timeout_seconds (float | None): Request timeout (default: 300s)
• poll_retry_budget_seconds (float | None): Wall-clock budget for retrying transient errors on the sandbox-status poll loop (default: 30s). Set to 0 to disable retry. Applied to returned Sandbox instances.
• poll_rpc_timeout_seconds (float | None): Per-call timeout for poll Get RPCs (default: 15s). Separate from timeout_seconds. Applied to returned Sandbox instances.

• OperationRef[list[Sandbox]]: OperationRef[list[Sandbox]]: Use .result() to block for results,
• OperationRef[list[Sandbox]]: or await directly in async contexts.

# Sync usage - active sandboxes only (default)
sandboxes = Sandbox.list(tags=["my-batch-job"]).result()
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")
    sb.stop().result()

# Include stopped sandboxes
all_sandboxes = Sandbox.list(
    tags=["my-batch-job"], include_stopped=True
).result()

# Async usage
sandboxes = await Sandbox.list(status="running")
for sb in sandboxes:
    result = await sb.exec(["echo", "hello"])

​

from_id

from_id(sandbox_id: str, *, base_url: str | None = None, timeout_seconds: float | None = None, poll_retry_budget_seconds: float | None = None, poll_rpc_timeout_seconds: float | None = None) -> OperationRef[Sandbox]

• sandbox_id (str): The ID of the existing sandbox
• base_url (str | None): Override API URL (default: CWSANDBOX_BASE_URL env or default)
• timeout_seconds (float | None): Request timeout (default: 300s)
• poll_retry_budget_seconds (float | None): Wall-clock budget for retrying transient errors on the sandbox-status poll loop (default: 30s). Set to 0 to disable retry. Applied to the returned Sandbox instance.
• poll_rpc_timeout_seconds (float | None): Per-call timeout for poll Get RPCs (default: 15s). Separate from timeout_seconds. Applied to the returned Sandbox instance.

• OperationRef[Sandbox]: OperationRef[Sandbox]: Use .result() to block for the Sandbox instance,
• OperationRef[Sandbox]: or await directly in async contexts.

• SandboxNotFoundError: If sandbox doesn’t exist

# Sync usage
sb = Sandbox.from_id("sandbox-abc123").result()
result = sb.exec(["python", "-c", "print('hello')"]).result()
sb.stop().result()

# Async usage
sb = await Sandbox.from_id("sandbox-abc123")
result = await sb.exec(["python", "-c", "print('hello')"])

​

delete

delete(sandbox_id: str, *, base_url: str | None = None, timeout_seconds: float | None = None, missing_ok: bool = False) -> OperationRef[None]

• sandbox_id (str): The sandbox ID to delete
• base_url (str | None): Override API URL (default: CWSANDBOX_BASE_URL env or default)
• timeout_seconds (float | None): Request timeout (default: 300s)
• missing_ok (bool): If True, suppress SandboxNotFoundError when sandbox doesn’t exist.

• OperationRef[None]: OperationRef[None]: Use .result() to block until complete.
• OperationRef[None]: Raises SandboxNotFoundError if not found (unless missing_ok=True),
• OperationRef[None]: SandboxError if deletion failed.

• SandboxNotFoundError: If sandbox doesn’t exist and missing_ok=False
• SandboxError: If deletion failed for other reasons

# Sync usage
Sandbox.delete("sandbox-abc123").result()

# Ignore if already deleted
Sandbox.delete("sandbox-abc123", missing_ok=True).result()

# Async usage
await Sandbox.delete("sandbox-abc123")

​

get_status

get_status() -> SandboxStatus

• SandboxStatus: SandboxStatus enum value

• SandboxNotRunningError: If sandbox has not been started

sb = Sandbox.run("sleep", "10")
status = sb.get_status()
print(f"Sandbox is {status}")  # SandboxStatus.PENDING or RUNNING

​

start

start() -> OperationRef[None]

• OperationRef[None]: OperationRef[None]: Use .result() to block until backend accepts.

sandbox = Sandbox(command="sleep", args=["infinity"])
sandbox.start().result()
print(f"Started sandbox: {sandbox.sandbox_id}")
sandbox.wait()  # Block until RUNNING

​

wait

wait(timeout: float | None = None) -> Sandbox

• timeout (float | None): Maximum seconds to wait. None means use default timeout.

• Sandbox: Self for method chaining. Check .status to determine final state.

• SandboxFailedError: If sandbox fails to start
• SandboxTerminatedError: If sandbox was terminated externally
• SandboxTimeoutError: If timeout expires

sb = Sandbox.run("sleep", "infinity").wait()
result = sb.exec(["echo", "ready"]).result()

​

wait_until_complete

wait_until_complete(timeout: float | None = None, *, raise_on_termination: bool = True) -> OperationRef[Sandbox]

• timeout (float | None): Maximum seconds to wait. None means use default timeout.
• raise_on_termination (bool): If True (default), raises SandboxTerminatedError when this client called stop() or the backend reports legacy TERMINATED status. External kills (infrastructure, lifetime limits, other clients) that result in COMPLETED are not detectable until the backend provides termination_reason metadata. Set to False to suppress SandboxTerminatedError entirely.

• OperationRef[Sandbox]: OperationRef[Sandbox]: Use .result() to block or await in async contexts.

• SandboxTimeoutError: If timeout expires
• SandboxTerminatedError: If sandbox was stopped by this client or reported as TERMINATED by backend (and raise_on_termination=True)
• SandboxFailedError: If sandbox failed

sb = Sandbox.run("python", "-c", "print('done')")
sb.wait_until_complete().result()
print(f"Exit code: {sb.returncode}")

​

stop

stop(*, snapshot_on_stop: bool = False, graceful_shutdown_seconds: float = DEFAULT_GRACEFUL_SHUTDOWN_SECONDS, missing_ok: bool = False) -> OperationRef[None]

• snapshot_on_stop (bool): If True, capture sandbox state before shutdown.
• graceful_shutdown_seconds (float): Time to wait for graceful shutdown.
• missing_ok (bool): If True, suppress SandboxNotFoundError when sandbox doesn’t exist.

• OperationRef[None]: OperationRef[None]: Use .result() to block until terminal.
• OperationRef[None]: Raises SandboxError on failure, SandboxNotFoundError if not found
• OperationRef[None]: (unless missing_ok=True).

sb.stop().result()  # Block until terminal (COMPLETED/FAILED)

# Ignore if already deleted
sb.stop(missing_ok=True).result()

# wait_until_complete() after stop() resolves when terminal
sb.stop()
sb.wait_until_complete().result()  # Polls through TERMINATING

​

exec

exec(command: Sequence[str], *, cwd: str | None = None, check: bool = False, timeout_seconds: float | None = None, stdin: bool = False) -> Process

• command (Sequence[str]): Command and arguments to execute
• cwd (str | None): Working directory for command execution. Must be an absolute path. When specified, the command is wrapped with a shell cd.
• check (bool): If True, raise SandboxExecutionError on non-zero returncode
• timeout_seconds (float | None): Timeout for command execution (after sandbox is RUNNING). Does not include time waiting for sandbox to reach RUNNING status.
• stdin (bool): If True, enable stdin streaming. Process.stdin will be a StreamWriter that can send input to the command. If False (default), stdin is closed immediately and Process.stdin is None.

• Process: Process handle with streaming stdout/stderr. Call .result() to block
• Process: for the final ProcessResult, or iterate over .stdout/.stderr for
• Process: real-time output. When stdin=True, Process.stdin is a StreamWriter.

• ValueError: If command is empty or cwd is invalid (empty or relative path)

# Get result directly
process = sb.exec(["echo", "hello"])
result = process.result()
print(result.stdout)

# With working directory
result = sb.exec(["ls", "-la"], cwd="/app").result()

# Stream output in real-time
process = sb.exec(["python", "script.py"])
for line in process.stdout:
    print(line)
result = process.result()

# With stdin streaming
process = sb.exec(["cat"], stdin=True)
process.stdin.write(b"hello world").result()
process.stdin.close().result()
result = process.result()

# Async usage
result = await sb.exec(["echo", "hello"])

​

shell

shell(command: Sequence[str] | None = None, *, width: int | None = None, height: int | None = None) -> TerminalSession

• command (Sequence[str] | None): Shell command to execute. Defaults to [“/bin/bash”]. Accepts a sequence like [“/bin/sh”] or [“/usr/bin/python3”].
• width (int | None): Initial terminal width in columns.
• height (int | None): Initial terminal height in rows.

• TerminalSession: TerminalSession handle with .output (StreamReader[bytes]),
• TerminalSession: .stdin (StreamWriter), and .resize(w, h).

• ValueError: If command is explicitly empty.

session = sandbox.shell(width=80, height=24)
session.stdin.writeline("echo hello").result()
for chunk in session.output:
    sys.stdout.buffer.write(chunk)
exit_code = session.wait()

​

read_file

read_file(filepath: str, *, timeout_seconds: float | None = None) -> OperationRef[bytes]

• filepath (str): Path to file in sandbox
• timeout_seconds (float | None): Timeout for the operation

• OperationRef[bytes]: OperationRef[bytes]: Use .result() to block and retrieve contents.

data = sb.read_file("/output/result.txt").result()

​

write_file

write_file(filepath: str, contents: bytes, *, timeout_seconds: float | None = None) -> OperationRef[None]

• filepath (str): Path to file in sandbox
• contents (bytes): File contents as bytes
• timeout_seconds (float | None): Timeout for the operation

• OperationRef[None]: OperationRef[None]: Use .result() to block until complete.

sb.write_file("/input/data.txt", b"content").result()

​

stream_logs

stream_logs(*, follow: bool = False, tail_lines: int | None = None, since_time: datetime | None = None, timestamps: bool = False, timeout_seconds: float | None = None) -> StreamReader[str]

• follow (bool): If True, continuously stream new logs (like tail -f). If False, stream existing logs and stop. Only running sandboxes support follow=True.
• tail_lines (int | None): Number of most recent lines to retrieve. If None, returns all available lines.
• since_time (datetime | None): Only return logs after this timestamp.
• timestamps (bool): If True, prefix each line with an ISO 8601 timestamp from the server.
• timeout_seconds (float | None): Client-side deadline for the gRPC call. Defaults to request_timeout_seconds when follow=False, and None (no timeout) when follow=True.

• StreamReader[str]: StreamReader yielding log lines as strings. Iterate synchronously
• StreamReader[str]: with for line in reader or asynchronously with
• StreamReader[str]: async for line in reader.

• SandboxNotRunningError: If follow=True and the sandbox has been stopped.
• SandboxError: If the log stream encounters an error.

# One-shot: get recent logs
for line in sandbox.stream_logs(tail_lines=100):
    print(line, end="")

# Follow mode: stream continuously
for line in sandbox.stream_logs(follow=True):
    print(line, end="")

# Retrieve logs from a stopped sandbox
sb = Sandbox.from_id("sbx-abc123").result()
for line in sb.stream_logs(tail_lines=50):
    print(line, end="")

# Async usage
async for line in sandbox.stream_logs(follow=True):
    print(line, end="")