Understanding Webhooks for AI Agents
Webhooks are user-defined HTTP callbacks that allow one system to notify another about events in real time. For AI agents—whether they are LLM-powered assistants, autonomous task runners, or multi-step reasoning systems—webhooks serve as the critical inbound communication channel. Instead of the agent constantly polling an API to check for new work, the external system sends an HTTP POST request to a URL the agent exposes, delivering the event payload immediately.
An AI agent might receive webhook events from a variety of sources: a GitHub repository firing a push event, a Stripe payment confirmation, a Slack slash command, or a custom workflow engine triggering the next step in a chain. The agent processes the payload, reasons about it, and takes action—all without latency-introducing polling loops.
Why Reliable Webhook Delivery Matters for AI Agents
AI agents differ from traditional webhook consumers in several ways that make reliability paramount:
- Non-deterministic processing time: An LLM call might take 500ms or 10 seconds depending on model load, token count, and reasoning depth. A naive webhook handler that doesn't acknowledge quickly will cause senders to retry, potentially creating duplicate work.
- Costly operations: Each agent invocation may burn tokens, call paid APIs, or trigger side effects. Duplicate or lost events translate directly to wasted resources or missed actions.
- Stateful reasoning: Agents often maintain conversation history or task state. A dropped event can leave the agent in an inconsistent state, waiting forever for input that was already sent.
- Chained workflows: In multi-agent systems, one agent's output may be another's webhook input. A failure cascade can paralyze an entire pipeline.
Building a reliable webhook receiver means guaranteeing exactly-once semantics (or at-least-once with idempotency), fast acknowledgment, durable persistence, and graceful error recovery.
Architecture Overview
The recommended architecture separates event ingestion from event processing. The webhook endpoint itself should be a thin, fast layer that captures the payload, persists it, acknowledges receipt, and then hands off the actual AI work to a background processor.
┌─────────────┐ POST /webhook ┌───────────────┐ ┌──────────────────┐
│ Webhook │──────────────────────▶│ Ingestion │────▶│ Queue / Stream │
│ Sender │◀────── 200 OK ───────│ Service │ │ (Redis, Kafka, │
└─────────────┘ └───────────────┘ │ SQS, etc.) │
└────────┬─────────┘
│
┌────────▼─────────┐
│ AI Agent Worker │
│ (Background) │
└──────────────────┘
Step-by-Step Implementation
1. The Ingestion Endpoint (FastAPI / Python)
The webhook receiver must be fast, stateless, and return a 2xx response before the sender's timeout expires. Here's a production-grade example using FastAPI with signature verification and idempotency key support.
# webhook_receiver.py
import hashlib
import hmac
import json
import os
from datetime import datetime, timezone
from typing import Optional
import aioredis
from fastapi import FastAPI, Request, HTTPException, Header
from pydantic import BaseModel
app = FastAPI()
# --- Configuration ---
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "your-signing-secret")
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
# --- Models ---
class WebhookEvent(BaseModel):
event_id: str
event_type: str
timestamp: str
payload: dict
# --- Redis connection ---
redis: Optional[aioredis.Redis] = None
@app.on_event("startup")
async def startup():
global redis
redis = await aioredis.from_url(REDIS_URL)
@app.on_event("shutdown")
async def shutdown():
if redis:
await redis.close()
# --- Signature verification ---
def verify_signature(raw_body: bytes, signature_header: str) -> bool:
"""Verify HMAC-SHA256 signature from the sender."""
if not signature_header:
return False
# Expected format: "sha256=abc123..."
expected_signature = hmac.new(
WEBHOOK_SECRET.encode(),
raw_body,
hashlib.sha256
).hexdigest()
received_signature = signature_header.replace("sha256=", "")
return hmac.compare_digest(expected_signature, received_signature)
# --- Idempotency check ---
async def is_duplicate(event_id: str) -> bool:
"""Check if event has already been processed."""
key = f"webhook:processed:{event_id}"
return await redis.exists(key) == 1
async def mark_processed(event_id: str):
"""Mark event as processed with a TTL for cleanup."""
key = f"webhook:processed:{event_id}"
await redis.setex(key, 86400 * 7, "processed") # 7 days TTL
# --- Main webhook endpoint ---
@app.post("/webhook")
async def receive_webhook(
request: Request,
x_signature: Optional[str] = Header(None, alias="X-Signature")
):
# 1. Read raw body for signature verification
raw_body = await request.body()
# 2. Verify signature
if not verify_significance(raw_body, x_signature or ""):
raise HTTPException(status_code=401, detail="Invalid signature")
# 3. Parse JSON
try:
body = json.loads(raw_body)
event = WebhookEvent(**body)
except (json.JSONDecodeError, Exception) as e:
raise HTTPException(status_code=400, detail=f"Invalid payload: {str(e)}")
# 4. Idempotency check - return 200 if already seen
if await is_duplicate(event.event_id):
return {"status": "already_processed", "event_id": event.event_id}
# 5. Persist to Redis stream for async processing
event_json = json.dumps(body)
await redis.xadd(
"webhook:events:stream",
{"data": event_json},
maxlen=100000 # cap stream size
)
# 6. Mark as seen to prevent duplicates
await mark_processed(event.event_id)
# 7. Return 200 immediately
return {"status": "accepted", "event_id": event.event_id}
2. The AI Agent Worker (Background Processor)
The worker reads from the persistent stream and invokes the AI agent. This decoupling means slow LLM calls never block the webhook endpoint. The worker uses a consumer group for reliable, load-balanced processing across multiple instances.
# agent_worker.py
import asyncio
import json
import os
from typing import Any, Dict
import aioredis
import openai
openai.api_key = os.getenv("OPENAI_API_KEY")
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
STREAM_NAME = "webhook:events:stream"
CONSUMER_GROUP = "agent-workers"
CONSUMER_NAME = f"worker-{os.getpid()}"
async def process_event(event: Dict[str, Any]) -> Dict[str, Any]:
"""
The actual AI agent logic. This is where you'd:
- Extract intent from the event
- Call an LLM with appropriate tools
- Execute actions based on the reasoning
"""
event_type = event.get("event_type")
payload = event.get("payload", {})
# Example: classify the event and decide action
system_prompt = f"""You are an AI agent processing a webhook event of type '{event_type}'.
Analyze the payload and determine what action to take. Respond with JSON:
{{"reasoning": "...", "action": "...", "parameters": {{}} }}"""
response = await openai.ChatCompletion.acreate(
model="gpt-4",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": json.dumps(payload)}
],
temperature=0.2
)
result = json.loads(response.choices[0].message.content)
# Execute the determined action (simplified)
print(f"[Agent] Event: {event_type}, Action: {result['action']}")
print(f"[Agent] Reasoning: {result['reasoning']}")
return result
async def worker_loop():
"""Main worker loop reading from Redis stream."""
redis = await aioredis.from_url(REDIS_URL)
# Create consumer group (idempotent)
try:
await redis.xgroup_create(
STREAM_NAME, CONSUMER_GROUP, id="$", mkstream=True
)
except aioredis.ResponseError as e:
if "BUSYGROUP" not in str(e):
raise
print(f"Worker {CONSUMER_NAME} starting...")
while True:
try:
# Read pending messages first (from crashes), then new ones
entries = await redis.xreadgroup(
CONSUMER_GROUP,
CONSUMER_NAME,
{STREAM_NAME: ">"}, # ">" = new messages
count=1,
block=5000 # 5 second block
)
if not entries:
# Also check for any pending messages assigned to this consumer
pending = await redis.xpending(
STREAM_NAME, CONSUMER_GROUP
)
if pending.get("pending", 0) > 0:
# Claim and process pending
claimed = await redis.xautoclaim(
STREAM_NAME,
CONSUMER_GROUP,
CONSUMER_NAME,
min_idle_time=60000, # 1 minute idle
count=1
)
if claimed and claimed[1]:
entries = [(STREAM_NAME, claimed[1])]
if not entries:
continue
for stream_name, messages in entries:
for message_id, fields in messages:
raw_data = fields.get(b"data", b"{}")
event = json.loads(raw_data.decode())
try:
result = await process_event(event)
# Acknowledge successful processing
await redis.xack(
STREAM_NAME, CONSUMER_GROUP, message_id
)
# Store result for later retrieval
event_id = event.get("event_id", "unknown")
await redis.setex(
f"webhook:result:{event_id}",
86400 * 7,
json.dumps(result)
)
except Exception as e:
print(f"Error processing {message_id}: {e}")
# Don't ack - message will be retried or claimed by another worker
# Log to dead letter queue for manual inspection
await redis.xadd(
"webhook:dead_letters:stream",
{
"original_event": raw_data.decode(),
"error": str(e),
"timestamp": str(asyncio.get_event_loop().time())
}
)
except asyncio.CancelledError:
break
except Exception as e:
print(f"Worker loop error: {e}")
await asyncio.sleep(1) # backoff
await redis.close()
if __name__ == "__main__":
asyncio.run(worker_loop())
3. The Full Integration (Docker Compose Example)
Bringing it together with Redis for the stream and the webhook receiver plus workers running in parallel.
# docker-compose.yml
version: "3.9"
services:
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
webhook-receiver:
build:
context: .
dockerfile: Dockerfile.receiver
ports:
- "8000:8000"
environment:
- REDIS_URL=redis://redis:6379
- WEBHOOK_SECRET=${WEBHOOK_SECRET}
depends_on:
- redis
restart: unless-stopped
agent-worker:
build:
context: .
dockerfile: Dockerfile.worker
environment:
- REDIS_URL=redis://redis:6379
- OPENAI_API_KEY=${OPENAI_API_KEY}
depends_on:
- redis
deploy:
replicas: 3 # multiple workers for throughput
restart: unless-stopped
volumes:
redis_data:
# Dockerfile.receiver
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY webhook_receiver.py .
CMD ["uvicorn", "webhook_receiver:app", "--host", "0.0.0.0", "--port", "8000"]
# Dockerfile.worker
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY agent_worker.py .
CMD ["python", "agent_worker.py"]
# requirements.txt
fastapi>=0.100.0
uvicorn>=0.23.0
aioredis>=2.0.0
openai>=0.27.0
pydantic>=2.0.0
Best Practices for Reliable Webhook Ingestion
1. Always Acknowledge Before Processing
The cardinal rule: return a 2xx response before doing any heavy work. The webhook sender (GitHub, Stripe, etc.) typically waits only 5–30 seconds for a response. If your agent takes 20 seconds to reason, the sender will timeout, assume failure, and retry—creating duplicate events. By persisting first and processing later, you decouple reliability from latency.
2. Implement Idempotency
Even with fast acknowledgment, network glitches can cause duplicate deliveries. Always check a unique event identifier (provided by the sender or derived from the payload) against a cache of recently processed events. Redis with a TTL is ideal: fast lookups and automatic cleanup. The code example above demonstrates this pattern with the is_duplicate() and mark_processed() functions.
3. Verify Signatures on Every Request
Webhook endpoints are publicly accessible URLs. Without verification, anyone could send fake events that trigger expensive LLM calls or corrupt agent state. HMAC-SHA256 signature verification using a shared secret is the industry standard. Always use constant-time comparison (hmac.compare_digest) to prevent timing attacks.
4. Use a Durable Buffer (Not In-Memory Queues)
In-memory queues (like Python's asyncio.Queue) lose all pending events on crash or restart. Redis streams, Kafka, RabbitMQ, or cloud equivalents (AWS SQS, Google Pub/Sub) provide persistence. Redis streams with consumer groups give you exactly the semantics needed: ordered, persistent, with explicit acknowledgment and automatic failover to other consumers.
5. Implement a Dead Letter Queue
Events that repeatedly fail processing shouldn't block the stream. After a configurable number of retries, move them to a dead letter stream for manual inspection. The worker code above logs failures to webhook:dead_letters:stream. Set up monitoring and alerts on this stream so you catch poison messages early.
6. Configure Retry Logic Thoughtfully
The sender's retry behavior varies. Stripe retries with exponential backoff over several days. GitHub retries for a few hours. Your internal retry logic should respect these windows. For AI agent failures, consider whether retrying with the same event is safe—if the agent already performed a side effect (e.g., sent an email), a retry could duplicate it. Use idempotency keys at the action level, not just the event level.
7. Monitor Webhook Health
Track these metrics:
- Ingestion latency: Time from receiving the HTTP request to persisting in the stream (target: <50ms)
- Processing latency: Time from persistence to agent completion
- Stream depth: Number of pending events (alert if growing)
- Dead letter growth: Events that cannot be processed
- Duplicate rate: How often idempotency checks catch repeats
Handling Webhook Events in the AI Agent
Once the event is safely in the processing pipeline, the agent needs to interpret it intelligently. Here's a pattern for structuring agent prompts around webhook events.
# agent_event_handler.py
from typing import Dict, Any
import json
import openai
AGENT_SYSTEM_PROMPT = """You are an AI agent that responds to incoming webhook events.
Your job is to:
1. Understand what the event represents
2. Decide if action is needed
3. Execute the action using available tools
4. Report the outcome
Available tools:
- send_email(to, subject, body): Send an email notification
- create_task(title, description, priority): Create a task in the project management system
- update_database(query): Execute a database update
- notify_slack(channel, message): Post to a Slack channel
- http_request(method, url, body): Make an HTTP request to an internal API
Always respond with a JSON object:
{
"analysis": "Brief analysis of the event",
"actions": [
{"tool": "tool_name", "args": {...}, "reason": "Why this action"}
],
"summary": "Human-readable summary of what was done"
}"""
async def handle_event_with_agent(event: Dict[str, Any]) -> Dict[str, Any]:
"""Process a webhook event through the AI agent."""
event_type = event.get("event_type", "unknown")
payload = event.get("payload", {})
messages = [
{"role": "system", "content": AGENT_SYSTEM_PROMPT},
{"role": "user", "content": f"""
Event Type: {event_type}
Event ID: {event.get('event_id', 'unknown')}
Timestamp: {event.get('timestamp', 'unknown')}
Payload:
{json.dumps(payload, indent=2)}
Analyze this event and determine what actions to take.
"""}
]
response = await openai.ChatCompletion.acreate(
model="gpt-4",
messages=messages,
temperature=0.3,
response_format={"type": "json_object"}
)
plan = json.loads(response.choices[0].message.content)
return plan
Security Considerations
Webhook endpoints for AI agents introduce unique security challenges beyond standard API security:
- Prompt injection via payload: Webhook payloads flow directly into LLM prompts. Malicious senders could craft payloads that override agent instructions. Always sanitize and structure event data before inserting it into prompts. Use structured input formats and validate against schemas.
- Rate limiting: Without limits, a flood of webhook events could exhaust your LLM API budget. Implement per-sender rate limiting at the ingestion layer.
- Tool invocation validation: If your agent can call tools based on webhook data, validate all parameters server-side. Don't let the LLM generate SQL queries or shell commands directly from untrusted input.
- Secret rotation: Rotate webhook signing secrets regularly. Support multiple active secrets during rotation windows so senders can transition without downtime.
Testing Webhook Reliability
Testing webhook handling requires simulating various failure modes. Here's a test harness that covers the critical paths.
# test_webhook_reliability.py
import asyncio
import hashlib
import hmac
import json
import time
from typing import List
import httpx
import pytest
from httpx import AsyncClient
# Test configuration
BASE_URL = "http://localhost:8000"
WEBHOOK_SECRET = "test-secret"
ENDPOINT = "/webhook"
def generate_signature(body: bytes) -> str:
"""Generate a valid signature for testing."""
sig = hmac.new(
WEBHOOK_SECRET.encode(),
body,
hashlib.sha256
).hexdigest()
return f"sha256={sig}"
def create_event(event_id: str, event_type: str = "test.event") -> dict:
"""Create a test webhook event."""
return {
"event_id": event_id,
"event_type": event_type,
"timestamp": str(int(time.time())),
"payload": {"test_key": "test_value", "nested": {"data": "more"}}
}
@pytest.mark.asyncio
async def test_successful_webhook():
"""Test basic webhook delivery."""
async with AsyncClient() as client:
event = create_event("evt-001")
body = json.dumps(event).encode()
headers = {"X-Signature": generate_signature(body)}
response = await client.post(
f"{BASE_URL}{ENDPOINT}",
content=body,
headers=headers
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "accepted"
assert data["event_id"] == "evt-001"
@pytest.mark.asyncio
async def test_idempotency():
"""Test that duplicate events are acknowledged but not reprocessed."""
async with AsyncClient() as client:
event = create_event("evt-002")
body = json.dumps(event).encode()
headers = {"X-Signature": generate_signature(body)}
# First delivery
response1 = await client.post(
f"{BASE_URL}{ENDPOINT}",
content=body,
headers=headers
)
assert response1.status_code == 200
assert response1.json()["status"] == "accepted"
# Duplicate delivery (simulates sender retry)
response2 = await client.post(
f"{BASE_URL}{ENDPOINT}",
content=body,
headers=headers
)
assert response2.status_code == 200
assert response2.json()["status"] == "already_processed"
@pytest.mark.asyncio
async def test_invalid_signature():
"""Test that requests with wrong signatures are rejected."""
async with AsyncClient() as client:
event = create_event("evt-003")
body = json.dumps(event).encode()
headers = {"X-Signature": "sha256=00000000000000000000000000000000"}
response = await client.post(
f"{BASE_URL}{ENDPOINT}",
content=body,
headers=headers
)
assert response.status_code == 401
@pytest.mark.asyncio
async def test_missing_signature():
"""Test that requests without signatures are rejected."""
async with AsyncClient() as client:
event = create_event("evt-004")
body = json.dumps(event).encode()
response = await client.post(
f"{BASE_URL}{ENDPOINT}",
content=body
)
assert response.status_code == 401
@pytest.mark.asyncio
async def test_malformed_payload():
"""Test handling of non-JSON bodies."""
async with AsyncClient() as client:
body = b"this is not json"
headers = {"X-Signature": generate_signature(body)}
response = await client.post(
f"{BASE_URL}{ENDPOINT}",
content=body,
headers=headers
)
assert response.status_code == 400
@pytest.mark.asyncio
async def test_concurrent_delivery():
"""Test that multiple unique events are all accepted."""
async with AsyncClient() as client:
tasks = []
for i in range(20):
event_id = f"evt-concurrent-{i:03d}"
event = create_event(event_id)
body = json.dumps(event).encode()
headers = {"X-Signature": generate_signature(body)}
tasks.append(
client.post(
f"{BASE_URL}{ENDPOINT}",
content=body,
headers=headers
)
)
responses = await asyncio.gather(*tasks)
statuses = [r.status_code for r in responses]
assert all(s == 200 for s in statuses)
@pytest.mark.asyncio
async def test_large_payload():
"""Test handling of larger payloads (e.g., 100KB)."""
async with AsyncClient() as client:
event = {
"event_id": "evt-large-001",
"event_type": "large.payload",
"timestamp": str(int(time.time())),
"payload": {
"data": "x" * 100000 # 100KB of data
}
}
body = json.dumps(event).encode()
headers = {"X-Signature": generate_signature(body)}
response = await client.post(
f"{BASE_URL}{ENDPOINT}",
content=body,
headers=headers,
timeout=30.0
)
assert response.status_code == 200
@pytest.mark.asyncio
async def test_end_to_end_processing():
"""
End-to-end test: send webhook, wait for worker to process,
and verify result is stored. Requires Redis and worker running.
"""
import aioredis
redis = await aioredis.from_url("redis://localhost:6379")
async with AsyncClient() as client:
event = create_event("evt-e2e-001", "user.signup")
body = json.dumps(event).encode()
headers = {"X-Signature": generate_signature(body)}
response = await client.post(
f"{BASE_URL}{ENDPOINT}",
content=body,
headers=headers
)
assert response.status_code == 200
# Poll for result (worker processes in background)
for _ in range(30): # wait up to 30 seconds
result = await redis.get("webhook:result:evt-e2e-001")
if result:
parsed = json.loads(result)
assert "analysis" in parsed or "action" in parsed
break
await asyncio.sleep(1)
else:
pytest.fail("Worker did not process event within 30 seconds")
await redis.close()
Deployment Checklist
Before exposing your webhook receiver to production traffic, verify:
- HTTPS only: Webhooks carry sensitive data and signing secrets. Never expose plain HTTP endpoints on the public internet.
- Load balancing: Run multiple receiver instances behind a load balancer with sticky sessions disabled—webhooks are stateless.
- Health checks: Expose a
/healthendpoint that verifies Redis connectivity so your orchestrator can restart unhealthy instances. - Graceful shutdown: On SIGTERM, stop accepting new requests, finish in-flight persistence, then exit. This prevents lost events during deployments.
- Logging: Log every webhook receipt with event_id, source IP, and processing outcome. These logs are invaluable for debugging missing or duplicate events.
- Alerting: Set alerts for elevated 4xx/5xx rates, growing stream depth, and dead letter queue growth.
Conclusion
Building a reliable webhook receiver for AI agents is fundamentally about decoupling fast, secure ingestion from slow, intelligent processing. By persisting events to a durable stream before the agent touches them, you eliminate the tension between webhook timeouts and LLM latency. Signature verification, idempotency checks, and structured dead letter handling turn a naive callback endpoint into a production-grade event pipeline. The patterns shown here—FastAPI ingestion, Redis streams with consumer groups, and background agent workers—give you a solid foundation that scales from a single agent to fleets of specialized AI workers processing millions of events. As AI agents grow more autonomous and interconnected, the reliability of their inbound communication becomes not just an operational concern but a fundamental requirement for trustworthy autonomous systems.