How to Integrate FastAPI for pension funds with PostgreSQL for production AI

Combining FastAPI for pension funds with PostgreSQL gives you a clean way to expose regulated pension workflows through APIs while keeping durable, queryable state in a relational store. For production AI systems, that means your agent can fetch member records, write audit trails, and persist model outputs without turning your app into an in-memory prototype.

Prerequisites

• •Python 3.11+
• •A running PostgreSQL 15+ instance
• •A FastAPI app wired for your pension-fund workflow endpoints
• •psycopg or asyncpg installed for PostgreSQL access
• •sqlalchemy[asyncio] if you want ORM + async sessions
• •Environment variables configured:
  • •DATABASE_URL
  • •PENSION_API_KEY
  • •PENSION_API_BASE_URL
• •A database role with least-privilege access
• •Network access from your FastAPI service to PostgreSQL

Integration Steps

1. •Set up your PostgreSQL connection layer.

Use an async engine so your API can handle concurrent requests without blocking. For production AI workloads, this matters when the agent is doing retrieval, tool calls, and persistence in the same request path.

# db.py
import os
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import declarative_base

DATABASE_URL = os.environ["DATABASE_URL"]

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Base = declarative_base()

2. •Define the tables you need for pension data and AI traces.

Keep the schema explicit. In regulated systems, you want member data, request metadata, and agent output separated so audits are straightforward.

# models.py
from sqlalchemy import String, Integer, DateTime, Text, func
from sqlalchemy.orm import Mapped, mapped_column
from db import Base

class PensionMember(Base):
    __tablename__ = "pension_members"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    member_number: Mapped[str] = mapped_column(String(64), unique=True, index=True)
    full_name: Mapped[str] = mapped_column(String(200), nullable=False)
    status: Mapped[str] = mapped_column(String(50), nullable=False)
    contribution_balance: Mapped[str] = mapped_column(String(50), nullable=False)

class AgentAuditLog(Base):
    __tablename__ = "agent_audit_logs"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    request_id: Mapped[str] = mapped_column(String(64), index=True)
    action: Mapped[str] = mapped_column(String(100), nullable=False)
    payload: Mapped[str] = mapped_column(Text, nullable=False)
    created_at: Mapped[DateTime] = mapped_column(DateTime(timezone=True), server_default=func.now())

3. •Wire FastAPI for pension funds to use the PostgreSQL session.

The pattern here is simple: inject a DB session into each request handler, call the pension-fund API if needed, then persist the result in PostgreSQL.

# main.py
import os
import httpx
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from db import AsyncSessionLocal, engine, Base
from models import PensionMember, AgentAuditLog

app = FastAPI()

PENSION_API_BASE_URL = os.environ["PENSION_API_BASE_URL"]
PENSION_API_KEY = os.environ["PENSION_API_KEY"]

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.get("/members/{member_number}")
async def get_member(member_number: str, db: AsyncSession = Depends(get_db)):
    stmt = select(PensionMember).where(PensionMember.member_number == member_number)
    result = await db.execute(stmt)
    member = result.scalar_one_or_none()

    if not member:
        raise HTTPException(status_code=404, detail="Member not found")

    return {
        "member_number": member.member_number,
        "full_name": member.full_name,
        "status": member.status,
        "contribution_balance": member.contribution_balance,
    }

4. •Add a handler that calls the pension service and stores the response in PostgreSQL.

This is where the integration becomes useful for production AI. Your agent can call the pension API for fresh data, then write both the business record and the audit trail into Postgres.

@app.post("/sync-member/{member_number}")
async def sync_member(member_number: str, db: AsyncSession = Depends(get_db)):
    headers = {"Authorization": f"Bearer {PENSION_API_KEY}"}

    async with httpx.AsyncClient(base_url=PENSION_API_BASE_URL) as client:
        resp = await client.get(f"/members/{member_number}", headers=headers)

    if resp.status_code != 200:
        raise HTTPException(status_code=502, detail="Pension API unavailable")

    data = resp.json()

    stmt = select(PensionMember).where(PensionMember.member_number == member_number)
    result = await db.execute(stmt)
    member = result.scalar_one_or_none()

    if member:
        member.full_name = data["full_name"]
        member.status = data["status"]
        member.contribution_balance = str(data["contribution_balance"])
    else:
        member = PensionMember(
            member_number=member_number,
            full_name=data["full_name"],
            status=data["status"],
            contribution_balance=str(data["contribution_balance"]),
        )
        db.add(member)

    db.add(
        AgentAuditLog(
            request_id=data.get("request_id", member_number),
            action="sync_member",
            payload=str(data),
        )
    )

    await db.commit()
    return {"ok": True, "member_number": member_number}

5. •Add a small AI-facing retrieval endpoint for downstream agents.

This is the pattern I use when an agent needs structured context before generating an answer or taking an action. PostgreSQL becomes the source of truth; FastAPI is just the control plane.

@app.get("/agent/context/{member_number}")
async def get_agent_context(member_number: str, db: AsyncSession = Depends(get_db)):
    stmt = select(PensionMember).where(PensionMember.member_number == member_number)
    result = await db.execute(stmt)
    
    member = result.scalar_one_or_none()
    if not member:
        raise HTTPException(status_code=404, detail="No context available")

    return {
        "member_number": member.member_number,
        "status": member.status,
        "contribution_balance": float(member.contribution_balance),
        "tooling_hint": "Use this record as authoritative context before generating advice.",
    }

Testing the Integration

Run a quick smoke test against your API after starting FastAPI and pointing it at PostgreSQL.

import httpx

base_url = "http://localhost:8000"

with httpx.Client() as client:
    sync_resp = client.post(f"{base_url}/sync-member/M12345")
    print("sync:", sync_resp.status_code, sync_resp.json())

    ctx_resp = client.get(f"{base_url}/agent/context/M12345")
    print("context:", ctx_resp.status_code, ctx_resp.json())

Expected output:

sync: 200 {'ok': True, 'member_number': 'M12345'}
context: 200 {'member_number': 'M12345', 'status': 'active', 'contribution_balance': 12500.0, 'tooling_hint': 'Use this record as authoritative context before generating advice.'}

Real-World Use Cases

• •Member servicing agents that answer balance and status questions using live pension records from PostgreSQL.
• •Compliance workflows that log every AI-driven action with timestamps and request IDs for audit review.
• •Claims or retirement planning assistants that fetch policy context from FastAPI endpoints and persist recommendations back into Postgres for later review.

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.