How to Integrate FastAPI for banking with LangChain for production AI
Combining FastAPI for banking with LangChain gives you a clean way to expose regulated banking workflows through HTTP while letting an LLM orchestrate customer support, document retrieval, and internal decision flows. The pattern is simple: FastAPI handles the API surface, auth, validation, and auditability; LangChain handles tool use, prompt orchestration, and agent logic.
This is the setup you want when building production AI for banking operations like balance inquiries, transaction explanations, dispute triage, or policy Q&A with controlled access to backend systems.
Prerequisites
- •Python 3.10+
- •A FastAPI app already running or ready to scaffold
- •
uvicorninstalled for local serving - •
langchain,langchain-openai, andpydanticinstalled - •An OpenAI API key or another LangChain-compatible model provider
- •Banking backend endpoints available behind your FastAPI layer
- •Basic auth in place for your banking APIs
- •Logging and request tracing enabled for audit trails
Install the core packages:
pip install fastapi uvicorn langchain langchain-openai pydantic httpx
Integration Steps
1) Define your banking API in FastAPI
Start by exposing a narrow set of banking operations through FastAPI. Keep the contract explicit with Pydantic models so LangChain only interacts with validated inputs.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
app = FastAPI(title="Banking API")
class AccountLookupRequest(BaseModel):
account_id: str = Field(..., min_length=6)
class AccountBalanceResponse(BaseModel):
account_id: str
currency: str
available_balance: float
@app.post("/accounts/balance", response_model=AccountBalanceResponse)
async def get_balance(req: AccountLookupRequest):
# Replace this with a real call to core banking or ledger service.
if req.account_id == "000000":
raise HTTPException(status_code=404, detail="Account not found")
return AccountBalanceResponse(
account_id=req.account_id,
currency="USD",
available_balance=12450.75,
)
This endpoint is your controlled banking boundary. Everything else in the system should talk to this API instead of touching core systems directly.
2) Wrap the FastAPI endpoint as a LangChain tool
LangChain agents work best when you expose narrow tools with clear input/output behavior. Use StructuredTool so the model gets typed parameters instead of free-form text parsing.
import httpx
from pydantic import BaseModel, Field
from langchain_core.tools import StructuredTool
class BalanceToolInput(BaseModel):
account_id: str = Field(..., description="Customer bank account identifier")
async def fetch_balance(account_id: str) -> str:
async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
resp = await client.post("/accounts/balance", json={"account_id": account_id})
resp.raise_for_status()
data = resp.json()
return (
f"Account {data['account_id']} has "
f"{data['available_balance']} {data['currency']} available."
)
balance_tool = StructuredTool.from_function(
coroutine=fetch_balance,
name="get_account_balance",
description="Fetches the current available balance for a bank account.",
args_schema=BalanceToolInput,
)
This keeps the LLM from inventing API calls. The tool is just a wrapper around your FastAPI endpoint.
3) Build a LangChain agent that can call the banking tool
Now connect the tool to an agent. Use a chat model and bind the tool so the model can decide when to call it.
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [balance_tool]
async def run_agent():
model_with_tools = llm.bind_tools(tools)
messages = [
HumanMessage(content="What is the balance for account 123456?")
]
response = await model_with_tools.ainvoke(messages)
print(response)
if __name__ == "__main__":
asyncio.run(run_agent())
In production, you would add guardrails before execution:
- •authenticate the caller at FastAPI
- •map user identity to permitted accounts
- •log every tool invocation
- •block sensitive actions unless explicitly approved
4) Add request validation and authorization at the FastAPI edge
Do not let LangChain become a backdoor into your banking systems. Enforce access control in FastAPI before any downstream call happens.
from fastapi import Depends, Header
def require_api_key(x_api_key: str = Header(...)):
if x_api_key != "bank-prod-secret":
raise HTTPException(status_code=401, detail="Unauthorized")
return True
@app.post("/accounts/balance", response_model=AccountBalanceResponse)
async def get_balance(req: AccountLookupRequest, authorized: bool = Depends(require_api_key)):
if req.account_id == "000000":
raise HTTPException(status_code=404, detail="Account not found")
return AccountBalanceResponse(
account_id=req.account_id,
currency="USD",
available_balance=12450.75,
)
If you skip this layer and trust the agent alone, you will eventually ship an access control bug. Banking systems need server-side enforcement every time.
5) Orchestrate multi-step banking workflows with LangChain
Once one tool works, extend it into workflows like balance lookup plus transaction explanation or dispute triage. LangChain chains can route between tools based on intent.
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "You are a banking assistant. Only use approved tools."),
("human", "{input}")
])
chain = prompt | llm.bind_tools([balance_tool])
async def answer(query: str):
result = await chain.ainvoke({"input": query})
return result
# Example:
# await answer("Check balance for account 123456")
For production AI, keep each step small:
- •one endpoint per sensitive operation
- •one tool per endpoint or business capability
- •one audit record per agent action
Testing the Integration
Run FastAPI first:
uvicorn main:app --reload --port 8000
Then test the tool directly:
import asyncio
async def test():
result = await fetch_balance("123456")
print(result)
asyncio.run(test())
Expected output:
Account 123456 has 12450.75 USD available.
If you want to verify end-to-end behavior through the agent layer, send a natural language request like:
import asyncio
async def test_agent():
response = await run_agent()
print(response)
asyncio.run(test_agent())
Expected behavior:
- •The model identifies that it needs account data.
- •It calls
get_account_balance. - •It returns a formatted banking response based on the API result.
Real-World Use Cases
- •Customer support agents that answer balance questions, recent activity summaries, and fee explanations without exposing core systems directly.
- •Internal ops assistants that help analysts triage payment disputes by querying approved banking APIs and summarizing results.
- •Compliance workflows that combine policy retrieval with transactional checks before escalating suspicious activity.
The main point is this: FastAPI gives you control and auditability; LangChain gives you orchestration. Put them together behind strict auth boundaries and you get an AI system that can actually survive contact with production banking requirements.
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