Prime Sandboxes SDK

Lightweight Python SDK for managing Prime Intellect sandboxes - secure remote code execution environments.

Features

Synchronous and async clients - Use with sync or async/await code
Full sandbox lifecycle - Create, list, execute commands, upload/download files, delete
Type-safe - Full type hints and Pydantic models
Authentication caching - Automatic token management
Bulk operations - Create and manage multiple sandboxes efficiently
No CLI dependencies - Pure SDK, ~50KB installed

Installation

uv pip install prime-sandboxes

Or with pip:

pip install prime-sandboxes

Quick Start

from prime_sandboxes import APIClient, SandboxClient, CreateSandboxRequest, StartCommand

# Initialize
client = APIClient(api_key="your-api-key")
sandbox_client = SandboxClient(client)

# Create a VM-backed sandbox.
request = CreateSandboxRequest(
    name="my-sandbox",
    docker_image="python:3.11-slim",
    cpu_cores=2,
    memory_gb=4,
)

sandbox = sandbox_client.create(request)
print(f"Created: {sandbox.id}")

# Boot commands use a structured argv contract; no shell is implied.
vm = sandbox_client.create(CreateSandboxRequest(
    name="vm-workload",
    docker_image="user-1/vm-image:latest",
    start_command=StartCommand(
        executable="/worker",
        args=["--platform", "linux/amd64"],
    ),
))


sandbox_client.wait_for_creation(sandbox.id)

# Execute commands
result = sandbox_client.execute_command(sandbox.id, "python --version")
print(result.stdout)

# Clean up
sandbox_client.delete(sandbox.id)

Async Usage

import asyncio
from prime_sandboxes import AsyncSandboxClient, CreateSandboxRequest

async def main():
    async with AsyncSandboxClient(api_key="your-api-key") as client:
        # Create sandbox
        sandbox = await client.create(CreateSandboxRequest(
            name="async-sandbox",
            docker_image="python:3.11-slim",
        ))

        # Wait and execute
        await client.wait_for_creation(sandbox.id)
        result = await client.execute_command(sandbox.id, "echo 'Hello from async!'")
        print(result.stdout)

        # Clean up
        await client.delete(sandbox.id)

asyncio.run(main())

List Platform Images

Use a platform admin or manager key with sandbox-read access to list platform images:

from prime_sandboxes import ImageBuildStatus, ImageClient

page = ImageClient().list(platform=True)
completed_images = [
    image.display_ref
    for image in page.data
    if image.status == ImageBuildStatus.COMPLETED
]

Image Builds

Dockerfile builds create VM artifacts on linux/amd64. The initial response includes upload_url and expires_in; upload the build context before calling start_build.

Source-image requests build VM artifacts directly from allowed public registry images. They do not return upload metadata. A single source returns build_id and build_ids. Comma-separated sources return BulkBuildImageResponse with ordered results: each entry has source_image, build (a BuildImageResponse or None), error, and retryable. There are no success or failed fields.

The server uses mixed wire casing: build_id, upload_url, and expires_in, but buildIds, fullImagePath, and sourceImage. SDK attributes use snake_case. The transfer_image method remains a compatibility name for POST /images/build:

from prime_sandboxes import ImageClient

images = ImageClient()
response = images.transfer_image("ubuntu:22.04")
print(response.build_ids)

All image builds support only linux/amd64. Docker Hub sources become public, org-less platform images automatically. Docker Hub source builds do not accept a custom destination, team, or private visibility. One comma-separated request cannot mix Docker Hub with other registries. Explicit non-Docker-Hub public registries can still use personal or team ownership, custom destinations, and public or private visibility. Allowed registries are Docker Hub, ghcr.io, quay.io, public.ecr.aws, registry.k8s.io, and mcr.microsoft.com. Google-hosted registries are rejected. Docker-Hub-only multi-source requests preserve source names and tags and force PUBLIC platform scope.

Use prime images push --source-image <reference> for one or comma-separated sources, or prime images push-bulk for manifests. Dockerfile platform publishing uses prime images push <name>:<tag> --platform-image; the primary build creates its VM artifact without a second publishing step.

Authentication

The SDK looks for credentials in this order:

Direct parameter: APIClient(api_key="sk-...")
Environment variable: export PRIME_API_KEY="sk-..."
Config file: ~/.prime/config.json (created by prime login CLI command)

Advanced Features

Environment Variables and Secrets

# Create sandbox with environment variables and secrets
request = CreateSandboxRequest(
    name="my-sandbox",
    docker_image="python:3.11-slim",
    environment_vars={
        "DEBUG": "true",
        "LOG_LEVEL": "info"
    },
    secrets={
        "API_KEY": "sk-secret-key-here",
        "DATABASE_PASSWORD": "super-secret-password"
    }
)

sandbox = sandbox_client.create(request)

Note: Secrets are never displayed in logs or outputs. When retrieving sandbox details, only the secret keys are shown with values masked as ***.

File Operations

# Upload a file
sandbox_client.upload_file(
    sandbox_id=sandbox.id,
    file_path="/app/script.py",
    local_file_path="./local_script.py"
)

# Download a file
sandbox_client.download_file(
    sandbox_id=sandbox.id,
    file_path="/app/output.txt",
    local_file_path="./output.txt"
)

Bulk Operations

# Create multiple sandboxes
sandbox_ids = []
for i in range(5):
    sandbox = sandbox_client.create(CreateSandboxRequest(
        name=f"sandbox-{i}",
        docker_image="python:3.11-slim",
    ))
    sandbox_ids.append(sandbox.id)

# Wait for up to 100 sandboxes with one batched lifecycle-status request per poll
statuses = sandbox_client.bulk_wait_for_creation(sandbox_ids)

# Delete by IDs or labels
sandbox_client.bulk_delete(sandbox_ids=sandbox_ids)
# OR by labels
sandbox_client.bulk_delete(labels=["experiment-1"])

Labels & Filtering

# Create with labels
sandbox = sandbox_client.create(CreateSandboxRequest(
    name="labeled-sandbox",
    docker_image="python:3.11-slim",
    labels=["experiment", "ml-training"],
))

# List with filters
sandboxes = sandbox_client.list(
    status="RUNNING",
    labels=["experiment"],
    page=1,
    per_page=50,
)

for s in sandboxes.sandboxes:
    print(f"{s.name}: {s.status}")

Long-Running Tasks

Use start_background_job to run long-running tasks that continue after the API call returns. Poll for completion with get_background_job.

from prime_sandboxes import APIClient, SandboxClient, CreateSandboxRequest

sandbox_client = SandboxClient(APIClient())

# Create sandbox with extended timeout
sandbox = sandbox_client.create(CreateSandboxRequest(
    name="training-job",
    docker_image="python:3.11-slim",
    timeout_minutes=1440,  # 24 hours
    cpu_cores=4,
    memory_gb=16,
))
sandbox_client.wait_for_creation(sandbox.id)

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

# VM sandboxes can check up to 100 SDK-started jobs across sandboxes with one
# platform request. Results preserve input order; completed jobs include the
# same bounded stdout/stderr tails as get_background_job().
statuses = sandbox_client.get_background_jobs([job])

# For latency-sensitive polling, status-only methods never download output.
# Fetch the hydrated result with get_background_job() after completion.
snapshots = sandbox_client.get_background_job_statuses([job])

# Poll for completion
import time
while True:
    status = sandbox_client.get_background_job(sandbox.id, job)
    if status.completed:
        print(f"Job finished with exit code: {status.exit_code}")
        print(status.stdout)
        break
    print("Still running...")
    time.sleep(30)

# Download results
sandbox_client.download_file(sandbox.id, "/app/model.pt", "./model.pt")

Status lookups are batched; gateways without the batch endpoint fall back to per-job polling. Once an exit code is observed, completion remains authoritative even if output retrieval exhausts its bounded retry deadline: the unavailable stream is None and its stdout_error or stderr_error field describes the retrieval failure.

Output downloads are deduplicated, cached within a bounded client-local LRU, and scheduled separately from completion polling. Advanced callers can tune the client-wide limits with background_job_output_concurrency, background_job_output_queue_size, and background_job_output_cache_bytes; the defaults are 20 active jobs, 200 queued jobs, and 64 MiB of cached streams.

Async version

import asyncio
from prime_sandboxes import AsyncSandboxClient, CreateSandboxRequest

async def run_training():
    async with AsyncSandboxClient() as client:
        sandbox = await client.create(CreateSandboxRequest(
            name="async-training",
            docker_image="python:3.11-slim",
            timeout_minutes=720,
        ))
        await client.wait_for_creation(sandbox.id)

        # Start background job
        job = await client.start_background_job(
            sandbox.id,
            "python train.py"
        )

        # Poll until done
        while True:
            status = await client.get_background_job(sandbox.id, job)
            if status.completed:
                print(status.stdout)
                break
            await asyncio.sleep(30)

        await client.delete(sandbox.id)

asyncio.run(run_training())