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.

In the Workflow, select the Agent stage and choose Advanced.
Under Webhook, click the edit icon, enter your endpoint URL, and press Enter.
Under Batching, set the Batch size and Maximum wait (minutes).
Save the Workflow. You can use the generated Signing secret to verify webhook signatures.

Removing the webhook URL also removes its batching settings. Switching the Agent stage to Auto removes the webhook configuration.

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

Webhook SDK support for Agents documentation is available here.

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.

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.
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.
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.
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.

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:

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:

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.

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:

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:

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

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.
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.
Copy the signing secret shown with that configuration into ENCORD_WEBHOOK_SECRET. If you change the configuration later, check the secret again.
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:

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.

Get Started

Data

Fundamentals

Annotate & Review

Label and Model Evaluation

Users and Workspaces

Task-Ready Notifications

Prerequisites

Build a FastAPI Receiver

Route Specific Secrets & Notifications

Notifications On Any Runtime

Configure, Deploy and Test

Responses & Errors

​Prerequisites

​Build a FastAPI Receiver

​Route Specific Secrets & Notifications

​Notifications On Any Runtime

​Configure, Deploy and Test

​Responses & Errors

Prerequisites

Build a FastAPI Receiver

Route Specific Secrets & Notifications

Notifications On Any Runtime

Configure, Deploy and Test

Responses & Errors