How to Integrate CrewAI for payments with FastAPI for multi-agent systems

Combining CrewAI for payments with FastAPI gives you a clean way to expose multi-agent payment workflows behind HTTP endpoints. That means you can take agent-driven tasks like invoice validation, fraud checks, payment routing, and reconciliation, then serve them as production APIs your internal systems can call.

Prerequisites

• •Python 3.10+
• •A FastAPI project set up with uvicorn
• •CrewAI installed and configured for your payment workflow
• •Access to your payment provider credentials or sandbox keys
• •Basic understanding of:
  • •FastAPI routers and request models
  • •CrewAI Agent, Task, and Crew objects
  • •Async HTTP APIs and JSON payloads

Install the core packages:

pip install fastapi uvicorn crewai pydantic

Integration Steps

1. •Define the payment workflow as CrewAI agents and tasks

Start by modeling the payment process in CrewAI. For a typical payment flow, split responsibilities across agents instead of putting everything into one prompt.

from crewai import Agent, Task, Crew, Process

payment_validator = Agent(
    role="Payment Validator",
    goal="Validate payment requests for required fields and business rules",
    backstory="You check invoice data, customer references, and amount limits before execution."
)

fraud_checker = Agent(
    role="Fraud Checker",
    goal="Detect suspicious payment patterns before execution",
    backstory="You inspect destination accounts, velocity patterns, and transaction risk."
)

settlement_agent = Agent(
    role="Settlement Agent",
    goal="Prepare the final payment instruction for execution",
    backstory="You format validated payments into a settlement-ready payload."
)

validate_task = Task(
    description="Validate the incoming payment request payload.",
    expected_output="A structured validation result with pass/fail and reasons.",
    agent=payment_validator
)

fraud_task = Task(
    description="Assess fraud risk for the validated payment request.",
    expected_output="A risk score and decision recommendation.",
    agent=fraud_checker
)

settle_task = Task(
    description="Create a settlement instruction if the request is approved.",
    expected_output="A final settlement payload ready for downstream execution.",
    agent=settlement_agent
)

payment_crew = Crew(
    agents=[payment_validator, fraud_checker, settlement_agent],
    tasks=[validate_task, fraud_task, settle_task],
    process=Process.sequential
)

2. •Create a FastAPI request model and endpoint

Use FastAPI to expose the workflow over HTTP. Keep the request schema strict so your agents receive predictable input.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional

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

class PaymentRequest(BaseModel):
    invoice_id: str = Field(..., example="INV-10021")
    customer_id: str = Field(..., example="CUST-2044")
    amount: float = Field(..., gt=0)
    currency: str = Field(default="USD")
    destination_account: str
    reference: Optional[str] = None

@app.post("/payments/process")
async def process_payment(request: PaymentRequest):
    try:
        result = payment_crew.kickoff(inputs=request.model_dump())
        return {
            "status": "accepted",
            "workflow_result": str(result)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

3. •Add real payment execution behind the agent output

CrewAI should decide or prepare the action; your backend should execute the actual transfer using your payment provider SDK or internal service. This keeps control logic separate from side effects.

import os
import requests

PAYMENT_API_URL = os.getenv("PAYMENT_API_URL", "https://sandbox.payments.internal/transfer")
PAYMENT_API_KEY = os.getenv("PAYMENT_API_KEY", "test-key")

def execute_payment(settlement_payload: dict):
    headers = {
        "Authorization": f"Bearer {PAYMENT_API_KEY}",
        "Content-Type": "application/json"
    }

    response = requests.post(
        PAYMENT_API_URL,
        json=settlement_payload,
        headers=headers,
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

Then wire it into your endpoint after CrewAI returns an approved instruction.

@app.post("/payments/submit")
async def submit_payment(request: PaymentRequest):
    crew_result = payment_crew.kickoff(inputs=request.model_dump())

    # In production, parse structured output from your task/agent contract.
    settlement_payload = {
        "invoice_id": request.invoice_id,
        "customer_id": request.customer_id,
        "amount": request.amount,
        "currency": request.currency,
        "destination_account": request.destination_account,
        "reference": request.reference,
        "decision_source": str(crew_result),
    }

    try:
        execution_result = execute_payment(settlement_payload)
        return {
            "status": "submitted",
            "payment_result": execution_result
        }
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Payment execution failed: {e}")

4. •Run the API as a service

Expose the FastAPI app with Uvicorn so other services can call it.

uvicorn main:app --reload --host 0.0.0.0 --port 8000

If you want this in production:

• •Run behind a reverse proxy like Nginx or an API gateway
• •Add auth middleware or JWT validation
• •Store secrets in Vault or environment variables
• •Log both agent decisions and downstream payment responses

5. •Structure outputs for downstream systems

Do not return raw agent text to callers that expect machine-readable results. Define a stable response contract.

from pydantic import BaseModel

class PaymentResponse(BaseModel):
    status: str
    invoice_id: str
    decision: str
    provider_reference: Optional[str] = None

@app.post("/payments/process-safe", response_model=PaymentResponse)
async def process_payment_safe(request: PaymentRequest):
    crew_result = payment_crew.kickoff(inputs=request.model_dump())

    # Replace this with parsed structured output from CrewAI task results.
    decision_text = str(crew_result)

    return PaymentResponse(
        status="processed",
        invoice_id=request.invoice_id,
        decision=decision_text[:200]
    )

Testing the Integration

Use curl or an HTTP client to verify that FastAPI accepts the payload and triggers the CrewAI workflow.

curl -X POST http://localhost:8000/payments/process \
  -H "Content-Type: application/json" \
  -d '{
    "invoice_id": "INV-10021",
    "customer_id": "CUST-2044",
      "amount": 1250.75,
      "currency": "USD",
      "destination_account": "ACC-998877",
      "reference": "Quarterly subscription"
  }'

Expected output:

{
  "status": "accepted",
  "workflow_result": "...CrewAI task output..."
}

If you wired execution too:

{
  "status": "submitted",
  "payment_result": {
     "...provider response..."
  }
}

Real-World Use Cases

• •

  Invoice-to-payment automation

  • •One agent validates invoice metadata.
  • •Another checks fraud risk.
  • •A third prepares settlement instructions for approval or direct execution.
• •

  Payment exception handling

  • •Agents classify failed transfers.
  • •FastAPI exposes retry and escalation endpoints.
  • •Your ops team gets structured decisions instead of unstructured tickets.
• •

  Treasury operations orchestration

  • •Use multi-agent workflows to route payments by currency, region, or business unit.
  • •Expose each workflow through FastAPI so ERP systems can trigger them programmatically.

The main pattern here is simple: let CrewAI reason about the workflow, and let FastAPI serve it as a controlled API surface. Keep business logic in agents where it helps coordination, but keep execution guarded in Python services you own.

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.