How to Integrate CrewAI for banking with FastAPI for multi-agent systems

Combining CrewAI for banking with FastAPI gives you a clean way to expose multi-agent workflows as production APIs. That matters when you need one request to trigger multiple specialized agents — for example, fraud review, KYC checks, and customer support triage — without turning your app into a monolith.

Prerequisites

• •Python 3.10+
• •fastapi
• •uvicorn
• •crewai
• •pydantic
• •Access to your CrewAI banking project, agents, and tasks
• •A configured LLM provider key in your environment, such as OPENAI_API_KEY
• •Basic familiarity with REST APIs and async web services

Install the dependencies:

pip install fastapi uvicorn crewai pydantic

Integration Steps

1. •Define your banking agents and tasks in CrewAI

Start by modeling the banking workflow as agents. Keep responsibilities narrow: one agent for compliance, one for fraud analysis, one for customer-facing summaries.

from crewai import Agent, Task, Crew, Process

fraud_agent = Agent(
    role="Fraud Analyst",
    goal="Detect suspicious transaction patterns",
    backstory="You analyze bank transactions for anomalies and risk indicators.",
    verbose=True,
)

compliance_agent = Agent(
    role="Compliance Officer",
    goal="Check KYC and AML policy alignment",
    backstory="You validate banking activity against regulatory requirements.",
    verbose=True,
)

fraud_task = Task(
    description="Review the transaction batch and identify suspicious items.",
    expected_output="A list of suspicious transactions with reasons.",
    agent=fraud_agent,
)

compliance_task = Task(
    description="Assess whether the case violates KYC or AML controls.",
    expected_output="A compliance assessment with risk level.",
    agent=compliance_agent,
)

crew = Crew(
    agents=[fraud_agent, compliance_agent],
    tasks=[fraud_task, compliance_task],
    process=Process.sequential,
)

2. •Wrap the CrewAI execution in a service layer

Do not call the crew directly from your route handler. Put it behind a function so you can add retries, logging, audit trails, and queueing later.

from typing import Dict, Any

def run_banking_crew(case_data: Dict[str, Any]) -> str:
    # In a real system you'd inject case_data into task descriptions or use structured inputs.
    result = crew.kickoff()
    return str(result)

If you need dynamic per-request context, build the tasks inside the function instead of reusing static task objects.

from crewai import Agent, Task, Crew, Process

def build_crew(case_id: str, transaction_amount: float) -> Crew:
    fraud_agent = Agent(
        role="Fraud Analyst",
        goal="Detect suspicious transaction patterns",
        backstory="You analyze bank transactions for anomalies and risk indicators.",
        verbose=True,
    )

    fraud_task = Task(
        description=f"Review case {case_id} with transaction amount {transaction_amount} and flag suspicious behavior.",
        expected_output="Fraud assessment with rationale.",
        agent=fraud_agent,
    )

    return Crew(
        agents=[fraud_agent],
        tasks=[fraud_task],
        process=Process.sequential,
    )

3. •Expose the workflow through FastAPI

Create an API endpoint that accepts banking case data and returns the CrewAI result. Use Pydantic models so your request/response contracts stay explicit.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Banking Multi-Agent API")

class BankingCaseRequest(BaseModel):
    case_id: str
    customer_id: str
    transaction_amount: float
    currency: str = "USD"

class BankingCaseResponse(BaseModel):
    case_id: str
    result: str

@app.post("/banking/case", response_model=BankingCaseResponse)
async def analyze_case(payload: BankingCaseRequest):
    crew_instance = build_crew(payload.case_id, payload.transaction_amount)
    result = crew_instance.kickoff()
    return BankingCaseResponse(case_id=payload.case_id, result=str(result))

This gives you a stable HTTP interface while keeping orchestration logic inside CrewAI.

4. •Add async-safe execution for long-running agent work

CrewAI runs are usually blocking. In FastAPI, push them into a threadpool so your event loop stays responsive.

from fastapi.concurrency import run_in_threadpool

@app.post("/banking/case/async", response_model=BankingCaseResponse)
async def analyze_case_async(payload: BankingCaseRequest):
    crew_instance = build_crew(payload.case_id, payload.transaction_amount)
    result = await run_in_threadpool(crew_instance.kickoff)
    return BankingCaseResponse(case_id=payload.case_id, result=str(result))

For production systems with higher volume, this route should eventually hand off to a job queue like Celery or RQ. The pattern stays the same: API accepts request, worker executes crew kickoff.

5. •Run the API locally

Start FastAPI with Uvicorn:

uvicorn main:app --reload --port 8000

Your service is now exposing a multi-agent banking workflow at /banking/case.

Testing the Integration

Use curl or any HTTP client to verify the endpoint returns a CrewAI-driven response.

curl -X POST "http://127.0.0.1:8000/banking/case" \
  -H "Content-Type: application/json" \
  -d '{
    "case_id": "CASE-10021",
    "customer_id": "CUST-7781",
    "transaction_amount": 12500.50,
    "currency": "USD"
  }'

Expected output:

{
  "case_id": "CASE-10021",
  "result": "Fraud assessment with rationale..."
}

If you want deeper validation during development, hit the interactive docs at:

• •http://127.0.0.1:8000/docs
• •http://127.0.0.1:8000/redoc

Real-World Use Cases

• •

  Fraud triage API
  Accepts transaction payloads and routes them through fraud detection and compliance agents before returning a risk summary.

• •

  KYC onboarding assistant
  Uses one agent to extract customer data from documents and another to validate policy completeness before account creation.

• •

  Dispute resolution workflow
  Takes chargeback or complaint cases and uses separate agents for evidence review, policy lookup, and customer response drafting.

The main pattern here is simple: let CrewAI handle orchestration between specialized banking agents, and let FastAPI handle transport, validation, and exposure to downstream systems. That separation keeps your system testable, auditable, and easier to scale when the number of workflows grows.

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.