Call the workflow run API

Once a workflow is deployed and active, call it with one image per request.

Endpoint

With a project API key, the project is inferred from the key:

POST /api/v1/workflows/{workflow_id}/run
Authorization: Bearer lty_…
Content-Type: application/json

Session users use the project-scoped path:

POST /api/v1/projects/{project_id}/workflows/{workflow_id}/run

There is no separate public hostname per workflow. The “endpoint” is Labelty’s API.

Request body

Provide one of:

Field	Description
image_base64	Base64-encoded image bytes
image_s3_key	Object key already in Labelty storage

Optional:

Field	Description
metadata	Arbitrary JSON object passed through to WorkflowInput.metadata

image_url and multipart upload are not supported on the current API.

cURL example

BASE="https://api.development.labelty.ai"
WID="<workflow_id>"
KEY="lty_..."

# Linux: base64 -w0
# macOS: base64 -i path/to/image.jpg
IMAGE_B64=$(base64 -w0 path/to/image.jpg)

curl -X POST "$BASE/api/v1/workflows/$WID/run" \
  -H "Authorization: Bearer $KEY" \
  -H "Content-Type: application/json" \
  --max-time 120 \
  -d "{\"image_base64\": \"$IMAGE_B64\", \"metadata\": {\"camera_id\": \"cam-4\"}}"

Why --max-time 120: the first cold call can take ~60s while the runner warms and installs heavy model dependencies.

Python example

import base64
import requests

BASE_URL = "https://api.development.labelty.ai"
WORKFLOW_ID = "<workflow_id>"
API_KEY = "lty_..."

with open("path/to/image.jpg", "rb") as f:
    image_b64 = base64.b64encode(f.read()).decode()

resp = requests.post(
    f"{BASE_URL}/api/v1/workflows/{WORKFLOW_ID}/run",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "image_base64": image_b64,
        "metadata": {"camera_id": "cam-4"},
    },
    timeout=120,
)
resp.raise_for_status()
print(resp.json())

JavaScript (Node) example

import { readFile } from "node:fs/promises";

const BASE_URL = "https://api.development.labelty.ai";
const WORKFLOW_ID = "<workflow_id>";
const API_KEY = "lty_...";

const imageB64 = (await readFile("path/to/image.jpg")).toString("base64");

const res = await fetch(`${BASE_URL}/api/v1/workflows/${WORKFLOW_ID}/run`, {
  method: "POST",
  headers: {
    Authorization: `Bearer ${API_KEY}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({ image_base64: imageB64 }),
  signal: AbortSignal.timeout(120_000),
});

if (!res.ok) {
  throw new Error(`Run failed: ${res.status} ${await res.text()}`);
}
console.log(await res.json());

Success response shape

{
  "message": "Workflow run complete",
  "data": {
    "predictions": [
      {
        "label_id": "<uuid>",
        "points": [[0.1, 0.2], [0.4, 0.2], [0.4, 0.5], [0.1, 0.5]],
        "metadata": { "confidence": 0.91, "model_name": "detector" },
        "name": "crack",
        "color": "#FF0000",
        "is_point": false
      }
    ],
    "routes": ["critical"],
    "route_deliveries": {
      "critical": { "status": "ok" }
    },
    "billing": {
      "duration_seconds": 1.234,
      "charged_usd": 0.01
    }
  }
}

How to read it:

• predictions — what your downstream system should consume
• routes — named destinations selected by workflow.py
• route_deliveries — whether DatasetOutput ingest succeeded; delivery errors do not drop the prediction payload
• billing — duration-based charge details when present

Common errors

Code	Meaning
400	Missing image fields; invalid model binding
403	Wrong auth for the route; plan lacks API access
404	Workflow / model / project not found
409	Workflow inactive or missing uploaded artifact
429	Another run already in progress on this backend process
502 / 503	Runner misconfigured or invoke failed

Monitoring

What exists today:

• Synchronous HTTP response (predictions + routes + billing)
• Wallet ledger entries for workflow runs
• In-app test panel for ad-hoc runs
• Ops logs in the runner / CloudWatch

What does not exist yet:

• Durable workflow_runs history API
• Webhooks for completion
• Guaranteed multi-tenant concurrency beyond process-local single-flight

Best practices

• Generate API keys per environment (dev/stage/prod) and rotate if leaked.
• Always set a client timeout ≥ 120s for cold starts.
• Keep workflow.py thin — put heavy ML in the model inference package.
• Treat DatasetOutput as an active-learning loop into Raw Data, not as your primary API response.