Qualia Python SDK

The official Python SDK for the Qualia robotics data and VLA platform.

Installation

pip install qualia-sdk

The native transfer engine behind client.data.upload / download is bundled in the wheel — no extra step.

Hugging Face imports need no client extra: the SDK submits a pinned revision and polls while the dataset streams server-side straight into storage.

Upgrading to 0.6.8

Hugging Face imports now run server-side. DataResource.preflight_huggingface and the max_files, max_bytes, work_dir, and quiet arguments to import_huggingface were client-side download controls and have been removed; the huggingface installation extra was retired with them. The service now applies its own bounded admission and buffering limits. Install plain qualia-sdk>=0.6.8, pass an exact repository, 40-hex revision, dataset name, and source_format, and use idempotency_key, poll_interval, or wait_timeout when recovery or polling control is needed. The terminal result contains dataset_id and dataset_version_id; it no longer exposes the former client-published git_oid or snapshot_digest fields.

CLI users should upgrade to qualia-cli>=0.1.3, which requires qualia-sdk>=0.6.8 and uses this server-side import contract.

Quick Start

from qualia import Qualia

# Initialize the client
client = Qualia(api_key="your-api-key")

# Or use the QUALIA_API_KEY environment variable
client = Qualia()

# List available VLA models
models = client.models.list()
for model in models:
    print(f"{model.id}: {model.name}")
    print(f"  Camera slots: {model.camera_slots}")

# Create a project
project = client.projects.create(name="My Robot Project")
print(f"Created project: {project.project_id}")

# Get the exact cameras for an immutable dataset version
version_id = "01234567-89ab-4cde-8fab-0123456789ab"
features = client.data.get_training_features(version_id)
print([camera.key for camera in features.camera_features])

# Start a finetune job
job = client.finetune.create(
    project_id=project.project_id,
    model_id="lerobot/smolvla_base",
    vla_type="smolvla",
    dataset_version_id=version_id,
    hours=2.0,
    camera_mappings={"cam_1": "observation.images.top"},
)
print(f"Started job: {job.job_id}")

# Check job status
status = client.finetune.get(job.job_id)
print(f"Status: {status.status}")
print(f"Current phase: {status.current_phase}")

# Cancel a job if needed
result = client.finetune.cancel(job.job_id)

Resources

Credits

# Get your credit balance
balance = client.credits.get()
print(f"Available credits: {balance.balance}")

Data

Publish and download robotics datasets.

# Stage and atomically publish a local dataset directory.
# source_format is one of: "lerobot_v2_1", "lerobot_v3" (default), "mcap_ros2".
result = client.data.upload("/data/gongsta", name="gongsta")
print(result["dataset_id"], result["episode_count"], result["total_frames"])

# Upload protocol v5 starts at qualia-sdk 0.6.9. The 0.6.8 wheel speaks v4;
# current deployments refuse it at the initial `begin` request, before any data
# transfer, with this same actionable floor:
#   pip install 'qualia-sdk>=0.6.9'
# V5 stages bytes directly to object storage, then submits the complete tree to
# one durable publication operation. Canonical bytes deduplicate across later
# attempts; interrupted transfers can also reuse locally remembered staging.

# LeRobot 2.1 remains a distinct source format; it is not aliased to v3.
client.data.upload(
    "/data/lerobot-2.1", name="legacy-recording", source_format="lerobot_v2_1"
)

# Upload a recorded robot MCAP session (one .mcap per episode directory;
# episode.json is optional metadata).
client.data.upload(
    "/data/session-2026-07-20", name="dk1-kitchen", source_format="mcap_ros2"
)

# Import one exact Hugging Face dataset commit. The SDK submits the pinned
# coordinates and polls while Hub bytes stream server-side into storage.
revision = "17c29c487d886878ddcb3c5649101b2860fbc047"
imported = client.data.import_huggingface(
    "qualia-robotics/openarm-cube-in-box-v1",
    revision,
    "openarm-cube-in-box",
    source_format="lerobot_v3",
)
print(imported["dataset_id"], imported["dataset_version_id"])

# Download a dataset (returns the written file paths)
paths = client.data.download("gongsta", dest="/data/gongsta-copy")

# List your datasets
page = client.data.list()
for d in page.items:
    print(d.artifact_id, d.path)

# Watch the account's in-flight uploads from any machine
for op in client.data.uploads():
    print(op.name, op.phase, op.progress_line or op.message)

# Get the immutable, exact camera features for mapping
features = client.data.get_training_features("01234567-89ab-4cde-8fab-0123456789ab")

# Explore one bounded page across the whole account lake. The page, totals, and
# bounded self-excluding facets share one database-statement MVCC snapshot.
from qualia import LakeExploreQuery, LakeNumericPredicate

lake = client.data.explore(
    LakeExploreQuery(
        tasks=["fold shirt"],
        tiers=["retrainable"],
        numeric_predicates=[LakeNumericPredicate(field="jerk_mean", op="lt", value=0.4)],
    )
)
for episode in lake.episodes:
    print(
        episode.dataset_name,
        episode.episode_index,
        episode.metrics,
        episode.channel_predicate_evidence,
    )

# Run native health analysis over one immutable DatasetVersion. The complete
# membership identity is re-proven; exact trajectory bytes are read for the
# declared deterministic sample (at most 120 members). This is not a byte scrub
# of unsampled artifacts or video, and model-based evidence is marked unavailable.
report = client.data.health_for_version("01234567-89ab-4cde-8fab-0123456789ab")
print(
    report.health_score,
    report.sampling.sampled_member_count,
    report.member_count,
    report.evidence_completeness.native_trajectory_metrics.status,
)
for flagged in report.flagged_episodes:
    print(flagged.episode_revision_id, flagged.reasons)

# Explain bad-data recommendations without mutating the source revision.
evidence = client.data.suggest_curation(["01234567-89ab-4cde-8fab-0123456789ab"])
for suggestion in evidence.suggestions:
    print(suggestion.episode_index, suggestion.risk_score, suggestion.reasons)

# Atomically publish the complete recommendation set as a new immutable
# DatasetVersion. Source bytes/history remain available. Preserve this UUID
# when retrying so a lost response recovers the same publication.
from uuid import uuid4

receipt = client.data.apply_curation(
    evidence,
    idempotency_key=uuid4(),
    definition_id="fedcba98-7654-4321-8fed-cba987654321",
    note="Reviewed automated bad-data evidence",
)
print(receipt.version_id, receipt.git_commit_id)

Resuming an interrupted upload

Run the same upload command again with the same dataset name. On deployments supporting staging reuse, the SDK remembers opaque staging handles before each object PUT. A retry asks the server to check the stored object's tenant, ObjectID claim, encoded length and envelope SHA-256, then copy it inside object storage into a fresh staging handle. Those bytes do not cross your uplink again. The copy remains unverified staging: normal worker verification and atomic publication still run. This does not skip verification (QUA-1704).

Hints are scoped to the gateway and authenticated repository and live under $XDG_CACHE_HOME/qualia/upload-staging-v1 (default ~/.cache/qualia/...). They contain no API keys, presigned URLs or dataset payloads. The cache is capped at 65,536 handles / 32 MiB and expires hints after seven days. Override its location with QUALIA_UPLOAD_STAGING_CACHE_DIR; set that variable to an empty string to disable it. A cache inside the upload source is automatically disabled.

This resumes complete objects, not partial PUTs, and requires the same local cache. Missing/deleted staging, unavailable caches and older deployments fall back to uploading the affected objects. Uploads made before this SDK recorded hints cannot recover their unpromoted staging this way.

Finetune

# Create a VLA finetune job (default job_type="vla")
job = client.finetune.create(
    project_id="...",
    model_id="lerobot/smolvla_base",  # Server-pinned built-in alias
    vla_type="smolvla",  # smolvla, pi05, act, sarm
    dataset_version_id="01234567-89ab-4cde-8fab-0123456789ab",
    hours=2.0,  # Training duration (max 168)
    camera_mappings={  # Map model slots to dataset keys
        "cam_1": "observation.images.top",
    },
    # Optional parameters:
    instance_type="gpu_1x_a100",  # From client.instances.list()
    region="us-east-1",
    batch_size=32,
    name="My training run",
)

# Create a reward model training job
reward_job = client.finetune.create(
    project_id="...",
    vla_type="sarm",
    dataset_version_id="01234567-89ab-4cde-8fab-0123456789ab",
    hours=2.0,
    camera_mappings={"cam_1": "observation.images.top"},
    job_type="reward",  # Train a reward model
)

# Create a VLA job with a trained reward model
vla_reward_job = client.finetune.create(
    project_id="...",
    model_id="lerobot/smolvla_base",
    vla_type="smolvla",
    dataset_version_id="01234567-89ab-4cde-8fab-0123456789ab",
    hours=2.0,
    camera_mappings={"cam_1": "observation.images.top"},
    job_type="vla_w_reward",  # VLA + reward model
    use_rabc=True,
    sarm_reward_model_id="your-org/sarm-reward-model",
    sarm_reward_model_revision="0123456789abcdef0123456789abcdef01234567",
    sarm_image_observation_key="observation.images.top",
)

# Get job status
status = client.finetune.get(job.job_id)

# Cancel a job
result = client.finetune.cancel(job.job_id)

Advanced: Custom Hyperparameters

You can customize model hyperparameters for fine-grained control over training. The SDK validates hyperparameters before submitting the job, so invalid configurations are caught early.

# 1. Get default hyperparameters for your model
params = client.finetune.get_hyperparams_defaults(
    vla_type="smolvla",
    model_id="lerobot/smolvla_base",
)

# 2. Customize the parameters as needed
params["training"]["learning_rate"] = 1e-5
params["training"]["num_epochs"] = 50

# 3. (Optional) Validate before creating the job
validation = client.finetune.validate_hyperparams(
    vla_type="smolvla",
    hyperparams=params,
)
if not validation.valid:
    for issue in validation.issues:
        print(f"  {issue.field}: {issue.message}")

# 4. Create the job with custom hyperparameters
# Note: create() internally calls validate_hyperparams() when vla_hyper_spec
# is provided. If validation fails, a ValueError is raised and no job is created.
job = client.finetune.create(
    project_id=project.project_id,
    model_id="lerobot/smolvla_base",
    vla_type="smolvla",
    dataset_version_id="01234567-89ab-4cde-8fab-0123456789ab",
    hours=2.0,
    camera_mappings={"cam_1": "observation.images.side"},
    vla_hyper_spec=params,
)

Instances

# List available GPU instances
instances = client.instances.list()
for inst in instances:
    print(f"{inst.id}: {inst.gpu_description} - {inst.credits_per_hour} credits/hr")
    print(f"  Specs: {inst.specs.gpu_count}x GPU, {inst.specs.memory_gib}GB RAM")
    print(f"  Regions: {[r.name for r in inst.regions]}")

Models

# List available VLA model types
models = client.models.list()
for model in models:
    print(f"{model.id}: {model.name}")
    print(f"  Base model: {model.base_model_id}")
    print(f"  Camera slots: {model.camera_slots}")

Projects

# Create a project
project = client.projects.create(
    name="My Project",
    description="Optional description",
)

# List all projects
projects = client.projects.list()
for p in projects:
    print(f"{p.name}: {len(p.jobs)} jobs")

# Delete a project (fails if it has active jobs)
client.projects.delete(project.project_id)

Configuration

Environment Variables

• QUALIA_API_KEY: Your API key (used if not passed to constructor)
• QUALIA_BASE_URL: Override the API base URL (default: https://api.qualiastudios.dev)

Custom HTTP Client

import httpx

# Use a custom httpx client for advanced configuration
custom_client = httpx.Client(
    timeout=60.0,
    limits=httpx.Limits(max_connections=10),
)

client = Qualia(api_key="...", httpx_client=custom_client)

Context Manager

# Automatically close the client when done
with Qualia(api_key="...") as client:
    models = client.models.list()

Error Handling

from qualia import (
    Qualia,
    QualiaError,
    QualiaAPIError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
)

try:
    client = Qualia(api_key="invalid-key")
    client.models.list()
except AuthenticationError as e:
    print(f"Auth failed: {e}")
except NotFoundError as e:
    print(f"Not found: {e}")
except ValidationError as e:
    print(f"Validation error: {e}")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
except QualiaAPIError as e:
    print(f"API error [{e.status_code}]: {e.message}")
except QualiaError as e:
    print(f"SDK error: {e}")

Requirements

• Python 3.10+
• httpx
• pydantic

License

MIT