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.

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.

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; a deliberate merge revert remains reverted 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 and undo are available in the dataset's Versions tab; existing 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 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. On a merged dataset, datasets.update(linked_task=) sets every upload it holds (QUA-3443), and episodes.assign_task links each episode in its upload: all or none per upload.

Upgrading to 0.8.0rc2

Dataset labels now expose canonical, combinable collection-method tags as labels.methods. Replace the complete tag set with client.datasets.update(dataset, methods=["teleop", "vr"]); pass methods=[] to explicitly clear it. The existing scalar method field remains available for older clients and integrations.

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

# 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(...).
# Retries see the same durable intent, so an explicit revert will not be
# undone by retrying the transfer. `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.
paths = client.data.download("gongsta", dest="/data/gongsta-copy")

# 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: reviewed, tagged and
# trained as one, every take keeping its upload, rulings, issues, notes, tags
# and trim. The uploads must share a robot interface, cameras, fps, method,
# robots, rigs and, for handheld, where their poses are (else 400
# FAILED_PRECONDITION naming what differs, with nothing written). They leave `list()` but stay readable by name and slug, and
# their versions, pins and mixture sources keep resolving. It cannot be undone.
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 with one label
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 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)
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

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

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.

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; a reverted append stays terminal until an explicit manual merge/redo.