Code Examples
Production-ready webhook implementations for handling Sumo purchases. Copy, adapt, and deploy.
Before You Start
Make sure you have these environment variables configured:
STRIPE_SECRET_KEY STRIPE_WEBHOOK_SECRET Full Webhook Implementation
Select your language to see a complete working example. Each implementation includes signature verification, Sumo detection, user creation, and access provisioning.
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import stripe
import os
app = FastAPI()
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET")
@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
payload = await request.body()
sig_header = request.headers.get("stripe-signature")
try:
event = stripe.Webhook.construct_event(
payload, sig_header, WEBHOOK_SECRET
)
except ValueError as e:
raise HTTPException(status_code=400, detail="Invalid payload")
except stripe.error.SignatureVerificationError as e:
raise HTTPException(status_code=400, detail="Invalid signature")
if event["type"] == "checkout.session.completed":
session = event["data"]["object"]
# Check if this is a Sumo purchase
if session.get("metadata", {}).get("sumo_campaign_id"):
await handle_sumo_purchase(session)
else:
await handle_direct_purchase(session)
return JSONResponse({"status": "ok"})
async def handle_sumo_purchase(session: dict):
"""Handle a purchase that originated from Sumo Marketplace."""
# Extract customer information
customer_email = session["customer_details"]["email"]
customer_name = session["customer_details"].get("name", "")
stripe_customer_id = session["customer"]
stripe_subscription_id = session.get("subscription")
# Extract Sumo metadata
metadata = session.get("metadata", {})
sumo_campaign_id = metadata.get("sumo_campaign_id")
sumo_product_tier_id = metadata.get("sumo_product_tier_id")
quantity = int(metadata.get("sumo_quantity", 1))
# 1. Find or create user in your database
user = await find_or_create_user(
email=customer_email,
name=customer_name,
stripe_customer_id=stripe_customer_id,
source="sumo"
)
# 2. Create subscription record
await create_subscription(
user_id=user.id,
stripe_subscription_id=stripe_subscription_id,
tier_id=sumo_product_tier_id,
quantity=quantity,
status="active"
)
# 3. Provision access based on the tier
await provision_access(user.id, sumo_product_tier_id, quantity)
# 4. Send welcome email
await send_welcome_email(
email=customer_email,
name=customer_name,
tier_id=sumo_product_tier_id
)
print(f"Sumo purchase complete for {customer_email}")Implementation Checklist
Signature Verification
Always verify the webhook signature before processing. This prevents replay attacks and ensures the webhook came from Stripe.
Sumo Detection
Check for sumo_campaign_id in metadata to identify Sumo-originated purchases vs your own checkout.
Idempotent User Creation
Use findOrCreate pattern to handle duplicate webhooks gracefully without creating duplicate users.
Return 200 Quickly
Return a successful response quickly. Queue heavy operations (emails, provisioning) for background processing if needed.
Functions You'll Need to Implement
The code examples call these app-specific functions. You'll need to implement them based on your database and business logic:
| Function | Purpose |
|---|---|
findOrCreateUser() | Find user by email or create a new one with Stripe customer ID |
createSubscription() | Store subscription record linking user to Stripe subscription |
provisionAccess() | Grant features/permissions based on purchased tier |
sendWelcomeEmail() | Send onboarding email with login instructions |
Using an AI Coding Assistant?
We've prepared an LLM-optimized prompt that you can give to Claude Code, Cursor, or any AI coding assistant. It contains all the context needed to implement Sumo integration in your existing codebase.
Get AI-Ready Prompt