How to Integrate CrewAI for fintech with FastAPI for AI agents

Combining CrewAI for fintech with FastAPI gives you a clean way to expose multi-agent workflows as production APIs. That matters when you need AI agents to handle things like transaction analysis, KYC review, fraud triage, or customer support without wiring everything into a monolith.

FastAPI handles request validation, routing, and async I/O. CrewAI for fintech handles the agent orchestration layer, so your API can trigger specialized agents and return structured results instead of raw model output.

Prerequisites

• •Python 3.10+
• •fastapi
• •uvicorn
• •crewai
• •pydantic
• •An LLM provider configured for CrewAI
• •API keys set in environment variables
• •Basic familiarity with REST endpoints and async Python

Install the dependencies:

pip install fastapi uvicorn crewai pydantic

Set your environment variables:

export OPENAI_API_KEY="your-key"

Integration Steps

1. •Create your FastAPI app and request schema

Start by defining the payload your fintech agent system will accept. Keep the contract strict so downstream agent execution is predictable.

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="Fintech Agent API")

class RiskReviewRequest(BaseModel):
    customer_id: str = Field(..., examples=["cus_12345"])
    transaction_amount: float = Field(..., gt=0)
    merchant_category: str = Field(..., examples=["electronics"])
    country: str = Field(..., examples=["KE"])

2. •Define CrewAI agents and tasks for the fintech workflow

Use separate agents for separate responsibilities. In fintech systems, that usually means one agent gathers context, another evaluates risk, and a final agent summarizes the decision.

from crewai import Agent, Task, Crew, Process

risk_analyst = Agent(
    role="Fintech Risk Analyst",
    goal="Assess transaction risk using available customer and transaction context",
    backstory="You review payment activity for fraud signals and compliance concerns.",
    verbose=True,
)

compliance_reviewer = Agent(
    role="Compliance Reviewer",
    goal="Check whether the transaction raises AML or policy concerns",
    backstory="You focus on suspicious patterns and regulatory red flags.",
    verbose=True,
)

risk_task = Task(
    description=(
        "Review this transaction: customer_id={customer_id}, amount={transaction_amount}, "
        "merchant_category={merchant_category}, country={country}. "
        "Return a concise risk assessment with a severity label."
    ),
    expected_output="A JSON-like summary with risk_level, reasons, and recommended_action.",
    agent=risk_analyst,
)

compliance_task = Task(
    description=(
        "Using the same transaction details, assess compliance concerns and note any escalation triggers."
    ),
    expected_output="A JSON-like summary with compliance_notes and escalation_needed.",
    agent=compliance_reviewer,
)

crew = Crew(
    agents=[risk_analyst, compliance_reviewer],
    tasks=[risk_task, compliance_task],
    process=Process.sequential,
)

3. •Expose the CrewAI workflow through a FastAPI endpoint

This is the bridge. The endpoint receives data from your client or internal system, passes it into CrewAI via kickoff, then returns the result.

from fastapi import HTTPException

@app.post("/review-transaction")
async def review_transaction(payload: RiskReviewRequest):
    try:
        result = crew.kickoff(inputs=payload.model_dump())
        return {
            "customer_id": payload.customer_id,
            "status": "completed",
            "agent_result": str(result),
        }
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc))

4. •Run the app and make it production-friendly

Use Uvicorn locally first. For real deployments, put this behind a reverse proxy and add auth before exposing it to internal teams.

# save as main.py

# then run:
# uvicorn main:app --reload --host 0.0.0.0 --port 8000

If you want better throughput under load, keep FastAPI async at the edge and move long-running CrewAI calls to a background worker or job queue. That avoids tying up request threads while agents think.

5. •Return structured output instead of string blobs

In production finance systems, plain text is not enough. Wrap the agent response in a response model so consumers can parse it reliably.

from pydantic import BaseModel

class RiskReviewResponse(BaseModel):
    customer_id: str
    status: str
    agent_result: str

@app.post("/review-transaction", response_model=RiskReviewResponse)
async def review_transaction(payload: RiskReviewRequest):
    result = crew.kickoff(inputs=payload.model_dump())
    return RiskReviewResponse(
        customer_id=payload.customer_id,
        status="completed",
        agent_result=str(result),
    )

Testing the Integration

Use curl or any HTTP client to verify the endpoint works end to end.

curl -X POST "http://localhost:8000/review-transaction" \
  -H "Content-Type: application/json" \
  -d '{
    "customer_id": "cus_12345",
    "transaction_amount": 2500,
    "merchant_category": "electronics",
    "country": "KE"
  }'

Expected output:

{
  "customer_id": "cus_12345",
  "status": "completed",
  "agent_result": "..."
}

If you want a quick programmatic test:

import requests

response = requests.post(
    "http://localhost:8000/review-transaction",
    json={
        "customer_id": "cus_12345",
        "transaction_amount": 2500,
        "merchant_category": "electronics",
        "country": "KE"
    },
)

print(response.status_code)
print(response.json())

Real-World Use Cases

• •

  Fraud triage API
  Trigger multiple agents to score transactions, explain anomalies, and recommend escalation to human analysts.

• •

  KYC document review service
  Accept uploaded metadata from your onboarding pipeline and have agents validate completeness, detect mismatches, and summarize missing fields.

• •

  Customer support for banking ops
  Route account questions into specialized agents that answer balance disputes, card issues, or payment trace requests through one FastAPI endpoint.

Practical Notes

Keep your prompts deterministic and your outputs structured. If you let multiple agents free-form their responses without schema discipline, your API will become hard to test and harder to trust.

For fintech workloads, also add:

• •request authentication
• •audit logging
• •timeout handling
• •retry policies for LLM calls
• •human approval gates for high-risk decisions

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.