How to Integrate CrewAI for banking with FastAPI for startups
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
.envfile 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
- •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.
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