How to Integrate FastAPI for payments with LangChain for startups
FastAPI gives you the payment API surface: create checkout sessions, confirm transactions, and expose webhooks. LangChain gives your agent the reasoning layer: it can decide when to charge a card, verify a subscription, or refund an order based on user intent and business rules.
That combination is useful when you want an AI agent that can do more than answer questions. It can quote a price, collect payment, validate the result, and continue the workflow without bouncing between disconnected services.
Prerequisites
- •Python 3.10+
- •A FastAPI app set up with:
- •
fastapi - •
uvicorn - •a payments provider SDK such as
stripe
- •
- •A LangChain setup with:
- •
langchain - •
langchain-openaior another chat model provider
- •
- •Environment variables configured:
- •
STRIPE_SECRET_KEY - •
STRIPE_WEBHOOK_SECRET - •
OPENAI_API_KEY
- •
- •A public webhook URL for local testing:
- •
ngrok,cloudflared, or a deployed staging endpoint
- •
Integration Steps
- •Install dependencies and create the FastAPI app.
pip install fastapi uvicorn stripe langchain langchain-openai pydantic
# app.py
from fastapi import FastAPI
app = FastAPI()
- •Add payment endpoints using the Stripe SDK.
Use Stripe Checkout for startup flows because it reduces PCI scope and keeps card handling out of your app.
# app.py
import os
import stripe
from pydantic import BaseModel
from fastapi import FastAPI, HTTPException
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
app = FastAPI()
class CheckoutRequest(BaseModel):
email: str
amount_cents: int
@app.post("/payments/checkout")
def create_checkout_session(payload: CheckoutRequest):
try:
session = stripe.checkout.Session.create(
mode="payment",
customer_email=payload.email,
line_items=[{
"price_data": {
"currency": "usd",
"product_data": {"name": "AI Agent Credit Pack"},
"unit_amount": payload.amount_cents,
},
"quantity": 1,
}],
success_url="https://yourapp.com/success?session_id={CHECKOUT_SESSION_ID}",
cancel_url="https://yourapp.com/cancel",
)
return {"checkout_url": session.url, "session_id": session.id}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
- •Add a webhook so your agent can trust payment state.
This is where you confirm completed payments before letting LangChain continue a paid action.
# app.py
from fastapi import Request
@app.post("/payments/webhook")
async def stripe_webhook(request: Request):
payload = await request.body()
sig_header = request.headers.get("stripe-signature")
secret = os.getenv("STRIPE_WEBHOOK_SECRET")
try:
event = stripe.Webhook.construct_event(payload, sig_header, secret)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Webhook error: {str(e)}")
if event["type"] == "checkout.session.completed":
session = event["data"]["object"]
# Persist payment status in your DB here
return {"status": "paid", "session_id": session["id"]}
return {"status": "ignored"}
- •Wire LangChain tools to call your payment API.
Use LangChain tools when the agent needs to decide whether to charge, then call your FastAPI endpoint through HTTP. The tool should stay thin; business rules belong in your API.
# agent.py
import os
import requests
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
PAYMENT_API_BASE = os.getenv("PAYMENT_API_BASE", "http://localhost:8000")
@tool
def create_payment_checkout(email: str, amount_cents: int) -> str:
"""Create a Stripe Checkout session via FastAPI."""
response = requests.post(
f"{PAYMENT_API_BASE}/payments/checkout",
json={"email": email, "amount_cents": amount_cents},
timeout=10,
)
response.raise_for_status()
data = response.json()
return data["checkout_url"]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [create_payment_checkout]
- •Build an agent flow that only proceeds after payment confirmation.
A startup-friendly pattern is: user asks for a paid action, agent creates checkout link, backend records intent, webhook confirms payment, then the agent unlocks the next step.
# orchestrator.py
from langchain.agents import initialize_agent, AgentType
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.OPENAI_FUNCTIONS,
verbose=True,
)
def handle_user_request(message: str):
prompt = f"""
You are a billing assistant.
If the user wants to buy credits, call create_payment_checkout.
User message: {message}
"""
return agent.invoke({"input": prompt})
Testing the Integration
Run FastAPI:
uvicorn app:app --reload --port 8000
Then test the checkout endpoint directly:
import requests
response = requests.post(
"http://localhost:8000/payments/checkout",
json={"email": "founder@startup.com", "amount_cents": 2500},
)
print(response.status_code)
print(response.json())
Expected output:
200
{
"checkout_url": "https://checkout.stripe.com/c/pay/cs_test_...",
"session_id": "cs_test_..."
}
If you want to test the LangChain side, invoke the tool through the agent and confirm it returns the same checkout URL.
Real-World Use Cases
- •Paid AI onboarding
- •Let an agent qualify a lead, generate a quote, and send them to Stripe Checkout before provisioning access.
- •Usage-based billing
- •Have LangChain detect when a user exceeds free-tier limits and trigger a top-up flow through FastAPI.
- •Refund and dispute workflows
- •Let an internal support agent summarize a ticket, check payment status via your API, and initiate refunds with guardrails.
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