Webhooks: From Setup to Production (Stripe API)

Conceptual Overview

A webhook is an HTTP callback that Stripe sends to your server when an event occurs in your Stripe account. Instead of polling the Stripe API repeatedly to check for changes—which wastes bandwidth and introduces latency—webhooks push real-time notifications directly to an endpoint you control.

When a customer completes a payment, disputes a charge, upgrades a subscription, or when a recurring billing cycle finishes, Stripe sends an HTTP POST request to your configured URL. The payload is a JSON Event object containing the event type, a unique identifier, and the relevant Stripe resource nested under data.object.

Why it matters:

• Real-time reactivity: Your application can fulfill orders, activate subscriptions, or send receipts the moment something happens.
• Reduced API load: Webhooks eliminate the need for continuous polling, which can exhaust rate limits and increase operational costs.
• Reliability for post-payment logic: Client-side redirects after a checkout are not guaranteed. Webhooks are the only reliable way to know that a payment has succeeded.
• Operational visibility: Stripe's Dashboard provides delivery history, retry logs, and manual resend for every webhook endpoint.

When Not to Use Webhooks

• Point-in-time snapshots: If you're building a reporting dashboard that refreshes once per day, a daily API fetch is simpler.
• Synchronous confirmation: If your use case is "charge the customer and immediately return a success/failure response," use the synchronous Stripe API call and treat webhooks as a fallback.
• No inbound HTTP capability: Serverless functions that only run on a schedule, or tightly firewalled environments, should use Stripe's EventBridge integration or a pull-based reconciliation job.
• Bursty workloads without a queue: If your handler performs heavy synchronous work, webhook deliveries may time out. Move processing to a background queue.

Rule of thumb: If your application needs to react to Stripe events in near-real-time and you control a server that can receive HTTP requests, webhooks are the correct approach.

Prerequisites

• Stripe account — Sign up here. All examples use test mode, which is free and unlimited for development.
• Stripe secret key — starts with sk_test_ in test mode:

  export STRIPE_SECRET_KEY="sk_test_..."

• Webhook signing secret — starts with whsec_:

  export STRIPE_WEBHOOK_SECRET="whsec_..."

• Security: Never hardcode API keys or signing secrets in source files. Hardcoding credentials in committed code is the most common cause of leaked keys.
• Python 3.9+ with the Stripe library: pip install stripe
• Stripe CLI (optional but recommended) — Installation instructions.

Step-by-Step Guide

1. Create a Webhook Endpoint (Python + Flask)

# webhook_handler_basic.py — minimal Stripe webhook receiver
import os
import stripe
from flask import Flask, request, jsonify

app = Flask(__name__)
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
endpoint_secret = os.environ["STRIPE_WEBHOOK_SECRET"]

@app.route("/webhook", methods=["POST"])
def webhook():
    payload = request.get_data(as_text=True)
    sig_header = request.headers.get("Stripe-Signature")

    try:
        event = stripe.Webhook.construct_event(payload, sig_header, endpoint_secret)
    except (ValueError, stripe.error.SignatureVerificationError) as e:
        return jsonify({"error": str(e)}), 400

    if event["type"] == "payment_intent.succeeded":
        payment_intent = event["data"]["object"]
        print(f"PaymentIntent succeeded: {payment_intent['id']}")
    elif event["type"] == "customer.subscription.deleted":
        subscription = event["data"]["object"]
        print(f"Subscription deleted: {subscription['id']}")
    else:
        print(f"Unhandled event type: {event['type']}")

    return jsonify({"received": True}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=4242, debug=True)

2. Forward Events to Your Local Endpoint

stripe login
stripe listen --forward-to http://localhost:4242/webhook

3. Simulate a Test Event

stripe trigger payment_intent.succeeded

4. Register a Production Endpoint

1. Go to the Webhooks tab in the Stripe Workbench.
2. Click Add endpoint and enter your production URL (must use HTTPS).
3. For Events to send, select only the event types you actually need.
4. Reveal and copy the signing secret. Set it as STRIPE_WEBHOOK_SECRET in your production environment.
5. Deploy your endpoint code to a publicly accessible server with HTTPS configured.

Complete Example

Below is a single, self-contained webhook handler that delivers everything the prose above describes: signature verification, idempotent processing via a per-request (thread-safe) database connection, per-type event dispatch with named handler functions, and a --backfill mode for recovering missed events.

"""
complete_webhook_handler.py

A production-oriented Stripe webhook handler with:
  - Signature verification (Stripe-Signature header)
  - Idempotency via a per-request database connection (thread-safe)
  - Per-type event dispatch with dedicated handler functions
  - --backfill mode for reconciliation of missed events

Usage (start webhook server):
    export STRIPE_SECRET_KEY="sk_test_..."
    export STRIPE_WEBHOOK_SECRET="whsec_..."
    stripe listen --forward-to http://localhost:4242/webhook
    python complete_webhook_handler.py

Usage (reconciliation backfill):
    python complete_webhook_handler.py --backfill
    python complete_webhook_handler.py --backfill --hours 48
"""

import argparse
import json
import logging
import os
import sqlite3
import sys
import time
from contextlib import contextmanager

import stripe
from flask import Flask, jsonify, request

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("stripe-webhook")

STRIPE_SECRET_KEY    = os.environ.get("STRIPE_SECRET_KEY")
STRIPE_WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET")
SQLITE_PATH          = os.environ.get("SQLITE_PATH", "webhooks.db")
DATABASE_URL         = os.environ.get("DATABASE_URL")
USE_SQLITE           = os.environ.get("USE_SQLITE", "true").lower() == "true" or not DATABASE_URL

if not STRIPE_SECRET_KEY or not STRIPE_WEBHOOK_SECRET:
    logger.error("STRIPE_SECRET_KEY and STRIPE_WEBHOOK_SECRET must be set.")
    sys.exit(1)

stripe.api_key = STRIPE_SECRET_KEY


# ── Thread-safe database helpers ─────────────────────────────────────────────
# A module-level sqlite3 connection is NOT thread-safe: Flask may handle
# concurrent requests on separate threads, causing "database is locked" errors
# or silent data corruption. Open and close a fresh connection per request.

@contextmanager
def get_db():
    """Yield a new database connection scoped to the current request."""
    if USE_SQLITE:
        conn = sqlite3.connect(SQLITE_PATH)
        conn.row_factory = sqlite3.Row
        conn.execute("""
            CREATE TABLE IF NOT EXISTS processed_events (
                stripe_event_id TEXT PRIMARY KEY,
                event_type      TEXT NOT NULL,
                payload         TEXT NOT NULL,
                status          TEXT NOT NULL DEFAULT 'pending',
                created_at      TEXT NOT NULL DEFAULT (datetime('now'))
            )
        """)
        conn.commit()
        try:
            yield conn
        finally:
            conn.close()
    else:
        import psycopg2
        conn = psycopg2.connect(DATABASE_URL)
        with conn:
            with conn.cursor() as cur:
                cur.execute("""
                    CREATE TABLE IF NOT EXISTS processed_events (
                        stripe_event_id TEXT PRIMARY KEY,
                        event_type      TEXT NOT NULL,
                        payload         JSONB NOT NULL,
                        status          TEXT NOT NULL DEFAULT 'pending',
                        created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
                    )
                """)
        try:
            yield conn
        finally:
            conn.close()


def claim_event(conn, event_id: str, event_type: str, payload: str) -> bool:
    """
    Insert the event atomically. Returns True if new, False if duplicate.
    The PRIMARY KEY constraint makes the INSERT fail for duplicates without
    requiring a prior SELECT — safe under concurrent load.
    """
    try:
        if USE_SQLITE:
            conn.execute(
                "INSERT INTO processed_events "
                "(stripe_event_id, event_type, payload, status) "
                "VALUES (?, ?, ?, 'processing')",
                (event_id, event_type, payload),
            )
            conn.commit()
        else:
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO processed_events "
                    "(stripe_event_id, event_type, payload, status) "
                    "VALUES (%s, %s, %s, 'processing')",
                    (event_id, event_type, json.dumps(json.loads(payload))),
                )
            conn.commit()
        return True
    except Exception:
        return False  # duplicate


def mark_done(conn, event_id: str) -> None:
    if USE_SQLITE:
        conn.execute(
            "UPDATE processed_events SET status = 'done' "
            "WHERE stripe_event_id = ?", (event_id,)
        )
        conn.commit()
    else:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE processed_events SET status = 'done' "
                "WHERE stripe_event_id = %s", (event_id,)
            )
        conn.commit()


# ── Per-type event dispatch ───────────────────────────────────────────────────

def on_payment_intent_succeeded(obj: dict) -> None:
    logger.info(
        f"PaymentIntent succeeded: id={obj['id']} "
        f"amount={obj['amount']} currency={obj['currency']}"
    )
    # TODO: fulfill order, send receipt, update your database

def on_payment_intent_failed(obj: dict) -> None:
    err = obj.get("last_payment_error", {})
    logger.warning(
        f"PaymentIntent failed: id={obj['id']} reason={err.get('message')}"
    )
    # TODO: notify customer, retry or mark order failed

def on_checkout_session_completed(obj: dict) -> None:
    logger.info(
        f"Checkout completed: id={obj['id']} customer={obj.get('customer')}"
    )
    # TODO: provision access, send welcome email

def on_subscription_created(obj: dict) -> None:
    logger.info(
        f"Subscription created: id={obj['id']} status={obj['status']}"
    )

def on_subscription_updated(obj: dict) -> None:
    logger.info(
        f"Subscription updated: id={obj['id']} status={obj['status']}"
    )
    # TODO: update entitlements if plan changed

def on_subscription_deleted(obj: dict) -> None:
    logger.info(
        f"Subscription cancelled: id={obj['id']} customer={obj['customer']}"
    )
    # TODO: revoke access, archive customer record

def on_invoice_paid(obj: dict) -> None:
    logger.info(f"Invoice paid: id={obj['id']} amount={obj['amount_paid']}")

def on_invoice_payment_failed(obj: dict) -> None:
    logger.warning(
        f"Invoice payment failed: id={obj['id']} customer={obj.get('customer')}"
    )
    # TODO: dunning email, retry or escalate

DISPATCH: dict = {
    "payment_intent.succeeded":      on_payment_intent_succeeded,
    "payment_intent.payment_failed": on_payment_intent_failed,
    "checkout.session.completed":    on_checkout_session_completed,
    "customer.subscription.created": on_subscription_created,
    "customer.subscription.updated": on_subscription_updated,
    "customer.subscription.deleted": on_subscription_deleted,
    "invoice.paid":                  on_invoice_paid,
    "invoice.payment_failed":        on_invoice_payment_failed,
}


def process_event(event: dict, raw_payload: str) -> None:
    event_id   = event["id"]
    event_type = event["type"]

    with get_db() as conn:
        if not claim_event(conn, event_id, event_type, raw_payload):
            logger.info(f"Duplicate — skipping {event_id}")
            return
        handler = DISPATCH.get(event_type)
        if handler:
            handler(event["data"]["object"])
        else:
            logger.info(f"No handler registered for {event_type}")
        mark_done(conn, event_id)

    logger.info(f"{event_id} ({event_type}) → done")


# ── Flask server ──────────────────────────────────────────────────────────────

app = Flask(__name__)

@app.route("/webhook", methods=["POST"])
def webhook():
    raw = request.get_data(as_text=True)
    sig = request.headers.get("Stripe-Signature")
    if not sig:
        return jsonify({"error": "Missing Stripe-Signature header"}), 400
    try:
        event = stripe.Webhook.construct_event(raw, sig, STRIPE_WEBHOOK_SECRET)
    except ValueError:
        return jsonify({"error": "Invalid payload"}), 400
    except stripe.error.SignatureVerificationError:
        return jsonify({"error": "Invalid signature"}), 400

    process_event(event, raw)
    return jsonify({"received": True}), 200


# ── Reconciliation backfill ───────────────────────────────────────────────────

def run_backfill(hours: int = 24) -> None:
    """
    Fetch the last `hours` hours of Stripe events and reprocess any that
    are missing or incomplete in the idempotency table.
    Run on a cron schedule or manually after an outage.
    """
    logger.info(f"Backfill: scanning last {hours}h of events...")
    created_after = int(time.time()) - (hours * 3600)
    processed = skipped = errors = 0

    for event in stripe.Event.list(
        limit=100,
        created={"gte": created_after}
    ).auto_paging_iter():
        if event["type"] not in DISPATCH:
            continue

        event_id = event["id"]
        with get_db() as conn:
            if USE_SQLITE:
                row = conn.execute(
                    "SELECT status FROM processed_events WHERE stripe_event_id = ?",
                    (event_id,),
                ).fetchone()
            else:
                with conn.cursor() as cur:
                    cur.execute(
                        "SELECT status FROM processed_events "
                        "WHERE stripe_event_id = %s", (event_id,)
                    )
                    row = cur.fetchone()

        if row and row[0] == "done":
            skipped += 1
            continue

        try:
            process_event(event, json.dumps(dict(event)))
            processed += 1
        except Exception as e:
            logger.error(f"Backfill error for {event_id}: {e}")
            errors += 1

    logger.info(
        f"Backfill done — processed: {processed}, "
        f"skipped (already done): {skipped}, errors: {errors}"
    )


# ── Entry point ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Stripe webhook server / backfill tool")
    parser.add_argument(
        "--backfill", action="store_true",
        help="Run reconciliation backfill instead of starting the HTTP server"
    )
    parser.add_argument(
        "--hours", type=int, default=24,
        help="Hours of history to scan during backfill (default: 24)"
    )
    args = parser.parse_args()

    if args.backfill:
        run_backfill(hours=args.hours)
    else:
        logger.info("Starting webhook server on :4242")
        app.run(host="0.0.0.0", port=4242, debug=False)

What this script does: Verifies the Stripe-Signature header on every request; uses a @contextmanager to open a fresh database connection per request (avoiding the thread-safety bug of a shared module-level connection); inserts each event ID atomically to guard against duplicates; routes events to named handler functions via the DISPATCH dict; and exposes a --backfill CLI flag that calls stripe.Event.list().auto_paging_iter() to recover any events missed during an outage.

Reference

Key Event Types

Webhook Delivery Properties

Production Checklist

• HTTPS endpoint: The webhook URL must use HTTPS with a valid TLS certificate.
• Signature verification enforced: Every event is verified with constructEvent before processing.
• Idempotent processing: The handler checks whether an event ID has been processed before running business logic.
• Fast 200 response: The endpoint acknowledges the event immediately and defers heavy work to a background queue.
• Only needed event types selected: Restrict the endpoint to specific events your application handles.
• Reconciliation cron running: A scheduled job fetches recent events as a safety net for missed deliveries.
• Separate test and live secrets: The whsec_ signing secret differs between test mode and live mode.
• Monitoring and alerting: Monitor webhook delivery rates and set up alerts for spikes in 400 or 500 responses.
• API version pinning: Pin your webhook endpoint to a specific Stripe API version.

Common Errors and Troubleshooting

1. Webhook signature verification failed

• Confirm you are using the correct whsec_ secret. Use the raw request body as a string, not a parsed JSON object.

2. Webhook timeouts

• Move any work that takes more than a few hundred milliseconds to a background queue.

3. Duplicate events processed

• Implement idempotency: store each event ID in a database and check it before performing any side effects.

4. Missing events

• Use the reconciliation backfill pattern: periodically call stripe.Event.list to fetch events and process any that are missing.

5. Hardcoded secret leaked

• Roll the compromised key in the Stripe Workbench immediately. Never write raw keys in source files.

6. Endpoint auto-disabled

• Fix the underlying error and re-enable the endpoint in the Dashboard. Use reconciliation to recover missed events.

7. Event order assumptions

• Do not rely on event ordering. Always read the current state of the Stripe object via the API rather than trusting the snapshot in the event payload alone.

For additional details, refer to Stripe Webhooks Documentation, Event Object Reference, and the Production Checklist.