How to Integrate CrewAI for pension funds with FastAPI for production AI

Combining CrewAI for pension funds with FastAPI gives you a clean path from agent orchestration to production HTTP endpoints. The practical win is simple: your pension workflows — document review, policy Q&A, member support, contribution anomaly checks — can run behind an API your internal systems can call.

FastAPI handles request validation, auth, and response shaping. CrewAI for pension funds handles multi-agent task execution, so you can turn a prompt into a controlled workflow instead of a single brittle model call.

Prerequisites

Before wiring this up, make sure you have:

• •Python 3.10+
• •fastapi
• •uvicorn
• •crewai
• •Access to the CrewAI for pension funds package or SDK you’re using in your environment
• •An LLM provider key configured in environment variables
• •A working virtual environment
• •Basic familiarity with REST APIs and Pydantic models

Install the core dependencies:

pip install fastapi uvicorn crewai pydantic

Set your environment variables:

export OPENAI_API_KEY="your-key"
export CREWAI_API_KEY="your-crewai-key"

Integration Steps

1) Define the agent roles and tasks

Start by modeling the pension-fund workflow as a crew. In production, keep agents narrow and tasks explicit.

from crewai import Agent, Task, Crew, Process

pension_analyst = Agent(
    role="Pension Fund Analyst",
    goal="Analyze pension-related requests and extract compliance-relevant details",
    backstory="You review pension fund documents, member requests, and policy references.",
    verbose=True,
)

compliance_reviewer = Agent(
    role="Compliance Reviewer",
    goal="Validate outputs against pension rules and flag risky statements",
    backstory="You ensure responses are suitable for regulated financial environments.",
    verbose=True,
)

analyze_request = Task(
    description=(
        "Review the user's pension query and produce a structured summary "
        "with risk flags, missing information, and recommended next action."
    ),
    expected_output="A JSON-like summary with analysis and compliance notes.",
    agent=pension_analyst,
)

review_output = Task(
    description="Review the analyst output for compliance issues and approve or reject it.",
    expected_output="A final compliance-reviewed response.",
    agent=compliance_reviewer,
)

2) Build the crew execution function

Wrap crew execution in a service function so FastAPI can call it cleanly. This keeps orchestration out of your route handlers.

from crewai import Crew, Process

def run_pension_crew(user_query: str) -> str:
    crew = Crew(
        agents=[pension_analyst, compliance_reviewer],
        tasks=[analyze_request, review_output],
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff(inputs={"user_query": user_query})
    return str(result)

If your SDK version exposes different kickoff semantics, keep the same pattern: build the crew once per request or reuse it through a service layer if your setup is stateless.

3) Expose the workflow through FastAPI

Now connect the orchestration layer to an API endpoint. Use Pydantic models so request validation happens before agent execution.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Pension AI API")

class PensionQueryRequest(BaseModel):
    query: str

class PensionQueryResponse(BaseModel):
    result: str

@app.post("/pension/analyze", response_model=PensionQueryResponse)
def analyze_pension_query(payload: PensionQueryRequest):
    result = run_pension_crew(payload.query)
    return PensionQueryResponse(result=result)

This is the core integration point. FastAPI receives the request, validates it, calls CrewAI for pension funds, then returns structured output.

4) Add production guardrails around the route

For real systems, don’t expose raw agent output without basic controls. Add timeouts, exception handling, and request logging.

import time
from fastapi import HTTPException

@app.post("/pension/analyze-safe", response_model=PensionQueryResponse)
def analyze_pension_query_safe(payload: PensionQueryRequest):
    start = time.time()

    try:
        result = run_pension_crew(payload.query)
        elapsed = time.time() - start

        if elapsed > 20:
            raise HTTPException(status_code=504, detail="Agent workflow timed out")

        return PensionQueryResponse(result=result)

    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"Agent processing failed: {exc}")

For regulated workloads, you’ll usually add:

• •Authentication middleware
• •Request IDs for traceability
• •Audit logging of prompts and outputs
• •Output filtering before returning results to clients

5) Run the API locally

Start FastAPI with Uvicorn.

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

If this is going into production behind a gateway, remove --reload and configure proper workers at the process manager level.

Testing the Integration

Use curl or Python requests to verify that FastAPI reaches CrewAI correctly.

import requests

response = requests.post(
    "http://127.0.0.1:8000/pension/analyze",
    json={
        "query": "A member wants to know whether their contribution history qualifies them for early retirement benefits."
    },
)

print(response.status_code)
print(response.json())

Expected output will look like this:

200
{
  "result": "..."
}

If you want to test locally from the browser side too:

curl -X POST "http://127.0.0.1:8000/pension/analyze" \
  -H "Content-Type: application/json" \
  -d '{"query":"Check whether this pension withdrawal request has any policy risks."}'

What matters here is not just that you get a response. You want to confirm:

• •The endpoint validates input correctly
• •The crew executes without runtime errors
• •The returned text is consistent enough for downstream systems to consume

Real-World Use Cases

• •

  Member service triage

  • •Route incoming pension questions to an agent crew that classifies intent, checks policy references, and drafts a response for human review.
• •

  Compliance pre-checks

  • •Use one agent to extract claims from member documents and another to verify them against fund rules before escalation.
• •

  Operational anomaly review

  • •Let agents inspect contribution patterns or benefit requests and flag cases that need manual investigation by operations teams.

The pattern here is stable: FastAPI gives you the API boundary, CrewAI gives you coordinated reasoning steps. Put them together behind auth and observability layers, and you have something that can survive real production traffic instead of just demo traffic.

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.