How to Integrate FastAPI for banking with PostgreSQL for AI agents

FastAPI for banking gives you a clean way to expose account, transaction, and policy workflows as APIs. PostgreSQL gives your AI agent a durable system of record for customer state, tool outputs, audit logs, and retrieval-ready context.

Put them together and you get an agent that can call banking operations through FastAPI, persist every decision in PostgreSQL, and recover state across sessions without guessing.

Prerequisites

• •Python 3.10+
• •A running FastAPI service for banking with endpoints like:
  • •POST /accounts
  • •GET /accounts/{account_id}
  • •POST /transactions
• •PostgreSQL 14+
• •A PostgreSQL database and user with read/write access
• •Installed Python packages:
  • •fastapi
  • •uvicorn
  • •psycopg2-binary
  • •sqlalchemy
  • •httpx
• •Environment variables configured:
  • •BANKING_API_BASE_URL
  • •BANKING_API_KEY
  • •DATABASE_URL

Integration Steps

1. •Set up your PostgreSQL connection layer.

Use SQLAlchemy for connection management and keep the raw DSN in an environment variable. For agent systems, you want a single persistence layer for tool calls, model outputs, and audit trails.

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

DATABASE_URL = os.environ["DATABASE_URL"]

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_events (
                id SERIAL PRIMARY KEY,
                conversation_id TEXT NOT NULL,
                event_type TEXT NOT NULL,
                payload JSONB NOT NULL,
                created_at TIMESTAMPTZ DEFAULT NOW()
            )
        """))

2. •Build a small FastAPI client for the banking service.

Treat the banking API as an external tool. Keep the client thin and explicit so your agent can call it predictably.

import os
import httpx

BANKING_API_BASE_URL = os.environ["BANKING_API_BASE_URL"]
BANKING_API_KEY = os.environ["BANKING_API_KEY"]

class BankingClient:
    def __init__(self):
        self.client = httpx.Client(
            base_url=BANKING_API_BASE_URL,
            headers={
                "Authorization": f"Bearer {BANKING_API_KEY}",
                "Content-Type": "application/json",
            },
            timeout=10.0,
        )

    def get_account(self, account_id: str):
        response = self.client.get(f"/accounts/{account_id}")
        response.raise_for_status()
        return response.json()

    def create_transaction(self, account_id: str, amount: float, currency: str):
        response = self.client.post(
            "/transactions",
            json={
                "account_id": account_id,
                "amount": amount,
                "currency": currency,
            },
        )
        response.raise_for_status()
        return response.json()

3. •Add a persistence function to store tool calls in PostgreSQL.

This is the part most teams skip. Don’t. If an agent moves money-related data around, every request and response should be traceable.

from sqlalchemy import text

def log_agent_event(conversation_id: str, event_type: str, payload: dict):
    with SessionLocal() as db:
        db.execute(
            text("""
                INSERT INTO agent_events (conversation_id, event_type, payload)
                VALUES (:conversation_id, :event_type, :payload::jsonb)
            """),
            {
                "conversation_id": conversation_id,
                "event_type": event_type,
                "payload": payload,
            },
        )
        db.commit()

4. •Wire FastAPI endpoints to call the banking API and write to PostgreSQL.

Your FastAPI app becomes the orchestration layer between the AI agent and the banking backend.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
banking_client = BankingClient()

class TransactionRequest(BaseModel):
    conversation_id: str
    account_id: str
    amount: float
    currency: str = "USD"

@app.post("/agent/transfer")
def transfer_funds(req: TransactionRequest):
    log_agent_event(
        req.conversation_id,
        "transfer_requested",
        req.model_dump(),
    )

    try:
        result = banking_client.create_transaction(
            account_id=req.account_id,
            amount=req.amount,
            currency=req.currency,
        )
    except httpx.HTTPStatusError as exc:
        log_agent_event(
            req.conversation_id,
            "transfer_failed",
            {"status_code": exc.response.status_code, "body": exc.response.text},
        )
        raise HTTPException(status_code=400, detail="Banking transaction failed")

    log_agent_event(
        req.conversation_id,
        "transfer_completed",
        result,
    )
    return {"status": "ok", "transaction": result}

5. •Expose a retrieval endpoint so your agent can fetch prior state from PostgreSQL.

Agents need memory that is queryable and auditable. Store events in Postgres; retrieve them when building prompts or deciding next actions.

@app.get("/agent/history/{conversation_id}")
def get_history(conversation_id: str):
    with SessionLocal() as db:
        rows = db.execute(
            text("""
                SELECT event_type, payload, created_at
                FROM agent_events
                WHERE conversation_id = :conversation_id
                ORDER BY created_at ASC
            """),
            {"conversation_id": conversation_id},
        ).mappings().all()

    return {"conversation_id": conversation_id, "events": list(rows)}

Testing the Integration

Run a basic end-to-end check by creating a transfer request through FastAPI and confirming that PostgreSQL recorded the event trail.

from fastapi.testclient import TestClient

client = TestClient(app)

def test_transfer_flow():
    init_db()

    response = client.post("/agent/transfer", json={
        "conversation_id": "conv_123",
        "account_id": "acct_456",
        "amount": 25.0,
        "currency": "USD"
    })

    assert response.status_code in (200, 400)

    history = client.get("/agent/history/conv_123")
    assert history.status_code == 200
    print(history.json())

Expected output:

{
  "conversation_id": "conv_123",
  "events": [
    {
      "event_type": "transfer_requested",
      "payload": {
        "conversation_id": "conv_123",
        "account_id": "acct_456",
        "amount": 25.0,
        "currency": "USD"
      }
    }
  ]
}

Real-World Use Cases

• •Banking support agents that check balances via FastAPI and persist every customer interaction in PostgreSQL for compliance review.
• •Fraud triage agents that call transaction endpoints from FastAPI and store risk signals in PostgreSQL for later analysis.
• •Loan assistant agents that read applicant state from PostgreSQL and trigger banking workflow APIs only when policy checks pass.

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.