How to Integrate FastAPI for pension funds with LangChain for multi-agent systems

Combining FastAPI for pension funds with LangChain gives you a clean way to expose regulated pension workflows as APIs while letting LLM agents orchestrate the messy parts: document retrieval, eligibility checks, member Q&A, and case routing. The useful pattern is simple: FastAPI handles the system boundary and policy enforcement, while LangChain coordinates multiple agents across pension data sources and tools.

Prerequisites

• •Python 3.10+
• •FastAPI installed and running
• •Uvicorn for local API serving
• •LangChain installed with your chosen model provider
• •Pydantic v2
• •Access to pension fund data sources:
  • •member records API
  • •contribution history service
  • •document store for policy PDFs and forms
• •An LLM API key configured in your environment
• •Basic understanding of:
  • •async Python
  • •REST endpoints
  • •LangChain tools and agents

Install the core packages:

pip install fastapi uvicorn langchain langchain-openai pydantic httpx

Integration Steps

1. •Create the FastAPI pension fund service.

Start by exposing the pension system as a set of typed endpoints. Keep this layer boring: validation, auth, logging, and deterministic business rules only.

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

app = FastAPI(title="Pension Fund API")

class MemberQuery(BaseModel):
    member_id: str = Field(..., min_length=5)

class PensionBalanceResponse(BaseModel):
    member_id: str
    balance: float
    status: str

def verify_token():
    # Replace with real JWT or mTLS auth
    return True

@app.get("/members/{member_id}/balance", response_model=PensionBalanceResponse)
async def get_balance(member_id: str, authorized: bool = Depends(verify_token)):
    if not authorized:
        raise HTTPException(status_code=401, detail="Unauthorized")

    # Replace with real pension admin backend call
    return PensionBalanceResponse(
        member_id=member_id,
        balance=128450.75,
        status="active",
    )

2. •Wrap FastAPI endpoints as LangChain tools.

LangChain agents should not talk to your database directly. Wrap your FastAPI endpoints behind tools so every agent action goes through the same policy layer.

import httpx
from langchain_core.tools import tool

PENSION_API_BASE = "http://localhost:8000"

@tool
async def get_member_balance(member_id: str) -> str:
    """Fetch a pension member's current balance from the pension fund API."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{PENSION_API_BASE}/members/{member_id}/balance")
        resp.raise_for_status()
        data = resp.json()
        return f"Member {data['member_id']} has balance {data['balance']} and status {data['status']}"

3. •Build a multi-agent workflow in LangChain.

For pensions, one agent usually isn’t enough. Use separate agents for retrieval, compliance-style reasoning, and customer response drafting. The router agent decides which specialist to call.

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a pension operations assistant. Use tools only when needed."),
    ("human", "{input}"),
])

tools = [get_member_balance]

agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

4. •Expose the agent through FastAPI.

This is where the two systems meet. Your API becomes the entry point for internal apps, portals, or downstream orchestration services that need multi-agent reasoning over pension data.

from pydantic import BaseModel

class AgentRequest(BaseModel):
    message: str

class AgentResponse(BaseModel):
    answer: str

@app.post("/agent/query", response_model=AgentResponse)
async def query_agent(payload: AgentRequest):
    result = await executor.ainvoke({"input": payload.message})
    return AgentResponse(answer=result["output"])

5. •Add a second specialist tool for document lookup.

A real pension workflow needs policy documents, contribution rules, or retirement eligibility guidance. Add another tool so the agent can switch between structured data and unstructured knowledge.

@tool
async def search_pension_policy(query: str) -> str:
    """Search pension policy documents for relevant guidance."""
    # Replace with vector store or document search backend
    docs = {
        "retirement age": "Normal retirement age is 65 unless early retirement conditions apply.",
        "contributions": "Employer contributions are capped at plan-specific limits.",
    }

    matches = [text for key, text in docs.items() if key in query.lower()]
    return matches[0] if matches else "No matching policy found."

tools = [get_member_balance, search_pension_policy]
agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

Testing the Integration

Run FastAPI first:

uvicorn main:app --reload --port 8000

Then verify both the direct API path and the LangChain path:

import asyncio
import httpx

async def test():
    async with httpx.AsyncClient() as client:
        # Direct FastAPI check
        r1 = await client.get("http://localhost:8000/members/MBR12345/balance")
        print("FastAPI:", r1.json())

        # Agent-driven check
        r2 = await client.post(
            "http://localhost:8000/agent/query",
            json={"message": "What is the balance for member MBR12345?"}
        )
        print("Agent:", r2.json())

asyncio.run(test())

Expected output:

FastAPI: {'member_id': 'MBR12345', 'balance': 128450.75, 'status': 'active'}
Agent: {'answer': 'Member MBR12345 has balance 128450.75 and status active'}

Real-World Use Cases

• •Member service copilot
  • •Answer balance questions, contribution status checks, and policy queries from a single conversational interface.
• •Claims and retirement case triage
  • •Route incoming requests to specialist agents that validate eligibility, fetch records, and draft next-step actions.
• •Compliance-aware document assistant
  • •Search plan rules, summarize policy PDFs, and keep all external access behind FastAPI-controlled endpoints.

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.