Stripe Webhook Agent: Handling Payments, Subscriptions, and Invoice Events

Why Stripe Webhooks Need an Agent

Stripe's webhook system delivers dozens of event types — from successful payments to failed charges, subscription renewals to invoice disputes. Each event type requires different handling logic: updating your database, notifying the customer, alerting the finance team, or triggering downstream workflows.

An AI agent sitting on these webhooks can go beyond simple if-else routing. It can analyze payment failure patterns across customers, draft personalized dunning emails for failed subscriptions, detect potentially fraudulent charges, and summarize billing activity for your finance team. The agent adds intelligence to what would otherwise be mechanical event processing.

Verifying Stripe Signatures

Stripe uses a specific signature scheme. Never skip verification — without it, anyone can send fake events to your endpoint.

flowchart LR
    CLIENT(["Client SDK"])
    GW["API Gateway<br/>auth plus rate limit"]
    APP["FastAPI app<br/>handlers and DI"]
    VAL["Pydantic validation"]
    SVC["Service layer<br/>business logic"]
    DB[(Database)]
    QUEUE[(Background queue)]
    OBS[(Tracing)]
    CLIENT --> GW --> APP --> VAL --> SVC
    SVC --> DB
    SVC --> QUEUE
    SVC --> OBS
    SVC --> CLIENT
    style GW fill:#4f46e5,stroke:#4338ca,color:#fff
    style APP fill:#f59e0b,stroke:#d97706,color:#1f2937
    style DB fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b

import os
import stripe
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks

app = FastAPI()

STRIPE_WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

@app.post("/stripe/webhook")
async def stripe_webhook(request: Request, background_tasks: BackgroundTasks):
    body = await request.body()
    signature = request.headers.get("Stripe-Signature", "")

    try:
        event = stripe.Webhook.construct_event(
            body, signature, STRIPE_WEBHOOK_SECRET
        )
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=401, detail="Invalid signature")
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid payload")

    background_tasks.add_task(route_stripe_event, event)
    return {"status": "accepted"}

The stripe.Webhook.construct_event method handles signature verification and payload parsing in one step. It raises specific exceptions for invalid signatures versus malformed payloads.

Event Router with Handler Registry

Map each Stripe event type to a handler function. Use a registry pattern so adding new handlers is a one-line change.

from openai import AsyncOpenAI
from datetime import datetime

llm = AsyncOpenAI()

EVENT_HANDLERS = {}

def handles(*event_types: str):
    def decorator(func):
        for event_type in event_types:
            EVENT_HANDLERS[event_type] = func
        return func
    return decorator

async def route_stripe_event(event: dict):
    event_type = event["type"]
    handler = EVENT_HANDLERS.get(event_type)
    if handler:
        await handler(event["data"]["object"], event)
    else:
        print(f"Unhandled event type: {event_type}")

The decorator pattern lets you annotate handler functions with the event types they handle, keeping registration close to the implementation.

Handling Payment Events

Process successful and failed payment intents with AI-powered analysis.

@handles("payment_intent.succeeded")
async def handle_payment_success(payment_intent: dict, event: dict):
    amount = payment_intent["amount"] / 100
    currency = payment_intent["currency"].upper()
    customer_id = payment_intent.get("customer")

    await update_order_status(payment_intent["id"], "paid")

    if amount > 500:
        await notify_sales_team(
            f"High-value payment received: {currency} {amount:.2f} "
            f"from customer {customer_id}"
        )

@handles("payment_intent.payment_failed")
async def handle_payment_failure(payment_intent: dict, event: dict):
    customer_id = payment_intent.get("customer")
    failure_code = payment_intent.get("last_payment_error", {}).get("code", "unknown")
    failure_msg = payment_intent.get("last_payment_error", {}).get("message", "")

    history = await get_customer_payment_history(customer_id)

    prompt = f"""A payment failed for customer {customer_id}.
Failure code: {failure_code}
Failure message: {failure_msg}
Recent payment history: {history}

Draft a brief, empathetic customer notification email explaining
the issue and suggesting next steps. Keep it under 100 words."""

    response = await llm.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    email_body = response.choices[0].message.content
    await send_customer_email(customer_id, "Payment Update", email_body)

Subscription Lifecycle Management

Track subscription changes and handle dunning for failed renewals.

@handles("customer.subscription.updated")
async def handle_subscription_update(subscription: dict, event: dict):
    status = subscription["status"]
    customer_id = subscription["customer"]
    plan = subscription["items"]["data"][0]["price"]["id"]

    await update_subscription_record(customer_id, status, plan)

    if status == "past_due":
        prompt = f"""A subscription for customer {customer_id} is now past due.
Plan: {plan}
Write a friendly dunning email reminding them to update their
payment method. Include urgency but remain professional."""

        response = await llm.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
        )
        await send_customer_email(
            customer_id, "Action Required: Update Payment Method",
            response.choices[0].message.content,
        )

@handles("customer.subscription.deleted")
async def handle_subscription_cancelled(subscription: dict, event: dict):
    customer_id = subscription["customer"]
    await deactivate_customer_access(customer_id)
    await schedule_winback_campaign(customer_id, delay_days=7)

State Management and Failure Recovery

Stripe guarantees at-least-once delivery, so your handlers must be idempotent. Track processed events and implement graceful failure recovery.

import redis.asyncio as redis

redis_client = redis.Redis(host="localhost", port=6379, db=1)

async def route_stripe_event_safe(event: dict):
    event_id = event["id"]
    lock_key = f"stripe:lock:{event_id}"
    processed_key = f"stripe:processed:{event_id}"

    if await redis_client.exists(processed_key):
        return

    lock = await redis_client.set(lock_key, "1", nx=True, ex=300)
    if not lock:
        return

    try:
        handler = EVENT_HANDLERS.get(event["type"])
        if handler:
            await handler(event["data"]["object"], event)
        await redis_client.set(processed_key, "1", ex=604800)  # 7 days
    except Exception as e:
        await redis_client.delete(lock_key)
        await store_failed_event(event, str(e))
        raise
    finally:
        await redis_client.delete(lock_key)

The lock prevents concurrent processing of the same event, and the processed key prevents reprocessing after success. Failed events are stored separately for manual review or automatic retry.

FAQ

Which Stripe events should I listen to at minimum?

Start with payment_intent.succeeded, payment_intent.payment_failed, customer.subscription.created, customer.subscription.updated, customer.subscription.deleted, and invoice.payment_failed. These cover the critical payment and subscription lifecycle events.

How do I handle Stripe webhook retries?

Stripe retries failed webhooks (non-2xx responses) with exponential backoff for up to 3 days. Your handler must be idempotent, and you should return 200 quickly even if processing takes time. Use the event ID for deduplication.

Should I use Stripe's event API instead of webhooks?

The Events API is useful for backfilling missed events or verifying webhook data. Best practice is to use webhooks for real-time processing and the Events API as a fallback to poll for any events your webhook receiver might have missed during downtime.

---

#Stripe #Webhooks #PaymentProcessing #AIAgents #FastAPI #AgenticAI #LearnAI #AIEngineering