How to Integrate FastAPI for fintech with PostgreSQL for production AI

FastAPI for fintech gives you the API layer to expose account, payment, and risk workflows. PostgreSQL gives you durable state for transactions, agent memory, audit trails, and retrieval data. Put them together and you get a production AI system that can accept requests from clients, call internal finance tools, persist decisions, and keep a full trace of what the agent did.

Prerequisites

• •Python 3.11+
• •A running PostgreSQL instance
• •A FastAPI app already scaffolded
• •pip or uv
• •These packages installed:
  • •fastapi
  • •uvicorn
  • •psycopg[binary] or asyncpg
  • •sqlalchemy[asyncio]
  • •pydantic
• •A PostgreSQL database URL like:
  • •postgresql+asyncpg://user:password@localhost:5432/fintech_ai
• •Basic knowledge of async Python

Integration Steps

1. •Set up the FastAPI app and database config.

Use environment variables for the database connection. In production AI systems, hardcoding credentials is how you end up with incidents.

from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
import os

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+asyncpg://postgres:postgres@localhost:5432/fintech_ai"
)

engine = create_async_engine(DATABASE_URL, echo=False)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

app = FastAPI(title="Fintech AI API")

2. •Define a table for agent transactions and decisions.

For production AI, don’t store only raw prompts. Store request metadata, model output, decision status, and timestamps so you can audit every action.

from sqlalchemy import String, Integer, Text, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class AgentDecision(Base):
    __tablename__ = "agent_decisions"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    customer_id: Mapped[str] = mapped_column(String(64), index=True)
    request_type: Mapped[str] = mapped_column(String(64))
    prompt: Mapped[str] = mapped_column(Text)
    result: Mapped[str] = mapped_column(Text)
    status: Mapped[str] = mapped_column(String(32), default="pending")
    created_at: Mapped[str] = mapped_column(DateTime(timezone=True), server_default=func.now())

Create the schema on startup:

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

3. •Add a dependency that gives each request a PostgreSQL session.

This is the part that keeps your API handlers clean. Each request gets its own DB session and closes it properly.

from typing import AsyncGenerator

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session

4. •Build a FastAPI endpoint that writes AI decisions to PostgreSQL.

This example simulates an AI underwriting or fraud-review step. In a real system you’d call your model or agent service here, then persist the outcome.

from fastapi import Depends
from sqlalchemy import select

class DecisionRequest(BaseModel):
    customer_id: str
    request_type: str
    prompt: str

@app.post("/agent/decide")
async def decide(payload: DecisionRequest, db: AsyncSession = Depends(get_db)):
    # Replace this with your actual model/agent call.
    if "fraud" in payload.prompt.lower():
        result_text = "Escalate to manual review"
        status = "review"
    else:
        result_text = "Approve automatically"
        status = "approved"

    row = AgentDecision(
        customer_id=payload.customer_id,
        request_type=payload.request_type,
        prompt=payload.prompt,
        result=result_text,
        status=status,
    )
    db.add(row)
    await db.commit()
    await db.refresh(row)

    return {
        "id": row.id,
        "customer_id": row.customer_id,
        "status": row.status,
        "result": row.result,
    }

5. •Add a read endpoint to fetch stored decisions for audits or agent memory.

This is where PostgreSQL becomes useful beyond persistence. You can use it as an audit store or as retrieval context for future agent calls.

@app.get("/agent/decisions/{customer_id}")
async def list_decisions(customer_id: str, db: AsyncSession = Depends(get_db)):
    stmt = select(AgentDecision).where(AgentDecision.customer_id == customer_id).order_by(AgentDecision.id.desc())
    result = await db.execute(stmt)
    rows = result.scalars().all()

    return [
        {
            "id": r.id,
            "request_type": r.request_type,
            "status": r.status,
            "result": r.result,
            "created_at": r.created_at.isoformat() if r.created_at else None,
        }
        for r in rows
    ]

Testing the Integration

Run the app:

uvicorn main:app --reload

Then test the write path with curl:

curl -X POST http://127.0.0.1:8000/agent/decide \
  -H "Content-Type: application/json" \
  -d '{
    "customer_id": "cust_1001",
    "request_type": "fraud_check",
    "prompt": "Review this transaction for fraud risk"
  }'

Expected output:

{
  "id": 1,
  "customer_id": "cust_1001",
  "status": "review",
  "result": "Escalate to manual review"
}

Then verify persistence:

curl http://127.0.0.1:8000/agent/decisions/cust_1001

Expected output:

[
  {
    "id": 1,
    "request_type": "fraud_check",
    "status": "review",
    "result": "Escalate to manual review",
    "created_at": "2026-04-21T10:15:00+00:00"
  }
]

Real-World Use Cases

• •Fraud triage agents that score incoming transactions in FastAPI and store every decision in PostgreSQL for compliance review.
• •Loan prequalification systems that persist applicant profiles, model outputs, and manual overrides in a single relational store.
• •Customer support agents that use PostgreSQL as long-term memory for prior cases, policy lookups, and escalation history.

If you need this pattern to scale cleanly, keep FastAPI thin and make PostgreSQL the source of truth for stateful agent behavior. That gives you traceability, replayability, and fewer surprises when finance teams ask why a decision was made.

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.