How to Integrate FastAPI for insurance with PostgreSQL for AI agents

FastAPI for insurance gives you the API surface to expose policy, claims, and underwriting workflows to an AI agent. PostgreSQL gives you the durable system of record for customer profiles, claim state, and agent memory. Put them together and you get an agent that can read policy data, make decisions with context, and write back actions without losing auditability.

Prerequisites

• •Python 3.11+
• •A running PostgreSQL instance
• •Access to your FastAPI for insurance service endpoint
• •pip installed
• •These Python packages:
  • •fastapi
  • •uvicorn
  • •psycopg[binary] or psycopg2-binary
  • •sqlalchemy
  • •pydantic
  • •httpx

Install dependencies:

pip install fastapi uvicorn psycopg[binary] sqlalchemy pydantic httpx

Set your environment variables:

export DATABASE_URL="postgresql://agent_user:agent_pass@localhost:5432/insurance_ai"
export INSURANCE_API_BASE_URL="http://localhost:8000"

Integration Steps

1. Define the PostgreSQL connection layer

Use SQLAlchemy so your agent can persist policy lookups, claim decisions, and tool outputs. Keep the database access isolated from the FastAPI client so each concern is testable.

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://agent_user:agent_pass@localhost:5432/insurance_ai"

engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

def init_db():
    with engine.begin() as conn:
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS agent_memory (
                id SERIAL PRIMARY KEY,
                conversation_id VARCHAR(64) NOT NULL,
                key VARCHAR(128) NOT NULL,
                value JSONB NOT NULL,
                created_at TIMESTAMP DEFAULT NOW()
            )
        """))

2. Create a client for the FastAPI for insurance service

Your AI agent will call FastAPI endpoints to fetch policy details or submit claim actions. Use httpx.AsyncClient so the agent can stay non-blocking under load.

import httpx

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

    async def get_policy(self, policy_id: str) -> dict:
        async with httpx.AsyncClient(base_url=self.base_url, timeout=10.0) as client:
            response = await client.get(f"/policies/{policy_id}")
            response.raise_for_status()
            return response.json()

    async def create_claim(self, payload: dict) -> dict:
        async with httpx.AsyncClient(base_url=self.base_url, timeout=10.0) as client:
            response = await client.post("/claims", json=payload)
            response.raise_for_status()
            return response.json()

3. Wire both systems into an agent workflow

This is the core pattern: read from PostgreSQL for context, call FastAPI for insurance for live business data, then persist the result back into PostgreSQL.

import json
from sqlalchemy import text

class InsuranceAgentService:
    def __init__(self, db_session_factory, insurance_client: InsuranceApiClient):
        self.db_session_factory = db_session_factory
        self.insurance_client = insurance_client

    async def process_claim_intent(self, conversation_id: str, policy_id: str) -> dict:
        policy = await self.insurance_client.get_policy(policy_id)

        decision = {
            "policy_id": policy_id,
            "coverage_status": policy.get("status"),
            "eligible": policy.get("status") == "active",
        }

        with self.db_session_factory() as db:
            db.execute(
                text("""
                    INSERT INTO agent_memory (conversation_id, key, value)
                    VALUES (:conversation_id, :key, :value::jsonb)
                """),
                {
                    "conversation_id": conversation_id,
                    "key": "policy_decision",
                    "value": json.dumps(decision),
                },
            )
            db.commit()

        return decision

4. Expose the workflow through a FastAPI endpoint

If your orchestration layer is itself FastAPI-based, expose a route that your AI agent runtime can call directly.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ClaimRequest(BaseModel):
    conversation_id: str
    policy_id: str

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

@app.post("/agent/process-claim")
async def process_claim(request: ClaimRequest):
    insurance_client = InsuranceApiClient(base_url="http://localhost:8000")
    service = InsuranceAgentService(SessionLocal, insurance_client)
    result = await service.process_claim_intent(
        conversation_id=request.conversation_id,
        policy_id=request.policy_id,
    )
    return result

5. Add a direct database write for claim escalation

When the insurance API says a claim needs manual review, store that state in PostgreSQL so downstream workers can pick it up reliably.

async def escalate_claim(conversation_id: str, claim_payload: dict):
    with SessionLocal() as db:
        db.execute(
            text("""
                INSERT INTO agent_memory (conversation_id, key, value)
                VALUES (:conversation_id, :key, :value::jsonb)
            """),
            {
                "conversation_id": conversation_id,
                "key": "claim_escalation",
                "value": json.dumps(claim_payload),
            },
        )
        db.commit()

Testing the Integration

Run this quick verification script against both systems.

import asyncio

async def main():
    init_db()
    client = InsuranceApiClient("http://localhost:8000")
    service = InsuranceAgentService(SessionLocal, client)

    result = await service.process_claim_intent(
        conversation_id="conv_123",
        policy_id="POLICY-10001"
    )

    print(result)

asyncio.run(main())

Expected output:

{
  "policy_id": "POLICY-10001",
  "coverage_status": "active",
  "eligible": true
}

If you want to verify persistence too:

with SessionLocal() as db:
    rows = db.execute(
        text("SELECT key, value FROM agent_memory WHERE conversation_id = :cid"),
        {"cid": "conv_123"},
    ).fetchall()
    print(rows)

Expected output:

[('policy_decision', {'policy_id': 'POLICY-10001', 'coverage_status': 'active', 'eligible': True})]

Real-World Use Cases

• •Claim triage agents that pull live policy status from FastAPI and store eligibility decisions in PostgreSQL for audit trails.
• •Underwriting assistants that query customer history from PostgreSQL before calling insurance APIs to prefill risk checks.
• •Customer support agents that keep short-term conversation memory in PostgreSQL while using FastAPI endpoints to retrieve coverage limits and claim status.

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.