> ## Documentation Index
> Fetch the complete documentation index at: https://docs.encord.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Task-Ready Notifications

You can add a webhook to an Advanced Agent stage so your service can learn when tasks are waiting without continuously polling Encord. The signed notification contains no task data; after receiving it, fetch the Agent stage queue with the SDK.

<div class="flex justify-center">
  <img src="https://storage.googleapis.com/docs-media.encord.com/static/img/webhook-batching-notifications.png" width="250" />
</div>

1. In the Workflow, select the Agent stage and choose **Advanced**.
2. Under **Webhook**, click the edit icon, enter your endpoint URL, and press <kbd>Enter</kbd>.
3. Under **Batching**, set the **Batch size** and **Maximum wait (minutes)**.
4. Save the Workflow. You can use the generated **Signing secret** to [verify webhook signatures](/platform-documentation/Annotate/annotate-webhooks-notifications#verifying-webhook-signatures).

<Info>
  Removing the webhook URL also removes its batching settings. Switching the Agent stage to **Auto** removes the webhook configuration.
</Info>

**Batch size and Minimum wait**:

Encord checks each agent stage's queue every five minutes. At each check, it sends a notification if either threshold is met.

* The **batch size** is met when the number of queued tasks reaches the set size.
* The **maximum wait** is met when the oldest queued task has been waiting longer than the set limit.

The maximum wait is counted from when each task enters the queue, not on a repeating timer. Once a threshold is met, Encord notifies again at every five-minute check until your agent processes the queue. After that, the next notification depends on when new tasks arrive.

For example, with a batch size of 10 and a maximum wait of 30 minutes, if 3 tasks arrive at 9:00:

| Time | Queue | Notification |
| - | - | - |
| 9:00–9:30 | 3 tasks; batch not full, none waiting 30 minutes | None |
| First check after 9:30 (by 9:35) | Oldest task passes 30 minutes | Sent, reason `max_wait_elapsed` |
| Each later check | Tasks still queued | Sent again every five minutes until processed |
| After processing | Queue empty | None until the batch fills or a new task waits 30 minutes |

<Note>
  Webhook SDK support for Agents documentation is available [here](/agents-documentation/Custom-Agents/Task-ready-notifications-sdk).
</Note>

<div class="flex justify-center">
  <img src="https://storage.googleapis.com/docs-media.encord.com/static/img/task-ready-notification-flow.svg" width="800" />
</div>

## Prerequisites

* **The package:** `encord-agents` v0.2.10 or later, plus FastAPI if you use the FastAPI receiver: `python -m pip install "encord-agents>=0.2.10" "fastapi[standard]"`.
* **An Agent stage:** a Workflow with an Agent stage whose Custom Agent runs in Advanced mode, so you can [register a webhook URL on it]().
* **Encord credentials for the runner:** the SSH key your Task Agent already uses to fetch and update tasks, set as `ENCORD_SSH_KEY_FILE` or `ENCORD_SSH_KEY`.
* **A publicly reachable HTTPS endpoint:** Encord must be able to POST to it. Because anyone who finds the URL can call it, every request is signed and your receiver must verify the signature.

The signing secret is shown in the Encord app alongside the stage's webhook configuration. Set it as `ENCORD_WEBHOOK_SECRET`, or pass it explicitly. Check it again whenever you change that configuration. Verification never needs your Encord credentials, so a receiver can check signatures before it holds any.

## Build a FastAPI Receiver

The receiver verifies each notification, checks it concerns a project this deployment serves, answers 200 straight away, then drains the stage in the background. It is the runnable example from `examples/task_agents/fastapi_task_notifications.py`.

1. **Create the app with `get_encord_app`.** It installs the exception handlers that turn failed verification into responses, so unverified requests never reach your route.
2. **Define the stage logic on a `Runner`.** This is the same `@runner.stage` function you would write for a polling Task Agent. It returns the pathway to move each task along.
3. **Inject the notification with `dep_task_notification`.** The dependency reads the raw request body, verifies the signature against `ENCORD_WEBHOOK_SECRET` and returns a `TaskNotification`.
4. **Check the Project, then hand off.** Verification proves the request came from Encord, not that it concerns a Project this deployment should act on. Return early for any other Project, then queue `run_stage` as a background task.

```python theme={"dark"}
import os
from uuid import UUID

from encord.objects.ontology_labels_impl import LabelRowV2
from fastapi import BackgroundTasks, Depends
from typing_extensions import Annotated

from encord_agents.core.webhooks import TaskNotification
from encord_agents.fastapi.cors import get_encord_app
from encord_agents.fastapi.notifications import dep_task_notification
from encord_agents.tasks import Runner

PROJECT_HASH = UUID(os.environ["ENCORD_PROJECT_HASH"])

app = get_encord_app()
runner = Runner(project_hash=str(PROJECT_HASH))


@runner.stage("<stage_name_or_uuid>")
def my_agent(lr: LabelRowV2) -> str:
    # Do whatever the stage should do to a task, then return the pathway name.
    return "<pathway_name>"


@app.post("/encord/task-ready")
def task_ready(
    notification: Annotated[TaskNotification, Depends(dep_task_notification)],
    background: BackgroundTasks,
) -> None:
    # Act only on the project this deployment serves.
    if notification.project_hash != PROJECT_HASH:
        return

    # Answer first, then drain the stage out of band.
    background.add_task(runner.run_stage, notification.stage_uuid)
```

Run it locally with:

```bash theme={"dark"}
ENCORD_WEBHOOK_SECRET=<secret> ENCORD_PROJECT_HASH=<project_hash> \
  uvicorn fastapi_task_notifications:app
```

**Why the work runs in the background:** Encord waits 180 seconds for a response and retries on timeout. It also re-notifies every few minutes while work remains. A handler that drains the stage inline would be notified again while the first drain is still running. Returning `None` answers 200, which is what Encord expects. Any other status is recorded as a failed delivery.

**Serving several Projects:** `run_stage` accepts a `project_hash`, so one deployment can drain stages across Projects.
As an additional enforcement step you may keep an explicit allow-list and pass the Project through only after checking it:

```python theme={"dark"}
SERVED_PROJECTS = {UUID("<project_hash_a>"), UUID("<project_hash_b>")}

@app.post("/encord/task-ready")
def task_ready(
    notification: Annotated[TaskNotification, Depends(dep_task_notification)],
    background: BackgroundTasks,
) -> None:
    if notification.project_hash not in SERVED_PROJECTS:
        return
    background.add_task(
        runner.run_stage,
        notification.stage_uuid,
        project_hash=notification.project_hash,
    )
```

**Building the app yourself:** if you create `FastAPI()` directly instead of using `get_encord_app`, call `add_notification_handlers(app)`. Without it, a request that fails verification surfaces as an unhandled 500 instead of a 401.

## Route Specific Secrets & Notifications

Use `dep_task_notification_with_args` when `ENCORD_WEBHOOK_SECRET` cannot cover every route, for example when one service receives notifications for two stages with different secrets. It returns a dependency that verifies against the secret you pass. You can also widen the timestamp tolerance, but only for a host whose clock cannot be kept closer than the default 300 seconds.

```python theme={"dark"}
import os

from encord_agents.fastapi import dep_task_notification_with_args

verify_stage_a = dep_task_notification_with_args(secret=os.environ["STAGE_A_WEBHOOK_SECRET"])

@app.post("/encord/stage-a/task-ready")
def stage_a_ready(
    notification: Annotated[TaskNotification, Depends(verify_stage_a)],
    background: BackgroundTasks,
) -> None:
    ...
```

The `TaskNotification` your route receives exposes these fields:

| Field | Type | Meaning |
| - | - | - |
| `project_hash` | `UUID` | The Project whose stage has work waiting. Check it against the projects you serve. |
| `stage_uuid` | `UUID` | The agent stage with work waiting. Pass it to `run_stage`. |
| `pending_count` | `int` | How many tasks were queued when the notification was raised. A hint only; fetch the queue to see what is there now. |
| `reason` | `AgentStageWorkReason` or `str` | `BATCH_SIZE_REACHED` or `MAX_WAIT_ELAPSED`. A condition not supported on your downloaded package arrives as a plain string. |
| `uid` | `UUID` | Identifies this delivery. Not a deduplication key: Encord re-notifies with a fresh `uid` while work remains. |
| `event_created_timestamp` | `datetime` | When Encord raised the event. |

To branch on the reason, compare against the enum and treat anything else as unknown. Use `isinstance`, because the enum subclasses `str`:

```python theme={"dark"}
from encord_agents.core.webhooks import AgentStageWorkReason

if notification.reason is AgentStageWorkReason.MAX_WAIT_ELAPSED:
    logger.info("Draining a partial batch of %s tasks", notification.pending_count)
elif not isinstance(notification.reason, AgentStageWorkReason):
    logger.info("Unrecognized reason: %s", notification.reason)
```

Because notifications repeat and counts go stale, make your stage logic safe to run more than once. `run_stage` works from the stage's queue as it stands, so an extra notification only triggers another look.

## Notifications On Any Runtime

`encord_agents.core.webhooks` works with any framework or serverless platform.

* `verify_and_parse_notification(body, headers, *, secret=None, tolerance_seconds=300)` verifies the request and returns a `TaskNotification`. Header lookup is case-insensitive, so a lowercased AWS Lambda headers dict works as-is.
* `verify_signature(body, *, signature, timestamp, secret=None, tolerance_seconds=300)` only checks the signature, if you extract the headers yourself.
* `parse_notification(body)` only reads the body. It trusts its input, so call it after verification.

Always pass the body exactly as received, as bytes. The signature covers the raw bytes, so a parsed and re-serialized body will not verify. Map each exception to the status Encord expects:

```python theme={"dark"}
from encord_agents.core.webhooks import (
    UnexpectedEventType,
    UnsupportedNotificationVersion,
    WebhookVerificationError,
    verify_and_parse_notification,
)

SERVED_PROJECTS = {...}

def handle(body: bytes, headers: dict[str, str]) -> int:
    try:
        notification = verify_and_parse_notification(body, headers)
    except WebhookVerificationError:
        return 401  # Not provably from Encord: do not read the body.
    except UnexpectedEventType:
        return 200  # Genuine, but a different event: acknowledge it.
    except UnsupportedNotificationVersion:
        return 501  # Upgrade encord-agents to read this envelope.

    if notification.project_hash in SERVED_PROJECTS:
        enqueue_drain(notification.project_hash, notification.stage_uuid)  # Your own queue or job.
    return 200
```

If you verify by hand, the signature is an HMAC-SHA256 hex digest of `<timestamp>.<body>` using the signing secret. It arrives in `X-Encord-Signature`, with the Unix timestamp in `X-Encord-Timestamp`.

## Configure, Deploy and Test

1. **Deploy the receiver** to a publicly reachable HTTPS URL, for example `https://agents.example.com/encord/task-ready`. Set `ENCORD_WEBHOOK_SECRET` and your SSH key in the deployment's environment.
2. **Register the URL on the agent stage.** In the workflow, open the agent stage's Custom Agent in Advanced mode, add the webhook URL, and set the minimum batch size and maximum wait. See [Configure batched webhook notifications](https://docs.encord.com/platform-documentation/Annotate/automated-labeling/annotate-custom-agents#configure-batched-webhook-notifications).
3. **Copy the signing secret** shown with that configuration into `ENCORD_WEBHOOK_SECRET`. If you change the configuration later, check the secret again.
4. **Send tasks to the stage** and watch the receiver's logs. You should see a 200 for each notification and the stage's tasks moving along their pathways.

To test locally before registering the URL, sign a sample notification the way Encord does and post it to your running app:

```python theme={"dark"}
import hashlib, hmac, json, os, time

import httpx

secret = os.environ["ENCORD_WEBHOOK_SECRET"]
body = json.dumps({
    "uid": "33333333-3333-3333-3333-333333333333",
    "version": 1,
    "source": "encord",
    "event_type": "agent_stage_work_available_event",
    "event_created_timestamp": "2026-09-01T12:00:00+00:00",
    "payload": {
        "project_hash": os.environ["ENCORD_PROJECT_HASH"],
        "stage_uuid": "<stage_uuid>",
        "pending_count": 5,
        "reason": "batch_size_reached",
    },
}).encode()

timestamp = str(int(time.time()))
signature = hmac.new(secret.encode(), f"{timestamp}.".encode() + body, hashlib.sha256).hexdigest()

response = httpx.post(
    "http://localhost:8000/encord/task-ready",
    content=body,
    headers={"X-Encord-Signature": signature, "X-Encord-Timestamp": timestamp},
)
print(response.status_code)  # 200
```

Change one byte of `body` after signing, or sign with another secret, and the same request returns 401.

## Responses & Errors

Encord reads only the status code, and records anything other than 200 as a failed delivery. With `get_encord_app` or `add_notification_handlers`, each outcome maps to a status automatically:

| Outcome | Raised as | Response |
| - | - | - |
| Verified agent-stage notification | Returned as `TaskNotification` | 200 once your route returns |
| Missing or malformed headers, stale timestamp, or wrong signature | `WebhookVerificationError` | 401; the failed check is logged, not returned |
| Genuine Encord event of another type | `UnexpectedEventType` | 200, so it is not logged as a failed delivery |
| Envelope version newer than the package reads | `UnsupportedNotificationVersion` | 501; upgrade `encord-agents` |
| No secret passed and `ENCORD_WEBHOOK_SECRET` unset | `PrintableError` | 500; fix the deployment's configuration |

A single URL can be registered on more than one workflow stage, which is why other event types are acknowledged rather than refused. Encord adds fields and reason values without changing the envelope version, so the parser tolerates them. It refuses only a new version, which Encord reserves for breaking changes.