How to Integrate CrewAI for fintech with FastAPI for production AI
Combining CrewAI for fintech with FastAPI gives you a clean way to expose multi-agent workflows as production APIs. That matters when you need an AI system that can triage transactions, summarize risk, or route customer requests through specialized agents without turning your app into a pile of scripts.
FastAPI handles the HTTP layer, validation, and async request lifecycle. CrewAI for fintech handles the agent orchestration, tool usage, and task execution behind the endpoint.
Prerequisites
- •Python 3.10+
- •A virtual environment set up
- •
fastapi - •
uvicorn - •
crewai - •Your CrewAI for fintech package or SDK configured with valid credentials
- •Access to the model provider used by your CrewAI agents
- •Basic familiarity with Pydantic models and REST APIs
Install the core packages:
pip install fastapi uvicorn crewai pydantic
If your fintech setup uses external tools like KYC, transaction monitoring, or document retrieval, make sure those APIs are reachable from your runtime.
Integration Steps
- •
Define the request and response contracts
Start with explicit schemas. In production AI, your API boundary should be stricter than your agent boundary.
from pydantic import BaseModel, Field
class FintechRequest(BaseModel):
customer_id: str = Field(..., examples=["cust_123"])
query: str = Field(..., examples=["Review this transaction for fraud risk"])
class FintechResponse(BaseModel):
result: str
status: str
- •
Create your CrewAI agents and tasks
Use a finance-focused analyst agent and a compliance reviewer agent. The exact role names can vary by your CrewAI setup, but the pattern stays the same: one agent gathers context, another produces a controlled output.
from crewai import Agent, Task, Crew, Process
risk_analyst = Agent(
role="Fintech Risk Analyst",
goal="Analyze customer requests and identify financial risk signals",
backstory="You review fintech workflows for fraud, AML, and unusual activity.",
verbose=True,
)
compliance_reviewer = Agent(
role="Compliance Reviewer",
goal="Produce compliant summaries suitable for internal review",
backstory="You rewrite findings in a concise form for ops teams.",
verbose=True,
)
analysis_task = Task(
description="Analyze the request: {query}. Return risk indicators and recommended next action.",
expected_output="A short risk assessment with actionable next steps.",
agent=risk_analyst,
)
review_task = Task(
description="Rewrite the analysis into a compliance-friendly summary.",
expected_output="A concise compliance summary.",
agent=compliance_reviewer,
)
- •
Wrap the crew execution inside a FastAPI endpoint
This is where FastAPI becomes the control plane. Keep the endpoint thin and push orchestration into a service function so you can test it independently.
from fastapi import FastAPI, HTTPException
app = FastAPI(title="CrewAI Fintech API")
def run_fintech_crew(query: str) -> str:
crew = Crew(
agents=[risk_analyst, compliance_reviewer],
tasks=[analysis_task, review_task],
process=Process.sequential,
verbose=True,
)
result = crew.kickoff(inputs={"query": query})
return str(result)
@app.post("/fintech/analyze", response_model=FintechResponse)
async def analyze_fintech_request(payload: FintechRequest):
try:
result = run_fintech_crew(payload.query)
return FintechResponse(result=result, status="ok")
except Exception as exc:
raise HTTPException(status_code=500, detail=str(exc))
- •
Add production-safe async handling
Crew executions can take time. If you call them directly in an async route and they block CPU or network I/O, your API latency will suffer. Use
run_in_threadpoolfor sync crew execution.
from fastapi.concurrency import run_in_threadpool
@app.post("/fintech/analyze-safe", response_model=FintechResponse)
async def analyze_fintech_request_safe(payload: FintechRequest):
try:
result = await run_in_threadpool(run_fintech_crew, payload.query)
return FintechResponse(result=result, status="ok")
except Exception as exc:
raise HTTPException(status_code=500, detail=str(exc))
- •
Run the app and expose it behind a proper server
Use Uvicorn locally and put it behind Gunicorn/Uvicorn workers in production if you need process isolation.
uvicorn main:app --reload --host 0.0.0.0 --port 8000
For production deployments:
gunicorn -k uvicorn.workers.UvicornWorker main:app -w 2 -b 0.0.0.0:8000
Testing the Integration
Send a request to confirm that FastAPI accepts input and CrewAI returns an orchestrated result.
import requests
payload = {
"customer_id": "cust_123",
"query": "Review this card payment of $4,800 to an overseas merchant"
}
response = requests.post("http://localhost:8000/fintech/analyze-safe", json=payload)
print(response.status_code)
print(response.json())
Expected output:
200
{
"result": "Compliance summary: The transaction is high-value and cross-border. Review merchant history, device fingerprinting, account age, and recent velocity patterns before approving.",
"status": "ok"
}
If you get a 500 error, check three things first:
- •Your model credentials are loaded correctly
- •Your CrewAI agents have valid task definitions
- •The endpoint is not blocking on external tools or network calls without timeout handling
Real-World Use Cases
- •
Fraud triage API
Accept transaction metadata via FastAPI and use CrewAI agents to classify risk levels, explain anomalies, and recommend escalation paths. - •
KYC document review
Upload customer documents through an API endpoint and let one agent extract entities while another checks policy compliance. - •
Customer support automation
Route banking questions to specialized agents that answer product queries, summarize account issues, or prepare handoff notes for human ops teams.
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