How to Integrate FastAPI for retail banking with LangChain for production AI
Combining FastAPI for retail banking with LangChain gives you a clean way to expose banking workflows as API endpoints while letting an LLM reason over customer requests, policy docs, and transaction context. The useful pattern is simple: FastAPI handles auth, routing, validation, and auditability; LangChain handles tool selection, retrieval, and response generation.
For retail banking, this is the difference between a chatbot that “sounds smart” and an agent that can actually check balances, explain fees, route disputes, and stay inside your control boundaries.
Prerequisites
- •Python 3.11+
- •FastAPI installed and running
- •Uvicorn for local API serving
- •LangChain installed with your model provider package
- •Pydantic v2
- •A bank API or internal service you can call for:
- •account balances
- •transaction history
- •fee schedules
- •customer profile lookup
- •An LLM API key set in environment variables
- •Basic familiarity with:
- •
FastAPI() - •
@app.get()/@app.post() - •LangChain
ChatOpenAI,ChatPromptTemplate, and tool calling
- •
Install the core packages:
pip install fastapi uvicorn langchain langchain-openai pydantic httpx
Integration Steps
- •Create the banking API layer in FastAPI
Start by defining the bank-facing endpoints. Keep them strict: typed inputs, typed outputs, and no direct LLM logic inside the route handlers.
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import List
app = FastAPI(title="Retail Banking AI API")
class BalanceResponse(BaseModel):
account_id: str
balance: float
currency: str = "USD"
class Transaction(BaseModel):
id: str
amount: float
description: str
class TransactionsResponse(BaseModel):
account_id: str
transactions: List[Transaction]
def get_current_customer_id() -> str:
# Replace with JWT/session lookup in production
return "cust_123"
@app.get("/accounts/{account_id}/balance", response_model=BalanceResponse)
def get_balance(account_id: str, customer_id: str = Depends(get_current_customer_id)):
if customer_id != "cust_123":
raise HTTPException(status_code=403, detail="Forbidden")
return BalanceResponse(account_id=account_id, balance=4820.55)
@app.get("/accounts/{account_id}/transactions", response_model=TransactionsResponse)
def get_transactions(account_id: str, customer_id: str = Depends(get_current_customer_id)):
if customer_id != "cust_123":
raise HTTPException(status_code=403, detail="Forbidden")
return TransactionsResponse(
account_id=account_id,
transactions=[
Transaction(id="tx1", amount=-42.10, description="Card purchase"),
Transaction(id="tx2", amount=-12.99, description="Subscription"),
],
)
This gives you the bank system of record behind stable HTTP contracts.
- •Wrap the FastAPI endpoints as LangChain tools
LangChain needs callable tools it can invoke during reasoning. Use StructuredTool or @tool for explicit schemas. For production banking flows, I prefer structured tools because they make inputs predictable.
import os
import httpx
from langchain_core.tools import StructuredTool
BANK_API_BASE = os.getenv("BANK_API_BASE", "http://localhost:8000")
async def fetch_balance(account_id: str) -> dict:
async with httpx.AsyncClient() as client:
resp = await client.get(f"{BANK_API_BASE}/accounts/{account_id}/balance")
resp.raise_for_status()
return resp.json()
async def fetch_transactions(account_id: str) -> dict:
async with httpx.AsyncClient() as client:
resp = await client.get(f"{BANK_API_BASE}/accounts/{account_id}/transactions")
resp.raise_for_status()
return resp.json()
balance_tool = StructuredTool.from_function(
coroutine=fetch_balance,
name="get_account_balance",
description="Fetch the current balance for a retail banking account.",
)
transactions_tool = StructuredTool.from_function(
coroutine=fetch_transactions,
name="get_account_transactions",
description="Fetch recent transactions for a retail banking account.",
)
Notice the boundary here: LangChain calls tools; FastAPI serves data. That separation keeps your audit trail clean.
- •Build a LangChain agent that uses those tools
Now connect an LLM to the tools so it can answer customer questions by calling the right endpoint instead of guessing.
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 retail banking assistant. Never invent balances or transactions."),
("human", "{input}")
])
tools = [balance_tool, transactions_tool]
llm_with_tools = llm.bind_tools(tools)
async def answer_banking_question(user_input: str):
messages = prompt.format_messages(input=user_input)
response = await llm_with_tools.ainvoke(messages)
return response
If you want full agent orchestration with tool execution loops, use LangChain’s agent utilities on top of this pattern. For many banking use cases, explicit tool invocation is easier to govern than a fully autonomous loop.
- •Expose a single AI endpoint through FastAPI
This is the endpoint your front end or contact-center system will call. It accepts user text and returns the model’s answer after tool use.
from pydantic import BaseModel
class AskRequest(BaseModel):
question: str
class AskResponse(BaseModel):
answer: str
@app.post("/ai/ask", response_model=AskResponse)
async def ask_ai(payload: AskRequest):
result = await answer_banking_question(payload.question)
return AskResponse(answer=result.content)
In production, add:
- •request IDs for tracing
- •rate limiting per customer/session
- •policy checks before tool execution
- •redaction for sensitive fields in logs
- •Add retrieval for policy and product documents
Banking answers often depend on internal policy docs rather than live account data. Use LangChain retrieval to ground responses in approved content.
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
[
"Overdraft fee is charged once per day.",
"Disputes must be filed within 60 days of statement date.",
"Wire transfers over $10k require manual review."
],
embedding=embeddings,
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
docs = retriever.invoke("What is the overdraft fee policy?")
That pattern keeps product knowledge separate from transactional APIs. It also reduces hallucination risk because the model can cite controlled source material.
Testing the Integration
Run FastAPI first:
uvicorn main:app --reload --port 8000
Then test the AI endpoint:
import asyncio
import httpx
async def test_ai():
async with httpx.AsyncClient() as client:
resp = await client.post(
"http://localhost:8000/ai/ask",
json={"question": "What is my current balance on account 12345?"}
)
print(resp.status_code)
print(resp.json())
asyncio.run(test_ai())
Expected output:
200
{
"answer": "Your current balance on account 12345 is $4,820.55."
}
If you see a generic answer without tool-backed data, check:
- •whether the tool call succeeded
- •whether your bank endpoints are reachable from the LangChain process
- •whether auth headers are being forwarded correctly
Real-World Use Cases
- •
Balance and transaction assistants
Let customers ask natural-language questions like “What did I spend on subscriptions last month?” while FastAPI enforces identity checks and LangChain handles query interpretation. - •
Dispute triage bots
Use LangChain to classify dispute intent and retrieve policy guidance, then use FastAPI endpoints to create tickets in your case management system. - •
Fee explanation agents
Combine live account data from FastAPI with policy retrieval in LangChain so customers get grounded explanations for overdrafts, transfer fees, or card charges.
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