How to Integrate FastAPI for insurance with PostgreSQL for production AI

FastAPI for insurance gives you a clean API layer for underwriting, claims, policy lookup, and document workflows. PostgreSQL gives you durable state, auditability, and queryable history, which is exactly what production AI agents need when they have to make decisions from policy data instead of chat history.

Prerequisites

• •Python 3.10+
• •FastAPI installed and running in your insurance service
• •PostgreSQL 14+ running locally or in production
• •A PostgreSQL database and user with read/write permissions
• •psycopg v3 or asyncpg installed
• •sqlalchemy and alembic if you want migrations
• •Environment variables configured:
  • •DATABASE_URL
  • •FASTAPI_INSURANCE_BASE_URL
  • •FASTAPI_INSURANCE_API_KEY
• •A FastAPI insurance app exposing endpoints for:
  • •policy lookup
  • •claim creation
  • •customer profile retrieval

Integration Steps

1. •

   Define the database connection layer

   Start with a single source of truth for your Postgres connection string. In production AI systems, this should be pooled and reused, not created per request.

   import os
   from sqlalchemy import create_engine
   from sqlalchemy.orm import sessionmaker, declarative_base
   
   DATABASE_URL = os.environ["DATABASE_URL"]
   
   engine = create_engine(
       DATABASE_URL,
       pool_size=10,
       max_overflow=20,
       pool_pre_ping=True,
   )
   
   SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
   Base = declarative_base()
   

2. •

   Create tables for AI agent state and insurance records

   Your agent needs persistent memory: requests received, policy context fetched, and decisions made. Store that alongside insurance entities so you can trace every action.

   from sqlalchemy import Column, Integer, String, DateTime, JSON, func
   
   class AgentInteraction(Base):
       __tablename__ = "agent_interactions"
   
       id = Column(Integer, primary_key=True)
       conversation_id = Column(String(64), nullable=False, index=True)
       customer_id = Column(String(64), nullable=False, index=True)
       intent = Column(String(128), nullable=False)
       payload = Column(JSON, nullable=False)
       result = Column(JSON, nullable=True)
       created_at = Column(DateTime(timezone=True), server_default=func.now())
   

   Run the migration or create the table directly during initial setup:

   Base.metadata.create_all(bind=engine)
   

3. •

   Call the FastAPI insurance service from your agent

   Use httpx to talk to the FastAPI insurance API. This keeps your AI agent decoupled from the insurance backend while still letting it fetch policy data and create claims.

   import os
   import httpx
   
   BASE_URL = os.environ["FASTAPI_INSURANCE_BASE_URL"]
   API_KEY = os.environ["FASTAPI_INSURANCE_API_KEY"]
   
   async def get_policy(policy_number: str) -> dict:
       headers = {"Authorization": f"Bearer {API_KEY}"}
   
       async with httpx.AsyncClient(base_url=BASE_URL, timeout=10.0) as client:
           response = await client.get(f"/policies/{policy_number}", headers=headers)
           response.raise_for_status()
           return response.json()
   

4. •

   Persist the AI workflow result into PostgreSQL

   Once the agent fetches policy data from FastAPI for insurance, store both the request context and the returned payload in Postgres. That gives you audit trails and makes downstream review possible.

    from sqlalchemy.orm import Session
   
    def save_interaction(
        db: Session,
        conversation_id: str,
        customer_id: str,
        intent: str,
        payload: dict,
        result: dict | None,
    ) -> None:
        record = AgentInteraction(
            conversation_id=conversation_id,
            customer_id=customer_id,
            intent=intent,
            payload=payload,
            result=result,
        )
        db.add(record)
        db.commit()
        db.refresh(record)
   

5. •

   Wire it into a FastAPI endpoint

   This is where the integration becomes useful in production. Your FastAPI route receives an insurance request, calls the external insurance service if needed, stores state in PostgreSQL, and returns a structured response to the client or another agent.

   from fastapi import FastAPI, Depends
   from sqlalchemy.orm import Session
   
   app = FastAPI()
   
   def get_db():
       db = SessionLocal()
       try:
           yield db
       finally:
           db.close()
   
   @app.post("/agent/claim-summary")
   async def claim_summary(
       conversation_id: str,
       customer_id: str,
       policy_number: str,
       db: Session = Depends(get_db),
   ):
       policy = await get_policy(policy_number)
   
       summary = {
           "policy_number": policy_number,
           "status": policy.get("status"),
           "coverage_limit": policy.get("coverage_limit"),
           "deductible": policy.get("deductible"),
       }
   
       save_interaction(
           db=db,
           conversation_id=conversation_id,
           customer_id=customer_id,
           intent="claim_summary",
           payload={"policy_number": policy_number},
           result=summary,
       )
   
       return summary
   

Testing the Integration

Use a real request path end-to-end: call FastAPI for insurance, persist to PostgreSQL, then read back the saved interaction.

import asyncio
from sqlalchemy import select

async def test_flow():
    db = SessionLocal()

    try:
      summary = await get_policy("POL-10001")
      save_interaction(
          db=db,
          conversation_id="conv-001",
          customer_id="cust-7788",
          intent="policy_lookup",
          payload={"policy_number": "POL-10001"},
          result=summary,
      )

      row = db.execute(
          select(AgentInteraction).where(AgentInteraction.conversation_id == "conv-001")
      ).scalar_one()

      print(row.intent)
      print(row.result["status"])
    finally:
      db.close()

asyncio.run(test_flow())

Expected output:

policy_lookup
active

Real-World Use Cases

• •Claims triage agents that fetch policy coverage from FastAPI for insurance and store triage decisions in PostgreSQL for audit and compliance.
• •Underwriting copilots that pull applicant history through API endpoints, enrich it with internal rules stored in Postgres tables, then generate recommendation summaries.
• •Customer service assistants that answer “Am I covered?” by querying live policy data while logging every interaction for review by ops teams.

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.