How to Integrate FastAPI for payments with LangChain for production AI
Combining FastAPI for payments with LangChain gives you a clean production path for AI systems that can charge users, gate premium actions, and trigger payment workflows from agent decisions. The pattern is simple: FastAPI handles the HTTP/payment boundary, while LangChain handles reasoning, tool selection, and orchestration.
This is useful when your agent needs to do things like create invoices, unlock paid reports, or route a user into a checkout flow after an LLM decides the request requires payment.
Prerequisites
- •Python 3.10+
- •A FastAPI app set up with:
- •
fastapi - •
uvicorn - •a payment provider SDK such as Stripe
- •
- •A LangChain setup with:
- •
langchain - •
langchain-openaior your preferred model provider
- •
- •API keys configured in environment variables:
- •
STRIPE_SECRET_KEY - •
OPENAI_API_KEY
- •
- •A webhook endpoint reachable from your payment provider
- •Basic familiarity with async Python
Integration Steps
- •Install dependencies and initialize your app
pip install fastapi uvicorn stripe langchain langchain-openai pydantic python-dotenv
Create a FastAPI app that will expose payment endpoints and agent endpoints in the same service.
from fastapi import FastAPI
app = FastAPI(title="Payments + LangChain Agent")
- •Create a payment service using the Stripe SDK
Use Stripe’s Python SDK directly inside your FastAPI service. In production, keep payment logic isolated in one module so your agent only calls safe wrapper functions.
import os
import stripe
from fastapi import HTTPException
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
def create_checkout_session(amount_cents: int, currency: str = "usd", success_url: str = "", cancel_url: str = ""):
try:
session = stripe.checkout.Session.create(
mode="payment",
line_items=[
{
"price_data": {
"currency": currency,
"product_data": {"name": "AI Agent Service"},
"unit_amount": amount_cents,
},
"quantity": 1,
}
],
success_url=success_url,
cancel_url=cancel_url,
)
return session
except stripe.error.StripeError as e:
raise HTTPException(status_code=400, detail=str(e))
Expose it through FastAPI:
from pydantic import BaseModel
class CheckoutRequest(BaseModel):
amount_cents: int
currency: str = "usd"
success_url: str
cancel_url: str
@app.post("/payments/checkout")
async def checkout(req: CheckoutRequest):
session = create_checkout_session(
amount_cents=req.amount_cents,
currency=req.currency,
success_url=req.success_url,
cancel_url=req.cancel_url,
)
return {"checkout_url": session.url, "session_id": session.id}
- •Wrap the payment endpoint as a LangChain tool
This is where LangChain becomes useful. The agent can decide when to initiate payment without hardcoding business rules into prompts.
from langchain_core.tools import tool
@tool
def start_payment(amount_cents: int, email: str) -> str:
"""
Create a checkout session for an AI service purchase.
"""
session = create_checkout_session(
amount_cents=amount_cents,
currency="usd",
success_url=f"https://yourapp.com/success?email={email}",
cancel_url="https://yourapp.com/cancel",
)
return f"Checkout session created for {email}: {session.url}"
If you want stricter control, define multiple tools instead of one generic tool. For example:
- •
start_payment_for_report - •
start_payment_for_subscription - •
refund_last_charge
That keeps the agent from inventing unsupported payment flows.
- •Build the LangChain agent and connect it to the tool
Use a chat model plus tools. The model can inspect user intent and call the payment tool when needed.
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [start_payment]
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.OPENAI_FUNCTIONS,
verbose=True,
)
@app.post("/agent/chat")
async def chat(payload: dict):
user_message = payload["message"]
result = await agent.ainvoke({"input": user_message})
return {"response": result["output"]}
A production note here: don’t let the LLM directly call Stripe primitives. Keep all billing rules inside your own tool layer so you can validate amounts, enforce limits, and log every transaction.
- •Add webhook handling for payment confirmation
The checkout session creation is only half the flow. You need webhook events so your system knows when payment succeeded and can unlock the AI feature.
from fastapi import Request
@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
payload = await request.body()
sig_header = request.headers.get("stripe-signature")
endpoint_secret = os.environ["STRIPE_WEBHOOK_SECRET"]
try:
event = stripe.Webhook.construct_event(payload, sig_header, endpoint_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 here: mark user as paid in DB
return {"status": "payment_confirmed", "session_id": session["id"]}
return {"status": "ignored"}
That webhook should update your database before any premium AI workflow is allowed to continue.
Testing the Integration
Run the app:
uvicorn main:app --reload
Then test both sides of the integration.
import requests
# Test checkout creation
resp = requests.post(
"http://localhost:8000/payments/checkout",
json={
"amount_cents": 2500,
"currency": "usd",
"success_url": "https://yourapp.com/success",
"cancel_url": "https://yourapp.com/cancel",
},
)
print(resp.json())
# Test agent route
resp2 = requests.post(
"http://localhost:8000/agent/chat",
json={"message": "Charge me $25 for premium report access."},
)
print(resp2.json())
Expected output:
{
"checkout_url": "https://checkout.stripe.com/c/pay/cs_test_...",
"session_id": "cs_test_..."
}
And for the agent:
{
"response": "Checkout session created for user@example.com: https://checkout.stripe.com/c/pay/cs_test_..."
}
Real-World Use Cases
- •Paid report generation
- •The agent qualifies the request, creates a Stripe checkout session via FastAPI, then releases the report after webhook confirmation.
- •Usage-based AI billing
- •Charge users before running expensive workflows like document extraction, compliance checks, or batch summarization.
- •Human-in-the-loop premium escalation
- •Let the agent identify high-value requests and route them through a paid approval or deposit flow before continuing.
The main pattern is stable across teams: FastAPI owns trust boundaries and money movement, LangChain owns decisioning. Keep those responsibilities separate and you get an AI system that’s easier to audit, test, and ship.
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.
Want the complete 8-step roadmap?
Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.
Get the Starter Kit