Webhook Receivers for AI Agents: Processing Inbound Events from External Services

What Webhook Receivers Do for AI Agents

Webhooks are the primary mechanism external services use to notify your system about events in real time. When Stripe processes a payment, when GitHub merges a pull request, when a CRM updates a contact — these services send HTTP POST requests to a URL you control. A webhook receiver is the endpoint that catches these requests and routes them to your AI agent for processing.

Building a reliable webhook receiver is harder than it looks. You need to verify that requests actually come from the claimed service, handle duplicate deliveries gracefully, process events asynchronously so the sender does not time out, and log everything for debugging. Getting any of these wrong means your agent either misses events or processes them incorrectly.

Designing the Webhook Endpoint

A well-designed webhook endpoint does four things in sequence: authenticate the request, parse the payload, enqueue the event for processing, and return a 200 response immediately.

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from pydantic import BaseModel
import hmac
import hashlib
import json

app = FastAPI()


class WebhookEvent(BaseModel):
    event_type: str
    payload: dict
    idempotency_key: str | None = None


def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)


@app.post("/webhooks/{provider}")
async def receive_webhook(
    provider: str,
    request: Request,
    background_tasks: BackgroundTasks,
):
    body = await request.body()
    signature = request.headers.get("X-Signature-256", "")
    secret = get_provider_secret(provider)

    if not verify_signature(body, signature, secret):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event_data = json.loads(body)
    background_tasks.add_task(process_webhook_event, provider, event_data)

    return {"status": "accepted"}

The verify_signature function uses HMAC-SHA256 comparison, which is constant-time to prevent timing attacks. The actual processing happens in a background task so the webhook sender gets a fast response.

Implementing Idempotency

Most webhook providers retry failed deliveries, which means your receiver will see the same event multiple times. Without idempotency handling, your agent might send duplicate emails, create duplicate records, or charge a customer twice.

import redis.asyncio as redis

redis_client = redis.Redis(host="localhost", port=6379, db=0)

IDEMPOTENCY_TTL = 86400  # 24 hours


async def is_duplicate(event_id: str) -> bool:
    key = f"webhook:processed:{event_id}"
    was_set = await redis_client.set(key, "1", nx=True, ex=IDEMPOTENCY_TTL)
    return was_set is None  # None means key already existed


async def process_webhook_event(provider: str, event_data: dict):
    event_id = event_data.get("id") or event_data.get("idempotency_key")
    if not event_id:
        event_id = hashlib.sha256(
            json.dumps(event_data, sort_keys=True).encode()
        ).hexdigest()

    if await is_duplicate(event_id):
        print(f"Skipping duplicate event: {event_id}")
        return

    handler = get_handler_for_provider(provider)
    await handler(event_data)

The Redis SET NX operation is atomic — even if two webhook retries arrive at the same millisecond, only one will succeed in setting the key. The TTL ensures the idempotency cache does not grow unbounded.

Payload Validation with Pydantic

Different providers send wildly different payload structures. Use Pydantic models to validate and normalize incoming data before your agent sees it.

from pydantic import BaseModel, field_validator
from typing import Literal


class StripeWebhookPayload(BaseModel):
    id: str
    type: str
    data: dict
    created: int

    @field_validator("type")
    @classmethod
    def validate_event_type(cls, v: str) -> str:
        allowed_prefixes = ["payment_intent.", "invoice.", "customer.subscription."]
        if not any(v.startswith(p) for p in allowed_prefixes):
            raise ValueError(f"Unhandled event type: {v}")
        return v


class GitHubWebhookPayload(BaseModel):
    action: str
    repository: dict
    sender: dict

Strict validation at the boundary means your downstream agent handlers can trust the data shape without additional defensive checks.

Async Processing with Task Queues

For high-volume webhook traffic, background tasks in FastAPI may not be sufficient. Use a proper task queue like Celery or ARQ to ensure events survive server restarts.

from arq import create_pool
from arq.connections import RedisSettings


async def enqueue_webhook(provider: str, event_data: dict):
    pool = await create_pool(RedisSettings(host="localhost"))
    await pool.enqueue_job(
        "process_webhook_task", provider, event_data
    )


async def process_webhook_task(ctx: dict, provider: str, event_data: dict):
    handler = get_handler_for_provider(provider)
    await handler(event_data)

ARQ persists jobs in Redis, so if your server crashes after accepting the webhook but before processing it, the job will still be picked up when the worker restarts.

FAQ

How do I test webhooks locally during development?

Use a tunneling service like ngrok or Cloudflare Tunnel to expose your local FastAPI server to the internet. Most providers also offer webhook testing tools in their dashboards that let you send sample events to your endpoint.

What status code should my webhook endpoint return?

Always return 200 or 202 as quickly as possible. Most providers treat any 2xx as success and any 4xx or 5xx as failure, triggering retries. Never return an error code because your AI processing is slow — accept the event first, process it asynchronously.

How long should I keep idempotency keys?

Match the provider's retry window. Stripe retries for up to 72 hours, GitHub for 3 days. A 24-hour to 7-day TTL on your idempotency keys covers most providers. Use longer TTLs for financial events where duplicate processing has severe consequences.

---

#Webhooks #AIAgents #FastAPI #Security #Idempotency #AgenticAI #LearnAI #AIEngineering