> For the complete documentation index, see [llms.txt](https://docs.unitlab.ai/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.unitlab.ai/api-sdk-cli/sdks/pattern-workflow-worker.md).

# Pattern: Workflow worker

Run your own code on every task of an Agent stage: read its annotations, save the result and complete the run.

## Agent stage worker

An [Agent stage](https://docs.unitlab.ai/documentation/workflows/agent-stage) sends each task to your service instead of waiting for someone to claim it. The webhook carries only IDs: your service reads the task's items with the SDK, saves new annotations, and completes the run.

### Example: refine building outlines

A Model stage outlined the buildings on a satellite image. Your service passes each item's annotations to your own model (`refine` below) and saves the result, and the task moves on to the next stage, for example Review. This example is a FastAPI app with one route, `POST /webhook`: set the stage's **HTTPS URL** to `https://<your host>/webhook`.

Install the packages with `pip install -U unitlab fastapi uvicorn`, save the code as `main.py`, and run it with `uvicorn main:app --host 0.0.0.0 --port 8091`.

```python
import os

from fastapi import BackgroundTasks, FastAPI, Request, Response
from unitlab import AgentWebhook, UnitlabClient, ValidationError

SECRET = os.environ["UNITLAB_AGENT_WEBHOOK_SECRET"]
client = UnitlabClient()  # reads UNITLAB_API_KEY
app = FastAPI()


def refine(result):
    ...


def work(task_id, run_id):
    try:
        task = client.get_workflow_task(task_id)
        for item in task.get_annotations():
            task.save_annotations(item, refine(item.document))
        task.complete_agent(run_id)
    except Exception as error:
        # read the task again: the first read may be what failed
        try:
            client.get_workflow_task(task_id).fail_agent(run_id, str(error))
        except ValidationError:
            pass  # a manager stopped the run or moved the task out: drop it


@app.post("/webhook")
async def webhook(request: Request, background: BackgroundTasks):
    body = await request.body()
    try:
        event = AgentWebhook.verify(body, request.headers, SECRET)
    except ValueError:
        return Response(status_code=401)  # wrong secret: Test connection says so
    if event.is_test:  # Test connection: any 2xx passes
        return Response(status_code=200)
    background.add_task(work, event.task.task_id, event.task.run_id)
    return Response(status_code=202)  # answer at once, the work runs after
```

* `task.get_annotations()` returns one entry per item (one per panel of a Data Group task) with `datasource_id`, `generic_type`, `source_url` (a link to the original file, valid for 24 hours; keep it out of logs) and `document`: the item's annotations as one UUEF document. See [Annotation JSON reference](/api-sdk-cli/apis/annotation-json-reference.md) for its fields. A polygon's `paths` hold the outline first, then one ring per hole, in pixels of the image.
* Answer within 10 seconds with any 2xx, then do the work. A slower answer counts as a timeout. FastAPI runs a background task after the answer is sent.
* A delivery is tried up to 5 times when your service does not answer in time, cannot be reached, or answers with a 5xx. Any other error answer fails the run. The same delivery can arrive twice: use `run_id` to drop duplicates.
* **Test connection** sends an `ai.unitlab.agent.test.v1` event with no task (`event.is_test`). Any 2xx answer (200 to 299) passes the test.
* `AgentWebhook.verify(body, headers, secret)` checks the Standard Webhooks headers and raises `ValueError` for an invalid signature. A stage without a signing secret sends unsigned events: read them with `AgentWebhook.parse(body)`.
* Once your service has answered, Unitlab does not send the task again. In production, hand the task to a durable queue before answering.
* A failed run stays in the Agent stage until a member with the **Manage data** permission clicks **Retry**, which sends the task again with a new `run_id`. A run that is not completed within 7 days fails too.

### When a manager stops the run or moves the task

While your worker runs, a manager can **Stop** the run or move the task out of the Agent stage. Your service is not notified; its next call for that task raises `unitlab.ValidationError` (HTTP `400`) and saves nothing:

* After a move, `save_annotations` raises "The task has left the Agent stage; its annotations can no longer be saved through the Agent API." and `complete_agent` / `fail_agent` raise "Agent result requires the current Agent stage".
* After a stop, the task is still in the Agent stage, so saves still go through, but `complete_agent` / `fail_agent` for the stopped run raise "Agent run is stale or not waiting for a result". The same happens for a result sent for an older run.

Catch `ValidationError`, stop working on that task and do not retry the call. There is nothing to clean up. If the task is retried or moved back into an Agent stage, your service receives a new event with a new `run_id`.

### Save as GeoJSON

A geospatial or pathology item can also be saved as a GeoJSON FeatureCollection ([GeoJSON (geospatial models)](/api-sdk-cli/apis/annotation-json-reference.md#geojson-for-geospatial-models)). Each feature names a project class by its value and keeps the id of the label it came from.

```python
collection = {
    "type": "FeatureCollection",
    "features": [
        {
            "type": "Feature",
            "id": "0b8e2c6a-5f4d-4c1e-9a7b-3d2f1e6c8a90",  # the label's id from the document you read
            "properties": {"class": 1, "markup_type": "polygon"},
            "geometry": {
                "type": "Polygon",
                "coordinates": [[[69.2401, 41.2995], [69.2410, 41.2995], [69.2410, 41.3001], [69.2401, 41.3001], [69.2401, 41.2995]]],
            },
        }
    ],
}
task.save_annotations(item, collection, format="geojson")
```

* `properties.class` is a project class value: the `value` of a class in the document's `classes` list. `properties.markup_type` is `polygon` or `mask`.
* A feature's `id` becomes the object's id. Give it the label's `id` to keep that object.
* Coordinates are longitude/latitude (RFC 7946); every ring ends on its first point. The raster's own map units and raster pixels are read too, and pathology slides use pixels.

### Save in the prediction format

An image item can also be saved in the [Prediction format (image models)](/api-sdk-cli/apis/annotation-json-reference.md#prediction-format-image-models), the JSON an image model connected under **My AI Models** returns.

```python
reply = {
    "classes": ["building"],
    "polygons": [{"point": [[120, 80], [340, 80], [340, 260], [120, 260]], "class": 0}],
}
task.save_annotations(item, reply, format="model")
```

* `classes` lists project class names. Each object's `class` is a position in that list, starting at 0; a class is matched by name and by the object's shape.
* Coordinates are pixels of the image. The saved objects get new ids.

### What a save does

* `format` is `uuef` (the default), `geojson` for geospatial and pathology items, or `model` for the [Prediction format (image models)](/api-sdk-cli/apis/annotation-json-reference.md#prediction-format-image-models) on image items, described in [Bring your own Models](https://docs.unitlab.ai/documentation/auto-labeling/bring-your-own-models#representative-image-request).
* A save **replaces the item's annotations completely**. Objects missing from the document are deleted. To keep an object, keep it in the document with its `id`.
* The newest save wins: it overwrites edits someone made after your read.
* Every object in the saved version is credited to the member who created the API key, and each save adds a version. Skip the save when nothing changed.
* A document that carries tags (`data_tags` in UUEF, a root `tags` list in GeoJSON or the prediction format; an empty list clears them) replaces the item's tags. Without that key, the item's tags are kept.
* Saving works only while the task is in an Agent stage. Reading works in any stage.
* A save the item cannot take raises an error and saves nothing. [Annotations and workflow API](/api-sdk-cli/apis/annotations-and-workflow-api.md#agent-stage) lists the cases.

### The event

Each event names one task. A Data Group task lists every panel in `datasource_ids`.

```json
{
  "specversion": "1.0",
  "id": "…",
  "source": "…",
  "type": "ai.unitlab.agent.task.ready.v1",
  "time": "2026-10-20T07:00:00+00:00",
  "datacontenttype": "application/json",
  "data": {
    "project_id": "…",
    "workflow_id": "…",
    "stage_id": "…",
    "task": {"task_id": "…", "run_id": "…", "datasource_ids": ["…"]}
  }
}
```

***

> **Continue with Unitlab:** [enterprise data annotation workflows](https://unitlab.ai/en/data-annotation)