Handling Webhooks in Python

Webhooks are the backbone of event-driven integration between services. Instead of polling an API every few seconds to check for changes, webhooks let services push updates to your application the moment something happens. This tutorial shows you how to receive, verify, and process webhooks in Python using Flask and FastAPI.

What Are Webhooks?

A webhook is an HTTP POST request sent from one application to another when a specific event occurs. Think of it as reverse API calls—instead of your code asking for data, the external service notifies you.

Consider Stripe processing a payment. Instead of querying Stripe’s API every 10 seconds to check if a payment succeeded, Stripe sends a POST request to your server the moment the payment completes. Your endpoint receives the payload, processes it, and responds.

Webhooks consist of three parts: the event type (what happened), the payload (the data), and the signature (proof the request actually came from the expected service). Understanding this triplet is essential before writing any webhook handler.

Receiving Webhooks with Flask

Flask makes webhook endpoints straightforward. A webhook handler is just another route that accepts POST requests.

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    payload = request.get_json()
    event_type = request.headers.get('X-Event-Type')
    
    print(f"Received event: {event_type}")
    print(f"Payload: {payload}")
    
    return jsonify({'status': 'received'}), 200

if __name__ == '__main__':
    app.run(port=5000)

This endpoint receives any POST request and extracts the JSON body. In production, you’d dispatch to different handlers based on event_type and process the payload accordingly.

One common pitfall: Flask’s debug server auto-reloads on file changes, which can cause your webhook handler to execute twice during development. Use reloader=False when testing webhooks locally:

app.run(port=5000, debug=True, use_reloader=False)

Receiving Webhooks with FastAPI

FastAPI provides similar functionality with type hints and async support out of the box.

from fastapi import FastAPI, Request, Header
from typing import Optional

app = FastAPI()

@app.post("/webhook")
async def handle_webhook(
    request: Request,
    x_event_type: Optional[str] = Header(None)
):
    payload = await request.json()
    
    print(f"Event: {x_event_type}")
    print(f"Payload: {payload}")
    
    return {"status": "received"}

The async def syntax lets you handle multiple webhook deliveries concurrently. This matters when your handler makes database calls or external API requests—FastAPI handles the I/O wait without blocking other requests.

FastAPI also validates incoming data automatically. If you define a Pydantic model for the expected payload, FastAPI returns a 422 error if the sender sends malformed data:

from pydantic import BaseModel

class PaymentEvent(BaseModel):
    event_id: str
    amount: int
    currency: str

@app.post("/webhook")
async def handle_payment(event: PaymentEvent):
    print(f"Payment received: {event.amount} {event.currency}")
    return {"status": "processed"}

Verifying Signatures

Signature verification is critical. Without it, anyone can fake a webhook and trick your system into processing fake events. Most services—Stripe, GitHub, Slack—sign their webhooks using a shared secret.

Stripe uses HMAC-SHA256. You compute the signature from the raw request body and compare it against the header Stripe sends.

import hmac
import hashlib
import json

def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected_sig = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected_sig, signature)

Hook this into your Flask route:

@app.route('/webhook', methods=['POST'])
def handle_stripe_webhook():
    signature = request.headers.get('Stripe-Signature', '')
    payload = request.get_data()
    
    if not verify_stripe_signature(payload, signature, STRIPE_WEBHOOK_SECRET):
        return jsonify({'error': 'invalid signature'}), 401
    
    event = json.loads(payload)
    # Process the verified event
    
    return jsonify({'status': 'received'}), 200

The hmac.compare_digest function runs in constant time, preventing timing attacks that could leak information about the expected signature.

GitHub uses a different approach. It sends the payload as-is and includes an X-Hub-Signature-256 header with an HMAC-SHA256 signature. The secret is your webhook secret configured in the GitHub UI.

import hmac
import hashlib

def verify_github_signature(payload: bytes, signature: str, secret: str) -> bool:
    computed = 'sha256=' + hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(computed, signature)

Always verify signatures before processing any webhook data. This single step prevents the majority of webhook-based attacks.

Handling Webhook Retries

Services resend webhooks if your server doesn’t respond with a 2xx status code. Stripe retries for 3 days, GitHub for 24 hours. Design your handlers to be idempotent—processing the same event multiple times shouldn’t cause duplicate actions.

Use the event ID to deduplicate:

processed_events = set()

def process_event(event_id: str, payload: dict):
    if event_id in processed_events:
        print(f"Duplicate event {event_id}, skipping")
        return
    
    # Process the event...
    processed_events.add(event_id)

In production, store processed_events in Redis or your database rather than a Python set, which loses state on restart.

Real-World Example: Stripe Payment Events

Here’s a practical Flask handler that processes multiple Stripe event types:

from flask import Flask, request, jsonify

app = Flask(__name__)

EVENT_HANDLERS = {
    'payment_intent.succeeded': handle_payment_success,
    'payment_intent.payment_failed': handle_payment_failure,
    'charge.refunded': handle_refund,
}

@app.route('/webhook', methods=['POST'])
def handle_stripe_webhook():
    signature = request.headers.get('Stripe-Signature', '')
    payload = request.get_data()
    
    if not verify_stripe_signature(payload, signature, STRIPE_WEBHOOK_SECRET):
        return jsonify({'error': 'invalid signature'}), 401
    
    event = json.loads(payload)
    event_type = event.get('type')
    event_id = event.get('id')
    
    handler = EVENT_HANDLERS.get(event_type)
    if handler:
        handler(event['data']['object'])
    
    return jsonify({'status': 'received'}), 200

def handle_payment_success(payment):
    print(f"Payment succeeded: {payment['id']}")
    # Update database, send receipt, etc.

def handle_payment_failure(payment):
    print(f"Payment failed: {payment['id']}")
    # Notify user, log for analysis

This pattern—routing events to handler functions based on type—scales well as you add more event types.

Real-World Example: GitHub Push Events

GitHub sends push events when code is pushed to a repository. Here’s a FastAPI handler:

from fastapi import FastAPI, Request, Header
import hmac
import hashlib

app = FastAPI()

@app.post("/webhook")
async def handle_github_webhook(
    request: Request,
    x_hub_signature_256: str = Header(None)
):
    payload = await request.body()
    
    if not verify_github_signature(
        payload, 
        x_hub_signature_256, 
        GITHUB_WEBHOOK_SECRET
    ):
        return {"error": "invalid signature"}, 401
    
    event_type = request.headers.get('X-GitHub-Event', 'push')
    
    if event_type == 'push':
        data = await request.json()
        commits = data.get('commits', [])
        print(f"Push with {len(commits)} commits")
    
    return {"status": "processed"}

GitHub validates that the payload actually came from your repository and hasn’t been tampered with. The signature uses your webhook secret, which you set in the repository settings.

Best Practices

1. Respond Quickly

Always return a 2xx status code quickly. Services interpret any non-2xx as a failure and retry the webhook. If your handler makes slow database calls, acknowledge receipt immediately and process asynchronously:

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    # Fast acknowledgment
    threading.Thread(target=process_async, args=(request.get_json(),)).start()
    return jsonify({'status': 'received'}), 200

2. Handle Duplicate Events

External services may send the same webhook multiple times due to network issues or retry logic. Design your processing to be idempotent—same input should produce same result regardless of how many times it’s received:

def process_payment(payment_id):
    # Check if already processed
    if PaymentLog.query.filter_by(webhook_id=payment_id).first():
        return  # Already handled, skip
    
    # Process and log
    fulfill_order(payment_id)
    PaymentLog(webhook_id=payment_id).save()

3. Verify Signatures

This is non-negotiable for security. Every major service provides a signature in the webhook headers. Always verify before processing:

• Stripe: Stripe-Signature header
• GitHub: X-Hub-Signature-256 header
• Slack: X-Slack-Signature header

Use constant-time comparison to prevent timing attacks:

import hmac

def verify(payload, signature, secret):
    expected = hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"v1={expected}", signature)

4. Log Everything

Keep records of received webhooks for debugging failed deliveries or investigating issues:

import logging

logger = logging.getLogger(__name__)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    logger.info(f"Webhook received: {request.headers.get('X-Event-ID')}")
    logger.debug(f"Payload: {request.get_json()}")
    
    # ... process ...

5. Use HTTPS

Never expose a webhook endpoint over HTTP in production. Services like Stripe and GitHub will refuse to send webhooks to non-HTTPS URLs. Use Let’s Encrypt for free TLS certificates, or your cloud provider’s automatic TLS.

6. Test Locally with Tunnels

Use tools like ngrok or Cloudflare Tunnel to expose your local development server to the internet for testing:

ngrok http 5000

This gives you a public URL you can configure as your webhook endpoint during development.

7. Store Secrets Safely

Never hardcode webhook secrets in your source code. Use environment variables:

import os

STRIPE_SECRET = os.environ.get('STRIPE_WEBHOOK_SECRET')
GITHUB_SECRET = os.environ.get('GITHUB_WEBHOOK_SECRET')

Use different secrets for development, staging, and production environments.

Summary

Webhooks transform your application from a passive data consumer into an event-driven system. Flask and FastAPI both handle webhook delivery cleanly, but FastAPI’s async support gives you an edge under load. Always verify signatures—that single step blocks most attacks. Design handlers to be idempotent, respond quickly, and test with real traffic using a tunnel tool. With these patterns, you’re ready to integrate with Stripe, GitHub, and dozens of other services that rely on webhooks for real-time communication.