How to Integrate Next.js for payments with Vercel AI SDK for RAG

Combining Next.js payments with Vercel AI SDK gives you a clean path to build AI agents that can answer questions, retrieve context, and trigger billing flows in the same workflow. In practice, that means a customer can ask a RAG-powered assistant for a policy summary, then immediately pay for a report, upgrade, or service without leaving the app.

Prerequisites

• •Python 3.10+
• •A Next.js app with payment handling set up using Stripe or your payment provider of choice
• •A Vercel AI SDK app exposing a chat/completion endpoint
• •API keys configured in environment variables:
  • •STRIPE_SECRET_KEY
  • •STRIPE_WEBHOOK_SECRET
  • •OPENAI_API_KEY or equivalent model key used by Vercel AI SDK
• •A vector store or document index for RAG
• •httpx, pydantic, and python-dotenv installed

pip install httpx pydantic python-dotenv

Integration Steps

1. •

   Expose your Next.js payment flow as an API contract

   Your Next.js app should own checkout/session creation. The Python side should call the payment endpoint, not reimplement payment logic.

   In Next.js, you typically create a route handler that uses Stripe’s checkout.sessions.create() method. From Python, you’ll call that endpoint when the agent decides a paid action is needed.

import os
import httpx
from pydantic import BaseModel

class CheckoutRequest(BaseModel):
    customer_id: str
    price_id: str
    success_url: str
    cancel_url: str

async def create_checkout_session(payload: CheckoutRequest) -> dict:
    url = os.environ["NEXTJS_PAYMENTS_URL"] + "/api/checkout"
    async with httpx.AsyncClient(timeout=20) as client:
        resp = await client.post(url, json=payload.model_dump())
        resp.raise_for_status()
        return resp.json()

2. •

   Call the Vercel AI SDK RAG endpoint from Python

   Keep retrieval and generation behind one API boundary. If your Vercel AI SDK route uses streamText() or generateText(), the Python service only needs to send the user query and optional metadata.

import os
import httpx

async def ask_rag_agent(query: str, user_id: str) -> dict:
    url = os.environ["VERCEL_AI_URL"] + "/api/chat"
    headers = {
        "Authorization": f"Bearer {os.environ['AI_SERVICE_TOKEN']}",
        "Content-Type": "application/json",
    }
    payload = {
        "messages": [
            {"role": "user", "content": query}
        ],
        "metadata": {
            "user_id": user_id,
            "source": "payments-rag-agent"
        }
    }

    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.post(url, json=payload, headers=headers)
        resp.raise_for_status()
        return resp.json()

3. •

   Use RAG output to decide whether to bill

   The agent should classify intent first. If the answer requires a paid document, premium report, or transaction-heavy action, trigger checkout through the Next.js payment API.

from pydantic import BaseModel

class AgentDecision(BaseModel):
    requires_payment: bool
    product_name: str | None = None
    price_id: str | None = None

async def process_user_request(query: str, user_id: str) -> dict:
    rag_response = await ask_rag_agent(query=query, user_id=user_id)
    decision = AgentDecision.model_validate(rag_response["decision"])

    if decision.requires_payment:
        checkout = await create_checkout_session(
            CheckoutRequest(
                customer_id=user_id,
                price_id=decision.price_id,
                success_url="https://app.example.com/success",
                cancel_url="https://app.example.com/cancel",
            )
        )
        return {
            "type": "payment_required",
            "checkout_url": checkout["url"],
            "answer_preview": rag_response.get("preview", "")
        }

    return {
        "type": "answer",
        "answer": rag_response["answer"]
    }

4. •

   Wire webhook confirmation back into your agent state

   Payment completion should update your backend state so the agent knows the user is entitled to premium retrieval or generation. In Next.js/Stripe this usually comes from a webhook event such as checkout.session.completed.

import os
import stripe

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

def verify_checkout_session(session_id: str) -> dict:
    session = stripe.checkout.Session.retrieve(session_id)
    return {
        "id": session.id,
        "payment_status": session.payment_status,
        "customer_email": session.customer_details.email if session.customer_details else None,
        "metadata": session.metadata,
    }

After verification, mark the entitlement in your database and let subsequent RAG calls include premium context.

5. •

   Pass entitlement context into Vercel AI SDK requests

   Once payment is confirmed, send entitlement flags to the RAG layer so it can unlock premium documents or higher token budgets.

async def ask_premium_rag(query: str, user_id: str) -> dict:
    url = os.environ["VERCEL_AI_URL"] + "/api/chat"
    payload = {
        "messages": [{"role": "user", "content": query}],
        "metadata": {
            "user_id": user_id,
            "entitlement": "premium",
            "retrieval_scope": ["public", "paid"],
            "max_sources": 8,
        }
    }

    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.post(url, json=payload)
        resp.raise_for_status()
        return resp.json()

Testing the Integration

Use one script to simulate the full flow: ask a question, receive either an answer or a checkout URL, then confirm payment and retry with premium access.

import asyncio

async def main():
    result = await process_user_request(
        query="Generate my claims summary report",
        user_id="user_123"
    )

    print("Initial response:", result)

    if result["type"] == "payment_required":
      print("Checkout URL:", result["checkout_url"])

      # Simulate post-payment verification
      session_info = verify_checkout_session("cs_test_123")
      print("Payment status:", session_info["payment_status"])

      premium_answer = await ask_premium_rag(
          query="Generate my claims summary report",
          user_id="user_123"
      )
      print("Premium response:", premium_answer)

asyncio.run(main())

Expected output:

Initial response: {'type': 'payment_required', 'checkout_url': 'https://checkout.stripe.com/...', 'answer_preview': 'I can generate this report once payment is completed.'}
Checkout URL: https://checkout.stripe.com/...
Payment status: paid
Premium response: {'answer': 'Here is your claims summary report...', 'sources': [...]} 

Real-World Use Cases

• •

  Paid document retrieval

  • •Let users ask for contracts, policy summaries, underwriting packs, or financial statements.
  • •The agent retrieves public context first, then gates premium documents behind checkout.
• •

  Usage-based advisory agents

  • •Charge for advanced insurance advice, loan prequalification reports, or compliance summaries.
  • •The Vercel AI SDK handles retrieval and response generation while Next.js handles billing.
• •

  Internal ops assistants

  • •Build an agent for sales or support teams that can fetch account data and trigger paid workflows like report exports or third-party lookups.
  • •This works well when you need auditability around both retrieval and monetization.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.