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

Combining FastAPI for pension funds with PostgreSQL gives you a clean way to expose pension data through API endpoints while keeping the underlying records queryable, auditable, and ready for AI agent workflows. For an AI agent system, this means the agent can fetch member data, contribution history, benefit estimates, and policy rules from a single backend without hardcoding business logic into the model layer.

Prerequisites

• •Python 3.10+
• •FastAPI installed
• •PostgreSQL 14+
• •psycopg2-binary or asyncpg
• •uvicorn for running the API
• •A PostgreSQL database with tables for members, contributions, and benefit calculations
• •Environment variables configured for:
  • •DATABASE_URL
  • •PENSION_API_KEY if your FastAPI for pension funds setup requires auth
• •Basic knowledge of:
  • •FastAPI route definitions
  • •SQL queries
  • •JSON response handling

Integration Steps

1. •Set up your PostgreSQL connection layer

Use a dedicated database module so your API routes do not manage raw connection details directly. For production systems, prefer connection pooling and parameterized queries.

# db.py
import os
from psycopg2.pool import SimpleConnectionPool

DATABASE_URL = os.getenv("DATABASE_URL")

pool = SimpleConnectionPool(
    minconn=1,
    maxconn=10,
    dsn=DATABASE_URL,
)

def get_conn():
    return pool.getconn()

def put_conn(conn):
    pool.putconn(conn)

2. •Define your pension fund data access functions

Keep SQL in a repository layer. This makes it easier for your AI agent to call stable functions instead of embedding queries in prompts or route handlers.

# repositories/pension_repo.py
from db import get_conn, put_conn

def get_member_by_id(member_id: int):
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT id, full_name, status, retirement_age
                FROM pension_members
                WHERE id = %s
                """,
                (member_id,),
            )
            row = cur.fetchone()
            if not row:
                return None

            return {
                "id": row[0],
                "full_name": row[1],
                "status": row[2],
                "retirement_age": row[3],
            }
    finally:
        put_conn(conn)

def get_contributions(member_id: int):
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT contribution_date, amount
                FROM contributions
                WHERE member_id = %s
                ORDER BY contribution_date DESC
                LIMIT 12
                """,
                (member_id,),
            )
            return [
                {"contribution_date": r[0].isoformat(), "amount": float(r[1])}
                for r in cur.fetchall()
            ]
    finally:
        put_conn(conn)

3. •Expose the data through FastAPI endpoints

This is where FastAPI for pension funds becomes the interface layer for your AI agents. The agent can call these endpoints to retrieve structured pension data instead of scraping internal systems.

# main.py
from fastapi import FastAPI, HTTPException
from repositories.pension_repo import get_member_by_id, get_contributions

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

@app.get("/members/{member_id}")
def read_member(member_id: int):
    member = get_member_by_id(member_id)
    if not member:
        raise HTTPException(status_code=404, detail="Member not found")
    return member

@app.get("/members/{member_id}/contributions")
def read_contributions(member_id: int):
    return {
        "member_id": member_id,
        "contributions": get_contributions(member_id),
    }

4. •Add an AI-agent-friendly service wrapper

Your agent should not talk to the database directly unless it absolutely has to. Wrap the API calls in a service that returns normalized payloads the agent can consume.

# services/pension_service.py
import requests

class PensionService:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip("/")

    def fetch_member_profile(self, member_id: int) -> dict:
        resp = requests.get(f"{self.base_url}/members/{member_id}", timeout=10)
        resp.raise_for_status()
        return resp.json()

    def fetch_member_contributions(self, member_id: int) -> dict:
        resp = requests.get(
            f"{self.base_url}/members/{member_id}/contributions",
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

5. •Wire it into an AI agent workflow

At this point your agent can decide when to query pension data and when to summarize it. In production, I’d keep the tool boundary explicit so retrieval is deterministic and auditable.

# agent_tools.py
from services.pension_service import PensionService

pension_service = PensionService(base_url="http://localhost:8000")

def build_retirement_summary(member_id: int) -> dict:
    profile = pension_service.fetch_member_profile(member_id)
    contributions = pension_service.fetch_member_contributions(member_id)

    total_contrib = sum(item["amount"] for item in contributions["contributions"])

    return {
        "member": profile,
        "last_12_months_total_contribution": total_contrib,
        "contribution_count": len(contributions["contributions"]),
    }

Testing the Integration

Run the FastAPI app:

uvicorn main:app --reload --port 8000

Then verify both PostgreSQL access and API output with a simple test script:

# test_integration.py
from agent_tools import build_retirement_summary

result = build_retirement_summary(1)
print(result)

Expected output:

{
  "member": {
    "id": 1,
    "full_name": "Amina Moyo",
    "status": "active",
    "retirement_age": 60
  },
  "last_12_months_total_contribution": 18450.0,
  "contribution_count": 12
}

If you want a direct API check:

curl http://localhost:8000/members/1/contributions

You should receive JSON with a member_id field and a contributions array.

Real-World Use Cases

• •

  Member support agents

  • •Let an internal AI assistant answer questions like “What’s my current contribution total?” by calling FastAPI endpoints backed by PostgreSQL.
• •

  Retirement planning copilots

  • •Build workflows that pull member history from PostgreSQL and generate retirement projections through your AI layer.
• •

  Compliance and audit assistants

  • •Expose read-only endpoints for contribution history, status changes, and benefit eligibility so agents can produce traceable summaries without direct table access.

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.