Automate with the REST API

Everything you do on screen can also be done through the /api/v1/… REST API. The screens use the same API. This page covers only the rules common to all APIs and leaves per-endpoint requests and responses to the Reference. The server's OpenAPI document is also at /openapi.json, and interactive docs are at /docs.

Every example was actually run against the capture stack (http://localhost:10000, tenant DEMO, account [email protected]). In your environment, change only the address and the account.

Authentication — cookie or JWT#

There are two ways to log in with the same account and password. Both take the form fields username (email) and password.

• Session lifetime is set by the server setting GEO_MLOPS_AUTH_TOKEN_LIFETIME (seconds). The default 0 means no expiry. In production, set a reasonable lifetime and use a dedicated account for automation.
• Log out with POST /auth/cookie/logout or POST /auth/jwt/logout.
• MLflow tokens and container tokens cannot be used for the REST API. They are only for /mlflow and /v2 respectively.

Selecting the tenant — X-Tenant#

An account can belong to several tenants, so tenant-scoped APIs must be told which tenant on every request, with the X-Tenant header or the ?tenant= query (case-insensitive; the header wins).

curl -sS -b cookies.txt "$API/api/v1/datasets?page_size=2"
# {"error_id":"fb61…","code":"bad_request","message":"tenant context required (X-Tenant header or ?tenant=)","detail":null}

curl -sS -b cookies.txt -H "X-Tenant: DEMO" "$API/api/v1/datasets?page_size=2"
# {"items":[…],"total":9,"page":1,"page_size":2}

Permissions are judged by your role in that tenant. An insufficient role gets 403; resources of a tenant you do not belong to get 404. Per-person APIs such as /users/me and /api/v1/stream/notifications need no tenant.

Error format#

Every failure has the same shape.

{"error_id": "7bea34c0…", "code": "bad_request", "message": "unknown sort 'bogus'",
 "detail": {"allowed": ["created_at", "name", "records", "size", "validation"]}}

Pagination — two kinds#

Numbered pages (most lists)#

Send page (starting at 1) and page_size, and you get {items, total, page, page_size}. Without page_size, the server default is used. Most lists also take search and sort under the same names.

curl -sS -b cookies.txt -H "X-Tenant: DEMO" \
  "$API/api/v1/datasets?page=2&page_size=2&sort=name&order=asc"
# {"items":[…2 items…],"total":9,"page":2,"page_size":2}

Cursors (time-ordered feeds)#

Feeds that keep piling up newest first, such as edge device logs, telemetry and inference records, use cursors. Send limit and cursor, and pass the response's next_cursor as the cursor of the next request. When next_cursor is null, you have reached the end. In these feeds total is not the overall count but the number of items in this response.

curl -sS -b cookies.txt -H "X-Tenant: DEMO" \
  "$API/api/v1/edge/devices/edge-demo-01/logs?limit=2"
# {"items":[…2 items…],"total":2,"next_cursor":"Mg=="}
curl -sS -b cookies.txt -H "X-Tenant: DEMO" \
  "$API/api/v1/edge/devices/edge-demo-01/logs?limit=2&cursor=Mg=="
# {"items":[…],"total":2,"next_cursor":"NA=="}

Treat cursors as opaque strings. Their shape may change.

Real-time streams — SSE#

Progress and notifications arrive as Server-Sent Events (text/event-stream), where the server keeps the connection open and keeps sending events. When you connect, a connected event comes first, then an event each time something happens.

curl -sS -N -b cookies.txt -H "X-Tenant: DEMO" "$API/api/v1/stream/alerts"
# event: connected
# data: {"channel": "alerts:DEMO"}

• Past events are not sent again. Read the current state through REST first, then attach the stream. For training curves, fetch everything through the metrics API and append the metrics increments.
• Event bodies are thin — about "what changed". Read the details again through REST.
• A browser EventSource cannot add headers. Log in with a cookie and pass the tenant as the ?tenant=DEMO query.
• If the connection drops, just reconnect. The operator must configure the front proxy not to cut long connections (turn off buffering, raise the time limit).

Python example — login · walking pages · SSE#

import os
import requests

API = os.environ.get("API", "http://localhost:10000")

# 1) Log in — get a JWT and use it in the Authorization header
r = requests.post(f"{API}/auth/jwt/login",
                  data={"username": os.environ["EMAIL"], "password": os.environ["PASSWORD"]})
r.raise_for_status()
s = requests.Session()
s.headers["Authorization"] = f"Bearer {r.json()['access_token']}"
s.headers["X-Tenant"] = "DEMO"          # required by every tenant-scoped API

# 2) Numbered pagination — walk to the end
page, names = 1, []
while True:
    body = s.get(f"{API}/api/v1/datasets",
                 params={"page": page, "page_size": 50, "sort": "name"}).json()
    names += [d["name"] for d in body["items"]]
    if page * body["page_size"] >= body["total"]:
        break
    page += 1
print(len(names), "datasets")

# 3) SSE — read a few events and close
with s.get(f"{API}/api/v1/stream/alerts", stream=True, timeout=(5, 30)) as resp:
    event = None
    for line in resp.iter_lines(decode_unicode=True):
        if line.startswith("event:"):
            event = line.split(":", 1)[1].strip()
        elif line.startswith("data:"):
            print(event, line.split(":", 1)[1].strip())
            break                      # this is an example, so stop at the first event

9 datasets
connected {"channel": "alerts:DEMO"}

Chunked upload#

Large files are not sent in one request, because they would hit the front proxy's body size and response time limits. The platform uses the same pattern — open a session → send chunks → finish — in two places. Finishing returns 202 immediately, and the server carries on with assembly and validation, so read the state again to confirm it has ended.

Use the chunk size the server returned. A larger chunk gets 413.

"""Uploads one file to a dataset in chunks (REST API example)."""

import hashlib
import os
import sys
import time

import requests

API = os.environ.get("API", "http://localhost:10000")
TENANT = os.environ.get("TENANT", "DEMO")
dataset_id, path = sys.argv[1], sys.argv[2]
size = os.path.getsize(path)
sha256 = hashlib.sha256(open(path, "rb").read()).hexdigest()

s = requests.Session()
s.headers["X-Tenant"] = TENANT
s.post(f"{API}/auth/cookie/login",
       data={"username": os.environ["EMAIL"],
             "password": os.environ["PASSWORD"]}).raise_for_status()

# 1) Open a session — the server decides chunk_size and returns it
r = s.post(f"{API}/api/v1/datasets/{dataset_id}/uploads",
           json={"filename": os.path.basename(path), "size": size, "sha256": sha256})
r.raise_for_status()
up = r.json()
upload_id, chunk = up["upload_id"], up["chunk_size"]

# 2) Send chunks — they are sent by index, so order does not matter and resending a chunk is fine
with open(path, "rb") as f:
    index = 0
    while data := f.read(chunk):
        s.put(f"{API}/api/v1/datasets/{dataset_id}/uploads/{upload_id}/chunks/{index}",
              data=data,
              headers={"Content-Type": "application/octet-stream"}).raise_for_status()
        index += 1

# 3) Finish — returns 202 immediately; the server carries on with assembly and validation
s.post(f"{API}/api/v1/datasets/{dataset_id}/uploads/{upload_id}:complete").raise_for_status()
while True:
    st = s.get(f"{API}/api/v1/datasets/{dataset_id}/uploads/{upload_id}").json()
    if st["state"] in ("done", "failed"):
        print(st["state"], st.get("result") or st.get("error"))
        break
    time.sleep(1)

EMAIL=[email protected] PASSWORD='<your-password>' \
  python3 upload_file.py ds-d3dd6730ad13 20260721_line3_0002.png
# done {'created': 1, 'skipped': 0, 'file_ids': ['b2597c97-…']}

"""Imports an image tar into the registry in chunks (REST API example)."""

import os
import sys
import time

import requests

API = os.environ.get("API", "http://localhost:10000")
TENANT = os.environ.get("TENANT", "DEMO")
path = sys.argv[1]
size = os.path.getsize(path)

s = requests.Session()
s.headers["X-Tenant"] = TENANT
# 1) Log in — cookie session (for JWT, call /auth/jwt/login and use the Authorization header)
r = s.post(f"{API}/auth/cookie/login",
           data={"username": os.environ["EMAIL"], "password": os.environ["PASSWORD"]})
r.raise_for_status()

# 2) Create an import session — announce the total size first
r = s.post(f"{API}/api/v1/registry/imports",
           json={"size_bytes": size, "filename": os.path.basename(path)})
r.raise_for_status()
job = r.json()
chunk = job["chunk_size"]
print("import", job["id"], "chunk", chunk)

# 3) Send chunks in order — Content-Range gives the position
with open(path, "rb") as f:
    offset = 0
    while offset < size:
        data = f.read(chunk)
        end = offset + len(data) - 1
        r = s.patch(
            f"{API}/api/v1/registry/imports/{job['id']}/chunks",
            data=data,
            headers={
                "Content-Type": "application/octet-stream",
                "Content-Range": f"bytes {offset}-{end}/{size}",
            },
        )
        if r.status_code == 416:  # restart from the position the server has
            offset = r.json()["detail"]["offset"]
            f.seek(offset)
            continue
        r.raise_for_status()
        offset = r.json()["received_bytes"]
        print(f"\r{offset}/{size}", end="", flush=True)
print()

# 4) Start the import — returns immediately; the server carries on with registration
s.post(f"{API}/api/v1/registry/imports/{job['id']}:start").raise_for_status()
while True:
    row = s.get(f"{API}/api/v1/registry/imports/{job['id']}").json()
    print(row["status"], row["phase"])
    if row["status"] in ("READY", "FAILED", "CANCELED"):
        print(row["image"] or row["error"])
        break
    time.sleep(3)

EMAIL=[email protected] PASSWORD='<your-password>' \
  python3 import_image.py mnist-trainer-v1.tar.gz
# import <import id> chunk 33554432
# 521930649/521930649
# RUNNING assemble
# RUNNING inspect
# READY done
# localhost:10000/demo/example/mnist-trainer:v1

See also#

• Full endpoint list: Reference
• On-screen procedure for importing a training container and registering a variant: Upload the image to the platform
• External data source API: DataOps integration