chamber-sdk 0.1.1.0

Python SDK for submitting and managing GPU workloads on Chamber

Chamber Python SDK

A Python SDK for submitting and managing GPU workloads on Chamber.

Installation

pip install chamber-sdk

For the auto-containerization feature (client.run()), install with the run extra:

pip install chamber-sdk[run]

Or install from source:

cd chamber-python-sdk
pip install -e ".[run]"

Quick Start

from chamber_sdk import ChamberClient, JobClass, JobStatus

# Initialize client with API token
client = ChamberClient(token="your-api-token")

# Or use token from environment variable (CHAMBER_TOKEN)
# or from Chamber CLI config (~/.chamber/token.json)
client = ChamberClient.from_config()

# Submit a GPU workload
job = client.submit_job(
    name="llm-training-v1",
    initiative_id="your-initiative-id",
    gpu_type="H100",
    requested_gpus=8,
    job_class=JobClass.RESERVED,
    priority=75,
    tags={"experiment": "v1", "owner": "ml-team"}
)
print(f"Submitted job: {job.id}")

# Check job status
job = client.get_workload(job.id)
print(f"Status: {job.status}")

# Wait for completion
result = client.wait_for_completion(job.id, poll_interval=30)
print(f"Final status: {result.status}")

Auto-Containerize & Run (One-Liner Submissions)

The client.run() method enables scientists to submit GPU workloads without expertise in Docker or Kubernetes. It automatically:

• Detects your project framework (PyTorch, TensorFlow, JAX)
• Generates optimized Dockerfiles with NVIDIA NGC base images
• Builds and pushes container images (with AWS ECR and Google Artifact Registry auto-authentication)
• Creates Kubernetes manifests (Job or RayJob for distributed training)
• Submits the workload to Chamber

Basic Usage

from chamber_sdk import ChamberClient

client = ChamberClient.from_config()

# One-liner to auto-containerize and submit
job = client.run(
    "./my-training-project",
    gpus=4,
    gpu_type="H100",
    team="ml-research",
    registry="123456.dkr.ecr.us-east-1.amazonaws.com",
)
print(f"Submitted: {job.id}")

Using a Configuration File

Create a .chamber.yaml in your project directory:

name: my-training-job
gpus: 4
gpu_type: H100
team: my-team-id
registry: 123456.dkr.ecr.us-east-1.amazonaws.com
entrypoint: train.py
entrypoint_args: --batch-size 32 --epochs 10
distributed: auto   # or "ray", "deepspeed", "none"
job_class: ELASTIC  # or "RESERVED"

# Environment variables
env:
  CUDA_VISIBLE_DEVICES: "0,1,2,3"
forward_env:         # Forward from your local environment
  - WANDB_API_KEY
  - HF_TOKEN

# Additional packages
extra_pip_packages:
  - wandb
  - tensorboard
extra_apt_packages:
  - ffmpeg

# Resource overrides
cpu: "24"
memory: 200Gi
shm_size: 64Gi

# Custom build commands
pre_build_commands:
  - pip install flash-attn --no-build-isolation
post_build_commands:
  - python -c "import torch; print(torch.cuda.is_available())"

# Files to exclude from container
ignore:
  - "*.ckpt"
  - "wandb/"
  - "outputs/"

Then submit with minimal arguments:

job = client.run("./my-project")  # Uses .chamber.yaml settings

Dry Run (Preview Without Executing)

Preview what would happen without actually building or submitting:

result = client.run("./my-project", dry_run=True)

print("=== Detected Profile ===")
print(f"Framework: {result.profile.framework}")
print(f"Entrypoint: {result.profile.entrypoint}")

print("\n=== Generated Dockerfile ===")
print(result.dockerfile)

print("\n=== K8s Manifest ===")
print(result.manifest)

print("\n=== Submit Payload ===")
print(result.submit_payload)

Save Generated Files

Write the generated Dockerfile and manifest to your project for inspection:

job = client.run(
    "./my-project",
    save_dockerfile=True,  # Writes Dockerfile.chamber
    save_manifest=True,    # Writes manifest.chamber.yaml
)

AWS ECR Auto-Authentication

When using AWS ECR, the SDK automatically:

1. Detects ECR registry URLs
2. Authenticates using your AWS CLI credentials
3. Creates the repository if it doesn't exist

# ECR authentication is handled automatically!
job = client.run(
    "./my-project",
    registry="123456789012.dkr.ecr.us-east-1.amazonaws.com",
    team="ml-team",
)

Requirements:

• AWS CLI installed and configured (aws configure)
• Permissions: ecr:GetAuthorizationToken, ecr:CreateRepository, ecr:BatchCheckLayerAvailability, etc.

Google Artifact Registry (Google Artifact Registry) Auto-Authentication

When using Google Artifact Registry (Google Artifact Registry), the SDK automatically:

1. Detects Google Artifact Registry registry URLs (pattern: {LOCATION}-docker.pkg.dev/{PROJECT}/{REPOSITORY})
2. Authenticates using your gcloud CLI credentials
3. Creates the repository if it doesn't exist

# Google Artifact Registry authentication is handled automatically!
job = client.run(
    "./my-project",
    registry="us-central1-docker.pkg.dev/my-gcp-project/ml-images",
    team="ml-team",
)

Requirements:

• gcloud CLI installed and authenticated (gcloud auth login)
• Permissions: artifactregistry.repositories.create, artifactregistry.repositories.get, artifactregistry.repositories.uploadArtifacts

Distributed Training

The SDK auto-detects distributed training frameworks:

# DeepSpeed (auto-detected from requirements.txt)
job = client.run(
    "./deepspeed-project",
    gpus=8,
    distributed="deepspeed",  # or "auto" to auto-detect
)

# Ray (creates RayJob manifest for large-scale training)
job = client.run(
    "./ray-project",
    gpus=32,
    distributed="ray",
)

# Accelerate
job = client.run(
    "./accelerate-project",
    gpus=4,
    distributed="auto",  # Detects accelerate from requirements
)

Custom Base Image

Override the auto-selected NGC base image:

job = client.run(
    "./my-project",
    base_image="nvcr.io/nvidia/pytorch:24.01-py3",
)

Using an Existing Dockerfile

Skip Dockerfile generation and use your own:

job = client.run(
    "./my-project",
    dockerfile="./Dockerfile.custom",
)

Wait for Completion

Block until the job finishes:

job = client.run(
    "./my-project",
    wait=True,
    poll_interval=30,  # Check every 30 seconds
    timeout=7200,      # Max 2 hours
)
print(f"Final status: {job.status}")

Progress Callbacks

Monitor progress during build and submission:

def on_progress(stage, message):
    print(f"[{stage}] {message}")

job = client.run(
    "./my-project",
    on_progress=on_progress,
)

Output:

[config] Resolving configuration...
[detect] Detecting project...
[detect] Framework: pytorch
[detect] Entrypoint: train.py
[dockerfile] Generating Dockerfile...
[manifest] Generating K8s manifest...
[docker] Authenticating to ECR (us-east-1)...
[docker] ECR authentication successful
[docker] Building image: 123456.dkr.ecr.us-east-1.amazonaws.com/my-project:a1b2c3d4e5f6
[docker] Build complete
[docker] Pushing image: 123456.dkr.ecr.us-east-1.amazonaws.com/my-project:a1b2c3d4e5f6
[docker] Push complete
[submit] Submitting workload...
[submit] Workload submitted: wl_abc123

All run() Parameters

Parameter	Type	Default	Description
directory	str	required	Path to project directory
gpus	int	1	Number of GPUs
gpu_type	str	"H100"	GPU type (H100, A100, L40S, etc.)
team	str	None	Team/initiative ID
name	str	directory name	Job name
entrypoint	str	auto-detect	Python entry point (train.py, main.py, etc.)
entrypoint_args	str	None	CLI arguments for entrypoint
registry	str	None	Container registry URL
base_image	str	auto-select	Override base Docker image
dockerfile	str	None	Path to existing Dockerfile
distributed	str	"auto"	"auto", "ray", "deepspeed", or "none"
job_class	str	"ELASTIC"	"RESERVED" or "ELASTIC"
env	dict	None	Environment variables
no_cache	bool	False	Force rebuild even if image exists
dry_run	bool	False	Preview without executing
save_dockerfile	bool	False	Save generated Dockerfile
save_manifest	bool	False	Save generated K8s manifest
wait	bool	False	Wait for job completion
poll_interval	float	10.0	Seconds between status checks
timeout	float	None	Max wait time in seconds
on_progress	callable	None	Progress callback(stage, message)

Framework Detection

The SDK detects ML frameworks from your requirements.txt:

Framework	Detected Packages	Base Image
PyTorch	torch, pytorch	nvcr.io/nvidia/pytorch:24.04-py3
TensorFlow	tensorflow, keras	nvcr.io/nvidia/tensorflow:24.04-tf2-py3
JAX	jax, jaxlib	nvcr.io/nvidia/jax:24.04-py3
Generic	(fallback)	nvcr.io/nvidia/cuda:12.4.1-devel-ubuntu22.04

Distributed Framework Detection

Framework	Detected Package	Launch Command
DeepSpeed	deepspeed	deepspeed --num_gpus N train.py
Accelerate	accelerate	accelerate launch train.py
Ray	ray	Creates RayJob K8s manifest
Horovod	horovod	(detected, uses default launcher)

API Endpoint

The SDK connects to https://api.usechamber.io/v1 by default. You can override this:

client = ChamberClient(
    token="ch.your-token",
    api_url="https://custom.api.example.com/v1"
)

Authentication

The SDK supports multiple authentication methods:

1. Direct Token

client = ChamberClient(token="ch.your-api-token")

2. Environment Variable

export CHAMBER_TOKEN="ch.your-api-token"

client = ChamberClient()  # Token loaded automatically

3. Chamber CLI Config

If you've logged in via the Chamber CLI (chamber login), the SDK will use the stored token:

client = ChamberClient.from_config()

Submitting Jobs

Basic Job Submission

job = client.submit_job(
    name="training-run",
    initiative_id="team-ml-research",
    gpu_type="A100",
    requested_gpus=4,
)

Reserved vs Elastic Jobs

# Reserved: Guaranteed capacity, non-preemptible
reserved_job = client.submit_job(
    name="critical-training",
    initiative_id="team-id",
    gpu_type="H100",
    requested_gpus=8,
    job_class=JobClass.RESERVED,
    priority=90,
)

# Elastic: Uses idle capacity, can be preempted
elastic_job = client.submit_job(
    name="experiment-run",
    initiative_id="team-id",
    gpu_type="A100",
    requested_gpus=4,
    job_class=JobClass.ELASTIC,
    priority=50,
)

Using Templates

# List available templates
templates = client.list_templates(scope="ORGANIZATION")
for t in templates:
    print(f"{t.name}: {t.gpu_type}")

# Submit using a template
job = client.submit_job(
    name="templated-job",
    initiative_id="team-id",
    gpu_type="H100",
    template_id="template-abc123",
)

Distributed Training

from chamber_sdk import ScalingMode

# Gang scheduling (all-or-nothing)
job = client.submit_job(
    name="distributed-training",
    initiative_id="team-id",
    gpu_type="H100",
    requested_gpus=32,
    gpus_per_pod=8,
    requested_pods=4,
    scaling_mode=ScalingMode.GANG,
)

# Elastic scaling
job = client.submit_job(
    name="elastic-training",
    initiative_id="team-id",
    gpu_type="H100",
    requested_gpus=64,
    gpus_per_pod=8,
    scaling_mode=ScalingMode.ELASTIC,
    min_pods=4,  # Scale between 4-8 pods
)

With Kubernetes Manifest

manifest = """
apiVersion: batch/v1
kind: Job
metadata:
  name: training-job
spec:
  template:
    spec:
      containers:
      - name: trainer
        image: your-image:latest
        resources:
          limits:
            nvidia.com/gpu: 8
"""

job = client.submit_job(
    name="k8s-training",
    initiative_id="team-id",
    gpu_type="H100",
    requested_gpus=8,
    k8s_manifest=manifest,
)

With Tags

job = client.submit_job(
    name="experiment-v2",
    initiative_id="team-id",
    gpu_type="A100",
    requested_gpus=4,
    external_id="exp-2024-001",  # Your tracking ID
    tags={
        "experiment": "transformer-v2",
        "dataset": "openwebtext",
        "owner": "ml-team",
    },
)

Listing and Filtering Workloads

List Running Jobs

response = client.list_workloads(status=JobStatus.RUNNING)
for job in response.items:
    print(f"{job.name}: {job.requested_gpus} GPUs on {job.gpu_type}")

Filter by Multiple Statuses

response = client.list_workloads(
    status=[JobStatus.RUNNING, JobStatus.QUEUED, JobStatus.PENDING]
)

Filter by Initiative

response = client.list_workloads(
    initiative_id="team-ml-research",
    status=JobStatus.RUNNING,
)

Iterate All Results (Auto-Pagination)

for job in client.iter_workloads(status=JobStatus.COMPLETED):
    print(f"{job.name} completed at {job.completed_at}")

Advanced Search

results = client.search_workloads(
    status=["RUNNING", "PENDING"],
    gpu_type=["H100"],
    priority_min=50,
    submitted_from="2024-01-01T00:00:00Z",
    query="training",
    sort_by="submitted_at",
    sort_order="desc",
    page_size=25
)

print(f"Found {results.total_count} workloads")
for job in results.items:
    print(f"{job.name}: {job.status}")

Aggregations

agg = client.get_workload_aggregations(
    dimension="status",
    initiative_id=["team-ml"]
)

print(f"Total: {agg.total}")
for bucket in agg.buckets:
    print(f"  {bucket.key}: {bucket.count}")

Managing Workloads

Cancel a Workload

cancelled = client.cancel_workload("job-id")
print(f"Status: {cancelled.status}")  # CANCELLED

Get Job Details

job = client.get_workload("job-id")
print(f"Status: {job.status}")
print(f"GPUs: {job.requested_gpus} x {job.gpu_type}")
print(f"Submitted: {job.submitted_at}")
if job.started_at:
    print(f"Started: {job.started_at}")

Get GPU Metrics

metrics = client.get_workload_metrics(
    "job-id",
    time_range="job_lifetime",
)
if metrics.gpu_utilization:
    print(f"GPU Utilization: {metrics.gpu_utilization.avg:.1f}%")
if metrics.memory_utilization:
    print(f"Memory: {metrics.memory_utilization.avg:.1f}%")

Get Global Metrics

metrics = client.get_global_metrics(
    time_range="last_24h",
    initiative_id="team-ml"
)
print(f"Active workloads: {metrics.active_workloads}")
print(f"Total GPU hours: {metrics.total_gpu_hours}")

Get Statistics

stats = client.get_workload_stats(
    time_range="last_7_days",
    initiative_id="team-id",
)
print(f"Total jobs: {stats.total}")
print(f"By status: {stats.by_status}")

Wait for Completion

# Block until job finishes
result = client.wait_for_completion(
    job.id,
    poll_interval=30,  # Check every 30 seconds
    timeout=3600,      # Max 1 hour
)

if result.status == JobStatus.COMPLETED:
    print("Job succeeded!")
elif result.status == JobStatus.FAILED:
    print(f"Job failed: {result.failure_reason}")

Teams

# List teams
teams = client.list_teams()
for team in teams:
    print(f"{team.name} ({team.id})")

# Create a team
team = client.create_team(
    name="ML Research",
    description="Machine learning research team",
    tags={"department": "engineering"}
)

# Get team details
team = client.get_team("team-id")

Templates

# List templates
templates = client.list_templates(
    scope="ORGANIZATION",
    include_system=True
)

# Get template details
template = client.get_template("template-id")
print(f"GPU Type: {template.gpu_type}")
print(f"GPUs: {template.requested_gpus}")

Allocations

# List allocations for a team
allocations = client.list_allocations(initiative_id="team-id")

# Create an allocation
allocation = client.create_allocation(
    initiative_id="team-id",
    reservation_id="reservation-123",
    allocated_instances=4
)

# Get allocation details
allocation = client.get_allocation("allocation-id")

Capacity Management

Check Available Capacity

capacity = client.get_capacity(initiative_id="team-id")

# Budget info
print(f"Allocated: {capacity.budget.allocated} GPU-hours")
print(f"Used: {capacity.budget.used} GPU-hours")
print(f"Available: {capacity.budget.available} GPU-hours")

# Pool details
for pool in capacity.pools:
    print(f"{pool.name}: {pool.available_gpus}/{pool.total_gpus} {pool.gpu_type} available")

Error Handling

from chamber_sdk import (
    ChamberClient,
    ChamberError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    DockerError,  # For run() errors
)

client = ChamberClient(token="your-token")

try:
    job = client.get_workload("invalid-id")
except NotFoundError:
    print("Job not found")
except AuthenticationError:
    print("Invalid or expired token")
except AuthorizationError:
    print("You don't have permission to view this job")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except ChamberError as e:
    print(f"API error ({e.status_code}): {e.message}")

Handling run() Errors

from chamber_sdk import ChamberClient, DockerError

client = ChamberClient.from_config()

try:
    job = client.run("./my-project", registry="123456.dkr.ecr.us-east-1.amazonaws.com")
except DockerError as e:
    # Docker not installed, daemon not running, build failed, push failed, ECR auth failed
    print(f"Docker error: {e}")
except FileNotFoundError as e:
    # No entrypoint found in project
    print(f"Project error: {e}")
except ValueError as e:
    # Missing required parameters (registry, team)
    print(f"Configuration error: {e}")

Multi-Organization Support

# Specify organization for multi-org users
client = ChamberClient(
    token="your-token",
    organization_id="org-123",
)

# Or per-request via from_config
client = ChamberClient.from_config(organization_id="org-456")

API Reference

ChamberClient

Method	Description
run(directory, ...)	Auto-containerize and submit - one-liner for scientists
submit_job(...)	Submit a new GPU workload
get_workload(id)	Get workload details
list_workloads(...)	List workloads with filters
iter_workloads(...)	Iterate all workloads (auto-pagination)
search_workloads(...)	Advanced workload search
get_workload_aggregations(...)	Get workload aggregations
cancel_workload(id)	Cancel a workload
get_workload_stats(...)	Get aggregated statistics
get_workload_metrics(id, ...)	Get GPU metrics for a workload
get_global_metrics(...)	Get organization-wide metrics
get_batch_workload_metrics(...)	Get metrics for multiple workloads
list_teams()	List accessible teams
create_team(...)	Create a new team
get_team(id)	Get team details
list_templates(...)	List workload templates
get_template(id)	Get template details
list_allocations(...)	List capacity allocations
create_allocation(...)	Create a new allocation
get_allocation(id)	Get allocation details
get_capacity(...)	Get capacity and budget info
wait_for_completion(id, ...)	Wait for job to finish
health()	Check API health

Job Statuses

Status	Description
PENDING	Submitted, awaiting scheduling
QUEUED	Scheduled, waiting for resources
STARTING	Resources allocated, job starting
RUNNING	Job is running
COMPLETED	Job finished successfully
FAILED	Job failed
PREEMPTED	Job was preempted (elastic only)
CANCELLED	Job was cancelled

Job Classes

Class	Description
RESERVED	Uses reserved capacity, non-preemptible
ELASTIC	Uses idle capacity, can be preempted
DISCOVERED	External workload discovered by Chamber

License

MIT