How to Integrate CrewAI for wealth management with FastAPI for production AI

Combining CrewAI for wealth management with FastAPI gives you a clean way to expose multi-agent financial workflows as production APIs. You get CrewAI handling research, analysis, and recommendation generation, while FastAPI gives you a typed, async HTTP layer that can sit behind auth, observability, and rate limits.

This is the right pattern when you want portfolio review, client risk profiling, or investment memo generation to run as an API instead of a one-off notebook.

Prerequisites

• •Python 3.10+
• •crewai
• •fastapi
• •uvicorn
• •pydantic
• •An LLM provider configured for CrewAI, such as OpenAI or Anthropic
• •Environment variables set for your model provider
• •Basic familiarity with REST APIs and async Python

Install the packages:

pip install crewai fastapi uvicorn pydantic

Set your API key:

export OPENAI_API_KEY="your-key"

Integration Steps

1. •Define your wealth management agents and task

Start by creating focused agents. In production, keep each agent narrow: one for portfolio analysis, one for market research, one for recommendation synthesis.

from crewai import Agent, Task, Crew, Process

portfolio_analyst = Agent(
    role="Portfolio Analyst",
    goal="Analyze client portfolios for risk, diversification, and concentration issues",
    backstory="You are a senior wealth management analyst focused on portfolio construction.",
    verbose=True,
)

market_researcher = Agent(
    role="Market Researcher",
    goal="Summarize relevant market conditions and macro risks",
    backstory="You provide concise market context for wealth advisors.",
    verbose=True,
)

recommendation_writer = Agent(
    role="Wealth Recommendation Writer",
    goal="Produce client-ready investment recommendations",
    backstory="You translate analysis into clear advisor-facing output.",
    verbose=True,
)

2. •Build a CrewAI workflow for the request

Use a task pipeline that accepts client data and produces a structured recommendation. For production systems, keep inputs explicit so the API layer can validate them before execution.

def build_wealth_crew(client_profile: dict) -> Crew:
    analysis_task = Task(
        description=f"""
        Review this client profile and identify portfolio risks:
        {client_profile}
        Focus on allocation gaps, concentration risk, and suitability concerns.
        """,
        expected_output="A concise risk analysis with actionable findings.",
        agent=portfolio_analyst,
    )

    market_task = Task(
        description="""
        Summarize current market conditions relevant to this client's holdings.
        Include rate risk, equity volatility, and sector exposure considerations.
        """,
        expected_output="A short market context summary.",
        agent=market_researcher,
    )

    recommendation_task = Task(
        description="""
        Combine the portfolio analysis and market context into a client-ready recommendation.
        Keep it practical and suitable for an advisor workflow.
        """,
        expected_output="A clear recommendation memo.",
        agent=recommendation_writer,
    )

    return Crew(
        agents=[portfolio_analyst, market_researcher, recommendation_writer],
        tasks=[analysis_task, market_task, recommendation_task],
        process=Process.sequential,
        verbose=True,
    )

3. •Expose the workflow through FastAPI

Create a request model and an endpoint that triggers the crew. This is where FastAPI gives you validation, routing, and response shaping.

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional

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

class ClientProfile(BaseModel):
    client_id: str
    age: int
    risk_tolerance: str
    holdings: List[str]
    cash_balance: float
    investment_horizon_years: int
    notes: Optional[str] = None

@app.post("/wealth/review")
async def review_portfolio(profile: ClientProfile):
    crew = build_wealth_crew(profile.model_dump())
    result = crew.kickoff()
    
    return {
        "client_id": profile.client_id,
        "recommendation": str(result),
    }

4. •Run the app with Uvicorn

Keep the API server separate from the agent logic. That makes it easier to scale workers independently later.

# main.py

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Start it:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

5. •Add production guardrails

For real deployments, add timeout handling, input limits, and logging around the crew execution. Crew runs can be expensive and slow if you let unbounded prompts through.

import logging
from fastapi import HTTPException

logger = logging.getLogger(__name__)

@app.post("/wealth/review")
async def review_portfolio(profile: ClientProfile):
    try:
        if len(profile.holdings) > 50:
            raise HTTPException(status_code=400, detail="Too many holdings in request")

        crew = build_wealth_crew(profile.model_dump())
        result = crew.kickoff()

        logger.info("Completed wealth review for %s", profile.client_id)

        return {
            "client_id": profile.client_id,
            "status": "completed",
            "recommendation": str(result),
        }
    except Exception as e:
        logger.exception("Wealth review failed")
        raise HTTPException(status_code=500, detail=str(e))

Testing the Integration

Use curl or any HTTP client to verify the endpoint works end to end.

curl -X POST "http://localhost:8000/wealth/review" \
  -H "Content-Type: application/json" \
  -d '{
    "client_id": "C12345",
    "age": 52,
    "risk_tolerance": "moderate",
    "holdings": ["SPY", "AGG", "AAPL"],
    "cash_balance": 25000,
    "investment_horizon_years": 10,
    "notes": "Wants lower volatility"
  }'

Expected output:

{
  "client_id": "C12345",
  "status": "completed",
  "recommendation": "..."
}

If you want a quick Python test instead:

import requests

payload = {
    "client_id": "C12345",
    "age": 52,
    "risk_tolerance": "moderate",
    "holdings": ["SPY", "AGG", "AAPL"],
    "cash_balance": 25000,
    "investment_horizon_years": 10,
}

response = requests.post("http://localhost:8000/wealth/review", json=payload)
print(response.status_code)
print(response.json())

Real-World Use Cases

• •

  Advisor copilot API
  Let internal advisor tools call /wealth/review to generate portfolio notes before client meetings.

• •

  Client onboarding intelligence
  Use CrewAI to analyze new-client intake forms and return suitability flags through FastAPI.

• •

  Investment memo generation
  Build an endpoint that turns holdings plus market context into standardized investment committee drafts.

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.