What Are Webhooks?

Webhooks are HTTP callbacks—instead of you asking for data, the service sends data to you when something happens.

Polling: "Is there anything new? Is there anything new? Is there anything new?"

Webhooks: "I'll tell you when something new happens."

How Webhooks Work

You register a URL with a service

When an event occurs, service POSTs to your URL

Your endpoint receives and processes the event

You respond with 200 OK

Service                          Your Server
   |                                  |
   |  POST /webhook                   |
   |  {"event": "new_message", ...}   |
   |--------------------------------->|
   |                                  | Process event
   |           200 OK                 |
   |<---------------------------------|

Setting Up Webhooks

Register Your Endpoint

Most services have a webhook configuration:

curl -X POST https://api.service.com/webhooks \
  -d '{"url": "https://your-server.com/webhook"}'

Create Your Handler

from flask import Flask, request

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    event = request.json
    
    if event['type'] == 'message.new':
        handle_new_message(event['data'])
    elif event['type'] == 'user.joined':
        handle_user_joined(event['data'])
    
    return 'OK', 200

Common Event Types

{
  "type": "message.created",
  "timestamp": "2025-02-01T10:00:00Z",
  "data": {
    "id": "msg_123",
    "content": "Hello!",
    "from": "user_456"
  }
}

Webhook Security

Verify Signatures

Services sign webhooks to prove authenticity:

import hmac
import hashlib

def verify_signature(payload, signature, secret):
    expected = hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(signature, expected)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Signature')
    
    if not verify_signature(request.data, signature, WEBHOOK_SECRET):
        return 'Invalid signature', 401
    
    # Process event...

Use HTTPS

Always use HTTPS for webhook endpoints. Never HTTP.

Validate Payloads

Don't trust the data blindly:

def handle_webhook(event):
    # Validate required fields
    if 'type' not in event or 'data' not in event:
        return 'Invalid payload', 400
    
    # Validate event type
    if event['type'] not in ALLOWED_EVENTS:
        return 'Unknown event type', 400

IP Allowlisting

Some services publish webhook source IPs:

ALLOWED_IPS = ['1.2.3.4', '5.6.7.8']

if request.remote_addr not in ALLOWED_IPS:
    return 'Forbidden', 403

Handling Webhooks Reliably

Respond Quickly

Webhooks expect fast responses (usually < 30 seconds):

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    event = request.json
    
    # Queue for async processing
    task_queue.enqueue(process_event, event)
    
    # Respond immediately
    return 'OK', 200

Handle Retries

Services retry failed webhooks. Handle duplicates:

processed_events = set()

def handle_event(event):
    event_id = event['id']
    
    if event_id in processed_events:
        return  # Already processed
    
    processed_events.add(event_id)
    # Process...

Idempotency

Same event processed twice should have same result:

def handle_payment(event):
    payment_id = event['data']['id']
    
    # Check if already processed
    if db.payment_exists(payment_id):
        return
    
    # Process payment
    db.create_payment(payment_id, event['data'])

Common Webhook Patterns

Event Routing

handlers = {
    'message.created': handle_new_message,
    'message.updated': handle_message_update,
    'user.joined': handle_user_joined,
    'user.left': handle_user_left,
}

def route_event(event):
    handler = handlers.get(event['type'])
    if handler:
        handler(event['data'])

Async Processing

from celery import Celery

celery = Celery('tasks')

@celery.task
def process_webhook(event):
    # Long-running processing here
    pass

@app.route('/webhook', methods=['POST'])
def webhook():
    event = request.json
    process_webhook.delay(event)  # Async
    return 'OK', 200

Dead Letter Queue

For failed processing:

def process_with_dlq(event):
    try:
        handle_event(event)
    except Exception as e:
        dead_letter_queue.put({
            'event': event,
            'error': str(e),
            'timestamp': datetime.now()
        })

Testing Webhooks

Local Development

Use tunneling services:

# ngrok
ngrok http 5000
# Gives you: https://abc123.ngrok.io

# Register this URL with the service

Manual Testing

# Simulate a webhook
curl -X POST http://localhost:5000/webhook \
  -H "Content-Type: application/json" \
  -d '{"type": "test", "data": {}}'

Webhook Testing Tools

Many services have "send test webhook" features in their dashboards.

Debugging Webhooks

Log Everything

import logging

@app.route('/webhook', methods=['POST'])
def webhook():
    logging.info(f"Received webhook: {request.json}")
    logging.info(f"Headers: {dict(request.headers)}")
    
    try:
        handle_event(request.json)
    except Exception as e:
        logging.error(f"Webhook processing failed: {e}")
        raise

Webhook Inspection Services

Tools like RequestBin let you see exactly what's being sent.

Check Service Logs

Most services log webhook delivery attempts and responses.

Common Issues

Timeout

Your handler takes too long:

Process asynchronously

Respond fast, process later

Signature Mismatch

Check:

Using raw body, not parsed JSON

Correct secret

Correct algorithm

Missing Events