How to Integrate FastAPI for fintech with PostgreSQL for AI agents

FastAPI for fintech gives you a clean API layer for payments, account actions, and internal workflows. PostgreSQL gives your AI agents durable state: customer profiles, transaction history, policy decisions, tool outputs, and audit logs. Put them together and you can build agents that don’t just answer questions — they can fetch financial data, persist decisions, and trigger controlled actions.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL instance
• •Access to your FastAPI fintech service
• •A PostgreSQL database and credentials
• •pip installed
• •Basic familiarity with:
  • •fastapi
  • •psycopg or asyncpg
  • •uvicorn
• •Environment variables configured:
  • •DATABASE_URL
  • •FINTECH_API_BASE_URL
  • •FINTECH_API_KEY

Install the packages:

pip install fastapi uvicorn psycopg[binary] sqlalchemy httpx pydantic

Integration Steps

1. •Define the database connection layer

Use PostgreSQL as the source of truth for agent state. For fintech workloads, keep connection handling explicit and pooled.

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

DATABASE_URL = os.getenv("DATABASE_URL")

engine = create_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

2. •Create a table for agent memory and audit trail

For fintech agents, store every action with enough context to replay decisions later.

from sqlalchemy import Table, Column, Integer, String, JSON, DateTime, MetaData, func

metadata = MetaData()

agent_events = Table(
    "agent_events",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("agent_id", String(64), nullable=False),
    Column("event_type", String(64), nullable=False),
    Column("payload", JSON, nullable=False),
    Column("created_at", DateTime(timezone=True), server_default=func.now()),
)

Create the table once at startup or through migrations:

from sqlalchemy import create_engine

engine = create_engine(DATABASE_URL)
metadata.create_all(engine)

3. •Build the FastAPI fintech client wrapper

If your fintech platform exposes REST endpoints through FastAPI, call it with httpx. Keep auth and timeouts explicit.

import os
import httpx

FINTECH_API_BASE_URL = os.getenv("FINTECH_API_BASE_URL")
FINTECH_API_KEY = os.getenv("FINTECH_API_KEY")

class FintechClient:
    def __init__(self):
        self.client = httpx.Client(
            base_url=FINTECH_API_BASE_URL,
            headers={"Authorization": f"Bearer {FINTECH_API_KEY}"},
            timeout=10.0,
        )

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

    def create_transfer(self, source_account: str, target_account: str, amount: float):
        response = self.client.post(
            "/transfers",
            json={
                "source_account": source_account,
                "target_account": target_account,
                "amount": amount,
            },
        )
        response.raise_for_status()
        return response.json()

4. •Wire FastAPI endpoints to PostgreSQL and the fintech service

This is where the integration becomes useful for an AI agent system. The API receives intent from the agent, checks state in PostgreSQL, then calls the fintech service.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import insert

app = FastAPI()
fintech_client = FintechClient()

@app.post("/agent/transfer")
def agent_transfer(payload: dict, db: Session = Depends(get_db)):
    agent_id = payload["agent_id"]
    source_account = payload["source_account"]
    target_account = payload["target_account"]
    amount = payload["amount"]

    balance_data = fintech_client.get_account_balance(source_account)
    if balance_data["available_balance"] < amount:
        raise HTTPException(status_code=400, detail="Insufficient funds")

    transfer_result = fintech_client.create_transfer(
        source_account=source_account,
        target_account=target_account,
        amount=amount,
    )

    stmt = insert(agent_events).values(
        agent_id=agent_id,
        event_type="transfer_created",
        payload={
            "source_account": source_account,
            "target_account": target_account,
            "amount": amount,
            "transfer_id": transfer_result["transfer_id"],
            "status": transfer_result["status"],
        },
    )
    db.execute(stmt)
    db.commit()

    return {"ok": True, "transfer": transfer_result}

5. •Add a retrieval endpoint for agent context

AI agents need memory. Use PostgreSQL to fetch previous actions before deciding what to do next.

from sqlalchemy import select

@app.get("/agent/{agent_id}/events")
def list_agent_events(agent_id: str, db: Session = Depends(get_db)):
    stmt = (
        select(agent_events)
        .where(agent_events.c.agent_id == agent_id)
        .order_by(agent_events.c.created_at.desc())
        .limit(20)
    )
    rows = db.execute(stmt).mappings().all()
    return {"agent_id": agent_id, "events": [dict(row) for row in rows]}

Testing the Integration

Run the app:

uvicorn main:app --reload --port 8000

Then verify both sides work together:

import httpx

payload = {
    "agent_id": "agent_123",
    "source_account": "acct_001",
    "target_account": "acct_002",
    "amount": 125.50,
}

response = httpx.post("http://localhost:8000/agent/transfer", json=payload)
print(response.status_code)
print(response.json())

events_response = httpx.get("http://localhost:8000/agent/agent_123/events")
print(events_response.json())

Expected output:

200
{
  "ok": true,
  "transfer": {
    "transfer_id": "trf_789",
    "status": "submitted"
  }
}
{
  "agent_id": "agent_123",
  "events": [
    {
      "id": 1,
      "agent_id": "agent_123",
      "event_type": "transfer_created",
      "...": "..."
    }
  ]
}

Real-World Use Cases

• •

  Payment operations agents

  • •Check balances in FastAPI-backed fintech services.
  • •Store approval history and transfer outcomes in PostgreSQL.
  • •Enforce auditability before any money movement.
• •

  Customer support agents

  • •Pull transaction history from PostgreSQL.
  • •Call FastAPI endpoints to freeze cards or open disputes.
  • •Persist every support action for compliance review.
• •

  Risk and fraud workflows

  • •Score suspicious activity using stored events in PostgreSQL.
  • •Trigger FastAPI endpoints for limits changes or step-up verification.
  • •Keep a full decision trail for regulators and internal teams.

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.