How to Integrate CrewAI for fintech with FastAPI for production AI

By Cyprian AaronsUpdated 2026-04-22
crewai-for-fintechfastapiproduction-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

  1. 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
  1. 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,
)
  1. 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))
  1. 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_threadpool for 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))
  1. 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

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