How to Integrate FastAPI for retail banking with PostgreSQL for startups

Combining FastAPI for retail banking with PostgreSQL gives you a clean path from API request to durable financial state. For startups building AI agent systems, this is the difference between a demo that answers questions and a system that can actually open accounts, fetch balances, and persist transaction history with auditability.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL 14+ instance
• •fastapi
• •uvicorn
• •psycopg2-binary or psycopg
• •sqlalchemy
• •pydantic
• •Access to your FastAPI for retail banking service credentials or local service URL
• •A .env file or secrets manager for database and API settings

Integration Steps

1. •

   Set up your FastAPI app and database config

   Keep config out of code. Use environment variables for the database URL and your banking API base URL.

   from fastapi import FastAPI
   from pydantic import BaseSettings
   
   class Settings(BaseSettings):
       database_url: str = "postgresql+psycopg2://bank_user:bank_pass@localhost:5432/banking"
       banking_api_base_url: str = "https://api.your-retail-banking-service.com"
   
   settings = Settings()
   app = FastAPI(title="Retail Banking Agent API")
   

2. •

   Create a PostgreSQL connection and schema

   Use SQLAlchemy for connection management and table creation. For retail banking, start with accounts and transactions tables.

   from sqlalchemy import create_engine, text
   from sqlalchemy.orm import sessionmaker
   
   engine = create_engine(settings.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 accounts (
                   id SERIAL PRIMARY KEY,
                   customer_id VARCHAR(64) NOT NULL,
                   account_number VARCHAR(32) UNIQUE NOT NULL,
                   balance NUMERIC(14, 2) NOT NULL DEFAULT 0,
                   created_at TIMESTAMP DEFAULT NOW()
               )
           """))
           conn.execute(text("""
               CREATE TABLE IF NOT EXISTS transactions (
                   id SERIAL PRIMARY KEY,
                   account_number VARCHAR(32) NOT NULL,
                   txn_type VARCHAR(20) NOT NULL,
                   amount NUMERIC(14, 2) NOT NULL,
                   reference VARCHAR(128),
                   created_at TIMESTAMP DEFAULT NOW()
               )
           """))
   
   init_db()
   

3. •

   Add a FastAPI endpoint that reads and writes PostgreSQL

   This is the core integration point. The endpoint accepts a request, writes the transaction to PostgreSQL, then returns the updated account state.

   from fastapi import Depends, HTTPException
   from pydantic import BaseModel
   from sqlalchemy.orm import Session
   
   class DepositRequest(BaseModel):
       account_number: str
       amount: float
       reference: str | None = None
   
   def get_db():
       db = SessionLocal()
       try:
           yield db
       finally:
           db.close()
   
   @app.post("/banking/deposit")
   def deposit_funds(payload: DepositRequest, db: Session = Depends(get_db)):
       account = db.execute(
           text("SELECT id, balance FROM accounts WHERE account_number = :account_number"),
           {"account_number": payload.account_number},
       ).fetchone()
   
       if not account:
           raise HTTPException(status_code=404, detail="Account not found")
   
       new_balance = float(account.balance) + payload.amount
   
       db.execute(
           text("UPDATE accounts SET balance = :balance WHERE account_number = :account_number"),
           {"balance": new_balance, "account_number": payload.account_number},
       )
   
       db.execute(
           text("""
               INSERT INTO transactions (account_number, txn_type, amount, reference)
               VALUES (:account_number, 'deposit', :amount, :reference)
           """),
           {
               "account_number": payload.account_number,
               "amount": payload.amount,
               "reference": payload.reference,
           },
       )
   
       db.commit()
       return {
           "account_number": payload.account_number,
           "new_balance": new_balance,
           "status": "posted",
       }
   

4. •

   Call the banking API from your agent workflow

   In a real startup setup, your AI agent may need to call an external retail banking service before persisting the result in PostgreSQL. Use httpx inside a service function so you can keep HTTP logic out of route handlers.

   import httpx
   
   async def open_retail_bank_account(customer_id: str, name: str):
       async with httpx.AsyncClient(base_url=settings.banking_api_base_url) as client:
           response = await client.post(
               "/v1/accounts",
               json={
                   "customer_id": customer_id,
                   "name": name,
                   "product_code": "retail-checking",
               },
               headers={"Authorization": f"Bearer {settings.banking_api_token}"},
           )
           response.raise_for_status()
           return response.json()
   

5. •

   Persist the banking API response in PostgreSQL

   Once the external retail banking system returns an account number, store it locally for fast lookup by your agent system.

   @app.post("/banking/open-account")
   async def open_account(customer_id: str, name: str, db: Session = Depends(get_db)):
       bank_account = await open_retail_bank_account(customer_id=customer_id, name=name)
   
       db.execute(
           text("""
               INSERT INTO accounts (customer_id, account_number, balance)
               VALUES (:customer_id, :account_number, :balance)
           """),
           {
               "customer_id": customer_id,
               "account_number": bank_account["account_number"],
               "balance": bank_account.get("opening_balance", 0),
           },
       )
       db.commit()
   
       return {
           "customer_id": customer_id,
           "account_number": bank_account["account_number"],
           "synced_to_postgres": True,
       }
   

Testing the Integration

Run the app:

uvicorn main:app --reload

Then verify both the API and PostgreSQL write path with a simple request:

import requests

payload = {
    "account_number": "ACC-10001",
    "amount": 250.00,
    "reference": "test-deposit"
}

response = requests.post("http://127.0.0.1:8000/banking/deposit", json=payload)
print(response.status_code)
print(response.json())

Expected output:

200
{'account_number': 'ACC-10001', 'new_balance': 250.0, 'status': 'posted'}

You can also confirm persistence directly in PostgreSQL:

SELECT account_number, balance FROM accounts WHERE account_number = 'ACC-10001';
SELECT txn_type, amount FROM transactions WHERE account_number = 'ACC-10001';

Real-World Use Cases

• •

  AI banking assistant

  • •Let an agent answer “What’s my balance?” by reading PostgreSQL and trigger deposits or transfers through FastAPI endpoints.
• •

  Loan onboarding workflow

  • •Use FastAPI to collect application data and store underwriting state in PostgreSQL so agents can resume incomplete applications.
• •

  Fraud monitoring dashboard

  • •Stream transaction events into PostgreSQL and expose them through FastAPI for internal risk tools and analyst workflows.

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.