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

Why this integration matters

If you're building multi-agent systems for pension funds, you need two things: a clean API layer for agent orchestration and a durable data store for member records, claims, contributions, and policy state. FastAPI gives you the HTTP surface for agent-to-service calls, while PostgreSQL gives you transactional storage and queryable history.

The practical win is this: one agent can validate requests, another can enrich pension data, and a third can persist decisions and audit trails in PostgreSQL without stepping on each other.

Prerequisites

• •Python 3.11+
• •FastAPI installed
• •Uvicorn installed
• •PostgreSQL 14+
• •psycopg or asyncpg installed
• •A running PostgreSQL database with credentials
• •Basic familiarity with async Python
• •An environment variable setup for database config

Install the packages:

pip install fastapi uvicorn psycopg[binary] pydantic

Set your database URL:

export DATABASE_URL="postgresql://postgres:postgres@localhost:5432/pension_agents"

Integration Steps

1. •Create the FastAPI app and define your pension fund payloads

Start by defining the request and response models your agents will use. For pension systems, keep the schema explicit so every agent sees the same contract.

from fastapi import FastAPI
from pydantic import BaseModel, Field

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

class PensionCase(BaseModel):
    member_id: str = Field(..., examples=["MF-100245"])
    fund_id: str = Field(..., examples=["PF-001"])
    contribution_amount: float = Field(..., gt=0)
    status: str = Field(default="pending")

class PensionCaseResponse(BaseModel):
    case_id: int
    member_id: str
    fund_id: str
    status: str

2. •Create a PostgreSQL connection helper

Use a connection pool so multiple agents can write concurrently without creating a new database connection per request.

import os
import psycopg
from psycopg_pool import ConnectionPool

DATABASE_URL = os.environ["DATABASE_URL"]

pool = ConnectionPool(conninfo=DATABASE_URL, min_size=1, max_size=10)

def init_db():
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS pension_cases (
                    id SERIAL PRIMARY KEY,
                    member_id TEXT NOT NULL,
                    fund_id TEXT NOT NULL,
                    contribution_amount NUMERIC(12,2) NOT NULL,
                    status TEXT NOT NULL DEFAULT 'pending',
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )
            """)
        conn.commit()

Call init_db() during startup so the table exists before any agent writes to it.

3. •Add an endpoint that persists pension cases into PostgreSQL

This is the core integration point. The FastAPI route receives a pension case from an agent, writes it to PostgreSQL using cursor.execute(), then returns the stored record ID.

from fastapi import FastAPI

@app.on_event("startup")
def startup():
    init_db()

@app.post("/pension-cases", response_model=PensionCaseResponse)
def create_pension_case(payload: PensionCase):
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO pension_cases (member_id, fund_id, contribution_amount, status)
                VALUES (%s, %s, %s, %s)
                RETURNING id
                """,
                (
                    payload.member_id,
                    payload.fund_id,
                    payload.contribution_amount,
                    payload.status,
                ),
            )
            case_id = cur.fetchone()[0]
        conn.commit()

    return PensionCaseResponse(
        case_id=case_id,
        member_id=payload.member_id,
        fund_id=payload.fund_id,
        status=payload.status,
    )

4. •Add a read endpoint for other agents to query state

Multi-agent systems need shared memory. A second agent can fetch the latest case state before deciding whether to approve, escalate, or enrich it.

from fastapi import HTTPException

@app.get("/pension-cases/{case_id}", response_model=PensionCaseResponse)
def get_pension_case(case_id: int):
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT id, member_id, fund_id, status
                FROM pension_cases
                WHERE id = %s
                """,
                (case_id,),
            )
            row = cur.fetchone()

    if not row:
        raise HTTPException(status_code=404, detail="Pension case not found")

    return PensionCaseResponse(
        case_id=row[0],
        member_id=row[1],
        fund_id=row[2],
        status=row[3],
    )

5. •Run the service and expose it to your agent runtime

Use Uvicorn to serve the FastAPI app. Other agents can now call these endpoints over HTTP and coordinate through PostgreSQL-backed state.

uvicorn main:app --reload --host 0.0.0.0 --port 8000

If you want an internal agent workflow, one agent can POST new cases while another polls or GETs them for decisioning.

Testing the Integration

Use curl or Python requests to verify both write and read paths work.

import requests

base_url = "http://localhost:8000"

create_resp = requests.post(
    f"{base_url}/pension-cases",
    json={
        "member_id": "MF-100245",
        "fund_id": "PF-001",
        "contribution_amount": 1250.50,
        "status": "pending"
    },
)

print("CREATE:", create_resp.status_code, create_resp.json())

case_id = create_resp.json()["case_id"]

get_resp = requests.get(f"{base_url}/pension-cases/{case_id}")
print("GET:", get_resp.status_code, get_resp.json())

Expected output:

CREATE: 200 {'case_id': 1, 'member_id': 'MF-100245', 'fund_id': 'PF-001', 'status': 'pending'}
GET: 200 {'case_id': 1, 'member_id': 'MF-100245', 'fund_id': 'PF-001', 'status': 'pending'}

Real-World Use Cases

• •

  Contribution processing pipelines
  One agent validates incoming contribution files through FastAPI while another stores normalized records in PostgreSQL for reconciliation and audit.

• •

  Claims and benefit review workflows
  An intake agent creates claim cases; an underwriting or rules agent reads the same record from PostgreSQL and updates status after evaluation.

• •

  Member servicing assistants
  A support agent answers policy questions through FastAPI endpoints while persisting interaction history and case notes in PostgreSQL for compliance 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.