How to Integrate FastAPI for banking with PostgreSQL for startups

FastAPI for banking gives you a clean API layer for transaction workflows, account operations, and agent-facing endpoints. PostgreSQL gives you durable state, auditability, and queryable history, which is exactly what you want when an AI agent needs to make decisions from banking data instead of guessing.

Prerequisites

• •Python 3.11+
• •A FastAPI app already scaffolded
• •PostgreSQL 14+ running locally or in your cloud environment
• •psycopg or asyncpg installed for PostgreSQL access
• •uvicorn for running FastAPI
• •Environment variables configured:
  • •DATABASE_URL
  • •BANK_API_KEY
  • •BANK_API_BASE_URL
• •A bank sandbox or internal FastAPI banking service with endpoints for:
  • •/accounts
  • •/transactions
  • •/balances

Integration Steps

1. •

   Install the dependencies

   Use async drivers if your agent will fan out requests and call tools in parallel.

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

2. •

   Create the PostgreSQL connection layer

   Keep database access isolated. For startup systems, the usual pattern is: API layer calls service layer, service layer writes audit logs and account snapshots into PostgreSQL.

   import os
   from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
   from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
   from sqlalchemy import String, Integer, DateTime, func
   
   DATABASE_URL = os.getenv("DATABASE_URL")
   
   engine = create_async_engine(DATABASE_URL, echo=True)
   SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
   
   class Base(DeclarativeBase):
       pass
   
   class TransactionLog(Base):
       __tablename__ = "transaction_logs"
   
       id: Mapped[int] = mapped_column(Integer, primary_key=True)
       account_id: Mapped[str] = mapped_column(String(64), index=True)
       action: Mapped[str] = mapped_column(String(64))
       amount_cents: Mapped[int] = mapped_column(Integer)
       status: Mapped[str] = mapped_column(String(32))
       created_at: Mapped[DateTime] = mapped_column(DateTime(timezone=True), server_default=func.now())
   

3. •

   Build the FastAPI banking client

   In most banking integrations, your FastAPI app is the orchestration layer that talks to a separate banking API. Use httpx.AsyncClient so your endpoints stay non-blocking.

   import os
   import httpx
   
   BANK_API_BASE_URL = os.getenv("BANK_API_BASE_URL")
   BANK_API_KEY = os.getenv("BANK_API_KEY")
   
   class BankingClient:
       def __init__(self):
           self.base_url = BANK_API_BASE_URL
           self.headers = {
               "Authorization": f"Bearer {BANK_API_KEY}",
               "Content-Type": "application/json",
           }
   
       async def get_balance(self, account_id: str):
           async with httpx.AsyncClient(base_url=self.base_url, headers=self.headers) as client:
               response = await client.get(f"/accounts/{account_id}/balance")
               response.raise_for_status()
               return response.json()
   
       async def create_transaction(self, payload: dict):
           async with httpx.AsyncClient(base_url=self.base_url, headers=self.headers) as client:
               response = await client.post("/transactions", json=payload)
               response.raise_for_status()
               return response.json()
   

4. •

   Wire FastAPI routes to both systems

   This is the core integration: call the banking service, persist the result in PostgreSQL, and return a stable API response for your agent or frontend.

   from fastapi import FastAPI, Depends
   from pydantic import BaseModel
   from sqlalchemy.ext.asyncio import AsyncSession
   
   app = FastAPI()
   banking_client = BankingClient()
   
   class TransferRequest(BaseModel):
       account_id: str
       destination_account_id: str
       amount_cents: int
   
   async def get_db():
       async with SessionLocal() as session:
           yield session
   
   @app.post("/banking/transfer")
   async def transfer_funds(payload: TransferRequest, db: AsyncSession = Depends(get_db)):
       bank_result = await banking_client.create_transaction({
           "source_account_id": payload.account_id,
           "destination_account_id": payload.destination_account_id,
           "amount_cents": payload.amount_cents,
           "type": "transfer",
       })
   
       log_row = TransactionLog(
           account_id=payload.account_id,
           action="transfer",
           amount_cents=payload.amount_cents,
           status=bank_result["status"],
       )
       db.add(log_row)
       await db.commit()
   
       return {
           "transaction_id": bank_result["transaction_id"],
           "status": bank_result["status"],
           "logged": True,
       }
   

5. •

   Add a read path for agents

   AI agents need a fast read model. Store recent transactions in PostgreSQL so the agent can retrieve context without hitting the banking system on every prompt cycle.

   from sqlalchemy import select
   
   @app.get("/banking/accounts/{account_id}/activity")
   async def account_activity(account_id: str, db: AsyncSession = Depends(get_db)):
       stmt = (
           select(TransactionLog)
           .where(TransactionLog.account_id == account_id)
           .order_by(TransactionLog.created_at.desc())
           .limit(20)
       )
       result = await db.execute(stmt)
       rows = result.scalars().all()
   
       return [
           {
               "action": row.action,
               "amount_cents": row.amount_cents,
               "status": row.status,
               "created_at": row.created_at.isoformat(),
           }
           for row in rows
       ]
   

Testing the Integration

Run the app:

uvicorn main:app --reload

Then test with a transfer request:

import httpx
import asyncio

async def test_transfer():
    async with httpx.AsyncClient(base_url="http://127.0.0.1:8000") as client:
        response = await client.post(
            "/banking/transfer",
            json={
                "account_id": "acct_123",
                "destination_account_id": "acct_456",
                "amount_cents": 2500,
            },
        )
        print(response.status_code)
        print(response.json())

asyncio.run(test_transfer())

Expected output:

200
{
  "transaction_id": "txn_789",
  "status": "completed",
  "logged": true
}

Real-World Use Cases

• •

  AI-powered cash flow assistant
  Let an agent inspect balances and recent activity from PostgreSQL before recommending transfers or flagging low-liquidity accounts.

• •

  Transaction monitoring and anomaly detection
  Write every banking event to PostgreSQL and have an agent score suspicious patterns like repeated micro-transfers or unusual destination accounts.

• •

  Ops dashboard for startup finance teams
  Expose FastAPI endpoints that serve live banking actions while PostgreSQL powers reporting views for reconciliation and audit trails.

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.