Guides

Workflows REST API

The UniteLabs REST API lets you manage the full workflow lifecycle programmatically — create and deploy workflows, trigger runs, monitor progress, and interact with paused runs.

Prerequisites

A UniteLabs account with a tenant ID and API credentials (client ID + secret)

Authentication

All API requests require a Bearer token obtained via OAuth2:

Terminal

curl -X POST "https://auth.unitelabs.io/realms/{tenant-id}/protocol/openid-connect/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials&client_id={client-id}&client_secret={client-secret}"

auth.py

import httpx

token_response = httpx.post(
    "https://auth.unitelabs.io/realms/{tenant-id}/protocol/openid-connect/token",
    data={
        "grant_type": "client_credentials",
        "client_id": "{client-id}",
        "client_secret": "{client-secret}",
    },
)
token = token_response.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}

All examples below assume BASE_URL = "https://api.unitelabs.io/{tenant-id}/v1" and headers set as above.

Workflows

Create a workflow

Terminal

curl -X POST "$BASE_URL/workflows" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "plate-qc-pipeline",
    "description": "QC check before dispensing",
    "entrypoint": "workflow.py:entrypoint",
    "dependencies": "pandas,httpx",
    "tags": ["qc", "production"],
    "enabled": true
  }'

Terminal

curl -X POST "$BASE_URL/workflows" \
  -H "Authorization: Bearer $TOKEN" \
  -F "name=plate-qc-pipeline" \
  -F "entrypoint=workflow.py:entrypoint" \
  -F "file=@workflow.zip"

create_workflow.py

response = httpx.post(
    f"{BASE_URL}/workflows",
    headers=headers,
    json={
        "name": "plate-qc-pipeline",
        "description": "QC check before dispensing",
        "entrypoint": "workflow.py:entrypoint",
        "dependencies": "pandas,httpx",
        "tags": ["qc", "production"],
        "enabled": True,
    },
)
workflow_id = response.json()["id"]

Response 201:

{
  "id": "d81023e8-87ab-4c9e-8394-0fc10ab562f3",
  "name": "plate-qc-pipeline",
  "entrypoint": "workflow.py:entrypoint",
  "dependencies": "pandas,httpx",
  "tags": ["qc", "production"],
  "enabled": true
}

List workflows

Terminal

# Most recent 10, sorted by creation date
curl "$BASE_URL/workflows?_take=10&_sort=-createdAt" \
  -H "Authorization: Bearer $TOKEN"

# Filter by name (partial match)
curl "$BASE_URL/workflows?name[like]=plate" \
  -H "Authorization: Bearer $TOKEN"

# Include recent runs in the response
curl "$BASE_URL/workflows?_include[]=recentRuns" \
  -H "Authorization: Bearer $TOKEN"

Paginate using the cursor from the previous response:

Terminal

curl "$BASE_URL/workflows?_take=20&_cursor=eyJpZCI6Ii4uLiJ9" \
  -H "Authorization: Bearer $TOKEN"

Get, update, delete

Terminal

# Get
curl "$BASE_URL/workflows/{workflowId}" -H "Authorization: Bearer $TOKEN"

# Update (partial — only include fields to change)
curl -X PATCH "$BASE_URL/workflows/{workflowId}" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"enabled": false, "tags": ["qc", "staging"]}'

# Delete
curl -X DELETE "$BASE_URL/workflows/{workflowId}" -H "Authorization: Bearer $TOKEN"
# Returns 204 No Content

Upload workflow files

Add or update individual source files without redeploying the whole workflow:

Terminal

# Create a file
curl -X POST "$BASE_URL/workflows/{workflowId}/files" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "workflow.py",
    "path": "/",
    "type": "file",
    "content": "from unitelabs.sdk.automate import workflow\n\n@workflow  # workflow\nasync def entrypoint():\n    pass\n"
  }'

# Update a file
curl -X PATCH "$BASE_URL/workflows/{workflowId}/files/workflow.py" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"content": "# updated content\n"}'

# List all files
curl "$BASE_URL/workflows/{workflowId}/files" -H "Authorization: Bearer $TOKEN"

Runs

Create a run

Start a workflow by its ID. Pass parameters to supply runtime values to the workflow function:

Terminal

curl -X POST "$BASE_URL/workflows/{workflowId}/runs" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "QC Run — Plate B3",
    "parameters": {
      "sample_id": "B3",
      "min_concentration": 250.0
    }
  }'

trigger_run.py

response = httpx.post(
    f"{BASE_URL}/workflows/{workflow_id}/runs",
    headers=headers,
    json={
        "name": "QC Run — Plate B3",
        "parameters": {
            "sample_id": "B3",
            "min_concentration": 250.0,
        },
    },
)
run_id = response.json()["id"]

Response 201:

{
  "id": "a3bb189e-8bf9-3888-9912-ace4e6543002",
  "workflowId": "d81023e8-87ab-4c9e-8394-0fc10ab562f3",
  "name": "QC Run — Plate B3",
  "status": "SCHEDULED",
  "statusSince": "2025-01-15T11:00:00.000Z",
  "parameters": { "sample_id": "B3", "min_concentration": 250.0 },
  "createdAt": "2025-01-15T11:00:00.000Z"
}

List and get runs

Terminal

# All runs for a workflow, newest first
curl "$BASE_URL/workflows/{workflowId}/runs?_sort=-createdAt" \
  -H "Authorization: Bearer $TOKEN"

# Filter by date range
curl "$BASE_URL/runs?created_at[gte]=2025-01-01T00:00:00Z&created_at[lte]=2025-01-31T23:59:59Z" \
  -H "Authorization: Bearer $TOKEN"

# Get a specific run
curl "$BASE_URL/runs/{runId}" -H "Authorization: Bearer $TOKEN"

Monitoring a run

Check status

Terminal

curl "$BASE_URL/runs/{runId}/status" -H "Authorization: Bearer $TOKEN"

Status	Description
`SCHEDULED`	Queued, not yet started
`RUNNING`	Actively executing
`PAUSED`	Suspended — waiting for operator action
`AWAITING_INPUT`	Suspended — waiting for structured data input
`COMPLETED`	Finished successfully
`FAILED`	Finished with an unhandled error
`CANCELLED`	Cancelled before completion

Execution timeline and logs

Terminal

# Phase and step timeline with timestamps
curl "$BASE_URL/runs/{runId}/timeline" -H "Authorization: Bearer $TOKEN"

# Execution logs
curl "$BASE_URL/runs/{runId}/logs" -H "Authorization: Bearer $TOKEN"

# Artifacts produced by the run
curl "$BASE_URL/runs/{runId}/artifacts" -H "Authorization: Bearer $TOKEN"

Poll until complete

poll.py

import time
import httpx

def wait_for_run(run_id: str, headers: dict, base_url: str, poll_interval: int = 5) -> dict:
    terminal = {"COMPLETED", "FAILED", "CANCELLED"}
    while True:
        run = httpx.get(f"{base_url}/runs/{run_id}", headers=headers).json()
        print(f"[{run['statusSince']}] {run['status']}")
        if run["status"] in terminal:
            return run
        time.sleep(poll_interval)

run = wait_for_run(run_id, headers, BASE_URL)
print(f"Run finished: {run['status']}")

Interacting with run states

Use POST /v1/runs/{runId}/status to drive a run externally — resume, provide input, or cancel.

Resume a paused run

Terminal

curl -X POST "$BASE_URL/runs/{runId}/status" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"status": "RUNNING"}'

Provide input on resume

When the run is in AWAITING_INPUT, include a data object keyed by phase ID:

Terminal

curl -X POST "$BASE_URL/runs/{runId}/status" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "status": "RUNNING",
    "data": {
      "quality_check": {
        "sample_id": "B3",
        "approved": true
      }
    }
  }'

Cancel a run

Terminal

curl -X POST "$BASE_URL/runs/{runId}/status" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"status": "CANCELLED"}'

For the workflow side — how to pause a run and wait for the resume signal from within your Python code — see Human in the loop.

Next steps

Run a workflow: trigger from the UI or UniteLabs SDK
Deploy a workflow: package and register a workflow before running it

Advanced error handling

Build a structured error taxonomy with typed error codes, user-facing messages, and automatic exception classification.

Overview

How UniteLabs handles data storage, secrets, and connectivity to external systems.