How to Integrate CrewAI for retail banking with FastAPI for startups
Combining CrewAI for retail banking with FastAPI gives you a clean way to expose banking agents as production HTTP services. For startups, that means you can turn internal workflows like customer support triage, loan pre-qualification, fraud review, and account inquiry into API endpoints your frontend or ops tools can call.
Prerequisites
- •Python 3.10+
- •
fastapi - •
uvicorn - •
crewai - •
pydantic - •Access to the retail banking tools or agents you want CrewAI to orchestrate
- •A local
.envfile or secrets manager for API keys and credentials - •Basic understanding of FastAPI routing and async request handling
Install the dependencies:
pip install fastapi uvicorn crewai pydantic python-dotenv
Integration Steps
- •Define your banking agent and task in CrewAI
Start by creating a CrewAI agent that handles a narrow banking workflow. Keep the scope tight: one agent, one job, one output format.
from crewai import Agent, Task, Crew
banking_agent = Agent(
role="Retail Banking Assistant",
goal="Analyze customer requests and return compliant banking guidance",
backstory="You assist with retail banking operations such as account queries, loan pre-checks, and fraud triage.",
verbose=True,
allow_delegation=False,
)
customer_task = Task(
description=(
"Review the customer's request: {request}. "
"Return a concise JSON-style response with fields: intent, risk_level, next_action."
),
expected_output="A structured response for downstream API consumption.",
agent=banking_agent,
)
crew = Crew(
agents=[banking_agent],
tasks=[customer_task],
)
- •Wrap the CrewAI execution in a service function
Your FastAPI app should not know CrewAI internals. Put orchestration behind a service layer so you can swap prompts, tools, or models without touching routes.
from crewai import Crew
def run_banking_workflow(request_text: str) -> str:
result = crew.kickoff(inputs={"request": request_text})
return str(result)
If you need stricter output handling, parse the result into a Pydantic model before returning it to the API layer.
from pydantic import BaseModel
class BankingResponse(BaseModel):
intent: str
risk_level: str
next_action: str
- •Expose the workflow through FastAPI
Now create an endpoint that accepts user input and returns the CrewAI output. Use Pydantic models for request validation so bad payloads fail fast.
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="Retail Banking Agent API")
class BankingRequest(BaseModel):
request: str
@app.post("/banking/analyze")
def analyze_banking_request(payload: BankingRequest):
result = run_banking_workflow(payload.request)
return {"result": result}
If your workflow is slow or calls external systems, make the route async-friendly and move execution to a background worker later. For startup MVPs, this sync version is fine as long as latency stays predictable.
- •Add structured responses and guardrails
For banking use cases, free-form text is not enough. Force the agent output into a predictable schema and reject malformed results before they hit clients.
import json
from fastapi import HTTPException
@app.post("/banking/analyze-structured")
def analyze_banking_request_structured(payload: BankingRequest):
raw_result = run_banking_workflow(payload.request)
try:
parsed = json.loads(raw_result)
validated = BankingResponse(**parsed)
return validated.model_dump()
except Exception as exc:
raise HTTPException(status_code=500, detail=f"Invalid agent output: {exc}")
This pattern matters because downstream systems like CRM tools, underwriting queues, or fraud dashboards need stable fields, not prose.
- •Run the app and wire environment config
Keep model keys and any tool credentials out of code. Use environment variables and load them before app startup.
from dotenv import load_dotenv
load_dotenv()
Start the server:
uvicorn main:app --reload --port 8000
Testing the Integration
Use curl or Python requests to verify that FastAPI receives input and CrewAI returns an answer through your endpoint.
import requests
response = requests.post(
"http://127.0.0.1:8000/banking/analyze-structured",
json={"request": "Customer wants to know if they qualify for a personal loan after two missed payments last year."}
)
print(response.status_code)
print(response.json())
Expected output:
{
"intent": "loan_prequalification",
"risk_level": "medium",
"next_action": "collect_income_docs_and_run_manual_review"
}
If you get a 500 error, check these first:
- •The agent output is not valid JSON
- •Your Pydantic schema does not match the returned fields
- •The CrewAI task prompt is too open-ended
- •Your FastAPI route is returning raw text instead of structured data
Real-World Use Cases
- •
Loan pre-screening API
Accept applicant questions from your web app, have CrewAI classify eligibility signals, then return next steps for underwriting. - •
Retail support triage
Route account access issues, card disputes, and fee explanations into different queues based on agent classification. - •
Fraud review assistant
Expose an endpoint that summarizes suspicious activity patterns and recommends whether to escalate to manual review.
The main pattern here is simple: let CrewAI handle reasoning and workflow orchestration, let FastAPI handle transport and validation. That split keeps your startup stack maintainable when traffic grows and compliance starts asking hard questions.
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