How to Integrate FastAPI for fintech with LangChain for production AI

FastAPI gives you the API surface for fintech systems: auth, validation, auditability, and low-latency request handling. LangChain gives you the orchestration layer for LLM-powered workflows: retrieval, tool calling, and structured output. Put them together and you can build production AI agents that answer account questions, summarize transactions, flag suspicious activity, and trigger controlled actions through your backend.

Prerequisites

• •Python 3.10+
• •FastAPI installed and running in a virtual environment
• •Uvicorn for local serving
• •LangChain installed with a model provider package
• •A working LLM API key, such as OpenAI or Anthropic
• •Pydantic configured for request/response schemas
• •Basic understanding of REST APIs and async Python

Install the core packages:

pip install fastapi uvicorn langchain langchain-openai pydantic

Integration Steps

1. •

   Define your FastAPI fintech contract

   Keep the API boundary strict. Your AI layer should not talk to raw database objects; it should call typed endpoints that return stable schemas.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Fintech AI API")

class AccountRequest(BaseModel):
    customer_id: str = Field(..., min_length=3)

class BalanceResponse(BaseModel):
    customer_id: str
    balance: float
    currency: str

@app.post("/accounts/balance", response_model=BalanceResponse)
async def get_balance(req: AccountRequest):
    mock_db = {
        "cust_001": {"balance": 1250.75, "currency": "USD"},
        "cust_002": {"balance": 9820.10, "currency": "EUR"},
    }

    if req.customer_id not in mock_db:
        raise HTTPException(status_code=404, detail="Customer not found")

    data = mock_db[req.customer_id]
    return BalanceResponse(customer_id=req.customer_id, **data)

2. •

   Create a LangChain tool that calls the FastAPI endpoint

   In production, your agent should call your API over HTTP like any other service. This keeps auth, rate limits, logging, and policy enforcement inside FastAPI.

import httpx
from langchain_core.tools import tool

BASE_URL = "http://localhost:8000"

@tool
async def fetch_balance(customer_id: str) -> str:
    """Fetch a customer's account balance from the fintech API."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            f"{BASE_URL}/accounts/balance",
            json={"customer_id": customer_id},
        )
        response.raise_for_status()
        data = response.json()

    return f"Customer {data['customer_id']} has {data['balance']} {data['currency']}"

3. •

   Build the LangChain agent with tool calling

   Use a chat model that supports tools. The key method here is bind_tools, which lets the model decide when to call your FastAPI-backed function.

import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
)

llm_with_tools = llm.bind_tools([fetch_balance])

async def ask_agent(question: str):
    messages = [HumanMessage(content=question)]
    result = await llm_with_tools.ainvoke(messages)
    return result

4. •

   Wire the agent into a FastAPI endpoint

   This is where production integration happens. Your app receives a user request, passes it to LangChain, then returns the model output after any tool calls.

from fastapi import Body

class AgentRequest(BaseModel):
    question: str

class AgentResponse(BaseModel):
    answer: str

@app.post("/agent/ask", response_model=AgentResponse)
async def agent_ask(payload: AgentRequest = Body(...)):
    result = await ask_agent(payload.question)

    # If your model returns tool calls first, you would execute them here.
    # For simple cases with direct responses:
    return AgentResponse(answer=result.content)

5. •

   Add controlled execution for real production flows

   For fintech use cases, don’t let the model directly mutate state without guardrails. Use explicit endpoints for actions like transfers or freezes, then require deterministic validation before execution.

class TransferRequest(BaseModel):
    from_account: str
    to_account: str
    amount: float

@app.post("/transfers")
async def create_transfer(req: TransferRequest):
    if req.amount <= 0:
        raise HTTPException(status_code=400, detail="Amount must be positive")

    return {
        "status": "queued",
        "transfer_id": "trf_12345",
        "from_account": req.from_account,
        "to_account": req.to_account,
        "amount": req.amount,
    }

Testing the Integration

Run FastAPI:

uvicorn main:app --reload --port 8000

Then test both layers together:

import asyncio

async def main():
    result = await ask_agent("What is the balance for customer cust_001?")
    print(result.content)

asyncio.run(main())

Expected output:

Customer cust_001 has 1250.75 USD

If you want to verify the raw API independently:

curl -X POST http://localhost:8000/accounts/balance \
  -H "Content-Type: application/json" \
  -d '{"customer_id":"cust_001"}'

Expected output:

{"customer_id":"cust_001","balance":1250.75,"currency":"USD"}

Real-World Use Cases

• •

  Customer support agent

  • •Answer balance questions, transaction status checks, and card-related FAQs by calling internal FastAPI endpoints through LangChain tools.
• •

  Fraud triage assistant

  • •Summarize suspicious activity patterns from event APIs and generate analyst-ready reports without exposing raw database access to the model.
• •

  Operations copilot

  • •Let internal staff query account state, queue transfers, or validate onboarding documents through tightly scoped endpoints with full audit logging.

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.