AI products have unique billing challenges. Users expect to pay based on usage, but predicting costs is hard. Here's how to build a billing system that works.
Billing Models for AI Products
1. Token-based Billing
Charge per API token consumed:
# Track usage
async def track_usage(user_id: str, tokens: int):
await supabase.from_('usage').insert({
'user_id': user_id,
'tokens': tokens,
'timestamp': datetime.now()
})
# Report to Stripe
stripe.billing.meter_events.create(
event_name="token_usage",
payload={
"value": tokens,
"stripe_customer_id": customer_id
}
)
2. Credit-based System
Pre-purchase credits that are consumed:
# Deduct credits
async def use_credits(user_id: str, amount: int):
result = await supabase.rpc('deduct_credits', {
'user_id': user_id,
'amount': amount
})
if result.data['remaining'] < amount * 0.2:
await send_low_credits_notification(user_id)
3. Hybrid Model (Recommended)
Base subscription + usage overage:
- Base plan includes X tokens/month
- Overage charged at Y per 1000 tokens
- Enterprise: custom volume pricing
Webhook Best Practices
Idempotency
Always handle duplicate webhooks:
async def handle_webhook(event: dict):
# Check if already processed
existing = await db.events.find_one({'stripe_event_id': event['id']})
if existing:
return {"status": "already_processed"}
# Process and store
await process_event(event)
await db.events.insert_one({'stripe_event_id': event['id']})
Retry Logic
Stripe retries failed webhooks. Return 200 even if processing fails asynchronously:
@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request, background_tasks: BackgroundTasks):
event = verify_stripe_signature(request)
background_tasks.add_task(process_stripe_event, event)
return {"received": True} # Always return 200
Handling Failed Payments
Implement dunning gracefully:
1. First failure: Retry automatically
2. Second failure: Email notification
3. Third failure: Downgrade to free tier
4. After 14 days: Suspend account
async def handle_payment_failed(customer_id: str, attempt: int):
if attempt == 1:
await stripe.payment_intents.confirm(payment_intent_id)
elif attempt == 2:
await send_payment_failed_email(customer_id)
elif attempt >= 3:
await downgrade_to_free(customer_id)
Shipfastai's Built-in Billing
All of this is pre-built in Shipfastai:
- Stripe Checkout integration
- Usage tracking with Supabase
- Webhook handlers
- Customer portal
- Invoice generation
Just add your Stripe keys and you're ready to monetize.