Building a Webhook-Driven Agent: Async Patterns for Slow Enterprise APIs

Most agent demos call an API and wait. That works until the API is an ERP that takes 40 seconds to answer, or a batch endpoint that takes ten minutes. Then the agent hangs, the request times out, a retry fires, and now you've created the same purchase order twice.

This post walks through the pattern I reach for instead: accept the task, hand the slow work to a queue, let the ERP call us back, and keep a polling job around for the days the callback never arrives. The code is Python with FastAPI and httpx. The ERP API is made up, so adapt the field names to yours.

Why synchronous waiting breaks

There are three problems with waiting inside the agent's request:

• Timeouts. Load balancers and clients give up long before a slow system finishes.
• Retries without protection. A client that retries a timed-out call can trigger the same side effect twice.
• Wasted capacity. A worker blocked on one slow call can't serve anyone else.

The fix is to split "accept the task" from "finish the task".

The shape of the solution

1. The client sends a task with an Idempotency-Key header and gets 202 Accepted right away.
2. A background worker submits the job to the ERP, including our callback URL.
3. The ERP calls /webhooks/erp when it finishes, with an HMAC signature.
4. A reconciler polls the ERP for jobs that have been waiting too long, in case the webhook got lost.

Setup and state

For the demo, state lives in dictionaries. In production, use a database so a restart doesn't lose tasks.

import asyncio, hashlib, hmac, os, time, uuid
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, Header, HTTPException, Request

ERP_URL = os.environ["ERP_URL"]
CALLBACK_URL = os.environ["CALLBACK_URL"]          # public URL of /webhooks/erp
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"].encode()

tasks: dict[str, dict] = {}     # task_id -> state
by_key: dict[str, str] = {}     # idempotency key -> task_id
by_job: dict[str, str] = {}     # ERP job id -> task_id
queue: asyncio.Queue = asyncio.Queue()

Accepting tasks safely

The idempotency key is what makes retries harmless. If the same key arrives twice, we return the original task instead of creating a new one.

@asynccontextmanager
async def lifespan(app: FastAPI):
    jobs = [asyncio.create_task(worker()), asyncio.create_task(reconcile())]
    yield
    for job in jobs:
        job.cancel()

app = FastAPI(lifespan=lifespan)


@app.post("/tasks", status_code=202)
async def create_task(body: dict, idempotency_key: str = Header(...)):
    if idempotency_key in by_key:
        return tasks[by_key[idempotency_key]]

    task_id = str(uuid.uuid4())
    tasks[task_id] = {
        "id": task_id, "status": "queued", "input": body,
        "result": None, "attempts": 0, "created": time.time(),
    }
    by_key[idempotency_key] = task_id
    await queue.put(task_id)
    return tasks[task_id]


@app.get("/tasks/{task_id}")
async def get_task(task_id: str):
    if task_id not in tasks:
        raise HTTPException(404)
    return tasks[task_id]

The worker

The worker submits the job and records the ERP's job id. We pass our own task_id as a reference, so if the ERP supports idempotent submission, a retried submit won't create a duplicate.

async def worker():
    async with httpx.AsyncClient(timeout=10) as client:
        while True:
            task_id = await queue.get()
            task = tasks[task_id]
            try:
                resp = await client.post(f"{ERP_URL}/jobs", json={
                    "payload": task["input"],
                    "callback_url": CALLBACK_URL,
                    "reference": task_id,
                })
                resp.raise_for_status()
                job_id = resp.json()["job_id"]
                by_job[job_id] = task_id
                task.update(status="waiting_for_erp", job_id=job_id)
            except httpx.HTTPError:
                task["attempts"] += 1
                if task["attempts"] < 3:
                    await asyncio.sleep(2 ** task["attempts"])
                    await queue.put(task_id)
                else:
                    task["status"] = "failed"
            finally:
                queue.task_done()

The webhook

Two rules matter here. Verify the signature before you trust anything, and treat duplicate deliveries as normal. Webhooks are usually "at least once", so the same event can arrive twice.

@app.post("/webhooks/erp")
async def erp_webhook(request: Request, x_signature: str = Header(...)):
    raw = await request.body()
    expected = hmac.new(WEBHOOK_SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, x_signature):
        raise HTTPException(401, "bad signature")

    event = await request.json()
    task_id = by_job.get(event["job_id"])
    if task_id is None:
        return {"ok": True}                      # unknown job, acknowledge anyway

    task = tasks[task_id]
    if task["status"] in ("done", "failed"):
        return {"ok": True}                      # duplicate delivery, ignore

    ok = event["state"] == "succeeded"
    task.update(status="done" if ok else "failed", result=event.get("result"))
    return {"ok": True}

Return a 2xx quickly. If your agent needs to do more work with the result, enqueue that instead of doing it inside the handler.

The polling fallback

Callbacks get lost: a deploy, a firewall change, a vendor outage. A reconciler that checks old jobs turns "silently stuck forever" into "late by a few minutes".

async def reconcile():
    async with httpx.AsyncClient(timeout=10) as client:
        while True:
            await asyncio.sleep(60)
            for task in list(tasks.values()):
                stale = time.time() - task["created"] > 300
                if task["status"] != "waiting_for_erp" or not stale:
                    continue
                try:
                    resp = await client.get(f"{ERP_URL}/jobs/{task['job_id']}")
                    resp.raise_for_status()
                    job = resp.json()
                except httpx.HTTPError:
                    continue
                if job["state"] in ("succeeded", "failed"):
                    ok = job["state"] == "succeeded"
                    task.update(status="done" if ok else "failed", result=job.get("result"))

Webhook or polling?

I don't think it's either/or. Webhooks give you speed and cheap traffic. Polling gives you certainty. Using both costs about twenty extra lines and removes a whole category of 3 a.m. questions.

What this doesn't cover

A few things you'd still need before production: persistent storage, a dead-letter list for tasks that fail permanently, metrics on queue depth and callback delay, and a way to cancel a task. I'd also add an alert for any task stuck in waiting_for_erp for more than an hour.

If you're designing the API side of an agent and want to compare options beyond webhooks, I wrote up REST, WebSocket, and webhook patterns for agents. I work on enterprise agent deployments at AI Hive. How do you handle lost callbacks today: polling, retries from the sender, or something else?