Push a dataset to New Theory

This page is for engineers integrating an external tool — a recording kit, an exporter, a data pipeline — that produces a LeRobot dataset and needs to get it into New Theory so newt finetune can train on it. The developer authenticates with their own nt_ API key; no infrastructure credential ever leaves New Theory.

There are two doors, and they end in the same place — a dataset staged under the developer's account, referenced by name:

• The CLI door. Shell out to newt finetune --dataset ./folder. One call uploads the local export and launches the run. Reach for this first.
• The HTTP door. Sign, PUT, verify, launch — the raw request sequence, for a native integration that doesn't want to depend on the CLI.

The CLI door

If your tool has the exported dataset folder on disk, the whole hand-off is one subprocess call. newt finetune detects that the argument is a local folder (it contains a path separator, or resolves to an existing directory), uploads it under the developer's account, and launches training against the staged name:

newt finetune --dataset ./datasets/my_task

The CLI validates the export locally before a byte moves, uploads every file in one signing round-trip, then launches and watches the run:

Detected a local folder — uploading 214 MB… staged as my_task
  … 342/342 files (214 MB / 214 MB) 100%
  uploaded 214 MB, staged as my_task
Launched fine-tune on dataset 'my_task'.
  job handle:   <handle>
  watch page:   https://newtheory-console.vercel.app/runs/<handle>

The staged name is the folder's basename (my_task above), so it must be a valid name: letters, digits, ., _, -, up to 128 characters. Rename the folder if it isn't.

The developer's key is read from NT_API_KEY, or from ~/.nt/credentials after newt login. Install the CLI with pip install "git+https://github.com/new-theory-research/newt-python.git".

If your integration is already Python and you want to upload without launching — for example, to stage a dataset now and let the developer launch later — call the upload primitive directly:

from newt.recording import NTCloudSink

sink = NTCloudSink("my_task", api_key="nt_…")
namespace = sink.upload_directory("./datasets/my_task")

upload_directory validates the export, signs every file in one round-trip, PUTs each one, and writes a completeness marker last. It raises on any failure rather than leaving a partial upload that looks complete. This needs the recording extra: pip install "newt[recording]".

What the dataset must contain

Both doors upload a LeRobot v3 dataset directory. Training reads the dataset's schema from meta/info.json; the intake gate rejects a dataset that doesn't meet the contract before spending any GPU time.

The directory layout:

my_task/
  meta/
    info.json         # schema: features, shapes, robot_type
    tasks.parquet     # the task strings, one per task index
    stats.json        # per-feature normalization stats (quantiles)
  data/
    chunk-000/
      file-000.parquet   # per-frame robot state + action
  videos/
    chunk-000/
      observation.images.cam0/…   # camera video, when features declare dtype "video"

This is a standard LeRobot v3 export — the same layout lerobot writes and the Rerun kit produces. Training reads all of it: the frame data from data/, the camera video from videos/, the instructions from meta/tasks.parquet, and the normalization stats from meta/stats.json. Export it complete; upload it as-is.

meta/info.json is the contract. Its features map is what the intake gate checks:

{
  "codebase_version": "v3.0",
  "robot_type": "so101",
  "features": {
    "observation.images.cam0": {"dtype": "video", "shape": [480, 640, 3], "names": ["height", "width", "channels"]},
    "observation.images.cam1": {"dtype": "video", "shape": [480, 640, 3], "names": ["height", "width", "channels"]},
    "observation.state": {"dtype": "float32", "shape": [6], "names": ["state"]},
    "action": {"dtype": "float32", "shape": [6], "names": ["action"]}
  }
}

Before it spends any GPU time, the intake gate reads meta/info.json and requires:

• features contains action, observation.state, and at least one observation.images.* camera.
• action and observation.state both have shape [6] — the SO-101 joint dimension.
• Cameras it can map to cam0/cam1 — it derives the rename from the dataset; a non-standard name is remapped, not rejected.

A dataset that misses the first two fails the run at gate: intake, with a message naming what tripped, before any GPU is spent. The gate also reads meta/tasks.parquet and rejects a dataset whose every episode's task string is empty, whitespace, or the exporter's placeholder.

The task strings must be real instructions, not the Rerun exporter's literal "task" placeholder — a policy trained on the placeholder learns nothing. Export with a real instruction (--task <instruction> when you export) and the right normalization stats (meta/stats.json with quantiles); the New Theory intake gate refuses a taskless dataset outright, so this is enforced twice.

The HTTP door

For a native integration, the sequence is four requests. Every request authenticates with the developer's key:

Authorization: Bearer nt_a1b2c3d4…

A real key is nt_ followed by 40 characters. A missing or malformed key is 401. All requests are against the console at https://newtheory-console.vercel.app.

1. Sign

POST /api/uploads/sign mints one short-lived upload URL per file, in a single round-trip. Send the dataset name and the list of file paths, each relative to the dataset directory:

POST /api/uploads/sign
Authorization: Bearer nt_…
Content-Type: application/json

{
  "dataset": "my_task",
  "paths": [
    "meta/info.json",
    "meta/tasks.parquet",
    "data/chunk-000/file-000.parquet",
    "videos/chunk-000/observation.images.cam0/file-000.mp4"
  ]
}

Response:

{
  "namespace": "u_9f3c…",
  "dataset": "my_task",
  "count": 4,
  "urls": [
    {
      "path": "meta/info.json",
      "url": "https://storage.example/upload-url-issued-by-new-theory…",
      "objectPath": "…",
      "expiresAt": "2026-07-15T18:42:00.000Z"
    }
  ]
}

Each entry pairs the path you sent with the url to PUT it to. expiresAt is when the URL stops working — minutes, not hours, so sign and upload in the same pass. objectPath is New Theory's internal storage path for the object; you don't need it — the top-level namespace is the durable handle for the upload.

Constraints:

• At most 1000 paths per request. More than that is 400 {"error": "bad_request", "max_batch_paths": 1000} — split into multiple sign calls. The list is never silently truncated.
• Each path is a /-joined set of safe segments (letters, digits, ., _, -), up to 512 characters. No .., no leading /. A malformed path is 400 {"error": "bad_request"}.
• The dataset name is one safe segment, up to 128 characters.
• The upload lands under the key owner's namespace, derived server-side. You cannot name it, and one developer's key can never write into another's.
• The dataset name must not already exist in your namespace. Uploads are create-only — signing against a name you've already staged (complete or partial) is 409 {"error": "dataset_name_exists", ...}, and no URLs are minted. Re-upload under a new name, e.g. my_task_v2.

There is also a single-file shape — {"dataset": "…", "path": "…"} → {"url", "objectPath", "expiresAt"} — but the batch shape above is what you want for a dataset directory: one round-trip instead of one per file.

2. Upload

PUT each file's bytes to its signed url. The content type is fixed:

PUT <url>
Content-Type: application/octet-stream

<raw file bytes>

Content-Type: application/octet-stream is required. It is bound into the URL's signature — a PUT with any other content type fails the signature check. No Authorization header goes on the PUT; the signed URL carries its own authorization. A 2xx means the object landed. Bytes never pass through the console — they go straight to storage over the signed URL.

3. Verify

GET /api/uploads/list?dataset=<name> lists what landed under the developer's account, so you can confirm every file arrived before launching:

GET /api/uploads/list?dataset=my_task
Authorization: Bearer nt_…

{
  "namespace": "u_9f3c…",
  "count": 4,
  "objects": [
    {
      "objectPath": "…",
      "path": "my_task/meta/info.json",
      "size": 4360,
      "updated": "2026-07-15T18:41:12.000Z"
    }
  ]
}

Here path includes the dataset name (my_task/meta/info.json) — it is the object's path within the developer's namespace. Match count against the number of files you uploaded. The list is scoped to the key owner's namespace; a developer only ever sees their own uploads.

4. Launch

POST /api/finetune launches training against the staged dataset name and returns a job handle. Training runs on New Theory's GPUs under the developer's key:

POST /api/finetune
Authorization: Bearer nt_…
Content-Type: application/json

{ "dataset": "my_task" }

{
  "job_handle": "<handle>",
  "dataset": "my_task",
  "uid": "<model-uid>",
  "status": "launched"
}

Optional body fields:

5. Poll (optional)

GET /api/finetune/status?job_handle=<handle> reports the run's state. Poll it until status is terminal:

GET /api/finetune/status?job_handle=<handle>
Authorization: Bearer nt_…

{ "status": "succeeded", "gate": null, "detail": null, "tag": "<model-tag>", "report_card": null, "model_status": "pending" }

status is launched / running while the run is going, then succeeded or failed. On failed, gate names the step that stopped it (intake, train, frame-check, …) and detail carries the pipeline's human-readable failure cause. model_status reports the trained model's admission state — pending, probed, live, or dead. A succeeded run is not servable until admission reaches live; the tag resolves for inference only then. Training runs for hours — poll on a slow interval (every 15 seconds is what the CLI uses).

Errors

End-to-end example

A complete native integration in Python, standard library plus requests. It uploads a dataset directory, verifies the upload, launches training, and polls to completion. Everything it needs is an nt_ key and a folder path.

import os
import sys
import time
from pathlib import Path

import requests

CONSOLE = "https://newtheory-console.vercel.app"
API_KEY = os.environ["NT_API_KEY"]        # nt_ + 40 chars
DATASET = "my_task"                        # staged name (folder basename)
EXPORT_DIR = Path("./datasets/my_task")    # LeRobot v3 dataset directory
AUTH = {"Authorization": f"Bearer {API_KEY}"}


def push_and_launch() -> str:
    # 1. Collect every file, as paths relative to the dataset directory.
    files = sorted(
        p.relative_to(EXPORT_DIR).as_posix()
        for p in EXPORT_DIR.rglob("*")
        if p.is_file()
    )
    if not files:
        sys.exit(f"no files under {EXPORT_DIR}")

    # 2. Sign every file in one round-trip (up to 1000 per call).
    signed = requests.post(
        f"{CONSOLE}/api/uploads/sign",
        headers={**AUTH, "Content-Type": "application/json"},
        json={"dataset": DATASET, "paths": files},
    )
    signed.raise_for_status()
    urls = {entry["path"]: entry["url"] for entry in signed.json()["urls"]}

    # 3. PUT each file to its signed URL — content type MUST be octet-stream.
    for rel in files:
        with open(EXPORT_DIR / rel, "rb") as fh:
            put = requests.put(
                urls[rel],
                data=fh.read(),
                headers={"Content-Type": "application/octet-stream"},
            )
        put.raise_for_status()
        print(f"uploaded {rel}")

    # 4. Verify the whole set landed before launching.
    listed = requests.get(
        f"{CONSOLE}/api/uploads/list",
        headers=AUTH,
        params={"dataset": DATASET},
    )
    listed.raise_for_status()
    landed = listed.json()["count"]
    if landed < len(files):
        sys.exit(f"only {landed}/{len(files)} files landed — not launching")

    # 5. Launch training against the staged name.
    launched = requests.post(
        f"{CONSOLE}/api/finetune",
        headers={**AUTH, "Content-Type": "application/json"},
        json={"dataset": DATASET},   # add "steps"/"name" here to override defaults
    )
    launched.raise_for_status()
    return launched.json()["job_handle"]


def watch(handle: str) -> None:
    while True:
        status = requests.get(
            f"{CONSOLE}/api/finetune/status",
            headers=AUTH,
            params={"job_handle": handle},
        ).json()
        state = status["status"]
        if state == "succeeded":
            model_status = status.get("model_status")
            print(f"trained — model tag: {status.get('tag')} (admission: {model_status})")
            if model_status != "live":
                print("not servable yet — the tag resolves once admission reaches 'live'")
            return
        if state == "failed":
            sys.exit(f"failed at gate: {status.get('gate')} — {status.get('detail')}")
        print(f"… {state}")
        time.sleep(15)


if __name__ == "__main__":
    handle = push_and_launch()
    print(f"launched: {handle}")
    watch(handle)

Where to go next

• CLI reference — every newt finetune flag, the --json output shape, and exit codes.
• Rerun SO-101 hackathon kit — the record, calibrate, and export reference the dataset comes from.