Connecting Agents to Webhooks
Receiving webhook events and triggering agent workflows in response.
Connecting Agents to Webhooks is a free AI Agents lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Agents learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
What Is a Webhook?
A webhook is an HTTP callback. When an event occurs in an external service, it sends a POST request to your endpoint with event data. Your agent processes the payload and acts.
Webhooks are push-based (events arrive when they happen) versus polling (you check repeatedly).
FastAPI Webhook Endpoint
FastAPI makes it easy to create a webhook receiver. Define a POST route, parse the JSON body, and hand off to your agent logic.
from fastapi import FastAPI, Request
from pydantic import BaseModel
app = FastAPI()
class WebhookPayload(BaseModel):
event: str
data: dict
@app.post('/webhook')
async def receive_webhook(payload: WebhookPayload):
print(f'Received event: {payload.event}')
print(f'Data: {payload.data}')
# Route to the right agent handler
if payload.event == 'email.received':
await handle_email_event(payload.data)
elif payload.event == 'file.uploaded':
await handle_file_event(payload.data)
return {'status': 'accepted'}
async def handle_email_event(data: dict):
print(f'Processing email from: {data.get("from")}')
async def handle_file_event(data: dict):
print(f'Processing file: {data.get("filename")}')Webhook Signature Verification
Always verify that webhook requests come from the expected sender. Most services sign their payloads with HMAC-SHA256 using a shared secret. Reject requests with invalid signatures.
import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
WEBHOOK_SECRET = 'your-webhook-secret-here'
def verify_signature(payload_bytes: bytes, signature_header: str) -> bool:
expected = hmac.new(
WEBHOOK_SECRET.encode(),
payload_bytes,
hashlib.sha256
).hexdigest()
received = signature_header.replace('sha256=', '')
return hmac.compare_digest(expected, received)
@app.post('/webhook/verified')
async def verified_webhook(request: Request):
payload_bytes = await request.body()
signature = request.headers.get('X-Signature', '')
if not verify_signature(payload_bytes, signature):
raise HTTPException(status_code=401, detail='Invalid signature')
# Safe to process
import json
data = json.loads(payload_bytes)
return {'status': 'verified', 'event': data.get('event')}Idempotency Keys
External services often retry failed webhook deliveries. An idempotency key is a unique ID sent with each event. Store processed keys and skip duplicates.
from fastapi import FastAPI, Request, HTTPException
import redis
import json
app = FastAPI()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
@app.post('/webhook/idempotent')
async def idempotent_webhook(request: Request):
payload = await request.json()
# Extract idempotency key from header or payload
idempotency_key = request.headers.get('Idempotency-Key') or payload.get('event_id')
if not idempotency_key:
raise HTTPException(status_code=400, detail='Missing idempotency key')
redis_key = f'webhook:processed:{idempotency_key}'
# Check if already processed
if r.exists(redis_key):
print(f'Duplicate event {idempotency_key}, skipping')
return {'status': 'duplicate', 'idempotency_key': idempotency_key}
# Process event
# ... agent logic here ...
# Mark as processed (expire after 24h)
r.setex(redis_key, 86400, '1')
return {'status': 'processed', 'idempotency_key': idempotency_key}Retry Deduplication Strategy
Beyond idempotency keys, consider deduplication windows. If you receive the same event content within a short window, it is likely a retry. Compare event hashes to detect and drop retries.
import hashlib
import json
from datetime import datetime
# In-memory store; use Redis in production
recent_hashes = {}
DEDUP_WINDOW_SECONDS = 300 # 5 minutes
def is_duplicate(payload: dict) -> bool:
# Hash the event content
content = json.dumps(payload, sort_keys=True)
event_hash = hashlib.md5(content.encode()).hexdigest()
now = datetime.utcnow().timestamp()
# Clean up old entries
expired = [h for h, ts in recent_hashes.items() if now - ts > DEDUP_WINDOW_SECONDS]
for h in expired:
del recent_hashes[h]
if event_hash in recent_hashes:
return True
recent_hashes[event_hash] = now
return False
# Test
payload = {'event': 'payment.completed', 'amount': 100}
print('First:', is_duplicate(payload)) # False
print('Second:', is_duplicate(payload)) # True (duplicate)Running the Agent Asynchronously
Webhook handlers should respond quickly (under 5 seconds) and process the agent logic in the background. Use BackgroundTasks in FastAPI to avoid timeouts.
from fastapi import FastAPI, BackgroundTasks
import asyncio
app = FastAPI()
async def run_agent_job(event: str, data: dict):
print(f'Agent starting for event: {event}')
await asyncio.sleep(2) # Simulate LLM call
print(f'Agent finished for event: {event}')
@app.post('/webhook/async')
async def async_webhook(request_data: dict, background_tasks: BackgroundTasks):
event = request_data.get('event', 'unknown')
data = request_data.get('data', {})
# Respond immediately
background_tasks.add_task(run_agent_job, event, data)
return {'status': 'accepted', 'message': 'Processing in background'}Parsing Complex Payloads
Different services send different payload shapes. Write dedicated parser functions for each service so your agent always receives a normalized event object.
from dataclasses import dataclass
from typing import Optional
@dataclass
class NormalizedEvent:
event_type: str
source: str
resource_id: str
metadata: dict
def parse_github_webhook(payload: dict) -> NormalizedEvent:
return NormalizedEvent(
event_type='github.' + payload.get('action', 'unknown'),
source='github',
resource_id=str(payload.get('repository', {}).get('id', '')),
metadata={
'repo': payload.get('repository', {}).get('full_name'),
'sender': payload.get('sender', {}).get('login')
}
)
def parse_stripe_webhook(payload: dict) -> NormalizedEvent:
return NormalizedEvent(
event_type=payload.get('type', 'unknown'),
source='stripe',
resource_id=payload.get('id', ''),
metadata={'amount': payload.get('data', {}).get('object', {}).get('amount')}
)
# Usage
github_payload = {'action': 'opened', 'repository': {'id': 123, 'full_name': 'user/repo'}, 'sender': {'login': 'alice'}}
event = parse_github_webhook(github_payload)
print(event)Webhook Response Codes Matter
Return the correct HTTP status. A 2xx tells the sender the webhook was accepted. A 4xx means a client error (bad payload). A 5xx or timeout causes the sender to retry.
- 200: Accepted and processed
- 202: Accepted for async processing
- 400: Bad request (missing fields)
- 401: Bad signature
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
@app.post('/webhook/proper-responses')
async def proper_webhook(request: Request):
try:
payload = await request.json()
except Exception:
raise HTTPException(status_code=400, detail='Invalid JSON body')
required_fields = ['event', 'data']
for field in required_fields:
if field not in payload:
raise HTTPException(status_code=400, detail=f'Missing field: {field}')
event = payload['event']
known_events = ['email.received', 'file.uploaded', 'payment.completed']
if event not in known_events:
# Acknowledge unknown events gracefully - do not retry
return JSONResponse(status_code=200, content={'status': 'ignored', 'reason': 'unknown event'})
# Start background processing
return JSONResponse(status_code=202, content={'status': 'accepted'})Testing Webhooks Locally
Use ngrok to expose your local server to the internet for testing. Run ngrok http 8000 to get a public URL that tunnels to your local FastAPI app.
# Start your FastAPI app
# uvicorn main:app --reload --port 8000
# In another terminal, start ngrok:
# ngrok http 8000
# You get: https://abc123.ngrok.io
# Now configure your webhook in Stripe/GitHub/etc. to:
# https://abc123.ngrok.io/webhook
# Test with curl:
import subprocess
def test_webhook_locally():
test_payload = '{"event": "email.received", "data": {"from": "test@example.com"}}'
# In real usage you would run this in terminal:
# curl -X POST http://localhost:8000/webhook \
# -H 'Content-Type: application/json' \
# -d '{"event": "email.received", "data": {"from": "test@example.com"}}'
print('Test payload:', test_payload)
print('Send to: http://localhost:8000/webhook')
test_webhook_locally()Logging Webhook Events
Log every incoming webhook with timestamp, source, event type, and processing result. This audit trail is essential for debugging missed events or duplicate processing issues.
import logging
import json
from datetime import datetime
import sys
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
stream=sys.stdout
)
logger = logging.getLogger('webhook')
def log_webhook_event(event_id: str, event_type: str, source: str, status: str, details: dict = None):
logger.info(json.dumps({
'timestamp': datetime.utcnow().isoformat(),
'event_id': event_id,
'event_type': event_type,
'source': source,
'status': status,
'details': details or {}
}))
# Usage in webhook handler
log_webhook_event(
event_id='evt_123',
event_type='email.received',
source='gmail',
status='processed',
details={'from': 'user@example.com', 'action_taken': 'reply_sent'}
)Rate Limiting Incoming Webhooks
Protect your webhook endpoint from being overwhelmed by using rate limiting. The slowapi library adds rate limiting to FastAPI with minimal code.
from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.post('/webhook/limited')
@limiter.limit('100/minute')
async def rate_limited_webhook(request: Request):
payload = await request.json()
return {'status': 'accepted', 'event': payload.get('event')}Knowledge Check: Webhooks
Test your understanding of webhook best practices for agents.
Webhooks in Production
In production, combine all the patterns: signature verification, idempotency keys, background processing, structured logging, and rate limiting. Deploy behind a reverse proxy like nginx for TLS termination and additional protection.
Frequently asked questions
Is the “Connecting Agents to Webhooks” lesson free?
Yes — the full text of “Connecting Agents to Webhooks” is free to read here on the web, and the AI Agents course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Agents course, upgrade to CoddyKit PRO.
What will I learn in “Connecting Agents to Webhooks”?
Receiving webhook events and triggering agent workflows in response. You practise AI Agents with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Agents?
No prior experience is required. AI Agents on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Connecting Agents to Webhooks” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Agents lesson?
Yes. Every AI Agents lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Trigger-Action Agent Patterns
- Connecting Agents to Webhooks
- Scheduling and Cron-Based Agents
- Building a Multi-App Automation Pipeline