How to Integrate FastAPI for wealth management with PostgreSQL for startups

FastAPI gives you a clean API layer for wealth management workflows. PostgreSQL gives you durable, queryable storage for portfolios, client profiles, transactions, and audit trails. Put them together and you get a backend that can power an AI agent system for startups without turning every request into a pile of ad hoc JSON.

Prerequisites

• •Python 3.10+
• •FastAPI installed
• •Uvicorn installed
• •psycopg2-binary or psycopg installed
• •PostgreSQL 14+
• •A running PostgreSQL database with credentials
• •Basic understanding of REST APIs and SQL
• •Optional: python-dotenv for local environment variables

Install the dependencies:

pip install fastapi uvicorn psycopg2-binary python-dotenv

Integration Steps

1. •Set up your PostgreSQL connection layer.

Use a small database module instead of opening connections inside route handlers. That keeps your FastAPI app testable and makes connection handling explicit.

# db.py
import os
import psycopg2
from psycopg2.extras import RealDictCursor

def get_connection():
    return psycopg2.connect(
        host=os.getenv("POSTGRES_HOST", "localhost"),
        dbname=os.getenv("POSTGRES_DB", "wealth_db"),
        user=os.getenv("POSTGRES_USER", "postgres"),
        password=os.getenv("POSTGRES_PASSWORD", "postgres"),
        port=os.getenv("POSTGRES_PORT", "5432"),
    )

def init_db():
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS clients (
                    id SERIAL PRIMARY KEY,
                    name TEXT NOT NULL,
                    email TEXT UNIQUE NOT NULL,
                    risk_profile TEXT NOT NULL,
                    created_at TIMESTAMP DEFAULT NOW()
                );
            """)
            cur.execute("""
                CREATE TABLE IF NOT EXISTS portfolios (
                    id SERIAL PRIMARY KEY,
                    client_id INTEGER REFERENCES clients(id) ON DELETE CASCADE,
                    symbol TEXT NOT NULL,
                    allocation NUMERIC(10,4) NOT NULL,
                    market_value NUMERIC(14,2) NOT NULL DEFAULT 0
                );
            """)
            conn.commit()
    finally:
        conn.close()

2. •Build the FastAPI service layer.

This is the API surface your agent system will call. In wealth management, keep endpoints narrow: create clients, fetch portfolios, and update allocations.

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from db import get_connection, init_db

app = FastAPI(title="Wealth Management API")

class ClientCreate(BaseModel):
    name: str
    email: EmailStr
    risk_profile: str

class PortfolioCreate(BaseModel):
    client_id: int
    symbol: str
    allocation: float
    market_value: float = 0

@app.on_event("startup")
def startup():
    init_db()

@app.post("/clients")
def create_client(payload: ClientCreate):
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO clients (name, email, risk_profile)
                VALUES (%s, %s, %s)
                RETURNING id, name, email, risk_profile;
                """,
                (payload.name, payload.email, payload.risk_profile),
            )
            row = cur.fetchone()
            conn.commit()
            return {
                "id": row[0],
                "name": row[1],
                "email": row[2],
                "risk_profile": row[3],
            }
    finally:
        conn.close()

@app.post("/portfolios")
def create_portfolio(payload: PortfolioCreate):
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT id FROM clients WHERE id = %s;", (payload.client_id,))
            if not cur.fetchone():
                raise HTTPException(status_code=404, detail="Client not found")

            cur.execute(
                """
                INSERT INTO portfolios (client_id, symbol, allocation, market_value)
                VALUES (%s, %s, %s, %s)
                RETURNING id;
                """,
                (payload.client_id, payload.symbol, payload.allocation, payload.market_value),
            )
            portfolio_id = cur.fetchone()[0]
            conn.commit()
            return {"id": portfolio_id}
    finally:
        conn.close()

3. •Add read endpoints for your AI agent.

Agents need retrieval more than anything else. Give them structured access to client and portfolio state so they can generate recommendations or trigger workflows.

# main.py continued
@app.get("/clients/{client_id}")
def get_client(client_id: int):
    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute("SELECT * FROM clients WHERE id = %s;", (client_id,))
            client = cur.fetchone()
            if not client:
                raise HTTPException(status_code=404, detail="Client not found")
            return dict(client)
    finally:
        conn.close()

@app.get("/clients/{client_id}/portfolio")
def get_portfolio(client_id: int):
    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute("""
                SELECT id, symbol, allocation, market_value
                FROM portfolios
                WHERE client_id = %s;
            """, (client_id,))
            rows = cur.fetchall()
            return {"client_id": client_id, "positions": [dict(r) for r in rows]}
    finally:
        conn.close()

4. •Wire the API into an AI agent workflow.

In a startup setting, the agent usually calls your FastAPI service as a tool. The pattern is simple: the model decides what to fetch; your service returns authoritative data from PostgreSQL.

# agent_client.py
import requests

BASE_URL = "http://localhost:8000"

def fetch_client_context(client_id: int):
    client_resp = requests.get(f"{BASE_URL}/clients/{client_id}")
    portfolio_resp = requests.get(f"{BASE_URL}/clients/{client_id}/portfolio")

    client_resp.raise_for_status()
    portfolio_resp.raise_for_status()

    return {
        "client": client_resp.json(),
        "portfolio": portfolio_resp.json(),
    }

context = fetch_client_context(1)
print(context)

5. •Run the service and persist real transactions.

For production use cases like rebalancing or contribution tracking, write transaction records in PostgreSQL inside one database transaction. That keeps your financial state consistent when requests fail halfway through.

@app.post("/rebalance/{client_id}")
def rebalance(client_id: int):
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT id FROM clients WHERE id = %s;", (client_id,))
            if not cur.fetchone():
                raise HTTPException(status_code=404, detail="Client not found")

            cur.execute("""
                UPDATE portfolios
                SET allocation = allocation * 1.05
                WHERE client_id = %s;
            """, (client_id,))
            conn.commit()
            return {"status": "rebalance_applied", "client_id": client_id}
    finally:
        conn.close()

Start it with:

uvicorn main:app --reload --port 8000

Testing the Integration

Use FastAPI’s built-in docs or a direct request to verify the full path from API to PostgreSQL works.

import requests

base_url = "http://localhost:8000"

create_client_payload = {
    "name": "Ava Chen",
    "email": "ava@example.com",
    "risk_profile": "moderate"
}

r1 = requests.post(f"{base_url}/clients", json=create_client_payload)
print(r1.status_code)
print(r1.json())

client_id = r1.json()["id"]

r2 = requests.post(
    f"{base_url}/portfolios",
    json={
        "client_id": client_id,
        "symbol": "VOO",
        "allocation": 0.75,
        "market_value": 25000.00,
    },
)
print(r2.status_code)
print(r2.json())

r3 = requests.get(f"{base_url}/clients/{client_id}/portfolio")
print(r3.status_code)
print(r3.json())

Expected output:

200
{'id': 1, 'name': 'Ava Chen', 'email': 'ava@example.com', 'risk_profile': 'moderate'}
200
{'id': 1}
200
{'client_id': 1, 'positions': [{'id': 1, 'symbol': 'VOO', 'allocation': '0.7500', 'market_value': '25000.00'}]}

Real-World Use Cases

• •Client onboarding agents that collect profile data through FastAPI and store it in PostgreSQL for KYC and suitability checks.
• •Portfolio monitoring agents that read holdings from PostgreSQL and trigger alerts when allocations drift outside policy.
• •Recommendation agents that query historical transactions from PostgreSQL before generating rebalancing or tax-loss harvesting suggestions.

If you want this to hold up in production startup environments, keep the API stateless and let PostgreSQL own the source of truth. FastAPI handles orchestration; PostgreSQL handles persistence; your agent layer sits on top and makes decisions from reliable data.

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.