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 ImageArtifactType, ImageBuildStatus, ImageClient

page = ImageClient().list(platform=True)
vm_images = [
    image.display_ref
    for image in page.data
    if image.artifact_type == ImageArtifactType.VM_SANDBOX
    and 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 SDK also accepts the old flat bulk response during rollout.

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:

1. Direct parameter: APIClient(api_key="sk-...")
2. Environment variable: export PRIME_API_KEY="sk-..."
3. 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")

get_background_jobs is VM-only. Container sandboxes retain the existing get_background_job polling behavior. 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())

Documentation

Full API reference: https://github.com/PrimeIntellect-ai/prime/tree/main/packages/prime-sandboxes

Related Packages

• prime - Full CLI + SDK with pods, inference, and more (includes this package)

License

MIT License - see LICENSE file for details