How to Integrate CrewAI for wealth management with FastAPI for RAG

CrewAI for wealth management gives you the orchestration layer for multi-step financial tasks. FastAPI gives you a clean HTTP boundary to expose retrieval and agent workflows as production services. Put them together, and you can build a RAG-backed advisor system that answers portfolio, policy, and client-document questions with traceable, API-driven execution.

Prerequisites

• •Python 3.10+
• •fastapi
• •uvicorn
• •crewai
• •crewai-tools
• •An embedding provider or vector store for RAG
• •Access to your document corpus:
  • •prospectus PDFs
  • •investment policy statements
  • •product sheets
  • •KYC / suitability notes
• •API keys configured in environment variables:
  • •OPENAI_API_KEY or your model provider key
  • •vector DB credentials if applicable

Install the core packages:

pip install fastapi uvicorn crewai crewai-tools pydantic

Integration Steps

1. •

   Define the RAG retrieval layer

   Keep retrieval separate from orchestration. FastAPI should expose a predictable search function that CrewAI can call as a tool.

from typing import List
from pydantic import BaseModel
from crewai_tools import tool

class RetrievedChunk(BaseModel):
    source: str
    content: str

@tool("wealth_rag_search")
def wealth_rag_search(query: str) -> List[dict]:
    """
    Search wealth management documents and return top matches.
    Replace this stub with your vector DB query.
    """
    results = [
        {"source": "ips_2024.pdf", "content": "Target allocation is 60/40 equities/fixed income."},
        {"source": "fund_fact_sheet.pdf", "content": "The fund has a 0.45% expense ratio."},
    ]
    return results

2. •

   Create the CrewAI agent for wealth management

   The agent should reason over retrieved context, not hallucinate from memory. Use a finance-specific role and constrain it to cite retrieved evidence.

from crewai import Agent, Task, Crew, Process

wealth_advisor = Agent(
    role="Wealth Management Analyst",
    goal="Answer client and advisor questions using retrieved portfolio documents.",
    backstory=(
        "You support wealth managers by analyzing IPS documents, fund factsheets, "
        "and client notes. You must ground answers in retrieved context."
    ),
    tools=[wealth_rag_search],
    verbose=True,
)

answer_task = Task(
    description=(
        "Use the retrieved context to answer the user's question about portfolio policy, "
        "fund details, or suitability constraints."
    ),
    expected_output="A concise answer with source references.",
    agent=wealth_advisor,
)

crew = Crew(
    agents=[wealth_advisor],
    tasks=[answer_task],
    process=Process.sequential,
)

3. •

   Expose the workflow through FastAPI

   FastAPI becomes your integration surface for chat apps, advisor dashboards, and internal automation.

from fastapi import FastAPI
from pydantic import BaseModel

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

class QueryRequest(BaseModel):
    question: str

class QueryResponse(BaseModel):
    answer: str

@app.post("/rag/query", response_model=QueryResponse)
def rag_query(payload: QueryRequest):
    result = crew.kickoff(inputs={"question": payload.question})
    return QueryResponse(answer=str(result))

4. •

   Wire in a real retrieval call before the agent runs

   In production, your endpoint should fetch relevant chunks first, then pass them into the task context. This keeps retrieval deterministic and makes responses easier to audit.

from fastapi import HTTPException

def retrieve_context(question: str) -> str:
    matches = wealth_rag_search(question)
    if not matches:
        raise HTTPException(status_code=404, detail="No relevant documents found")

    formatted = []
    for item in matches:
        formatted.append(f"[{item['source']}] {item['content']}")
    return "\n".join(formatted)

@app.post("/rag/query-v2", response_model=QueryResponse)
def rag_query_v2(payload: QueryRequest):
    context = retrieve_context(payload.question)

    result = crew.kickoff(
        inputs={
            "question": payload.question,
            "context": context,
        }
    )
    return QueryResponse(answer=str(result))

5. •

   Run FastAPI and connect it to your client

   Start the service with Uvicorn and call it from any internal app or agent gateway.

uvicorn main:app --reload --port 8000

If you want the CrewAI task to explicitly use the retrieved context, update the task description:

answer_task = Task(
    description=(
        "Answer the question using only this context:\n"
        "{context}\n\n"
        "Question:\n"
        "{question}\n\n"
        "Cite which document each claim came from."
    ),
    expected_output="Grounded answer with citations.",
    agent=wealth_advisor,
)

Testing the Integration

Use curl or Python requests to verify the endpoint returns grounded output.

import requests

resp = requests.post(
    "http://localhost:8000/rag/query-v2",
    json={"question": "What is the portfolio allocation target?"}
)

print(resp.status_code)
print(resp.json())

Expected output:

{
  "answer": "The target allocation is 60/40 equities/fixed income. Source: ips_2024.pdf"
}

If you see an empty answer or uncited claims, check these first:

• •The retriever is returning chunks
• •The task prompt includes {context}
• •The agent is not being asked to answer without evidence
• •Your model provider key is set correctly

Real-World Use Cases

• •Advisor copilots that answer questions like:
  • •“Can this client hold alternatives?”
  • •“What does their IPS say about concentration limits?”
• •Client service automation for document-heavy workflows:
  • •fund fact lookup
  • •fee explanations
  • •policy-based suitability checks
• •Internal compliance assistants that search approved sources before generating responses:
  • •pre-trade rule summaries
  • •disclosure lookup
  • •audit-friendly Q&A trails

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.