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.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.

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 = UPLOAD") replaces it and returns qualia.v1.Dataset. 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/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 (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)

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/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. 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) or a mixture (kept up to date from other datasets). 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. 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")

# 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

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

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.

Requirements

• Python 3.10+
• httpx
• pydantic

License

MIT