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

Why this integration matters

CrewAI for wealth management is useful when you need multi-agent workflows that can research portfolios, summarize market context, and draft client-ready recommendations. FastAPI gives you the HTTP layer to expose those agent workflows as production endpoints that your internal systems, advisors, or client portals can call.

The combination is simple: CrewAI handles orchestration and task execution, while FastAPI turns those workflows into a service with auth, validation, and observability.

Prerequisites

• •Python 3.10+
• •A working FastAPI project
• •CrewAI installed and configured
• •An LLM provider key set in environment variables
• •Basic familiarity with Pydantic models
• •uvicorn for local API runs

Install the dependencies:

pip install fastapi uvicorn crewai crewai-tools pydantic python-dotenv

Set your environment variables:

export OPENAI_API_KEY="your-key"

Integration Steps

1. •Create a CrewAI workflow for wealth management

Start by defining agents and tasks. In wealth management, keep responsibilities narrow: one agent for portfolio analysis, another for compliance-aware recommendation drafting.

from crewai import Agent, Task, Crew, Process

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

recommendation_writer = Agent(
    role="Wealth Recommendation Writer",
    goal="Draft concise client-ready investment recommendations",
    backstory="You translate portfolio findings into advisor-friendly language.",
    verbose=True,
)

analysis_task = Task(
    description=(
        "Review the client's portfolio summary and identify risks, "
        "diversification gaps, and rebalancing opportunities."
    ),
    expected_output="A structured analysis with risks and opportunities.",
    agent=portfolio_analyst,
)

draft_task = Task(
    description=(
        "Using the analysis, draft a short wealth management recommendation "
        "for an advisor to review."
    ),
    expected_output="A client-ready recommendation memo.",
    agent=recommendation_writer,
)

wealth_crew = Crew(
    agents=[portfolio_analyst, recommendation_writer],
    tasks=[analysis_task, draft_task],
    process=Process.sequential,
)

2. •Wrap the crew execution in a service function

Do not call the crew directly from your route handler. Put it behind a service function so you can test it independently and add retries later.

from typing import Dict, Any

def run_wealth_workflow(client_profile: Dict[str, Any]) -> str:
    portfolio_summary = f"""
Client name: {client_profile['name']}
Risk tolerance: {client_profile['risk_tolerance']}
Portfolio: {client_profile['portfolio']}
Goals: {client_profile['goals']}
"""

    result = wealth_crew.kickoff(inputs={"client_profile": portfolio_summary})
    return str(result)

3. •Expose the workflow through FastAPI

Now create a request model and endpoint. Use Pydantic to validate input before it reaches your agents.

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

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

class ClientProfile(BaseModel):
    name: str = Field(..., min_length=1)
    risk_tolerance: str
    goals: List[str]
    portfolio: dict

@app.post("/wealth/recommendation")
async def generate_recommendation(profile: ClientProfile):
    output = run_wealth_workflow(profile.model_dump())
    return {
        "client": profile.name,
        "recommendation": output,
    }

4. •Add startup wiring and run the API locally

If your workflow needs initialization logic later — tool registration, vector store connections, or config loading — keep it in FastAPI startup events or lifespan handlers.

import uvicorn

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

Run it:

uvicorn main:app --reload

5. •Add basic error handling around the crew call

Production systems need predictable failure modes. Catch runtime errors from the agent execution path and return a clean API response.

from fastapi import HTTPException

@app.post("/wealth/recommendation/safe")
async def generate_recommendation_safe(profile: ClientProfile):
    try:
        output = run_wealth_workflow(profile.model_dump())
        return {"client": profile.name, "recommendation": output}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent workflow failed: {str(e)}")

Testing the Integration

Use FastAPI’s TestClient to verify the endpoint returns a response without spinning up a server.

from fastapi.testclient import TestClient

client = TestClient(app)

def test_recommendation_endpoint():
    payload = {
        "name": "Amina Patel",
        "risk_tolerance": "moderate",
        "goals": ["retirement", "capital preservation"],
        "portfolio": {
            "equities": 55,
            "bonds": 35,
            "cash": 10
        }
    }

    response = client.post("/wealth/recommendation", json=payload)
    assert response.status_code == 200
    assert "recommendation" in response.json()

print(test_recommendation_endpoint())

Expected output:

None

If you want to inspect the body instead of just asserting success:

response = client.post("/wealth/recommendation", json=payload)
print(response.json())

Expected shape:

{
  "client": "Amina Patel",
  "recommendation": "...agent-generated wealth management output..."
}

Real-World Use Cases

• •

  Advisor copilot API
  Expose an endpoint that summarizes client portfolios and drafts talking points before advisor meetings.

• •

  Rebalancing suggestion service
  Build an internal API that reviews holdings against policy rules and suggests reallocation actions.

• •

  Client onboarding assistant
  Accept new-client profiles through FastAPI and have CrewAI agents produce suitability summaries, next steps, and follow-up questions.

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.