Sandbox SDK Guide

The SDK ships two clients with identical methods: SandboxClient for synchronous scripts and AsyncSandboxClient for concurrent workloads. Both are importable from prime_sandboxes.

Sync Client

SandboxClient is the simplest way to get started — no async/await needed.

from prime_sandboxes import SandboxClient, CreateSandboxRequest, APIClient

client = SandboxClient(APIClient())

# Create and wait
sandbox = client.create(
    CreateSandboxRequest(
        name="sdk-demo",
        docker_image="python:3.11-slim",
        labels=["experiment"],
        timeout_minutes=120,
        environment_vars={"LOG_LEVEL": "debug"},
        secrets={"API_KEY": "sk-abc123"},
    )
)
client.wait_for_creation(sandbox.id)

# Run a command
result = client.execute_command(sandbox.id, "python -c 'print(42)'")
print(result.stdout.strip())

# Upload / download files
client.upload_file(sandbox.id, "/workspace/data.csv", "./data.csv")
client.download_file(sandbox.id, "/workspace/output.csv", "./output.csv")

# Clean up
client.delete(sandbox.id)

Pass idempotency_key to CreateSandboxRequest to make create retries safe: requests that share the same key are deduplicated, so a network failure mid-create can be re-run without producing a second sandbox. Every method shown in the async sections below has an identical synchronous counterpart on SandboxClient.

Async Client

Most sandbox automations spin up more than one environment. The async client lets you fan out creates, waits, commands, and teardown without juggling threads.

import asyncio
from prime_sandboxes import AsyncSandboxClient, CreateSandboxRequest

async def launch_demo() -> None:
    async with AsyncSandboxClient() as sandboxes:
        request = CreateSandboxRequest(
            name="sdk-demo",
            docker_image="python:3.11-slim",
            labels=["experiment", "ml-pipeline", "team-research"],
            timeout_minutes=120,
        )
        sandbox = await sandboxes.create(request)
        await sandboxes.wait_for_creation(sandbox.id)
        result = await sandboxes.execute_command(sandbox.id, "python -c 'print(42)'")
        print(result.stdout.strip())
        await sandboxes.delete(sandbox.id)

asyncio.run(launch_demo())

Launch a Fleet

async def create_many(images: list[str]) -> None:
    async with AsyncSandboxClient() as sandboxes:
        requests = [
            CreateSandboxRequest(
                name=f"batch-{i}", docker_image=image, idempotency_key=f"batch-{i}"
            )
            for i, image in enumerate(images, start=1)
        ]
        created = await asyncio.gather(*[sandboxes.create(req) for req in requests])
        await sandboxes.bulk_wait_for_creation([sbx.id for sbx in created])
        print("Ready:", ", ".join(sbx.name for sbx in created))

# asyncio.run(create_many(["python:3.11-slim", "node:20-slim"]))

bulk_wait_for_creation polls via the list endpoint, backing off automatically if the API throttles you.

Run Commands

async def smoke_test(sandbox_id: str) -> None:
    async with AsyncSandboxClient() as sandboxes:
        results = await sandboxes.execute_command(
            sandbox_id,
            "python -c 'import platform; print(platform.python_version())'",
        )
        print("stdout:", results.stdout.strip())

Command responses include stdout, stderr, and exit code so you can short-circuit pipelines when something breaks.

Move Data In and Out

async def sync_artifacts(sandbox_id: str) -> None:
    async with AsyncSandboxClient() as sandboxes:
        await sandboxes.upload_file(sandbox_id, "/workspace/model.bin", "./artifacts/model.bin")
        await sandboxes.download_file(sandbox_id, "/workspace/report.csv", "./reports/report.csv")
        contents = await sandboxes.read_file(sandbox_id, "/workspace/report.csv")
        print(contents.content[:80])

Note: File uploads are limited to 16 MiB per file. Reads are also size-limited; use download_file for large files. Uploads/downloads use short-lived gateway tokens stored in a local cache. Call sandboxes.clear_auth_cache() if you rotate credentials or hit 401s.

Start Command

Pass start_command to run your own process when the sandbox boots. There is no shell, so the executable and each argument are separate values:

from prime_sandboxes import StartCommand

sandbox = await sandboxes.create(
    CreateSandboxRequest(
        name="api-server",
        docker_image="python:3.11-slim",
        start_command=StartCommand(executable="python", args=["serve.py", "--port", "8000"]),
    )
)

To get shell features (pipes, variables, &&), invoke a shell explicitly:

start_command=StartCommand(executable="/bin/sh", args=["-c", "python serve.py --port 8000"])

The start command is one-shot: when the process exits, the sandbox stops. Omit it to keep the sandbox up and ready for execute_command calls.

Environment Variables & Secrets

Pass configuration and credentials when creating a sandbox:

sandbox = await sandboxes.create(
    CreateSandboxRequest(
        name="configured-runner",
        docker_image="python:3.11-slim",
        environment_vars={
            "APP_ENV": "staging",
            "LOG_LEVEL": "debug",
        },
        secrets={
            "DB_PASSWORD": "hunter2",
            "API_KEY": "sk-abc123",
        },
    )
)

Environment variables are stored in plain text. Secrets are encrypted at rest and never returned in API responses — use them for API keys, passwords, and other sensitive values. Both are injected into the sandbox as standard environment variables. You can also pass per-command environment variables to execute_command:

result = await sandboxes.execute_command(
    sandbox.id,
    "echo $CUSTOM_VAR",
    env={"CUSTOM_VAR": "hello"},
)

Network Isolation

Sandboxes can restrict outbound traffic at creation time. Use an allowlist or a denylist — the two are mutually exclusive.

async def isolated_sandbox() -> None:
    async with AsyncSandboxClient() as sandboxes:
        # Deny all egress, including DNS
        request = CreateSandboxRequest(
            name="isolated-runner",
            docker_image="python:3.11-slim",
            network_allowlist=[],
        )
        sandbox = await sandboxes.create(request)
        await sandboxes.wait_for_creation(sandbox.id)

        # Outbound calls now fail with a network error
        await sandboxes.execute_command(
            sandbox.id,
            "python -c 'import urllib.request; urllib.request.urlopen(\"https://example.com\")'",
        )

        await sandboxes.delete(sandbox.id)

Allow only specific destinations with network_allowlist=["api.openai.com"], or block specific destinations and allow the rest with network_denylist=["evil.example.com"]. Entries are domains (example.com, *.example.com) or IPv4 addresses/CIDRs; IPv6 is not supported.

Long-Running Commands

execute_command accepts a timeout parameter in seconds (default 300):

async def run_long_command(sandbox_id: str) -> None:
    async with AsyncSandboxClient() as sandboxes:
        result = await sandboxes.execute_command(
            sandbox_id,
            "python preprocessing.py --dataset large",
            timeout=1800,
        )
        print(f"Exit code: {result.exit_code}")

For tasks that should survive the current connection, use background jobs instead. They keep running in the sandbox while you poll for completion.

Background Jobs

Use start_background_job for tasks that run independently of the command connection. The job continues running in the sandbox while you poll for completion.

async def run_training_job() -> None:
    async with AsyncSandboxClient() as sandboxes:
        sandbox = await sandboxes.create(
            CreateSandboxRequest(
                name="training-job",
                docker_image="python:3.11-slim",
                timeout_minutes=1440,  # 24 hours
                cpu_cores=4,
                memory_gb=16,
            )
        )
        await sandboxes.wait_for_creation(sandbox.id)

        # Start a long-running job in the background
        job = await sandboxes.start_background_job(
            sandbox.id, "python train.py --epochs 100"
        )
        print(f"Job started: {job.job_id}")

        # Poll for completion
        while True:
            status = await sandboxes.get_background_job(sandbox.id, job)
            if status.completed:
                print(f"Exit code: {status.exit_code}")
                print(status.stdout)
                break
            print("Still running...")
            await asyncio.sleep(30)

        # Download results
        await sandboxes.download_file(sandbox.id, "/app/model.pt", "./model.pt")
        await sandboxes.delete(sandbox.id)

The timeout_minutes parameter controls how long the sandbox stays alive. For VM sandboxes, any negative value removes the lifetime deadline. Background jobs persist across API calls until completion or sandbox termination.

Idle Timeout

Set idle_timeout_minutes on CreateSandboxRequest to have the sandbox terminate itself when nothing has touched it for a while:

from prime_sandboxes import SandboxClient, CreateSandboxRequest, APIClient

client = SandboxClient(APIClient())

sandbox = client.create(
    CreateSandboxRequest(
        name="idle-aware",
        docker_image="python:3.11-slim",
        timeout_minutes=-1,
        idle_timeout_minutes=15,
    )
)

Any execute_command, upload_file, download_file, or file-read call resets the idle clock. Long-running execs stay pinned for their full duration.

Disabled by default. Omit the field to keep the legacy lifetime-only behavior.
Must be between 1 and 1,440 minutes. For a finite lifetime, it must not exceed timeout_minutes.
SSH sessions do not count as activity.

When a sandbox shuts down for this reason, the response object exposes termination_reason="idle_timeout".

Error Handling

The SDK raises typed exceptions so you can handle specific failure modes. All exceptions are importable directly from prime_sandboxes.

from prime_sandboxes import (
    # Sandbox lifecycle errors
    SandboxNotRunningError,
    SandboxOOMError,
    SandboxTimeoutError,
    SandboxImagePullError,
    # Operation errors
    CommandTimeoutError,
    UploadTimeoutError,
    DownloadTimeoutError,
    # API errors
    APIError,
    UnauthorizedError,
    PaymentRequiredError,
    APITimeoutError,
)

Sandbox Lifecycle Errors

These are raised when a sandbox is no longer in RUNNING state. They form a hierarchy — catch the base class for broad handling, or specific subclasses for targeted recovery.

Exception	Cause	Typical Fix
`SandboxNotRunningError`	Operation attempted on a non-running sandbox (terminated, errored, or timed out)	Check sandbox status before operating on it
`SandboxOOMError`	Sandbox killed due to out-of-memory	Increase `memory_gb` in `CreateSandboxRequest` or optimize memory usage
`SandboxTimeoutError`	Sandbox exceeded its `timeout_minutes` and was terminated	Increase the timeout or split work into smaller tasks
`SandboxImagePullError`	Docker image could not be pulled	Verify the image name, tag, and that you can access the image

SandboxOOMError, SandboxTimeoutError, and SandboxImagePullError are all subclasses of SandboxNotRunningError.

Operation Errors

Raised during specific operations when the sandbox is still running but the operation itself fails.

Exception	Cause	Typical Fix
`CommandTimeoutError`	`execute_command` exceeded its `timeout` parameter	Increase the per-command timeout, or use `start_background_job` for long tasks
`UploadTimeoutError`	File upload timed out	Check file size (16 MiB limit) and network conditions
`DownloadTimeoutError`	File download timed out	Check file size and network conditions
`SandboxFileNotFoundError`	File does not exist at the given path	Check the path before uploading, downloading, or reading
`SandboxFileTooLargeError`	Read exceeded the read size limit	Use `download_file` instead of `read_file` for large files

API Errors

Raised for HTTP-level failures when communicating with the platform API.

Exception	Cause	Typical Fix
`APIError`	Base class for all API errors (non-2xx response, malformed response, network failure)	Inspect the error message for details
`UnauthorizedError`	Invalid or expired API key (HTTP 401)	Check `PRIME_API_KEY` or re-run `prime login`
`PaymentRequiredError`	Insufficient balance (HTTP 402)	Top up your account balance
`APITimeoutError`	API request timed out before receiving a response	Retry the request; check network connectivity

Clean Exit

Delete a single sandbox or use bulk_delete to tear down many at once by IDs or labels:

async def teardown() -> None:
    async with AsyncSandboxClient() as sandboxes:
        # Delete one
        await sandboxes.delete("sbx_123")

        # Bulk delete by IDs
        result = await sandboxes.bulk_delete(sandbox_ids=["sbx_456", "sbx_789"])
        print(f"Deleted: {result.succeeded}, Failed: {result.failed}")

        # Bulk delete by labels — removes all sandboxes matching ALL given labels
        result = await sandboxes.bulk_delete(labels=["experiment", "staging"])
        print(f"Deleted: {result.succeeded}, Failed: {result.failed}")

You must pass either sandbox_ids or labels, not both. For a full script, see prime/examples/sandbox_async_demo.py, which covers create → wait → run → delete.

Getting Started

Lab

Libraries

Compute

Sync Client

Async Client

Launch a Fleet

Run Commands

Move Data In and Out

Start Command

Environment Variables & Secrets

Network Isolation

Long-Running Commands

Background Jobs

Idle Timeout

Error Handling

Sandbox Lifecycle Errors

Operation Errors

API Errors

Clean Exit

​Sync Client

​Async Client

​Launch a Fleet

​Run Commands

​Move Data In and Out

​Start Command

​Environment Variables & Secrets

​Network Isolation

​Long-Running Commands

​Background Jobs

​Idle Timeout

​Error Handling

​Sandbox Lifecycle Errors

​Operation Errors

​API Errors

​Clean Exit

Sync Client

Async Client

Launch a Fleet

Run Commands

Move Data In and Out

Start Command

Environment Variables & Secrets

Network Isolation

Long-Running Commands

Background Jobs

Idle Timeout

Error Handling

Sandbox Lifecycle Errors

Operation Errors

API Errors

Clean Exit