How to Integrate FastAPI for retail banking with PostgreSQL for production AI

Retail banking AI systems need two things to be useful in production: a reliable API surface and durable state. FastAPI gives you the request/response layer for customer-facing and internal banking workflows, while PostgreSQL gives you transactional storage for accounts, sessions, audit trails, and agent memory.

The useful pattern here is simple: FastAPI handles the banking workflow, PostgreSQL stores the source of truth, and your AI agent reads and writes through that API boundary instead of talking directly to random services.

Prerequisites

• •Python 3.11+
• •FastAPI installed
• •Uvicorn installed
• •PostgreSQL 14+
• •psycopg or psycopg2-binary
• •A running PostgreSQL database with credentials
• •Basic knowledge of REST APIs and SQL
• •An .env file or secret manager for database credentials

Install the core packages:

pip install fastapi uvicorn psycopg[binary] python-dotenv pydantic

Integration Steps

1. •

   Define your banking data model in PostgreSQL

   Start with a table that can support account lookup, balance checks, and agent interactions. Keep it boring and explicit.

CREATE TABLE IF NOT EXISTS bank_accounts (
    id BIGSERIAL PRIMARY KEY,
    account_number VARCHAR(32) UNIQUE NOT NULL,
    customer_name VARCHAR(255) NOT NULL,
    balance NUMERIC(18,2) NOT NULL DEFAULT 0,
    currency CHAR(3) NOT NULL DEFAULT 'USD',
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS agent_audit_log (
    id BIGSERIAL PRIMARY KEY,
    request_id UUID NOT NULL,
    action VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

2. •

   Create a PostgreSQL connection layer

   Use a connection pool so your FastAPI app does not open a new connection on every request.

import os
from contextlib import contextmanager

from dotenv import load_dotenv
from psycopg_pool import ConnectionPool

load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL")

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

@contextmanager
def get_db():
    with pool.connection() as conn:
        with conn.cursor() as cur:
            yield conn, cur

3. •

   Build the FastAPI banking endpoints

   This is the API layer your AI agent will call. Use Pydantic models for validation and keep database access behind small helper functions.

from decimal import Decimal
from uuid import UUID, uuid4

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Retail Banking API")

class BalanceResponse(BaseModel):
    account_number: str
    balance: Decimal
    currency: str

class TransferRequest(BaseModel):
    from_account: str = Field(..., min_length=5)
    to_account: str = Field(..., min_length=5)
    amount: Decimal = Field(..., gt=0)

@app.get("/accounts/{account_number}/balance", response_model=BalanceResponse)
def get_balance(account_number: str):
    with get_db() as (conn, cur):
        cur.execute(
            "SELECT account_number, balance, currency FROM bank_accounts WHERE account_number = %s",
            (account_number,),
        )
        row = cur.fetchone()

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

    return BalanceResponse(
        account_number=row[0],
        balance=row[1],
        currency=row[2],
    )

4. •

   Add a transactional transfer endpoint

   For production AI systems, this is where correctness matters. Use a single transaction so debit and credit succeed or fail together.

@app.post("/transfers")
def transfer_funds(req: TransferRequest):
    request_id = uuid4()

    with get_db() as (conn, cur):
        try:
            cur.execute(
                "SELECT balance FROM bank_accounts WHERE account_number = %s FOR UPDATE",
                (req.from_account,),
            )
            source = cur.fetchone()

            cur.execute(
                "SELECT balance FROM bank_accounts WHERE account_number = %s FOR UPDATE",
                (req.to_account,),
            )
            target = cur.fetchone()

            if not source or not target:
                raise HTTPException(status_code=404, detail="Account not found")

            if source[0] < req.amount:
                raise HTTPException(status_code=400, detail="Insufficient funds")

            cur.execute(
                "UPDATE bank_accounts SET balance = balance - %s, updated_at = NOW() WHERE account_number = %s",
                (req.amount, req.from_account),
            )
            cur.execute(
                "UPDATE bank_accounts SET balance = balance + %s, updated_at = NOW() WHERE account_number = %s",
                (req.amount, req.to_account),
            )
            cur.execute(
                """
                INSERT INTO agent_audit_log (request_id, action, payload)
                VALUES (%s, %s, %s)
                """,
                (request_id, "transfer_funds", {"from": req.from_account, "to": req.to_account, "amount": str(req.amount)}),
            )

            conn.commit()
        except Exception:
            conn.rollback()
            raise

    return {"status": "ok", "request_id": str(request_id)}

5. •

   Expose the app and wire it into your agent runtime

   Your AI agent should call the FastAPI endpoint instead of writing SQL directly. That keeps authorization, validation, and audit logging centralized.

import httpx

async def check_customer_balance(account_number: str):
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        response = await client.get(f"/accounts/{account_number}/balance")
        response.raise_for_status()
        return response.json()

async def initiate_transfer(from_account: str, to_account: str, amount: str):
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        response = await client.post(
            "/transfers",
            json={
                "from_account": from_account,
                "to_account": to_account,
                "amount": amount,
            },
        )
        response.raise_for_status()
        return response.json()

Run the service:

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

Testing the Integration

Use a quick smoke test against a seeded database record.

import asyncio

async def main():
    balance = await check_customer_balance("ACC10001")
    print(balance)

if __name__ == "__main__":
    asyncio.run(main())

Expected output:

{
  "account_number": "ACC10001",
  "balance": "12500.00",
  "currency": "USD"
}

If you want to validate the transfer path too:

async def test_transfer():
    result = await initiate_transfer("ACC10001", "ACC20002", "150.00")
    print(result)

# Expected:
# {"status": "ok", "request_id": "..."}

Real-World Use Cases

• •

  Customer self-service agents

  • •Let an AI assistant answer balance questions and trigger transfers through approved FastAPI endpoints backed by PostgreSQL audit logs.
• •

  Fraud-aware banking workflows

  • •Store interaction history in PostgreSQL and have FastAPI enforce transaction rules before an agent can move money or change customer details.
• •

  Agent memory for regulated workflows

  • •Persist case state in PostgreSQL so an AI agent can resume a loan servicing or dispute-resolution flow without losing context between requests.

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.