How to Integrate FastAPI for wealth management with PostgreSQL for RAG

Combining FastAPI for wealth management with PostgreSQL gives you a clean path from client-facing API to retrieval-backed intelligence. In practice, that means your wealth platform can accept portfolio queries, fetch structured client data, and ground agent responses in PostgreSQL-stored documents, notes, and market research.

This is the pattern you want when building an AI agent for advisors: FastAPI handles the request/response layer, while PostgreSQL becomes the durable retrieval store for RAG over portfolios, policies, suitability notes, and compliance content.

Prerequisites

• •Python 3.10+
• •FastAPI installed
• •Uvicorn for running the API
• •PostgreSQL 14+ running locally or in a managed environment
• •psycopg or psycopg2-binary installed
• •A PostgreSQL database created for your RAG workload
• •Basic knowledge of REST endpoints and SQL
• •Optional: pgvector extension if you want embedding-based retrieval

Integration Steps

1. •Set up the FastAPI service for wealth management endpoints.

Start with a small API that exposes portfolio lookup and advisor query routes. In a wealth management system, this usually means one endpoint for structured client data and another for agent-driven retrieval.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

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

class ClientQuery(BaseModel):
    client_id: str
    question: str

@app.get("/clients/{client_id}/portfolio")
def get_portfolio(client_id: str):
    # Replace with real service call or DB lookup
    if client_id == "missing":
        raise HTTPException(status_code=404, detail="Client not found")
    return {
        "client_id": client_id,
        "assets": [
            {"symbol": "AAPL", "weight": 0.18},
            {"symbol": "BND", "weight": 0.42},
        ],
    }

@app.post("/agent/query")
def agent_query(payload: ClientQuery):
    return {
        "client_id": payload.client_id,
        "question": payload.question,
        "status": "received",
    }

2. •Connect FastAPI to PostgreSQL using a proper connection pool.

Use PostgreSQL as the persistence layer for client notes, policy docs, and retrieved chunks. For production, use a pool so each request does not open a fresh connection.

from contextlib import contextmanager
import psycopg

DATABASE_URL = "postgresql://wealth_user:secret@localhost:5432/wealth_rag"

@contextmanager
def get_conn():
    conn = psycopg.connect(DATABASE_URL)
    try:
        yield conn
    finally:
        conn.close()

def init_db():
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS rag_documents (
                    id SERIAL PRIMARY KEY,
                    client_id TEXT NOT NULL,
                    title TEXT NOT NULL,
                    content TEXT NOT NULL,
                    metadata JSONB DEFAULT '{}'::jsonb
                )
            """)
            conn.commit()

3. •Store retrievable wealth management content in PostgreSQL.

RAG only works if your source material is normalized enough to retrieve reliably. Store advisor notes, suitability summaries, compliance policies, and meeting transcripts as chunks tied to a client_id.

from fastapi import Depends

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

@app.post("/documents")
def add_document(payload: dict):
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO rag_documents (client_id, title, content, metadata)
                VALUES (%s, %s, %s, %s)
                RETURNING id
                """,
                (
                    payload["client_id"],
                    payload["title"],
                    payload["content"],
                    payload.get("metadata", {}),
                ),
            )
            doc_id = cur.fetchone()[0]
            conn.commit()
    return {"id": doc_id}

4. •Add a retrieval endpoint that pulls relevant context from PostgreSQL.

For a first-pass RAG implementation, use keyword matching or metadata filters before adding embeddings. This keeps the integration simple and gives you something testable before you introduce vector search.

@app.get("/documents/search")
def search_documents(client_id: str, q: str):
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT id, title, content
                FROM rag_documents
                WHERE client_id = %s
                  AND (title ILIKE %s OR content ILIKE %s)
                ORDER BY id DESC
                LIMIT 5
                """,
                (client_id, f"%{q}%", f"%{q}%"),
            )
            rows = cur.fetchall()

    return {
        "client_id": client_id,
        "results": [
            {"id": r[0], "title": r[1], "content": r[2]}
            for r in rows
        ],
    }

5. •Wire the retrieval result into your agent response path.

This is where FastAPI becomes the orchestration layer. The endpoint accepts a question, retrieves context from PostgreSQL, and returns grounded text that your LLM can consume next.

@app.post("/agent/wealth-answer")
def wealth_answer(payload: ClientQuery):
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT title, content
                FROM rag_documents
                WHERE client_id = %s
                  AND content ILIKE %s
                ORDER BY id DESC
                LIMIT 3
                """,
                (payload.client_id, f"%{payload.question}%"),
            )
            docs = cur.fetchall()

    context = "\n\n".join([f"{title}: {content}" for title, content in docs])

    return {
        "client_id": payload.client_id,
        "question": payload.question,
        "context": context,
        "answer_hint": "Pass 'context' into your LLM prompt here.",
    }

Testing the Integration

Run the API:

uvicorn main:app --reload

Then verify document storage and retrieval:

import requests

base_url = "http://127.0.0.1:8000"

doc_payload = {
    "client_id": "c123",
    "title": "Risk Tolerance Review",
    "content": "Client prefers moderate risk and has a 10-year investment horizon.",
    "metadata": {"source": "advisor_note"}
}

r1 = requests.post(f"{base_url}/documents", json=doc_payload)
print(r1.json())

r2 = requests.get(f"{base_url}/documents/search", params={"client_id": "c123", "q": "risk"})
print(r2.json())

Expected output:

{"id": 1}
{
  "client_id": "c123",
  "results": [
    {
      "id": 1,
      "title": "Risk Tolerance Review",
      "content": "Client prefers moderate risk and has a 10-year investment horizon."
    }
  ]
}

Real-World Use Cases

• •Advisor copilot that answers questions like “What changed in this client’s risk profile since last review?” using FastAPI endpoints backed by PostgreSQL document retrieval.
• •Compliance-aware assistant that searches stored policy excerpts before generating responses about suitability checks or disclosure requirements.
• •Portfolio servicing workflow that combines structured account data from API routes with unstructured meeting notes stored in PostgreSQL for better recommendations.

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.