How to Integrate CrewAI for pension funds with FastAPI for startups

Combining CrewAI for pension funds with FastAPI gives you a clean way to expose agent-driven workflows over HTTP without turning your app into a pile of glue code. For startups building retirement, benefits, or wealth-adjacent products, this setup lets you route user requests into specialized agents, then return structured answers through an API your frontend or partners can consume.

Prerequisites

• •Python 3.10+
• •A FastAPI project with uvicorn installed
• •CrewAI installed and configured in your environment
• •API keys or credentials for any LLM provider your CrewAI agents use
• •A basic understanding of:
  • •FastAPI()
  • •path operations like @app.post(...)
  • •Pydantic request/response models
• •A working project structure, for example:
  • •app/main.py
  • •app/agents.py
  • •app/schemas.py

Integration Steps

1. •

   Install the dependencies

   Start with the core packages. If your CrewAI setup uses tools or providers, install those too.

   pip install fastapi uvicorn crewai pydantic
   

2. •

   Define request and response schemas

   Keep the API contract strict. Pension workflows should not accept free-form payloads without validation.

   # app/schemas.py
   from pydantic import BaseModel, Field
   
   class PensionQueryRequest(BaseModel):
       member_id: str = Field(..., examples=["MEM-10293"])
       query: str = Field(..., examples=["What is my projected retirement balance at age 65?"])
   
   class PensionQueryResponse(BaseModel):
       answer: str
       agent_name: str
       status: str = "success"
   

3. •

   Create a CrewAI agent and task for pension fund queries

   Use CrewAI’s Agent, Task, and Crew objects to encapsulate the workflow. In a production system, this is where you’d add domain rules, compliance instructions, and retrieval tools.

   # app/agents.py
   from crewai import Agent, Task, Crew, Process
   
   pension_agent = Agent(
       role="Pension Fund Analyst",
       goal="Answer pension fund member questions accurately and concisely",
       backstory=(
           "You work with retirement plan data, benefit summaries, "
           "and member-facing explanations."
       ),
       verbose=True,
       allow_delegation=False,
   )
   
   def run_pension_crew(member_id: str, query: str) -> str:
       task = Task(
           description=(
               f"Member ID: {member_id}\n"
               f"Question: {query}\n"
               "Provide a concise response suitable for a startup pension portal."
           ),
           expected_output="A clear pension-related answer with no unsupported claims.",
           agent=pension_agent,
       )
   
       crew = Crew(
           agents=[pension_agent],
           tasks=[task],
           process=Process.sequential,
           verbose=True,
       )
   
       result = crew.kickoff()
       return str(result)
   

4. •

   Expose the agent through FastAPI

   This is the integration point. FastAPI receives the request, validates it, calls the CrewAI workflow, then returns JSON.

   # app/main.py
   from fastapi import FastAPI, HTTPException
   from app.schemas import PensionQueryRequest, PensionQueryResponse
   from app.agents import run_pension_crew
   
   app = FastAPI(title="Pension AI API")
   
   @app.post("/pension/query", response_model=PensionQueryResponse)
   async def pension_query(payload: PensionQueryRequest):
       try:
           answer = run_pension_crew(
               member_id=payload.member_id,
               query=payload.query,
           )
           return PensionQueryResponse(
               answer=answer,
               agent_name="Pension Fund Analyst",
           )
       except Exception as exc:
           raise HTTPException(status_code=500, detail=str(exc))
   

5. •

   Run the API and wire it into your startup stack

   Use Uvicorn locally first. In production, put this behind your normal ASGI deployment setup.

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

Testing the Integration

Use FastAPI’s test client to verify that the endpoint accepts input and returns a structured response.

# tests/test_pension_api.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_pension_query():
    response = client.post(
        "/pension/query",
        json={
            "member_id": "MEM-10293",
            "query": "What is my projected retirement balance at age 65?"
        },
    )

    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "success"
    assert data["agent_name"] == "Pension Fund Analyst"
    assert "answer" in data

print(client.post(
    "/pension/query",
    json={
        "member_id": "MEM-10293",
        "query": "Can I increase my monthly contribution?"
    },
).json())

Expected output:

{
  "answer": "...",
  "agent_name": "Pension Fund Analyst",
  "status": "success"
}

Real-World Use Cases

• •

  Member self-service assistant

  • •Answer contribution questions, retirement projections, vesting timelines, and plan rules through an API consumed by a web portal or mobile app.
• •

  Advisor support backend

  • •Let internal advisors query plan-specific information using an agent layer while FastAPI handles authentication and request routing.
• •

  Claims and benefits triage

  • •Route incoming benefit-related questions to specialized agents that summarize policy context before escalating to human review.

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.