Production Stripe Webhooks in Python: Verification, Idempotency & Serverless Architecture

Stripe handles billions of dollars in volume, and its webhook system is designed for high resilience: if your endpoint fails to acknowledge a webhook with an HTTP 2xx status code within 30 seconds, Stripe marks the delivery as failed and will retry sending it over the next 72 hours with exponential backoff.

If you run your Stripe webhook handler on a traditional web server (like a Flask or FastAPI instance on a VPS), you face three common failure modes:
1. Downtime during deployments: If your app is restarting while Stripe delivers a checkout.session.completed event, the request fails.
2. Slow synchronous processing: If your handler attempts to generate a PDF invoice, write to a primary database, and send three emails before returning HTTP 200, Stripe can easily time out on slower connections.
3. Replay & duplicate executions: Stripe explicitly guarantees at-least-once delivery. If a temporary network blip prevents Stripe from receiving your 200 OK, it will send the identical event again. Without idempotency guards, you may double-charge or double-provision your users.

In this guide, we will implement a production-ready Python Stripe webhook handler using stripe-python and deploy it on a serverless micro-endpoint.

The Stripe Webhook Architecture

Stripe Billing / Checkout Engine
                 │
                 ▼
       HTTPS POST to Webhook URL
   (Contains 'Stripe-Signature' Header)
                 │
                 ▼
      LiteLambda Micro-Sandbox
      ┌────────────────────────────────────────────────────────┐
      │ 1. Read Raw Request Body (Byte-for-byte exact)         │
      │ 2. stripe.Webhook.construct_event()                    │
      │    - Verifies HMAC-SHA256 signature against whsec_xxx  │
      │    - Checks timestamp against tolerance window (replay)│
      │ 3. Check KV Idempotency Key (event['id'])             │
      │ 4. Dispatch Event (checkout, subscription, invoice)    │
      │ 5. Return HTTP 200 {"received": True} (< 80ms)         │
      └────────────────────────────────────────────────────────┘

Verifying the Stripe Signature in Python

Stripe signs webhook events using HMAC-SHA256. The Stripe-Signature header contains:
- t: The UNIX timestamp of when the event was dispatched.
- v1: The cryptographic HMAC signature.

Crucial rule: Never parse the JSON body before verifying the signature. Python's json.loads() can reorder dictionary keys or alter whitespace, which alters the hash and causes signature verification to fail immediately.

Here is the proper verification pattern using the official stripe SDK:

import os
import stripe

def verify_and_construct_event(raw_body: str, sig_header: str) -> stripe.Event:
    """
    Validates Stripe signature and returns the typed Event object.
    Raises ValueError or SignatureVerificationError if invalid.
    """
    webhook_secret = os.environ.get("STRIPE_WEBHOOK_SECRET")
    if not webhook_secret:
        raise ValueError("Missing STRIPE_WEBHOOK_SECRET environment variable")

    # construct_event validates the HMAC and verifies that the event 
    # timestamp is within the 300-second tolerance window (replay protection)
    return stripe.Webhook.construct_event(
        payload=raw_body,
        sig_header=sig_header,
        secret=webhook_secret,
        tolerance=300
    )

Complete Production Handler: Subscriptions, Checkouts & Idempotency

The following complete handler processes the most critical Stripe events for SaaS businesses:
- checkout.session.completed: Customer completed payment; provision their account.
- customer.subscription.updated: Subscription plan upgraded, downgraded, or renewed.
- customer.subscription.deleted: Customer churned or cancelled; revoke access.
- invoice.payment_failed: Grace period alert; trigger automated dunning email.

import os
import stripe
import requests

def handler(event, context):
    """
    Production Stripe Webhook Handler.
    Runs in an isolated LiteLambda sandbox.
    """
    raw_body = event.get("body", "")
    headers = event.get("headers", {})
    sig_header = headers.get("stripe-signature") or headers.get("Stripe-Signature")

    if not sig_header:
        return {"status": 400, "error": "Missing Stripe-Signature header"}

    # 1. Verify Signature
    try:
        stripe_event = stripe.Webhook.construct_event(
            payload=raw_body,
            sig_header=sig_header,
            secret=os.environ["STRIPE_WEBHOOK_SECRET"]
        )
    except ValueError as e:
        print(f"Invalid payload: {e}")
        return {"status": 400, "error": "Invalid payload"}
    except stripe.error.SignatureVerificationError as e:
        print(f"Signature verification failed: {e}")
        return {"status": 400, "error": "Invalid signature"}

    event_id = stripe_event.get("id")
    event_type = stripe_event.get("type")
    data_object = stripe_event.get("data", {}).get("object", {})

    # 2. Enforce Idempotency
    # Ensure this exact Stripe Event ID has not already been handled
    idempotency_key = f"stripe_evt_{event_id}"
    if hasattr(context, "kv"):
        if context.kv.get(idempotency_key):
            print(f"Stripe event {event_id} already processed. Acknowledging 200.")
            return {"status": 200, "received": True, "duplicate": True}
        # Keep key for 7 days (Stripe retries up to 72 hours)
        context.kv.set(idempotency_key, "1", ttl=604800)

    print(f"🔔 Received Stripe Event: {event_type} [ID: {event_id}]")

    # 3. Route Events
    if event_type == "checkout.session.completed":
        handle_checkout_completed(data_object)

    elif event_type == "customer.subscription.updated":
        handle_subscription_updated(data_object)

    elif event_type == "customer.subscription.deleted":
        handle_subscription_cancelled(data_object)

    elif event_type == "invoice.payment_failed":
        handle_invoice_failed(data_object)

    # 4. Immediate 200 Acknowledgment
    return {"status": 200, "received": True}


def handle_checkout_completed(session: dict):
    customer_id = session.get("customer")
    customer_email = session.get("customer_details", {}).get("email")
    client_reference_id = session.get("client_reference_id") # Your internal user ID
    amount_total = session.get("amount_total")
    currency = session.get("currency")

    print(f"✅ Provisioning access for User: {client_reference_id} ({customer_email}) - {amount_total/100} {currency.upper()}")

    # Call your primary backend to activate subscription
    call_internal_api("/v1/billing/activate", {
        "user_id": client_reference_id,
        "stripe_customer_id": customer_id,
        "email": customer_email,
        "status": "active"
    })


def handle_subscription_updated(subscription: dict):
    customer_id = subscription.get("customer")
    status = subscription.get("status")
    current_period_end = subscription.get("current_period_end")

    print(f"🔄 Subscription {subscription.get('id')} status changed to: {status}")
    call_internal_api("/v1/billing/sync-subscription", {
        "stripe_customer_id": customer_id,
        "status": status,
        "renews_at": current_period_end
    })


def handle_subscription_cancelled(subscription: dict):
    customer_id = subscription.get("customer")
    print(f"⚠️ Subscription cancelled for customer: {customer_id}")
    call_internal_api("/v1/billing/revoke-access", {
        "stripe_customer_id": customer_id
    })


def handle_invoice_failed(invoice: dict):
    customer_email = invoice.get("customer_email")
    attempt_count = invoice.get("attempt_count")
    hosted_invoice_url = invoice.get("hosted_invoice_url")

    print(f"❌ Payment failed for invoice {invoice.get('id')} (Attempt {attempt_count})")
    # Trigger your dunning alert or notification
    # send_dunning_email(customer_email, hosted_invoice_url)


def call_internal_api(endpoint: str, payload: dict):
    backend_url = os.environ.get("BACKEND_API_URL")
    api_key = os.environ.get("INTERNAL_API_SECRET")
    if backend_url and api_key:
        try:
            res = requests.post(
                f"{backend_url}{endpoint}",
                json=payload,
                headers={"Authorization": f"Bearer {api_key}"},
                timeout=10
            )
            res.raise_for_status()
        except requests.RequestException as e:
            print(f"Error calling internal API {endpoint}: {e}")

Never Miss a Billing Event

Traditional web apps bundle webhook routes inside monolithic frameworks. If your database pool exhausts or a worker thread locks up, Stripe webhooks fail.

On LiteLambda, your webhook handler lives in an isolated serverless function with guaranteed 99.99% uptime, built-in key-value caching, and automatic retry protection.

Deploy your Stripe Webhook on LiteLambda in 60s →

Step-by-Step Deployment Guide

1. Requirements

Save the above code as stripe_handler.py. In the same directory, create requirements.txt:

stripe>=7.0.0
requests>=2.31.0

2. Deploy to LiteLambda

pip install litelambda-cli
litelambda login

litelambda deploy stripe-webhook stripe_handler.py \
  --requirements requirements.txt \
  --type webhook \
  --env STRIPE_WEBHOOK_SECRET="whsec_xxxxxxxxxxxxxxxxxxxx" \
  --env BACKEND_API_URL="https://api.yourdomain.com" \
  --env INTERNAL_API_SECRET="secret_token_12345"

You receive your public live webhook endpoint:

https://run.litelambda.in/w/stripe-webhook/live_wh_78e90c1a2

3. Add Endpoint to Stripe Dashboard

1. Navigate to Developers → Webhooks in the Stripe Dashboard.
2. Click + Add endpoint.
3. In Endpoint URL, paste your LiteLambda HTTPS URL.
4. Select the events you want to listen to:
5. checkout.session.completed
6. customer.subscription.updated
7. customer.subscription.deleted
8. invoice.payment_failed
9. Click Add endpoint.
10. Under Signing secret, click Reveal and copy your whsec_... value into your LiteLambda environment variables.

4. Local Testing with Stripe CLI

You can test this endpoint locally before going live using the official Stripe CLI:

stripe listen --forward-to https://run.litelambda.in/w/stripe-webhook/live_wh_78e90c1a2

In another terminal, trigger a test payment:

stripe trigger checkout.session.completed

Check the real-time execution logs in your LiteLambda console to watch the signature verify and the access provision in real time.

Architectural Comparison

Summary

Payment and subscription webhooks are mission-critical. A single missed subscription update or dropped cancellation event creates billing discrepancies and customer churn.

Decoupling your Stripe webhook handler into an isolated serverless function gives you high availability, automatic scaling, and peace of mind knowing your payment events are processed reliably 24/7.

Deploy your Stripe webhook on LiteLambda →