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

Combining FastAPI for retail banking with PostgreSQL gives you a clean path from customer-facing banking APIs to durable agent memory and transaction state. For AI agents, that means one system can expose account actions through FastAPI, persist audit trails and conversation context in PostgreSQL, and make decisions from real banking data instead of ephemeral prompts.

Prerequisites

• •Python 3.11+
• •A running PostgreSQL instance
• •A FastAPI app for your retail banking service
• •psycopg or asyncpg installed for PostgreSQL access
• •uvicorn for local API execution
• •Environment variables configured:
  • •DATABASE_URL
  • •BANKING_API_BASE_URL
  • •BANKING_API_KEY

Install the core packages:

pip install fastapi uvicorn psycopg[binary] httpx pydantic

Integration Steps

1. •

   Define the banking API client and database connection

   Keep the FastAPI banking layer and PostgreSQL layer separate. Your agent should call the banking API through HTTP, then write the result into PostgreSQL for persistence and retrieval.

import os
import httpx
import psycopg
from psycopg.rows import dict_row

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

def get_db_connection():
    return psycopg.connect(DATABASE_URL, row_factory=dict_row)

def get_banking_client():
    return httpx.Client(
        base_url=BANKING_API_BASE_URL,
        headers={"Authorization": f"Bearer {BANKING_API_KEY}"},
        timeout=10.0,
    )

2. •

   Create the PostgreSQL tables for agent state and audit logs

   Store customer context, tool calls, and transaction outcomes. In banking systems, you want traceability first and cleverness second.

DDL = """
CREATE TABLE IF NOT EXISTS agent_sessions (
    session_id TEXT PRIMARY KEY,
    customer_id TEXT NOT NULL,
    last_intent TEXT,
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS banking_audit_log (
    id BIGSERIAL PRIMARY KEY,
    session_id TEXT NOT NULL,
    customer_id TEXT NOT NULL,
    action TEXT NOT NULL,
    request_payload JSONB NOT NULL,
    response_payload JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);
"""

with get_db_connection() as conn:
    with conn.cursor() as cur:
        cur.execute(DDL)
    conn.commit()

3. •

   Call the FastAPI retail banking endpoint from your agent

   This example assumes your retail banking service exposes an endpoint like POST /v1/transfers built with FastAPI’s standard routing primitives: @app.post(...), Request, HTTPException, and Pydantic models.

from pydantic import BaseModel

class TransferRequest(BaseModel):
    from_account_id: str
    to_account_id: str
    amount: float
    currency: str = "USD"
    memo: str | None = None

def initiate_transfer(payload: TransferRequest) -> dict:
    with get_banking_client() as client:
        response = client.post("/v1/transfers", json=payload.model_dump())
        response.raise_for_status()
        return response.json()

4. •

   Persist the result in PostgreSQL after the API call

   Once the FastAPI service returns a transaction reference, write both the request and response to PostgreSQL. This gives your AI agent durable memory and a clean audit trail.

import json

def log_transfer(session_id: str, customer_id: str, request_data: dict, response_data: dict):
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO banking_audit_log (
                    session_id, customer_id, action, request_payload, response_payload
                )
                VALUES (%s, %s, %s, %s::jsonb, %s::jsonb)
                """,
                (
                    session_id,
                    customer_id,
                    "transfer_funds",
                    json.dumps(request_data),
                    json.dumps(response_data),
                ),
            )
        conn.commit()

5. •

   Wrap it in an agent-friendly orchestration function

   Your agent should fetch context from PostgreSQL, call the bank API via FastAPI, then update state. That pattern keeps side effects explicit and testable.

def transfer_funds_for_agent(session_id: str, customer_id: str):
    transfer = TransferRequest(
        from_account_id="CHK-001",
        to_account_id="SAV-002",
        amount=250.00,
        memo="Monthly savings sweep",
    )

    result = initiate_transfer(transfer)
    log_transfer(
        session_id=session_id,
        customer_id=customer_id,
        request_data=transfer.model_dump(),
        response_data=result,
    )

    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO agent_sessions (session_id, customer_id, last_intent)
                VALUES (%s, %s, %s)
                ON CONFLICT (session_id)
                DO UPDATE SET last_intent = EXCLUDED.last_intent, updated_at = NOW()
                """,
                (session_id, customer_id, "transfer_funds"),
            )
        conn.commit()

    return result

Testing the Integration

Run a quick end-to-end check by calling the orchestration function directly.

if __name__ == "__main__":
    output = transfer_funds_for_agent(
        session_id="sess_1001",
        customer_id="cust_7788",
    )
    print(output)

Expected output:

{
  "transaction_id": "txn_9f1c2a",
  "status": "pending",
  "from_account_id": "CHK-001",
  "to_account_id": "SAV-002",
  "amount": 250.0,
  "currency": "USD"
}

You should also verify the audit row landed in PostgreSQL:

with get_db_connection() as conn:
    with conn.cursor() as cur:
        cur.execute(
            "SELECT action, request_payload, response_payload FROM banking_audit_log ORDER BY id DESC LIMIT 1"
        )
        row = cur.fetchone()
        print(row)

Real-World Use Cases

• •

  Agent-assisted transfers with auditability

  • •The agent can initiate internal transfers through FastAPI and store every request/response pair in PostgreSQL for compliance review.
• •

  Customer support copilots

  • •An AI agent can read account metadata from PostgreSQL, call FastAPI endpoints for balance checks or card status changes, then summarize outcomes for support staff.
• •

  Fraud triage workflows

  • •The agent can score suspicious activity using historical events in PostgreSQL and trigger FastAPI actions like account holds or step-up verification 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.