How to Integrate CrewAI for banking with FastAPI for production AI
Combining CrewAI for banking with FastAPI gives you a clean way to expose multi-agent banking workflows behind a production HTTP API. That matters when you need an agent system that can triage customer requests, route compliance checks, and generate banker-ready summaries without wiring everything into a monolith.
Prerequisites
- •Python 3.10+
- •
crewaiinstalled and configured for your banking use case - •
fastapianduvicorninstalled - •A working LLM provider configured in environment variables
- •Access to any internal banking tools your agents will call, such as:
- •KYC lookup service
- •transaction search API
- •policy/compliance rules engine
- •Basic understanding of:
- •CrewAI
Agent,Task, andCrew - •FastAPI path operations and Pydantic models
- •CrewAI
Install the core packages:
pip install crewai fastapi uvicorn pydantic
Integration Steps
- •Define the banking agent and task
Start by modeling the work the agent should do. In banking, keep the agent narrowly scoped: one for customer support triage, one for compliance review, one for account summary generation.
from crewai import Agent, Task, Crew, Process
banking_analyst = Agent(
role="Banking Operations Analyst",
goal="Analyze customer banking requests and produce compliant summaries",
backstory=(
"You work in a regulated banking environment. "
"You must be precise, concise, and compliant."
),
verbose=True,
allow_delegation=False,
)
review_task = Task(
description=(
"Review the customer request: {request}. "
"Return a structured summary, risk flags, and next action."
),
expected_output="A JSON-like summary with risk_flags and next_action",
agent=banking_analyst,
)
- •Wrap the CrewAI workflow in a service function
Keep CrewAI orchestration out of your route handlers. Put it behind a plain Python function so you can test it independently and reuse it from jobs or workers later.
def run_banking_workflow(request_text: str) -> str:
crew = Crew(
agents=[banking_analyst],
tasks=[review_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff(inputs={"request": request_text})
return str(result)
This is the key integration point. Crew.kickoff() is where your agent graph actually executes, and FastAPI will just call this function.
- •Expose the workflow through FastAPI
Now create an API boundary around the workflow. Use Pydantic models so your request and response shapes stay explicit.
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="Banking AI Agent API")
class BankingRequest(BaseModel):
request_text: str
class BankingResponse(BaseModel):
output: str
@app.post("/banking/analyze", response_model=BankingResponse)
def analyze_banking_request(payload: BankingRequest):
output = run_banking_workflow(payload.request_text)
return BankingResponse(output=output)
This gives you an HTTP endpoint that can sit behind an API gateway, auth layer, rate limiter, or internal service mesh.
- •Add production-friendly error handling
In production AI systems, failures are normal: model timeouts, malformed inputs, downstream tool issues. Catch them at the boundary and return predictable responses.
from fastapi import HTTPException
@app.post("/banking/analyze-safe", response_model=BankingResponse)
def analyze_banking_request_safe(payload: BankingRequest):
try:
output = run_banking_workflow(payload.request_text)
return BankingResponse(output=output)
except Exception as exc:
raise HTTPException(
status_code=500,
detail=f"Banking workflow failed: {str(exc)}"
)
If you need stricter control, add:
- •request validation rules
- •timeout handling
- •structured logging with correlation IDs
- •retries only around safe idempotent calls
- •Run the API and wire it into your internal consumers
Start the server with Uvicorn:
uvicorn main:app --reload --host 0.0.0.0 --port 8000
Then call it from another service, backend job, or frontend proxy using standard HTTP.
Testing the Integration
Use FastAPI’s test client to verify the endpoint returns a valid response.
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_banking_analysis():
response = client.post(
"/banking/analyze",
json={
"request_text": "Customer reports an unauthorized card charge of $240 on 2026-04-20."
},
)
assert response.status_code == 200
assert "output" in response.json()
print(test_banking_analysis())
Expected output:
None
That None is normal here because the test uses assertions instead of returning data. If you want to inspect the payload directly:
response = client.post(
"/banking/analyze",
json={"request_text": "Customer requests a statement for mortgage underwriting."},
)
print(response.status_code)
print(response.json())
Expected output shape:
200
{'output': '...agent-generated summary...'}
Real-World Use Cases
- •
Customer request triage
- •Classify inbound messages into fraud, disputes, account servicing, lending support, or complaints.
- •Route each case to the right human queue with a structured summary.
- •
Compliance review assistant
- •Summarize customer interactions and flag risky language before escalation.
- •Generate audit-friendly notes for internal reviewers.
- •
Operations copilot
- •Turn raw banking tickets into concise action items.
- •Produce standardized outputs for CRM updates, case management systems, or analyst dashboards.
The pattern is simple: keep CrewAI focused on reasoning and task execution, then let FastAPI handle transport, validation, and service boundaries. That separation is what makes this workable in production instead of just a notebook demo.
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