Ваш AI-агент готов — время на нём зарабатывать. Разбираем 3 модели монетизации через Stripe: ежемесячные подписки с лимитами токенов, pay-per-use с поминутной тарификацией и prepaid-кредиты. Полная интеграция с FastAPI + Stripe Webhooks.
Архитектура монетизации: Stripe Checkout → Webhook → Database → Agent Middleware с проверкой кредитов
Первый шаг — создать Stripe Checkout Session. Пользователь нажимает «Купить», ваш FastAPI-эндпоинт создаёт сессию и возвращает URL. Stripe сам показывает красивую форму оплаты, поддерживает карты, Apple Pay, Google Pay и десятки локальных методов.
import stripe from fastapi import FastAPI, HTTPException from pydantic import BaseModel import os stripe.api_key = os.getenv("STRIPE_SECRET_KEY") app = FastAPI() class CheckoutRequest(BaseModel): user_id: str price_id: str # ID цены из Stripe Dashboard success_url: str = "https://myagent.app/success" cancel_url: str = "https://myagent.app/pricing" @app.post("/create-checkout") async def create_checkout(req: CheckoutRequest): try: session = stripe.checkout.Session.create( mode="subscription", # или "payment" для разовых line_items=[{ "price": req.price_id, "quantity": 1 }], client_reference_id=req.user_id, success_url=f"{req.success_url}?session_id={{CHECKOUT_SESSION_ID}}", cancel_url=req.cancel_url, metadata={"user_id": req.user_id} ) return {"url": session.url} except stripe.error.StripeError as e: raise HTTPException(status_code=400, detail=str(e))
Stripe отправляет webhook-события на ваш сервер при каждой важной операции: оплата прошла, подписка продлена, платёж не удался. Обрабатываем checkout.session.completed для начисления токенов/активации, и customer.subscription.deleted для отключения доступа.
from stripe import Webhook import json WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET") @app.post("/stripe-webhook") async def stripe_webhook(request: Request): payload = await request.body() sig_header = request.headers.get("stripe-signature") try: event = Webhook.construct_event(payload, sig_header, WEBHOOK_SECRET) except ValueError as e: raise HTTPException(400, "Invalid payload") except stripe.error.SignatureVerificationError: raise HTTPException(400, "Invalid signature") if event["type"] == "checkout.session.completed": session = event["data"]["object"] user_id = session["client_reference_id"] # Определяем тариф по price_id price_id = session["line_items"]["data"][0]["price"]["id"] plans = { "price_basic": {"credits": 1000, "tier": "basic"}, "price_pro": {"credits": 10000, "tier": "pro"}, "price_enterprise": {"credits": 100000, "tier": "enterprise"} } plan = plans.get(price_id, {"credits": 500, "tier": "free"}) # Обновляем пользователя в базе await db.execute( """UPDATE users SET credits = credits + :credits, plan_tier = :tier, subscription_id = :sub_id WHERE id = :uid""", {"credits": plan["credits"], "tier": plan["tier"], "sub_id": session["subscription"], "uid": user_id} ) elif event["type"] == "customer.subscription.deleted": sub = event["data"]["object"] await db.execute( "UPDATE users SET plan_tier='free' WHERE subscription_id=:sid", {"sid": sub["id"]} ) return {"status": "ok"}
Вместо фиксированной подписки можно брать деньги за фактическое использование. Stripe поддерживает metered billing: вы сообщаете usage (потраченные токены), Stripe автоматически выставляет счёт в конце периода. Реализуем middleware, считающий токены через tiktoken.
import tiktoken enc = tiktoken.get_encoding("cl100k_base") # Кодировка GPT-4/3.5 class TokenCounter: def __init__(self, user_id: str): self.user_id = user_id self.input_tokens = 0 self.output_tokens = 0 def count_input(self, text: str): self.input_tokens += len(enc.encode(text)) def count_output(self, text: str): self.output_tokens += len(enc.encode(text)) def total_cost(self) -> float: # Цены: input $3/1M токенов, output $15/1M токенов (GPT-4o) return (self.input_tokens * 3 + self.output_tokens * 15) / 1_000_000 async def report_to_stripe(self): # Отправляем usage в Stripe (subscription_item из Dashboard) stripe.SubscriptionItem.create_usage_record( self.user_stripe_sub_item_id, quantity=self.input_tokens + self.output_tokens, action="increment" ) # Middleware для FastAPI @app.middleware("http") async def token_meter(request: Request, call_next): user_id = request.headers.get("X-User-Id") if user_id: counter = TokenCounter(user_id) request.state.token_counter = counter response = await call_next(request) if user_id and hasattr(request.state, "token_counter"): await counter.report_to_stripe() return response
Самая гибкая модель — prepaid-кредиты. Пользователь покупает пакет токенов (например, 10 000 за $10), и агент списывает их по мере использования. Создаём credits-менеджер с атомарными операциями и middleware, блокирующим запросы при недостатке кредитов.
from sqlalchemy import select, update import asyncio async def deduct_credits(user_id: str, amount: int): """Атомарное списание кредитов с проверкой баланса.""" async with db.begin() as conn: result = await conn.execute( update(User).where( User.id == user_id, User.credits >= amount # Блокировка на уровне БД! ).values(credits=User.credits - amount) ) if result.rowcount == 0: raise HTTPException( status_code=402, detail="Недостаточно кредитов. Пополните баланс." ) async def check_and_run_agent(user_id: str, prompt: str): """Защитная обёртка: проверяем кредиты → выполняем → списываем.""" # Оцениваем стоимость (приблизительно) estimated_tokens = len(enc.encode(prompt)) + 500 # + запас на ответ # Предварительная блокировка (оптимистичная) await deduct_credits(user_id, estimated_tokens) try: result = await agent.ainvoke({"messages": [("user", prompt)]}) actual_cost = len(enc.encode(result["messages"][-1].content)) # Возвращаем переплату if actual_cost < estimated_tokens: refund = estimated_tokens - actual_cost await db.execute( update(User).where(User.id == user_id).values(credits=User.credits + refund) ) return result except Exception: # Возвращаем средства при ошибке await db.execute( update(User).where(User.id == user_id).values(credits=User.credits + estimated_tokens) ) raise
Stripe Customer Portal позволяет пользователям самостоятельно управлять подпиской: менять тариф, обновлять карту, просматривать историю платежей. Это избавляет вас от написания биллинг-интерфейса. Создаём portal session и возвращаем URL.
@app.post("/create-portal-session") async def create_portal(user_id: str): # Получаем stripe_customer_id из базы user = await db.get_user(user_id) portal_session = stripe.billing_portal.Session.create( customer=user.stripe_customer_id, return_url="https://myagent.app/dashboard", configuration="bpc_123456" # Настраивается в Stripe Dashboard ) return {"url": portal_session.url}
Перед выкаткой в прод тестируем webhook'и локально. Stripe CLI пробрасывает события на localhost, а тестовые карты (4242 4242...) позволяют прогнать все сценарии: успешная оплата, отказ, 3D Secure, возврат.
# Установка Stripe CLI # https://stripe.com/docs/stripe-cli # Локальный проброс webhook-событий stripe listen --forward-to localhost:8000/stripe-webhook # Триггерим тестовое событие stripe trigger checkout.session.completed # Тестовые карты: # Успех: 4242 4242 4242 4242 # Отказ: 4000 0000 0000 0002 # 3D Secure: 4000 0000 0000 3220 # CVC: любой 3 цифры Exp: любая будущая дата # Для продакшена — переменные окружения: # STRIPE_SECRET_KEY=sk_live_... # STRIPE_WEBHOOK_SECRET=whsec_... # STRIPE_PUBLISHABLE_KEY=pk_live_...
Мы реализовали полноценную систему монетизации AI-агента через Stripe: Checkout для приёма платежей, webhook-обработчик для активации подписок, metered billing для pay-per-use, кредитную систему с атомарным списанием и Customer Portal для самообслуживания. Архитектура готова к продакшену: middleware проверяет кредиты на каждом запросе, Stripe CLI позволяет протестировать все сценарии локально. Дальше — дашборд аналитики доходов, A/B-тестирование цен и интеграция с налоговым учётом через Stripe Tax.