How to Integrate FastAPI for insurance with PostgreSQL for multi-agent systems

FastAPI for insurance plus PostgreSQL is a solid base for agentic insurance systems that need structured intake, policy lookup, claims orchestration, and durable memory. FastAPI gives you the HTTP surface for agent tools and workflows, while PostgreSQL gives you transactional storage for customer profiles, claim state, audit logs, and inter-agent coordination.

When you combine them, one agent can collect a claim, another can validate coverage, and a third can persist decisions without losing state between steps.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL instance
• •pip installed
• •A FastAPI app scaffolded for your insurance workflow
• •psycopg or asyncpg installed for PostgreSQL access
• •uvicorn for local API execution
• •Basic knowledge of async Python

Install the core packages:

pip install fastapi uvicorn psycopg[binary] pydantic

Integration Steps

1. •Create the FastAPI app and database connection settings

Use environment variables for connection details. In production insurance systems, hardcoding DB credentials is not acceptable.

from fastapi import FastAPI
from pydantic import BaseModel
import os

app = FastAPI(title="Insurance Agent API")

DB_HOST = os.getenv("DB_HOST", "localhost")
DB_NAME = os.getenv("DB_NAME", "insurance_db")
DB_USER = os.getenv("DB_USER", "postgres")
DB_PASSWORD = os.getenv("DB_PASSWORD", "postgres")
DB_PORT = os.getenv("DB_PORT", "5432")

2. •Create a PostgreSQL connection helper

For a multi-agent system, keep DB access in one place so every agent uses the same transaction pattern.

import psycopg

def get_conn():
    return psycopg.connect(
        host=DB_HOST,
        dbname=DB_NAME,
        user=DB_USER,
        password=DB_PASSWORD,
        port=DB_PORT,
    )

3. •Initialize tables for policies, claims, and agent events

This schema is enough to support multiple agents sharing state. Policies store coverage data, claims store workflow state, and events store traceability.

def init_db():
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS policies (
                    id SERIAL PRIMARY KEY,
                    policy_number TEXT UNIQUE NOT NULL,
                    customer_name TEXT NOT NULL,
                    coverage_type TEXT NOT NULL,
                    active BOOLEAN DEFAULT TRUE
                );
            """)
            cur.execute("""
                CREATE TABLE IF NOT EXISTS claims (
                    id SERIAL PRIMARY KEY,
                    claim_number TEXT UNIQUE NOT NULL,
                    policy_number TEXT NOT NULL REFERENCES policies(policy_number),
                    status TEXT NOT NULL DEFAULT 'received',
                    amount NUMERIC(12, 2) NOT NULL,
                    created_at TIMESTAMP DEFAULT NOW()
                );
            """)
            cur.execute("""
                CREATE TABLE IF NOT EXISTS agent_events (
                    id SERIAL PRIMARY KEY,
                    agent_name TEXT NOT NULL,
                    event_type TEXT NOT NULL,
                    payload JSONB NOT NULL,
                    created_at TIMESTAMP DEFAULT NOW()
                );
            """)
        conn.commit()

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

4. •Expose an insurance workflow endpoint

This endpoint lets one agent submit a claim while another agent later validates it against policy data. The endpoint writes both the business record and the event trail.

class ClaimRequest(BaseModel):
    claim_number: str
    policy_number: str
    amount: float
    agent_name: str = "claim-intake-agent"

@app.post("/claims")
def create_claim(payload: ClaimRequest):
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO claims (claim_number, policy_number, amount)
                VALUES (%s, %s, %s)
                RETURNING id, status;
                """,
                (payload.claim_number, payload.policy_number, payload.amount),
            )
            claim_id, status = cur.fetchone()

            cur.execute(
                """
                INSERT INTO agent_events (agent_name, event_type, payload)
                VALUES (%s, %s, %s::jsonb);
                """,
                (
                    payload.agent_name,
                    "claim_created",
                    f'{{"claim_number":"{payload.claim_number}","policy_number":"{payload.policy_number}"}}',
                ),
            )
        conn.commit()

    return {"claim_id": claim_id, "status": status}

5. •Add a lookup endpoint for other agents

A validation agent needs to read from PostgreSQL before making decisions. This is where multi-agent coordination becomes practical.

@app.get("/policies/{policy_number}")
def get_policy(policy_number: str):
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT policy_number, customer_name, coverage_type, active
                FROM policies
                WHERE policy_number = %s;
                """,
                (policy_number,),
            )
            row = cur.fetchone()

    if not row:
        return {"found": False}

    return {
        "found": True,
        "policy_number": row[0],
        "customer_name": row[1],
        "coverage_type": row[2],
        "active": row[3],
    }

Testing the Integration

Run the app:

uvicorn main:app --reload

Then verify the flow with a simple script:

import requests

base_url = "http://127.0.0.1:8000"

# seed a policy directly for testing
requests.post(f"{base_url}/seed-policy", json={
    "policy_number": "POL-10001",
    "customer_name": "Jane Doe",
    "coverage_type": "auto"
})

resp = requests.post(f"{base_url}/claims", json={
    "claim_number": "CLM-90001",
    "policy_number": "POL-10001",
    "amount": 1250.75,
})

print(resp.status_code)
print(resp.json())

policy_resp = requests.get(f"{base_url}/policies/POL-10001")
print(policy_resp.json())

Expected output:

200
{'claim_id': 1, 'status': 'received'}
{'found': True, 'policy_number': 'POL-10001', 'customer_name': 'Jane Doe', 'coverage_type': 'auto', 'active': True}

Real-World Use Cases

• •

  Claims intake agents

  • •One agent collects FNOL data through FastAPI.
  • •PostgreSQL stores the claim record and event history for audit and downstream processing.
• •

  Policy validation pipelines

  • •A validation agent checks eligibility against policy tables before approving payout workflows.
  • •This keeps underwriting rules consistent across multiple agents.
• •

  Agent coordination with durable memory

  • •Each agent writes its actions to agent_events.
  • •Another agent can resume work from the last known state after retries or failures.

If you want this production-ready for an insurance platform, add migrations with Alembic, use async DB calls with AsyncConnection, and enforce row-level locking around claim updates so two agents never approve the same case twice.

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.