Qualia Python SDK

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

Draft a Task from person-picked episodes (staged)

Create a Task manually with client.tasks.create("Pick a box"), even without any episodes. To ask for a proposal, first pick every reference yourself: 1–3 success takes and optionally 1–2 takes per named failure/recovery mode. Nothing samples, suggests or drops references. Each episode is a full accounts/{account}/datasets/{dataset}/episodes/{episode} name:

task = client.tasks.create(
    "Pick a box",
    references=[
        {
            "role": "SUCCESS",
            "episode": "accounts/<account>/datasets/<dataset>/episodes/<episode>",
        },
    ],
)
price = client.task_drafts.estimate(task.name)  # no reservation or model call
run = client.task_drafts.create(task.name)  # reserves credits
result = client.task_drafts.wait(run.name)  # or get/list/cancel
if result.state == "SUCCEEDED":
    # Inspect result.proposal and its procedure; edit before saving if needed.
    saved = client.task_drafts.apply(result)
    # Publish only when a person explicitly decides to:
    # client.tasks.publish(saved.name, etag=saved.etag)

max_cost= is an optional credit cap for estimate/create. Drafting does not change or publish the Task; apply(..., edited=...) saves reviewed text and graph, and publish is separate. Pick or drop one reference with add_reference and delete_reference; an API key acts as the person who created it. This API is staged: the vlm-task-draft template is disabled until rollout authorization; calls to price/start fail until it is enabled, and live billing/graph-check QA is still pending.

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.

Upload errors marked X-Qualia-Retryable: false stop after one request. Typed ingest configuration and protocol failures show the gateway's readable cause in Python and the CLI. The transfer exception retains code, retryable, detail and status_code; ordinary unavailable and timeout failures still retry. The advisory failure heartbeat sends the readable cause without Python class names, internal gateway routes or serialized error objects. It remains one line of at most 300 characters, with signed URL query strings removed. Reporting is best-effort and never replaces the caller's original exception.

0.8.0rc8: what the dashboard does, from an API key

The SDK and qualia-cli==0.3.0rc8 cover the dashboard's curation, task, check, run, version and export actions. Members and integrations need a key made with their opt-in scope. It needs the server with these routes. See Use Qualia from an agent.

0.8.0rc7: robots and their grippers from a script

client.robots registers robots, imports their URDFs and gives a robot a gripper of its own; the matching CLI is qualia-cli==0.3.0rc7 (qualia robot ...). See Robots (qualia.v1). It needs the server that has robots own their gripper and CopyRobot.

0.8.0rc6: set a handheld dataset up from a script

client.datasets gains setup, poses, resolve_poses and wait_resolve, and update(reach_settings=); client.checks gains reach_status and cancel_reach. Together they do what the dashboard's Setup strip does (pick the robot and the solver, resolve the poses at the TCP, run and follow the solver), and the matching CLI is qualia-cli==0.3.0rc6 (qualia dataset setup, qualia dataset poses, qualia check reach-status). See Set up a handheld (UMI) dataset. They need no server change.

Upgrading to 0.8.0rc4

This source candidate adds saved upload destinations shared with the CLI's local rig configuration. Qualia(workspace=...) selects the workspace; data.upload(..., merge_target=...) chooses an existing merged dataset and standalone=True overrides a saved destination for one upload. Saved choices apply only to the matching API host, account and workspace. The server binds the intent to the logical source upload across transfer retries; one a merge revert released before merges were final stays released instead of silently appending again.

The matching CLI is qualia-cli==0.3.0rc4. After these candidates are published, install this SDK with pip install 'qualia-sdk==0.8.0rc4' and use a deployment that supports the merged-upload lifecycle. The examples below show the scope and override options. Merge history is in the dataset's Versions tab; a whole merge cannot be reverted, individual uploads can be removed, and frozen versions keep their source references.

0.8.0rc3: links to Tasks

Episodes, uploads and mixture sources can name the Task they are about (linked_task, accounts/{a}/tasks/{t}), beside the free-text task label training reads. Nothing is removed: episodes.assign_task takes linked_task= (the label then defaults to the task's name, so task is optional), episodes.link_task links or unlinks one episode, datasets.update(linked_task=) links an upload, and mixtures.add_source(linked_task=) follows every upload (and merged dataset) linked to a task. A task can still be deleted: its sources then read state == "TASK_DELETED", resolve to nothing, and are only removed (mixtures.remove_source).

An upload's Task is its episodes' too (QUA-3413): an episode with no link of its own reads the upload's, so datasets.update(linked_task=) links every one in a single request. Episodes that link the old Task themselves keep it unless you pass relink_episodes=True; episodes.assign_task still overrides the upload's Task for the episodes it names. A merged dataset's Task is its own (QUA-3511): datasets.update(linked_task=) sets it on the merged dataset alone, and its episodes with no link of their own read it, as an upload's read the upload's (its uploads keep theirs, which nothing reads any more); relink_episodes=True works there as on an upload. Linking by label links a merged dataset by its own task label, as it links an upload, and never an upload a merge took. episodes.assign_task through it links each episode in its upload: all or none per upload.

Upgrading to 0.8.0rc2

Use client.datasets.update(dataset, tags=["umi", "real-world", "human", "gello"]) to change collection classifications through their saved metadata. Nonempty custom-only input (tags=["warehouse-a"]), or unchanged classifications with edited customs, replaces custom tags while preserving declarations and inference. When classifications are present, echo them to remove all customs without declaring methods. tags=[] clears customs and explicitly records Unknown in every collection dimension. The SDK reads the owner and uses its etag unless one is supplied; stale writes fail. These edits require a server that returns independent collection metadata. method and methods remain compatibility views. Merged datasets own their tags and keep them on append. See the collection contract and calibrated TCP export.

Upgrading to 0.8.0rc1

This release removes the old version and episode-edit helpers from client.data. Their replacements on client.versions and client.episodes speak qualia.v1, so they need your account (see 0.7.0rc7 below), and an episode is named through the upload it was recorded in: pass dataset= with a bare id, or the full name accounts/{a}/datasets/{d}/episodes/{e}.

A qualia.v1 tag now carries dataset_version_id: the immutable DatasetVersion UUID that client.finetune.create, materialize_version, health_for_version and get_training_features take. Tag an upload or a mixture with versions.create. A definition built from filters or an explicit episode selection is still pinned with data.pin_version(def_id): saved queries have no qualia.v1 equivalent, so that path stays on /v1.

What changed in the shapes:

• Times are seconds from the start of the recording (time_seconds, start_seconds, end_seconds), not microseconds or trim_start_s.
• Batch edits return EpisodeBatch: episodes holds each edited qualia.v1.Episode as it now stands, and skipped holds SkippedEpisode(episode=<name>, reason=...).
• An edit is checked against the etag the helper read, not an expected_parent_rev; a stale one raises QualiaAPIError with code "aborted". qualia.v1 edits carry no revision note, tier or review_state (rulings are episodes.keep/reject).
• A split preview (validate_only=True) names its children with throwaway ids; a split reports dropped_lead_frame_count/dropped_tail_frame_count.

client.events.stream now follows one job and needs job_id= (a project-restricted key also passes project_id=). The server no longer has an account-wide stream, so the client-side job filter is gone. The stream carries log, telemetry and capacity-wait lines, not status: pipeline.pipeline_status events no longer arrive on it, so a loop that broke on one never ends. Use client.finetune.watch(job_id), which reads the status with finetune.get and reopens the stream until the job ends.

Removed with them: the models EpisodeMarker(s), EpisodeSplit, EpisodeSplitBatch, SplitChild, EpisodeUnsplit, EpisodeRevisionCommit, EpisodeTrim, EpisodeTrimBatch, EpisodeSubtask(s), EpisodeSubtaskBatch, SkippedEpisode, DatasetVersionPage and DatasetVersionSourcePage (the qualia.v1 ones live in qualia.v1). Definitions (create_definition and friends) and list_episode_revisions are unchanged.

Upgrading to 0.7.0rc7

New resources speak the qualia.v1 API, the one the dashboard reads: client.datasets, client.episodes, client.versions and client.mixtures (see below). They need your account: pass account=, set QUALIA_ACCOUNT, or run qualia auth login --account <uuid>.

client.data.list() is unchanged: it still reads the old /v1/datasets/list route and returns qualia.Dataset. It is now the legacy call; client.datasets.list(filter="kind != MIXTURE") replaces it and returns qualia.v1.Dataset (an upload merged into another dataset is left out; the merged dataset, kind MERGED, stands in for it). The fields map like this:

search= becomes display_name:"<text>" in the filter. The server cannot filter on the format, so to keep one subtype, check source_format on each item.

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/pi05_base",
    vla_type="pi05",
    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 op: {status.current_op}")
for op in status.ops:  # one per pipeline op, in run order
    print(op.name, op.status)

# 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", "mcap_ros2"; when
# omitted it is read from the tree (meta/info.json, or any .mcap).
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().
# While it transfers, stderr shows "X / Y sent": source bytes already in storage,
# whether or not the server has verified them yet (verification follows the
# transfer). on_progress events carry that figure as "bytes" and the verified
# part as "bytes_verified".

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

# Qualia() also reads a saved `qualia dataset destination set` from the shared
# local CLI config when its API host, account, and workspace match. Select a
# workspace explicitly with the constructor or QUALIA_WORKSPACE; this is sent
# as the gateway's X-Qualia-Workspace header.
client = Qualia(
    api_key="your-api-key", account="<account-uuid>", workspace="<project-uuid>"
)

# Pin one upload to an existing merged dataset, or override the saved default.
result = client.data.upload(
    "/data/week-2",
    name="week-2",
    source_format="lerobot_v3",
    merge_target="accounts/<account-uuid>/datasets/<merged-uuid>",
)
print(result["merge_append"])  # appended or action_needed

# A failed append leaves the published upload receipt and dataset available in
# result; review it and append explicitly with client.datasets.merge(...).
# An append compares no label and no Task: the upload keeps its own operator,
# task label and Task, and the merged dataset its own. Retries see the same
# durable intent. `standalone=True` opts out for one upload.
client.data.upload(
    "/data/calibration",
    name="calibration",
    source_format="lerobot_v3",
    standalone=True,
)

# A handheld (UMI) recording converts to gripper poses as it lands: `rig=`
# names a gripper (its name, id or rig key) whose rig and TCP calibration the
# server writes into the upload's meta/ingest.toml. An upload into a merged
# dataset whose uploads agree on one rig takes it without `rig=`. The local
# file is never written; result["ingest_config"] says what was composed.
result = client.data.upload(
    "/data/umi-session",
    name="umi-session",
    source_format="mcap_ros2",
    rig="umi-handheld",
)

# 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.
# These are the files as recorded: trims and splits are not applied, and when the
# dataset has any it warns (qualia.UncuratedDownloadWarning) with the curated path.
paths = client.data.download("gongsta", dest="/data/gongsta-copy")
# The curated episodes: a tagged version, as a verified LeRobot tree.
paths = client.data.download(
    "gongsta", dest="/data/gongsta-1.0", version=f"{dataset_id}:1.0"
)

# List your datasets (legacy route; client.datasets.list is the qualia.v1 replacement)
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)

Conflicting generated paths

Uploads refuse meta/ingest_report.json before hashing, packing or contacting the gateway: ingest generates this path for LeRobot 2.1, LeRobot v3 and MCAP. MCAP also reserves meta/info.json for derived episode metadata; LeRobot uses that file as source input. Case aliases and file/directory collisions are refused too. Other metadata remains valid, subject to the existing path contract and trusted meta/qds namespace rule.

The error names the source and generated paths. Keep the original tree and rename the conflicting file in a separate copy before uploading it. The SDK does not skip, delete or overwrite the conflicting file. Episode-dependent generated paths are checked against the actual worker output; a collision there is terminal and names the path even when client preflight was bypassed.

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.

Cancelled uploads

An upload in progress can be cancelled from the dashboard (or by any client acting on the account). The server cannot stop your process, so it tells the SDK: the next transfer heartbeat answers cancelled, and data.upload stops and raises UploadCancelledError (its repo_id names the repository). Nothing is published. Bytes already sent are kept for the usual retention, so running the same upload again can resume it (see above) rather than start over.

from qualia import UploadCancelledError

try:
    client.data.upload("/data/session", name="dk1-kitchen")
except UploadCancelledError as e:
    print(f"cancelled by the user: {e.repo_id}")

The SDK names its transfer session to the server in the x-qualia-upload-session header on the publication request, so a cancel that arrives between heartbeats still refuses the publication. An SDK from before upload cancellation (QUA-3009) never reads the heartbeat's answer and keeps sending bytes after a cancel; the server refuses its publication instead (QUA-3025), and it fails with a 409 saying the upload was cancelled by the user.

Finetune

# Create a VLA finetune job (default job_type="vla")
job = client.finetune.create(
    project_id="...",
    model_id="lerobot/pi05_base",  # Server-pinned built-in alias
    vla_type="pi05",  # pi05 or act
    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",
)

# ACT trains from scratch, so it takes no model_id
act_job = client.finetune.create(
    project_id="...",
    vla_type="act",
    dataset_version_id="01234567-89ab-4cde-8fab-0123456789ab",
    hours=2.0,
    camera_mappings={"cam_1": "observation.images.top"},
)

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

# A pipeline with map nodes runs each map's body once per item; status.maps
# gives each map's op_run_id and item counts (skipped = the credit budget ran
# out before the item started). Page its items, or walk them all:
for m in status.maps:
    print(m.node_id, f"{m.counts.completed}/{m.counts.total}", m.counts.failed)
    page = client.finetune.map_items(job.job_id, m.op_run_id, status="failed")
    for item in client.finetune.iter_map_items(job.job_id, m.op_run_id):
        print(item.ordinal, item.attempt, item.status, item.error)

# Follow a job's log, telemetry and capacity-wait lines until it ends. The
# stream has no replay and the server ends it now and then: watch reads the
# status with get() and reopens it until the job completes, fails or is
# cancelled. Read get() afterwards for the full history.
for event in client.finetune.watch(job.job_id):
    if message := event.data.get("message"):
        print(message)

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

Finetuning supports ACT and Pi0.5. SmolVLA and SARM finetuning, the reward and vla_w_reward job types and RA-BC (use_rabc, sarm_*) were retired: create() raises ValueError for them before any request is sent. Jobs and models trained with them stay readable, so job.metadata.vla_type and job_type are plain strings that may still hold "smolvla", "sarm", "reward" or "vla_w_reward".

A failed job says who caused it: status.failure_cause is user when something you set or authored made it fail (status.failure_subject names it, e.g. dataset_version:<id> or parameter:base_model; status.failure_reason says why), and platform otherwise. status.credits is the settlement (reserved, charged, refunded, reason). A user failure is charged its usage, up to the reservation; a platform failure is refunded in full. See the finetune reference.

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="pi05",
    model_id="lerobot/pi05_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="pi05",
    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/pi05_base",
    vla_type="pi05",
    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. A step's one-line description is its condition (what an action gets done, what an outcome looks like); there is no separate conditions list. 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=[
        # the start needs no description: the scene says how an attempt begins
        Step(id="start", kind=StepKind.START),
        # 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

Datasets, episodes, versions and mixtures (qualia.v1)

These four resources speak the qualia.v1 API, the one the dashboard uses. A dataset is an upload (one recording session), a mixture (kept up to date from other datasets) or a merge of uploads (several uploads reviewed, versioned and trained as one). Episodes, versions (tags) and a mixture's sources nest under a dataset. Everything lives under your account: pass account= or set QUALIA_ACCOUNT, or use full resource names. One episode can be read through several datasets, so a bare episode id also needs dataset=; its full name (accounts/{a}/datasets/{d}/episodes/{e}) carries the scope.

The returned models (qualia.v1.Dataset, qualia.v1.Episode, ...) and the REST routes are generated from proto/qualia/v1 into qualia/v1/_gen.py when the wheel is built (in a q-core checkout, run ./proto/gen.sh sdk first). Attributes are the proto field names; enums are value names such as "KEPT", listed in qualia.v1.ENUMS.

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

# Datasets: uploads and mixtures, with any AIP-160 filter
uploads = client.datasets.list(filter='kind = UPLOAD AND labels.task = "fold"')
mix = client.datasets.create("Folding, all rigs")
client.datasets.update(mix.name, description="every folding session")

# Merge a week of one bench's sessions into one dataset of its own: reviewed,
# tagged, mixed and trained as one, every take keeping its rulings, issues,
# notes and trim. The uploads must share cameras, features, channel names, fps
# and where their poses are (else 400 FAILED_PRECONDITION naming what differs,
# with nothing written); where they differ in a robot, method, reach settings or
# channel key, validate_only=True lists it and resolutions= pick one value for
# all; where they differ in a label or Task, the merged dataset takes their
# commonest value unless resolutions= pick another, as its own, and each upload
# keeps its own. Collection tags are seeded once on C, a method resolution
# changes only C, and appending preserves its existing tags. A merge is
# final: it cannot be undone,
# and the uploads become the merged dataset's history. They leave `list()`;
# their reads (`get()`, which names the merged dataset, versions, episodes,
# issues) keep answering, and every write goes through the merged dataset.
week = client.datasets.merge([u.name for u in uploads.items], display_name="Fold bench")
week.source_datasets  # the uploads, in the order merged
client.datasets.merge([next_upload.name], into=week.name)  # add one later

# Mixtures read sources: a dataset by name, or every upload and merged dataset
# with one label (a merged dataset by its own labels and method tags)
client.mixtures.add_source(mix.name, dataset=uploads.items[0].name, follow="head")
client.mixtures.add_source(mix.name, label=("task", "fold"))  # follows stable

# Tasks: link what was recorded against one. The label stays the instruction
# training reads; a link changes no revision or version.
fold = client.tasks.list().items[0]
client.datasets.update(uploads.items[0].name, linked_task=fold.name)
client.episodes.assign_task(
    uploads.items[0].name, filter="state = KEPT", linked_task=fold.name
)
client.mixtures.add_source(
    mix.name, linked_task=fold.name
)  # every upload or merged dataset linked to it
linked = client.episodes.list("-", filter=f'linked_task = "{fold.name}"')
# Publishing a task links the unlinked labels that are its name; near misses
# (case, spacing, punctuation) are previewed, then linked by pick
preview = client.episodes.link_task_labels(fold.name, validate_only=True)
client.episodes.link_task_labels(fold.name, [m.label for m in preview.matches])

# Episodes: list, edit, rule. "-" lists across the whole account.
page = client.episodes.list(mix.name, filter="state = UNREVIEWED")
episode = client.episodes.get(page.items[0].name, view="analysis")
client.episodes.update(episode.name, tags=["clean"], etag=episode.etag)
# Many episodes in one call: all of them change or none do, one revision each.
# Up to 500 episodes of one upload; trims and task links are one per call
# (update(), link_task()). A refusal names its edit ("edits[3] (...): ...").
from uuid import uuid4

from qualia.resources.episodes import EpisodeEdit

takes = client.episodes.list(uploads.items[0].name, filter="state = UNREVIEWED").items
client.episodes.update_many(
    uploads.items[0].name,
    [EpisodeEdit(episode=e.name, tags=["clean"], etag=e.etag) for e in takes],
    request_id=uuid4(),  # a retry with the same id answers the first reply
)
client.episodes.keep(mix.name, [e.name for e in page.items])
client.episodes.reject(mix.name, filter='open_issue_codes:"jerky"', reason="jerky")
client.episodes.split_at_markers(
    uploads.items[0].name, filter="child_count = 0", labels=["success"]
)
# What the studio plays: short-lived camera URIs, and the joints sampled evenly
videos = client.episodes.sign_videos(episode.name)  # .unpublished for a split child
series = client.episodes.sample_series(episode.name, max_points=400)  # .unpublished too
# streams= adds "TCP" for a handheld take; version= reads a tag as it froze the take
series = client.episodes.sample_series(episode.name, streams=["STATE", "TCP"])
frozen = client.episodes.get(episode.name, version="stable")
batch = client.episodes.batch_sample_series(mix.name, [e.name for e in page.items[:50]])

# The take's edit history, newest first, then revision 0
history = client.episodes.revisions(episode.name, limit=20)
# Split around windows to keep; what lies outside every window is dropped
client.episodes.split(
    uploads.items[0].name,
    keep={takes[0].name: [(0.5, 6.0), (8.0, 14.0)]},
    validate_only=True,
)
# One take's own Task link, with the etag read; "" falls back to the dataset's
client.episodes.link_task(episode.name, fold.name, etag=episode.etag)

# Marker edits, one revision per episode; skipped ones say why
client.episodes.trim_to_markers(uploads.items[0].name, ids, start_labels=["success"])
client.episodes.name_subtasks(uploads.items[0].name, ids, labels=["grasp", "place"])

# Versions: tag what a dataset yields now; cite it as <slug>:<tag>
diff = client.versions.compare(mix.name)
tag = client.versions.create(
    mix.name, diff.suggested_version, note=diff.draft_note, digest=diff.target_digest
)
print(tag.reference)  # "folding-all-rigs:1.0"
print(tag.dataset_version_id)  # the UUID training takes
commits = client.versions.list_source_commits(tag.name)  # the repositories it pins

Lists page with limit/cursor; page_info.total_count counts every match where the API reports it. A stale etag raises QualiaAPIError with status 409 and code "aborted".

Training takes a tag's dataset_version_id.

A saved dataset version can have no kept episodes (episode_count = 0). It remains readable, and the source episodes stay stored. Training and export refuse that version.

Deleting an upload whose episodes are Task references removes those references (the Tasks stay), so it is refused (400, failed_precondition) until you say so:

upload = client.datasets.get(uploads.items[0].name)
for held in upload.task_references:  # one per Task, with its episode count
    print(held.display_name, held.episode_count)
print(upload.task_referenced_episode_count)  # distinct episodes in all
client.datasets.delete(upload.name, force=True)

Set up a handheld (UMI) dataset (qualia.v1)

client.datasets does what the dashboard's Setup strip does: pick the robot a handheld upload plays on, pick the solver and placement, and move the poses to the robot's TCP. Nothing here needs the dashboard.

# Where are the poses? at is "root" (the gripper root's <side>.ee.* poses),
# "tcp" (each hand’s TCP too) or "recorded" (a raw recording’s trackers). None means the dataset holds no handheld poses.
poses = client.datasets.poses(upload.name)
poses.at, poses.rig  # "root", "umi-handheld"
poses.capture_gripper  # the workspace's gripper for that rig: its calibration is what a resolve takes
poses.current  # at the TCP: resolved with the capture gripper's calibration as it is now?

# Pick the robot and the solver. Only what you give is written; the rest of the
# reach settings are kept, and repeating the call writes nothing.
setup = client.datasets.setup(
    upload.name,
    robot=robot.name,  # an id or accounts/{a}/robots/{r}
    solver="pyroki",  # "pyroki" | "dls" | "" (follow the robot's model)
    base=[0.0, 0.0, 0.0],  # metres in the robot's base frame, within 5 m
    heading_deg=0.0,
)
setup.changed  # ["target_robot", "reach_settings"]
setup.resolve  # the resolve it started, or None
setup.resolve_skipped  # why none started: "robot-without-tcp", "up-to-date", ...

Picking a robot starts a resolve at the TCP by itself, exactly when the dataset is an upload with handheld poses, every arm of the robot has a TCP, the workspace has a gripper for the upload's rig with a tracker-to-TCP calibration, and the poses are not at the TCP yet (or are there with another calibration). The resolve uses that capture gripper's calibration, never the robot's own gripper's. A robot with no TCP, a mixture and a merged dataset leave the poses alone, and setup never undoes poses. resolve=False never resolves; resolve=True resolves with the capture gripper whatever the robot.

A resolve updates the saved pose file and publishes the result:

setup = client.datasets.setup(
    upload.name,
    robot=robot.name,
    wait=True,
    on_progress=lambda r: print(r.state, r.progress),
)
# or start it, and wait later
started = client.datasets.resolve_poses(upload.name, "tcp", wait=False)
client.datasets.wait_resolve(started.operation)  # polls Datasets.GetReingest
client.datasets.resolve_poses(upload.name, "root")  # back to the gripper root

wait defaults to two hours of polling for resolve_poses; when it runs out (TimeoutError) the work keeps running server-side, so wait for it with wait_resolve instead of resolving again. A resolve that fails raises QualiaError with the server's reason; a ruling or trim made while it runs is one cause, and resolving again is the way on.

For a merged dataset, prepare its uploads explicitly. Picking a robot does not move them. The precheck names each upload's state and any blocker; pass its membership revision to move eligible uploads or retry a selected upload. The server uses each capture rig's current calibration. Missing saved pose files cannot be converted by this operation. A move of an upload with split episodes is blocked until pose correction can preserve its pieces (QUA-3786). The precheck currently requires a write-scoped credential because it shares the move RPC; read-scoped prechecks remain a follow-up. Removing a member makes it a standalone upload again; its data and review decisions remain, and the merged dataset keeps its history and old versions.

check = client.datasets.precheck_merged_poses(merged.name)
for member in check.members:
    print(member.display_name, member.state, member.reason)
started = client.datasets.move_merged_poses(
    merged.name, membership_revision=check.membership_revision
)
# To retry one upload, pass members=[member.upload]. To remove it from the
# current dataset and record the removal, use the latest revision:
client.datasets.remove_merged_member(
    merged.name, member.upload, membership_revision=check.membership_revision
)

Run solver is never pressed by itself. Press it with run_solver=True on setup (it first waits for a resolve it started), or checks.run(upload.name, "reach"), then follow it:

client.datasets.setup(upload.name, run_solver=True)
# settings, current_key, last_run, stale
status = client.checks.reach_status(upload.name)
status.last_run.state, status.last_run.done, status.last_run.total
client.checks.cancel_reach(upload.name)  # takes already solved keep their results

datasets.update(upload.name, reach_settings={"solver": "dls", "base_m": [0.1, 0, 0], "home_heading_rad": 0.0}) writes the settings directly; it replaces all three, so what you leave out follows the robot's model. Saving them starts no solve.

Robots (qualia.v1)

client.robots registers robots, imports their URDFs and gives a robot a gripper of its own to line up and set the jaw of; copy the robot to try another gripper. create_from_urdf registers and imports in one call, set_gripper says what a robot carries in one call, and every method takes a display name as well as an id. A step that fails after something was written says, in the error's message, what it left and how to remove it. The methods, errors and a walkthrough are in Robots.

Checks (qualia.v1)

client.checks runs a check over a dataset, follows its run and reads what it found. The VLM check (vlm) has a model read each episode against the published Task it links: confident failures become issues, uncertain readings notes. Ask what a run would cost first; asking reserves nothing, and a run is never charged more than it reserved.

estimate = client.checks.estimate(dataset.name, max_episodes=5, max_cost=20)
# Match the trial's selection and ceiling when starting it.
started = client.checks.run(
    dataset.name, model=estimate.models[0], max_episodes=5, max_cost=20
)
run = client.checks.wait(started.check_runs[-1])  # QUEUED → RUNNING → SUCCEEDED | FAILED
client.checks.list_runs("vlm", dataset=dataset.name, state="FAILED")
for issue in client.checks.findings(dataset.name).items:
    f = (
        issue.finding
    )  # verdict, confidence, explanation, moments (recording seconds), model
    print(f.verdict, f.confidence, issue.text)

model is for the VLM check only, and must be one the deployment offers (estimate(...).models, the default first); another is refused with 400. max_episodes samples linked episodes across the scope, and max_cost caps credits reserved and spent. Both are VLM-only positive integers. Omit them to check every eligible episode without a ceiling.

The checks catalog (QUA-3524) holds versioned checks such as task.graph. check_ids, categories and all select a union of them, run as one run with one model call per episode for every selected VLM check:

for check in client.checks.list(dataset.name).items:  # runnable, last_result
    print(check.check_id, check.runnable, check.unavailable_reason)
answer = client.checks.estimate(dataset.name, categories=["task"], max_episodes=50)
started = client.checks.run(
    dataset.name, categories=["task"], max_episodes=50, max_cost=answer.estimate.credits
)
client.checks.wait(started.check_run)
client.checks.list_findings(dataset.name, "task.graph", verdict="FAIL")

A catalog run needs max_cost covering the bound estimate answers; a retry with the same request_id returns the first run.

The reach check ("reach", Run solver) is pressed with checks.run(dataset.name, "reach"); checks.reach_status and checks.cancel_reach follow and stop it (see Set up a handheld (UMI) dataset).

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 (tasks, datasets, episodes, versions and mixtures)

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.

For structured finetune HTTP 400 refusals, the SDK exposes detail.message directly as QualiaAPIError.message when the JSON detail is an object with a string message.

Requirements

• Python 3.10+
• httpx
• pydantic

License

MIT

Saved uploader destinations require a known matching account, normalized API host and workspace. Selecting a different API key without an explicit/confirmed account does not inherit the saved destination or its workspace. Automatic upload completion uses the server's upload intent; an append reverted before whole merges became non-revertible stays terminal (a whole merge cannot be reverted).