How to Integrate CrewAI for retail banking with FastAPI for production AI

Combining CrewAI for retail banking with FastAPI gives you a clean way to expose multi-agent banking workflows as production HTTP services. That means you can take tasks like customer support triage, transaction dispute analysis, KYC document review, or loan pre-screening and serve them behind a proper API that your web app, internal tools, or orchestration layer can call.

Prerequisites

• •Python 3.10+
• •A virtual environment tool like venv, poetry, or uv
• •FastAPI and an ASGI server:
  • •fastapi
  • •uvicorn[standard]
• •CrewAI installed in your project
• •An LLM provider configured for CrewAI, such as OpenAI or another supported provider
• •Environment variables set for credentials:
  • •OPENAI_API_KEY or the provider key you use
• •A clear retail banking workflow to automate:
  • •example: fraud triage, account opening review, complaint routing, loan eligibility checks

Integration Steps

1. •

   Install the dependencies and create a service boundary.

   Keep CrewAI inside a service layer. FastAPI should only handle request/response concerns, while CrewAI handles agent orchestration and task execution.

   pip install fastapi uvicorn crewai crewai-tools pydantic
   

2. •

   Define your retail banking agents and task graph.

   In CrewAI, you usually define Agent, Task, and Crew. For retail banking, keep the agents narrow: one for compliance review, one for customer context summarization, and one for resolution drafting.

   # banking_crew.py
   from crewai import Agent, Task, Crew, Process
   
   compliance_agent = Agent(
       role="Retail Banking Compliance Analyst",
       goal="Review customer cases for policy and regulatory risk",
       backstory="You analyze retail banking cases with focus on KYC, AML, and complaint handling.",
       verbose=True,
       allow_delegation=False,
   )
   
   support_agent = Agent(
       role="Customer Case Analyst",
       goal="Summarize customer issues and identify next best action",
       backstory="You work on retail banking support cases and turn raw notes into actionable summaries.",
       verbose=True,
       allow_delegation=False,
   )
   
   resolution_task = Task(
       description=(
           "Analyze this retail banking case: {case_text}. "
           "Return a short summary, risk flags, and recommended next action."
       ),
       expected_output="JSON with summary, risk_flags, recommendation",
       agent=support_agent,
   )
   
   compliance_task = Task(
       description=(
           "Review the same case for compliance concerns: {case_text}. "
           "Identify any KYC/AML or policy issues."
       ),
       expected_output="JSON with compliance_findings",
       agent=compliance_agent,
   )
   
   crew = Crew(
       agents=[support_agent, compliance_agent],
       tasks=[resolution_task, compliance_task],
       process=Process.sequential,
       verbose=True,
   )
   

3. •

   Wrap the CrewAI run inside a FastAPI endpoint.

   The key pattern is simple: accept input with Pydantic, call crew.kickoff(inputs=...), return structured output. For production systems, keep the endpoint synchronous unless you have a strong reason to introduce async complexity around model calls.

   # main.py
   from fastapi import FastAPI, HTTPException
   from pydantic import BaseModel
   from banking_crew import crew
   
   app = FastAPI(title="Retail Banking AI Service")
   
   class CaseRequest(BaseModel):
       case_id: str
       case_text: str
   
   class CaseResponse(BaseModel):
       case_id: str
       result: str
   
   @app.post("/cases/analyze", response_model=CaseResponse)
   def analyze_case(payload: CaseRequest):
       try:
           result = crew.kickoff(
               inputs={
                   "case_text": payload.case_text
               }
           )
           return CaseResponse(case_id=payload.case_id, result=str(result))
       except Exception as e:
           raise HTTPException(status_code=500, detail=str(e))
   

4. •

   Add validation and predictable output formatting.

   Production AI breaks when outputs are free-form. Force the agent to emit JSON-like structure in the task prompt and validate it before returning to clients.

   # schemas.py
   from pydantic import BaseModel
   from typing import List
   
   class BankingAnalysis(BaseModel):
       summary: str
       risk_flags: List[str]
       recommendation: str
       compliance_findings: List[str]
   
   
   # parser_example.py
   import json
   
   def parse_crew_output(raw_output: str):
       # If your crew returns markdown or text around JSON,
       # normalize it here before parsing.
       cleaned = raw_output.strip()
       return json.loads(cleaned)
   

5. •

   Run the service and wire it into your application stack.

   Start FastAPI with Uvicorn and point your internal systems at the endpoint. This is where you add auth middleware, rate limiting, request logging, and observability hooks.

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

Testing the Integration

Use FastAPI’s TestClient to verify that the endpoint triggers the CrewAI workflow correctly.

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_analyze_case():
    response = client.post(
        "/cases/analyze",
        json={
            "case_id": "CASE-1001",
            "case_text": (
                "Customer reports an unauthorized card transaction of $240 "
                "and requests immediate reversal."
            )
        },
    )
    assert response.status_code == 200
    data = response.json()
    assert data["case_id"] == "CASE-1001"
    assert "result" in data

if __name__ == "__main__":
    print(client.post(
        "/cases/analyze",
        json={
            "case_id": "CASE-1001",
            "case_text": "Customer reports an unauthorized card transaction of $240."
        },
    ).json())

Expected output:

{
  "case_id": "CASE-1001",
  "result": "...CrewAI analysis output..."
}

If you want stricter verification in CI:

• •mock the LLM provider behind CrewAI
• •assert that crew.kickoff() is called with the right inputs
• •validate that malformed outputs fail fast before reaching clients

Real-World Use Cases

• •

  Dispute triage API
  Classify card disputes by urgency, extract key facts from customer messages, and route cases to chargeback ops or fraud review.

• •

  Account opening review service
  Use one agent to summarize application data and another to flag missing KYC fields or suspicious patterns before onboarding continues.

• •

  Complaint resolution assistant
  Turn inbound complaints into structured case notes with recommended actions for contact center teams and relationship managers.

The production pattern here is straightforward: CrewAI owns reasoning flow, FastAPI owns service delivery. Keep them separated at the code level so you can scale agents independently without turning your API layer into a prompt script.

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.