How to Integrate CrewAI for banking with FastAPI for startups

By Cyprian AaronsUpdated 2026-04-22
crewai-for-bankingfastapistartups

CrewAI for banking gives you the orchestration layer for multi-agent financial workflows. FastAPI gives you the HTTP surface to expose those workflows to your product, internal tools, or partner systems.

Put them together and you can build bank-grade AI services like KYC triage, loan document review, transaction dispute handling, and compliance-first customer support behind a clean API.

Prerequisites

  • Python 3.10+
  • fastapi
  • uvicorn
  • crewai
  • Access to your banking-specific CrewAI package or extension
  • An LLM provider key configured in environment variables
  • Basic understanding of async Python and REST APIs
  • A local .env file for secrets

Install the core dependencies:

pip install fastapi uvicorn crewai python-dotenv pydantic

Set your environment variables:

export OPENAI_API_KEY="your-key"
export BANKING_API_KEY="your-banking-key"

Integration Steps

1) Define your banking agents and tasks

Start by modeling the workflow as agents. In banking, keep roles narrow: one agent for document extraction, one for risk review, one for compliance validation.

from crewai import Agent, Task, Crew, Process

document_agent = Agent(
    role="Banking Document Analyst",
    goal="Extract key fields from banking documents accurately",
    backstory="Specializes in KYC, account opening, and loan application documents.",
    verbose=True,
)

risk_agent = Agent(
    role="Risk Reviewer",
    goal="Assess financial risk based on extracted data",
    backstory="Evaluates anomalies, missing fields, and policy violations.",
    verbose=True,
)

compliance_task = Task(
    description="Review customer onboarding data for missing KYC fields and risk flags.",
    expected_output="A structured compliance summary with pass/fail status and reasons.",
    agent=risk_agent,
)

analysis_task = Task(
    description="Extract applicant name, address, income, and ID number from submitted documents.",
    expected_output="Structured JSON-like extraction output.",
    agent=document_agent,
)

crew = Crew(
    agents=[document_agent, risk_agent],
    tasks=[analysis_task, compliance_task],
    process=Process.sequential,
)

2) Wrap the CrewAI workflow in a service function

Keep the orchestration out of your route handlers. Your FastAPI endpoints should call a small service function that runs the crew and returns normalized data.

from typing import Any

def run_banking_review(payload: dict[str, Any]) -> str:
    input_text = f"""
    Customer name: {payload.get('name')}
    Document text: {payload.get('document_text')}
    Product type: {payload.get('product_type')}
    """
    
    result = crew.kickoff(inputs={"input_text": input_text})
    return str(result)

If you need structured output in production, define a response schema and post-process the crew result before returning it.

3) Expose the workflow through FastAPI

Now create a FastAPI app with a request model and a POST endpoint. This is where startup teams usually get value fast: one endpoint powers web apps, admin tools, or webhook consumers.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Banking AI Orchestrator")

class BankingReviewRequest(BaseModel):
    name: str
    document_text: str
    product_type: str

class BankingReviewResponse(BaseModel):
    status: str
    result: str

@app.post("/banking/review", response_model=BankingReviewResponse)
async def banking_review(request: BankingReviewRequest):
    result = run_banking_review(request.model_dump())
    
    return BankingReviewResponse(
        status="completed",
        result=result,
    )

For startup systems that need low latency under load, move heavy CrewAI execution into background jobs or a task queue later. Start simple first.

4) Add basic validation and error handling

Banking workflows fail when inputs are incomplete. Validate required fields before calling the crew so you don’t waste tokens on bad requests.

from fastapi import HTTPException

@app.post("/banking/review/validated", response_model=BankingReviewResponse)
async def banking_review_validated(request: BankingReviewRequest):
    if len(request.document_text.strip()) < 50:
        raise HTTPException(
            status_code=400,
            detail="document_text is too short for reliable analysis",
        )

    try:
        result = run_banking_review(request.model_dump())
        return BankingReviewResponse(status="completed", result=result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Workflow failed: {str(e)}")

This pattern keeps API behavior predictable and makes debugging much easier when agents return malformed output.

5) Run FastAPI and connect it to your client app

Start the server with Uvicorn:

uvicorn main:app --reload --port 8000

Then call it from another Python service or frontend backend:

import requests

payload = {
    "name": "Jane Doe",
    "document_text": "Jane Doe lives at 12 River Road. ID number AB1234567. Monthly income is $8,000.",
    "product_type": "personal_loan",
}

response = requests.post("http://localhost:8000/banking/review", json=payload)
print(response.status_code)
print(response.json())

Testing the Integration

Use FastAPI’s test client to verify the endpoint responds correctly without running a full server.

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_banking_review():
    payload = {
        "name": "Jane Doe",
        "document_text": "Jane Doe lives at 12 River Road. ID number AB1234567. Monthly income is $8,000.",
        "product_type": "personal_loan",
    }

    response = client.post("/banking/review/validated", json=payload)

    assert response.status_code == 200
    assert response.json()["status"] == "completed"

test_banking_review()
print("Integration test passed")

Expected output:

Integration test passed

If you want deeper verification, mock crew.kickoff() and assert that the endpoint returns your normalized response shape.

Real-World Use Cases

  • KYC onboarding assistant
    Use CrewAI agents to extract identity details, flag missing fields, and expose the workflow through a FastAPI endpoint for your onboarding portal.

  • Loan pre-screening API
    Accept applicant data via FastAPI, run CrewAI-based document review and risk checks, then return an approval recommendation to your underwriting system.

  • Dispute intake triage
    Let customers submit chargeback or transaction dispute details through an API while agents classify severity, extract evidence requirements, and route cases downstream.


Keep learning

By Cyprian Aarons, AI Consultant at Topiax.

Want the complete 8-step roadmap?

Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.

Get the Starter Kit

Related Guides