How to Integrate LangChain for payments with Stripe for RAG

Integrating LangChain for payments with Stripe gives you a clean way to build AI agents that can both reason over retrieved context and trigger real billing events. In a RAG system, that means your agent can answer questions, generate quotes, unlock premium documents, or charge for usage without leaving the workflow.

The pattern is simple: LangChain handles the retrieval and orchestration layer, while Stripe handles payment intents, checkout sessions, and webhook-driven state changes. The value is in connecting them so payment status can influence what the agent retrieves, what it says, and what actions it is allowed to take.

Prerequisites

• •Python 3.10+
• •A Stripe account with:
  • •STRIPE_SECRET_KEY
  • •STRIPE_WEBHOOK_SECRET
  • •at least one product/price configured
• •A LangChain-based app already using RAG
• •Installed packages:
  • •langchain
  • •langchain-openai
  • •stripe
  • •fastapi
  • •uvicorn
• •An embeddings/vector store setup for retrieval
• •Basic familiarity with:
  • •PaymentIntent
  • •Checkout Session
  • •webhook verification

Integration Steps

1. •Install dependencies and configure environment variables.

pip install langchain langchain-openai stripe fastapi uvicorn
export OPENAI_API_KEY="your-openai-key"
export STRIPE_SECRET_KEY="sk_test_..."
export STRIPE_WEBHOOK_SECRET="whsec_..."

2. •Create a Stripe client and a payment service wrapper.

Use Stripe’s Python SDK directly. For most agent flows, you want two primitives: creating a checkout session for one-time purchases and creating a payment intent for custom amounts.

import os
import stripe

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

def create_checkout_session(customer_email: str, price_id: str, success_url: str, cancel_url: str):
    return stripe.checkout.Session.create(
        mode="payment",
        customer_email=customer_email,
        line_items=[{"price": price_id, "quantity": 1}],
        success_url=success_url,
        cancel_url=cancel_url,
    )

def create_payment_intent(amount_cents: int, currency: str = "usd"):
    return stripe.PaymentIntent.create(
        amount=amount_cents,
        currency=currency,
        automatic_payment_methods={"enabled": True},
    )

3. •Wire payment state into your LangChain RAG flow.

A practical pattern is to gate retrieval based on whether the user has paid. Your retriever can stay the same; the access policy changes before you call it.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a support agent. Only answer using retrieved context."),
    ("human", "User question: {question}\n\nContext:\n{context}")
])

def build_rag_answer(question: str, context: str):
    chain = prompt | llm
    return chain.invoke({"question": question, "context": context})

def get_context_for_user(user_id: str, retriever):
    # Replace this with your own entitlement check from DB or webhook-updated state.
    has_paid = True

    if not has_paid:
        return "Access denied. Payment required to retrieve premium context."

    docs = retriever.invoke(user_id)
    return "\n\n".join(doc.page_content for doc in docs)

4. •Add a FastAPI endpoint to start payment and another to receive Stripe webhooks.

This is where the integration becomes production-grade. The webhook updates your entitlement store after Stripe confirms payment.

import json
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.post("/create-checkout-session")
async def checkout_session(payload: dict):
    session = create_checkout_session(
        customer_email=payload["email"],
        price_id=payload["price_id"],
        success_url=payload["success_url"],
        cancel_url=payload["cancel_url"],
    )
    return {"id": session.id, "url": session.url}

@app.post("/stripe/webhook")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    try:
        event = stripe.Webhook.construct_event(
            payload=payload,
            sig_header=sig_header,
            secret=os.environ["STRIPE_WEBHOOK_SECRET"],
        )
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

    if event["type"] == "checkout.session.completed":
        session = event["data"]["object"]
        # Persist entitlement in your DB here.
        print(f"Payment completed for {session['customer_email']}")

    return {"received": True}

5. •Connect payment-aware retrieval to the final response generation.

In practice, your agent should check entitlement first, then retrieve documents only if the user is allowed. That keeps paid content out of the prompt path.

def answer_user_question(user_id: str, question: str, retriever):
    context = get_context_for_user(user_id=user_id, retriever=retriever)

    if context.startswith("Access denied"):
        return {
            "status": "payment_required",
            "message": context,
            "next_action": "create_checkout_session",
        }

    response = build_rag_answer(question=question, context=context)
    return {
        "status": "ok",
        "answer": response.content,
    }

Testing the Integration

Run the app and hit both paths: payment creation and gated RAG answering.

# Example test script
class DummyDoc:
    def __init__(self, page_content):
        self.page_content = page_content

class DummyRetriever:
    def invoke(self, user_id):
        return [DummyDoc("Premium policy terms include expedited claims handling."), DummyDoc("SLA response time is under 2 hours.")]

result = answer_user_question(
    user_id="user_123",
    question="What does my premium plan include?",
    retriever=DummyRetriever()
)

print(result)

Expected output:

{
  'status': 'ok',
  'answer': '...'
}

If entitlement is disabled in get_context_for_user, expected output becomes:

{
  'status': 'payment_required',
  'message': 'Access denied. Payment required to retrieve premium context.',
  'next_action': 'create_checkout_session'
}

Real-World Use Cases

• •Paid document retrieval in insurance:
  • •users pay to access premium policy wording, claims guidance, or underwriting summaries.
• •Usage-based AI support:
  • •charge per report generated by an internal agent that summarizes claim files or loan packs.
• •Premium analyst workflows:
  • •let a bank customer or broker pay to unlock deeper RAG answers over research notes and product disclosures.

The main thing to keep straight is control flow. Stripe should be the source of truth for money movement; LangChain should only retrieve premium context after your app confirms entitlement from webhook-updated state or your billing database.

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.