How to Integrate FastAPI for wealth management with PostgreSQL for AI agents

FastAPI gives you the request/response layer for wealth-management workflows. PostgreSQL gives your AI agents durable state: client profiles, portfolio snapshots, suitability checks, audit logs, and retrieval-ready transaction history.

That combo is what turns an agent from a chat demo into a system that can answer questions, trigger actions, and keep records your compliance team can inspect later.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL instance
• •A FastAPI app installed and configured
• •psycopg or psycopg2-binary
• •sqlalchemy
• •uvicorn
• •Environment variables for:
  • •DATABASE_URL
  • •FASTAPI_WEALTH_API_KEY if your wealth-management API requires auth
• •A basic schema for:
  • •clients
  • •portfolios
  • •recommendations
  • •audit_events

Integration Steps

1. •Set up your PostgreSQL connection layer.

Use SQLAlchemy for connection pooling and clean session handling. For AI agents, this matters because you do not want every tool call opening raw connections directly.

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

DATABASE_URL = "postgresql+psycopg://agent_user:agent_pass@localhost:5432/wealth_db"

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

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

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

2. •Create the tables your agent needs to persist state.

Keep the schema small and explicit. For wealth management, you usually need client identity, portfolio data, and an audit trail for every recommendation or action the agent makes.

from sqlalchemy import MetaData, Table, Column, Integer, String, Numeric, DateTime, Text, ForeignKey
from sqlalchemy.sql import func

metadata = MetaData()

clients = Table(
    "clients",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(120), nullable=False),
    Column("risk_profile", String(32), nullable=False),
)

portfolios = Table(
    "portfolios",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("client_id", Integer, ForeignKey("clients.id"), nullable=False),
    Column("symbol", String(16), nullable=False),
    Column("quantity", Numeric(18, 6), nullable=False),
)

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

metadata.create_all(bind=engine)

3. •Expose a FastAPI endpoint that reads from PostgreSQL and returns agent-ready context.

This is the bridge. Your agent calls FastAPI; FastAPI queries PostgreSQL and returns structured JSON that the model can reason over.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import select

app = FastAPI()

@app.get("/clients/{client_id}/context")
def get_client_context(client_id: int, db=Depends(get_db)):
    client_row = db.execute(
        select(clients).where(clients.c.id == client_id)
    ).mappings().first()

    if not client_row:
        raise HTTPException(status_code=404, detail="Client not found")

    portfolio_rows = db.execute(
        select(portfolios).where(portfolios.c.client_id == client_id)
    ).mappings().all()

    return {
        "client": dict(client_row),
        "portfolio": [dict(row) for row in portfolio_rows],
        "instructions": [
            "Use risk profile before making any recommendation",
            "Write every decision to audit_events",
        ],
    }

4. •Add an action endpoint that writes agent decisions back to PostgreSQL.

Do not let the model “decide” without persistence. Every recommendation should be stored with enough context to reconstruct why it happened.

from pydantic import BaseModel

class RecommendationIn(BaseModel):
    client_id: int
    event_type: str
    payload: str

@app.post("/audit/recommendation")
def write_recommendation(item: RecommendationIn, db=Depends(get_db)):
    result = db.execute(
        audit_events.insert().values(
            client_id=item.client_id,
            event_type=item.event_type,
            payload=item.payload,
        )
    )
    db.commit()

    return {
        "status": "stored",
        "audit_event_id": result.inserted_primary_key[0],
    }

5. •Wire the endpoint into your AI agent tool layer.

If you are using an agent framework that supports HTTP tools or function calling, point it at your FastAPI routes. The agent fetches context first, then posts its decision after reasoning.

import requests

BASE_URL = "http://localhost:8000"

def fetch_client_context(client_id: int):
    resp = requests.get(f"{BASE_URL}/clients/{client_id}/context")
    resp.raise_for_status()
    return resp.json()

def store_recommendation(client_id: int, payload: str):
    resp = requests.post(
        f"{BASE_URL}/audit/recommendation",
        json={
            "client_id": client_id,
            "event_type": "portfolio_recommendation",
            "payload": payload,
        },
    )
    resp.raise_for_status()
    return resp.json()

Testing the Integration

Run FastAPI:

uvicorn main:app --reload

Then verify both read and write paths:

context = fetch_client_context(1)
print(context["client"]["name"])
print(context["client"]["risk_profile"])

result = store_recommendation(
    1,
    '{"action":"rebalance","reason":"excess concentration in one equity"}'
)
print(result)

Expected output:

Alice Chen
moderate
{'status': 'stored', 'audit_event_id': 12}

If that works end-to-end, your agent can now pull live wealth data from PostgreSQL through FastAPI and persist decisions back into the same system.

Real-World Use Cases

• •

  Portfolio review agents
  Fetch holdings from PostgreSQL through FastAPI, summarize concentration risk, and store a recommendation trail for compliance review.

• •

  Client servicing assistants
  Let an internal assistant answer “What’s my current allocation?” using live database-backed context instead of stale cached data.

• •

  Suitability and policy checks
  Combine stored client risk profiles with product rules before allowing an agent to suggest trades or model portfolios.

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.