API Annotation Pipeline - Avala Documentation

Automate your entire annotation lifecycle through the Avala API. This guide walks through the complete flow from raw data to exported annotations.

This guide requires a PRO or ENTERPRISE organization plan. Contact your Avala account manager to upgrade.

Prerequisites

An Avala account with an API key (see Authentication)
API key with scopes: datasets.read, datasets.write, projects.read, projects.write, exports.create, work_batches.read
A template project set up by your Avala team (defines annotation types, task config, and quality settings)

Overview

1. Upload Data     POST /datasets/manual-upload/
        |
2. Clone Template  POST /projects/create-from-existing/
        |
3. Approve         POST /projects/{uid}/approve/
        |
4. Monitor         GET  /work-batches/  +  Webhooks
        |
5. Export           POST /exports/

Step 1: Upload Your Data

Upload a dataset with files via presigned S3 URLs.

import requests

API_KEY = "YOUR_API_KEY"
BASE = "https://api.avala.ai/api/v1"
headers = {"X-Avala-Api-Key": API_KEY}

# Create dataset and get upload URLs
response = requests.post(
    f"{BASE}/datasets/manual-upload/",
    headers={**headers, "Content-Type": "application/json"},
    json={
        "name": "batch-2026-04-09",
        "data_type": "images",
        "files": [
            {"name": "frame_001.png", "size": 1024000},
            {"name": "frame_002.png", "size": 1024000},
        ],
    },
)
upload_data = response.json()
dataset_uid = upload_data["uid"]

# Upload each file to its presigned URL
for file_info in upload_data["files"]:
    with open(f"./data/{file_info['name']}", "rb") as f:
        requests.put(file_info["upload_url"], data=f)

For large uploads, use the Python SDK’s avala datasets upload CLI command which handles retries and parallelism automatically.

Step 2: Clone a Template Project

Create a new project by cloning a pre-configured template onto your dataset. The template defines the annotation types, task configuration, and quality settings.

# Clone template project onto your dataset
response = requests.post(
    f"{BASE}/projects/create-from-existing/",
    headers={**headers, "Content-Type": "application/json"},
    json={
        "source_project_uid": "TEMPLATE_PROJECT_UID",
        "name": f"batch-2026-04-09",
        "image_dataset_uid": dataset_uid,
    },
)
project = response.json()
project_uid = project["uid"]
# Status is "pending-approval"

Your Avala team will provide the source_project_uid for your template. Each template is pre-configured for your annotation type (bounding boxes, polygons, cuboids, etc.).

Step 3: Approve the Project

Approving starts annotation work. This triggers the full pipeline: task creation, work batches, and coworker assignment.

response = requests.post(
    f"{BASE}/projects/{project_uid}/approve/",
    headers=headers,
)

if response.status_code == 200:
    project = response.json()
    print(f"Project approved: {project['status']}")  # "active"
elif response.status_code == 403:
    error = response.json()
    if error.get("code") == "plan_insufficient":
        print("Upgrade to PRO or ENTERPRISE plan")
    elif error.get("code") == "scope_required":
        print("API key needs 'projects.write' scope")

Approving a project starts billable annotation work. Use pause or cancel to stop work after approval.

Step 4: Monitor Progress

Polling

Check project metrics to track annotation progress:

response = requests.get(
    f"{BASE}/projects/{project_uid}/metrics/",
    headers=headers,
)
metrics = response.json()
print(f"Progress: {metrics.get('completed_tasks', 0)}/{metrics.get('total_tasks', 0)}")

Webhooks (recommended)

Set up webhooks to receive real-time notifications instead of polling:

# Create a webhook subscription
requests.post(
    f"{BASE}/webhooks/",
    headers={**headers, "Content-Type": "application/json"},
    json={
        "target_url": "https://your-app.example.com/avala-webhook",
        "events": [
            "result.submitted",   # Annotator submits work
            "result.accepted",    # QC approves annotation
            "export.completed",   # Export is ready to download
        ],
    },
)

See Webhooks for event types, signature verification, and retry behavior.

Step 5: Export Results

Once annotations are complete, export them:

# Create an export
response = requests.post(
    f"{BASE}/exports/",
    headers={**headers, "Content-Type": "application/json"},
    json={
        "dataset_uid": dataset_uid,
        "format": "avala_lidar_labels",  # or "coco", "yolo", etc.
    },
)
export_uid = response.json()["uid"]

# Poll for completion (or use the export.completed webhook)
import time
while True:
    resp = requests.get(f"{BASE}/exports/{export_uid}/", headers=headers)
    export = resp.json()
    if export["status"] == "completed":
        print(f"Download: {export['download_url']}")
        break
    time.sleep(10)

Complete Example

Here is the full pipeline in one script:

import time
import requests

API_KEY = "YOUR_API_KEY"
BASE = "https://api.avala.ai/api/v1"
TEMPLATE_UID = "YOUR_TEMPLATE_PROJECT_UID"
headers = {"X-Avala-Api-Key": API_KEY}

# 1. Upload
upload = requests.post(
    f"{BASE}/datasets/manual-upload/",
    headers={**headers, "Content-Type": "application/json"},
    json={"name": "batch-2026-04-09", "data_type": "images", "files": [{"name": "frame.png", "size": 1024}]},
).json()

# 2. Clone
project = requests.post(
    f"{BASE}/projects/create-from-existing/",
    headers={**headers, "Content-Type": "application/json"},
    json={"source_project_uid": TEMPLATE_UID, "name": "batch-2026-04-09", "image_dataset_uid": upload["uid"]},
).json()

# 3. Approve
resp = requests.post(f"{BASE}/projects/{project['uid']}/approve/", headers=headers)
resp.raise_for_status()

# 4. Monitor (poll every 60s using project metrics)
while True:
    metrics = requests.get(f"{BASE}/projects/{project['uid']}/metrics/", headers=headers).json()
    total = metrics.get("total_tasks", 0)
    completed = metrics.get("completed_tasks", 0)
    if total > 0 and completed >= total:
        break
    print(f"Progress: {completed}/{total} tasks complete...")
    time.sleep(60)

# 5. Export
export = requests.post(
    f"{BASE}/exports/",
    headers={**headers, "Content-Type": "application/json"},
    json={"dataset_uid": upload["uid"], "format": "avala_lidar_labels"},
).json()

print(f"Export started: {export['uid']}")

Error Handling

Your pipeline should handle these error codes:

Code	HTTP	Meaning	Action
`plan_insufficient`	403	Org needs PRO/ENTERPRISE plan	Contact Avala to upgrade
`scope_required`	403	API key missing required scope	Create new key with correct scopes
—	400	Invalid state transition	Check project status before approving
—	404	Resource not found	Verify UIDs are correct
—	429	Rate limited	Retry with exponential backoff

See Error Codes for the full reference.

Next Steps

Webhooks for real-time event notifications
Python SDK for a higher-level client
Authentication for API key scopes and management

​Prerequisites

​Overview

​Step 1: Upload Your Data

​Step 2: Clone a Template Project

​Step 3: Approve the Project

​Step 4: Monitor Progress

​Polling

​Webhooks (recommended)

​Step 5: Export Results

​Complete Example

​Error Handling

​Next Steps

Prerequisites

Overview

Step 1: Upload Your Data

Step 2: Clone a Template Project

Step 3: Approve the Project

Step 4: Monitor Progress

Polling

Webhooks (recommended)

Step 5: Export Results

Complete Example

Error Handling

Next Steps