How to Integrate FastAPI for wealth management with PostgreSQL for multi-agent systems

FastAPI gives you the API layer for wealth-management workflows: client profiles, portfolio views, suitability checks, and advisor actions. PostgreSQL gives you the durable state layer for multi-agent systems: shared memory, task queues, audit trails, and agent-to-agent coordination without losing consistency.

When you combine them, you get a clean split: FastAPI handles request/response orchestration, while PostgreSQL stores the facts your agents need to act on. That’s the pattern you want for regulated systems where multiple agents must read the same customer context and write back decisions safely.

Prerequisites

• •Python 3.10+
• •FastAPI installed
• •Uvicorn installed
• •PostgreSQL 14+
• •psycopg or asyncpg installed
• •A running PostgreSQL database with credentials
• •Basic understanding of REST APIs and SQL
• •An AI agent runtime that can call HTTP endpoints and read/write state

Install the Python packages:

pip install fastapi uvicorn psycopg[binary] pydantic

Integration Steps

1. •

   Create the PostgreSQL schema for shared agent state

   For multi-agent systems, don’t store just “customers.” Store agent-ready entities: tasks, portfolio snapshots, and decisions. Keep writes idempotent so agents can retry safely.

import psycopg

DB_URL = "postgresql://postgres:postgres@localhost:5432/wealth_agents"

schema_sql = """
CREATE TABLE IF NOT EXISTS client_profiles (
    client_id TEXT PRIMARY KEY,
    full_name TEXT NOT NULL,
    risk_profile TEXT NOT NULL,
    updated_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE IF NOT EXISTS agent_tasks (
    task_id UUID PRIMARY KEY,
    client_id TEXT NOT NULL REFERENCES client_profiles(client_id),
    agent_name TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    payload JSONB NOT NULL,
    result JSONB,
    created_at TIMESTAMPTZ DEFAULT now(),
    updated_at TIMESTAMPTZ DEFAULT now()
);
"""

with psycopg.connect(DB_URL) as conn:
    with conn.cursor() as cur:
        cur.execute(schema_sql)
    conn.commit()

2. •

   Build the FastAPI service that reads and writes PostgreSQL

   Use FastAPI endpoints as the contract between your agents and your wealth-management backend. In production, this is where you enforce validation, authorization, and audit logging.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import psycopg

app = FastAPI()
DB_URL = "postgresql://postgres:postgres@localhost:5432/wealth_agents"

class ClientProfileIn(BaseModel):
    client_id: str
    full_name: str
    risk_profile: str

@app.post("/clients")
def upsert_client(profile: ClientProfileIn):
    sql = """
    INSERT INTO client_profiles (client_id, full_name, risk_profile)
    VALUES (%s, %s, %s)
    ON CONFLICT (client_id)
    DO UPDATE SET full_name = EXCLUDED.full_name,
                  risk_profile = EXCLUDED.risk_profile,
                  updated_at = now()
    RETURNING client_id, full_name, risk_profile;
    """
    with psycopg.connect(DB_URL) as conn:
        with conn.cursor() as cur:
            cur.execute(sql, (profile.client_id, profile.full_name, profile.risk_profile))
            row = cur.fetchone()
        conn.commit()

    return {"client_id": row[0], "full_name": row[1], "risk_profile": row[2]}

3. •

   Add an endpoint for agent task orchestration

   Multi-agent systems need a shared queue or task table. One agent can create a task for portfolio review; another can pick it up and write back its result.

from uuid import UUID
from datetime import datetime

class AgentTaskIn(BaseModel):
    task_id: str
    client_id: str
    agent_name: str
    payload: dict

@app.post("/tasks")
def create_task(task: AgentTaskIn):
    sql = """
    INSERT INTO agent_tasks (task_id, client_id, agent_name, payload)
    VALUES (%s::uuid, %s, %s, %s::jsonb)
    RETURNING task_id, status;
    """
    with psycopg.connect(DB_URL) as conn:
        with conn.cursor() as cur:
            cur.execute(sql, (task.task_id, task.client_id, task.agent_name, json.dumps(task.payload)))
            row = cur.fetchone()
        conn.commit()

    return {"task_id": str(row[0]), "status": row[1]}

4. •

   Expose a retrieval endpoint for downstream agents

   Agents should fetch only what they need. For wealth management workflows this usually means current profile data plus open tasks.

import json

@app.get("/clients/{client_id}")
def get_client(client_id: str):
    sql = "SELECT client_id, full_name, risk_profile FROM client_profiles WHERE client_id = %s;"
    
    with psycopg.connect(DB_URL) as conn:
        with conn.cursor() as cur:
            cur.execute(sql, (client_id,))
            row = cur.fetchone()

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

    return {
        "client_id": row[0],
        "full_name": row[1],
        "risk_profile": row[2],
    }

5. •

   Wire an agent worker to call FastAPI and persist results in PostgreSQL

   This is the real integration point. The worker reads from the API layer using httpx, computes a recommendation, then stores the outcome back into PostgreSQL through FastAPI or directly if you control both sides.

import httpx
import psycopg

BASE_URL = "http://localhost:8000"

def run_rebalancing_agent(client_id: str):
    with httpx.Client() as client:
        profile_resp = client.get(f"{BASE_URL}/clients/{client_id}")
        profile_resp.raise_for_status()
        profile = profile_resp.json()

        recommendation = {
            "action": "rebalance",
            "target_allocation": {"equities": 60, "bonds": 30, "cash": 10},
            "reason": f"Risk profile is {profile['risk_profile']}"
        }

        # In production prefer a dedicated /tasks/{id}/result endpoint.
        with psycopg.connect(DB_URL) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    UPDATE agent_tasks
                    SET status = 'completed',
                        result = %s::jsonb,
                        updated_at = now()
                    WHERE client_id = %s AND agent_name = 'rebalancing_agent'
                      AND status = 'pending'
                    """,
                    (json.dumps(recommendation), client_id),
                )
            conn.commit()

if __name__ == "__main__":
    run_rebalancing_agent("client-001")

Testing the Integration

Start the API:

uvicorn main:app --reload --port 8000

Then verify both write and read paths:

import httpx

base_url = "http://localhost:8000"

client_payload = {
    "client_id": "client-001",
    "full_name": "Amina Patel",
    "risk_profile": "moderate"
}

task_payload = {
    "task_id": "550e8400-e29b-41d4-a716-446655440000",
    "client_id": "client-001",
    "agent_name": "rebalancing_agent",
    "payload": {"objective": "review allocation"}
}

with httpx.Client() as client:
    r1 = client.post(f"{base_url}/clients", json=client_payload)
    r2 = client.post(f"{base_url}/tasks", json=task_payload)
    r3 = client.get(f"{base_url}/clients/client-001")

print(r1.json())
print(r2.json())
print(r3.json())

Expected output:

{'client_id': 'client-001', 'full_name': 'Amina Patel', 'risk_profile': 'moderate'}
{'task_id': '550e8400-e29b-41d4-a716-446655440000', 'status': 'pending'}
{'client_id': 'client-001', 'full_name': 'Amina Patel', 'risk_profile': 'moderate'}

Real-World Use Cases

• •

  Portfolio review agents

  • •One agent pulls current holdings from FastAPI.
  • •Another stores suitability checks and recommendations in PostgreSQL for auditability.
• •

  Advisor workflow automation

  • •Agents create follow-up tasks after market events.
  • •PostgreSQL keeps task state consistent across workers.
• •

  Client servicing copilots

  • •A support agent reads profile data through FastAPI.
  • •It writes interaction summaries into PostgreSQL so other agents can continue the thread without reprocessing history.

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.