Python SDK Reference

from boltz_lab import BoltzPlatformClient

# Using environment variables / config file (recommended)
client = BoltzPlatformClient()

# With explicit API key
client = BoltzPlatformClient(api_key="sk-...")

# With custom timeout
client = BoltzPlatformClient(timeout=60.0)  # seconds

async with BoltzPlatformClient() as client:
    status = await client.get_prediction_status("…")

job = await client.submit_job_from_yaml(
    "path/to/job.yaml",
    prediction_name="My Prediction",
)

job_spec = {
    "sequences": [
        {
            "protein": {
                "id": ["A"],
                "sequence": "MVTPEGNVS...",
                "modifications": []
            }
        },
        {
            "ligand": {
                "id": ["B"],
                "smiles": "N[C@@H](Cc1ccc(O)cc1)C(=O)O"
            }
        }
    ],
    "constraints": [],
    "properties": [{"affinity": {"binder": "B"}}],
}

job = await client.submit_job_from_dict(job_spec, prediction_name="Dict Prediction")

job = await client.submit_prediction(
    complex_data={"sequences": [...], "constraints": []},
    prediction_name="Raw Submit",
)

job = await client.submit_job_from_yaml(
    "affinity.yaml",
    flags={"recycling_steps": 5, "diffusion_samples": 2},
)

status = await client.get_prediction_status(prediction_id)
print(status.prediction_status)
print(status.prediction_stage_description)

from boltz_lab.models import JobStatus

print([s.value for s in JobStatus])
# ['PENDING', 'CREATED', 'RUNNING', 'COMPLETED', 'FAILED', 'CANCELLED', 'TIMED_OUT']

from boltz_lab.models import JobStatus

# List all predictions (default limit=20, offset=0)
resp = await client.list_predictions()

# Filter by status
completed = await client.list_predictions(status=JobStatus.COMPLETED)

# Pagination
page1 = await client.list_predictions(limit=10, offset=0)
page2 = await client.list_predictions(limit=10, offset=10)

url = await client.get_prediction_output_url(prediction_id)

# Download as tar.gz archive (default)
path = await client.download_results(prediction_id)

# Download into a directory
path = await client.download_results(prediction_id, output_dir="results/")

# Download as JSON (writes prediction_results to a .json file)
json_path = await client.download_results(
    prediction_id,
    output_dir="results/",
    output_format="json",
)

# Custom filename (without extension)
custom_path = await client.download_results(
    prediction_id,
    output_dir="results/",
    output_filename="my_results",
)

def on_progress(s):
    print(s.prediction_status, s.prediction_stage_description)

final_status = await client.wait_for_prediction(
    prediction_id,
    polling_interval=5,
    timeout=600,
    progress_callback=on_progress,
)

status = await job.get_status()

await job.wait_for_completion(
    polling_interval=10,
    timeout=600,
    progress_callback=lambda s: print(s.prediction_status),
)

path = await job.download_results(
    output_dir="results/",
    output_format="archive",
)

from boltz_lab import BoltzAPIError, BoltzAuthenticationError, BoltzNotFoundError, BoltzTimeoutError
from boltz_lab.exceptions import BoltzValidationError, BoltzConnectionError

try:
    job = await client.submit_job_from_yaml("job.yaml")
    await job.wait_for_completion(timeout=300)
except BoltzAuthenticationError:
    print("Invalid API key")
except BoltzValidationError as e:
    print(f"Invalid job specification: {e}")
except BoltzTimeoutError:
    print("Job timed out")
except BoltzAPIError as e:
    print(f"API error: {e}")
    if e.response_data:
        print(f"Details: {e.response_data}")

import asyncio
from pathlib import Path
from boltz_lab import BoltzPlatformClient, BoltzAPIError

async def run_prediction():
    try:
        async with BoltzPlatformClient() as client:
            # Submit job
            print("Submitting job...")
            job = await client.submit_job_from_yaml(
                "affinity.yaml",
                prediction_name="Tyrosine Affinity Prediction",
            )
            print(f"Job ID: {job.prediction_id}")

            # Wait with progress updates
            print("Waiting for completion...")
            def progress(status):
                print(f"  → {status.prediction_stage_description}")

            await job.wait_for_completion(
                polling_interval=5,
                timeout=600,
                progress_callback=progress
            )

            # Download results
            print("Downloading results...")
            output_dir = Path("results")
            output_dir.mkdir(exist_ok=True)

            path = await job.download_results(str(output_dir))
            print(f"✓ Results saved to: {path}")

    except BoltzAPIError as e:
        print(f"Error: {e}")
        if e.response_data:
            print(f"Details: {e.response_data}")

if __name__ == "__main__":
    asyncio.run(run_prediction())