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.

Newer servers may add fields to their responses; an older SDK ignores the fields it does not know. Request arguments stay strict, so a misspelt key still fails before anything is sent.

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.

Git checkouts no longer fall back to Git LFS

The transfer filter that qualia git install wires into a cloned dataset fetches a file's bytes only by the x-qds-oid content hash the filter itself writes into the pointer. Qualia's git remote no longer serves the Git LFS batch API, so there is no fallback: checking out a stock git-lfs pointer (only version, oid sha256: and size) fails with pointer carries no x-qds-oid content hash, and git lfs pull against the remote fails too. To fix a file, put its original bytes back in the working tree, run qualia git install, then git add --renormalize <path>, commit and push; the new pointer carries x-qds-oid.

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"])
# result["ingest_verdict"] is "published", "degraded" or "underivable" (bytes
# stored, no episode derived); pass raise_on_underivable=True to raise instead.
# The report card lists outlier episodes, such as a take that barely moves,
# separately from flagged ones, and an outlier never flags an episode.
# An MCAP upload also returns result["channel_summary"]: the topics that fed the
# state and action columns, with "action: none (no action column)" when there is
# none. A rig that needs to declare them writes <session>/meta/ingest.toml; see
# the data.upload docstring and qualia.ingest_config_schema().

# 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.
# LeRobot trees are checked before any byte moves. Structure the server always
# refuses (unreadable meta/info.json, a codebase_version that does not match
# source_format, no parquet under data/, a v2.1 tree without episodes.jsonl or
# tasks.jsonl) raises UploadBlockedError, and allow_underivable=True cannot
# override it. A missing video only warns, naming the episode and camera:
# that episode will be flagged, and you should probably exclude it.
# These checks match the server's own reading: a camera is any feature whose
# dtype is "video" in any letter case, and a .parquet shard counts in any case.
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). Damaged or empty bags upload with a
# warning; a session no bag can derive raises UploadBlockedError before any
# byte moves (allow_underivable=True uploads it anyway).
# A recording cut short (killed recorder) uploads its readable part, with a
# source_truncated warning, when that part carries state the server can
# derive; the pre-flight blocks it as source_underivable only when it can
# prove there is none. The result's `validation` report card marks each
# episode `ok` or `flagged` (older cards say `accepted`, which reads as ok).
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). An interrupted download
# keeps its progress in .gongsta-copy.qualia-download-partial; run it again to resume.
paths = client.data.download("gongsta", dest="/data/gongsta-copy")

# List your datasets
page = client.data.list()
for d in page.items:
    # verdict: published | degraded | underivable; None = recorded before verdicts (published)
    print(
        d.artifact_id,
        d.path,
        d.qds and d.qds.ingest_verdict,
        d.qds and d.qds.verdict_reason,
    )

# 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)

Tasks

A task says what a robot should do and how success shows, as five pieces of prose: description, scene, steps (one per line), success and failure. It has one editable draft; publishing writes an immutable, numbered revision, so later edits never change what was published. Tasks live under your account (accounts/{account}/tasks/{task}): pass account= or set QUALIA_ACCOUNT, or use full resource names.

from qualia import Qualia, TaskSpec

client = Qualia(account="<account-uuid>")

task = client.tasks.create(
    "Restock shelf",
    draft=TaskSpec(description="Move every box from the tray to the shelf"),
)

# Save single fields; the rest of the stored draft stays. "" clears a field.
task = client.tasks.update(
    task.id,
    steps="Pick a box from the tray\nPlace it upright on the shelf",
    success="The tray is empty and every box stands upright",
    etag=task.etag,  # refuses a stale write
)

# Or replace the whole draft
task = client.tasks.update(task.id, draft=task.draft, display_name="Restock")

# Publishing needs a description and a success
revision = client.tasks.publish(task.id, etag=task.etag, note="first cut")
client.tasks.revisions(task.id)  # newest first
client.tasks.revision(task.id, 1)

client.tasks.delete(task.id, force=True)  # account owners; force drops revisions

Lists page with limit/cursor like the other resources. A revision's content_hash is the sha256 of its spec: one hash, one content.

A task may also carry an optional procedure beside the prose: the task as a state machine of steps (one start, actions, success and failure outcomes) and transitions, a DAG that checks report against. It saves on its own like a prose field, and publishing checks it is well formed when there is one:

from qualia import Procedure, Step, StepKind, Transition

procedure = Procedure(
    steps=[
        Step(id="start", kind=StepKind.START, description="Boxes in the tray"),
        # exactly 6; min_count alone is "at least", max_count alone "up to"
        Step(
            id="place",
            description="Place a box upright on the shelf",
            min_count=6,
            max_count=6,
        ),
        Step(id="shelved", kind=StepKind.SUCCESS, description="All boxes shelved"),
        Step(id="fallen", kind=StepKind.FAILURE, description="A box fell over"),
    ],
    transitions=[
        Transition(source="start", target="place"),
        Transition(source="place", target="shelved"),
        Transition(source="place", target="fallen"),
    ],
)
task = client.tasks.update(task.id, procedure=procedure, etag=task.etag)
task.draft.procedure.edges()  # [("start", "place"), ("place", "shelved"), ("place", "fallen")]
client.tasks.update(task.id, procedure=Procedure())  # an empty one clears it

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)
• QUALIA_ACCOUNT: Account UUID for account-scoped resources such as tasks

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}")

On the account-scoped routes (/v1/accounts/..., e.g. tasks) QualiaAPIError.code carries the API's code name: ABORTED (409) is a stale etag, so re-read and retry; FAILED_PRECONDITION (400) is a business-rule refusal such as publishing a draft without a description or success.

Requirements

• Python 3.10+
• httpx
• pydantic

License

MIT