How to Integrate CrewAI for payments with FastAPI for production AI

Combining CrewAI for payments with FastAPI gives you a clean pattern for exposing payment-aware AI workflows over HTTP. The practical use case is simple: let an agent decide what to do, then trigger a payment flow through an API endpoint that your app, ops team, or downstream systems can call safely.

Prerequisites

• •Python 3.10+
• •fastapi
• •uvicorn
• •crewai
• •A CrewAI account/project configured for payments
• •API keys stored as environment variables
• •Basic familiarity with async Python and REST APIs

Install the dependencies:

pip install fastapi uvicorn crewai

Set your environment variables:

export CREWAI_API_KEY="your_crewai_api_key"
export CREWAI_PROJECT_ID="your_project_id"

Integration Steps

1. •Create a FastAPI app and initialize your payment client

Start by wiring FastAPI and the CrewAI client in one place. Keep the client creation at startup so you do not re-initialize it per request.

import os
from fastapi import FastAPI
from crewai import Client

app = FastAPI()

crewai_client = Client(
    api_key=os.environ["CREWAI_API_KEY"],
    project_id=os.environ["CREWAI_PROJECT_ID"],
)

2. •Define a request model for payment-triggered agent jobs

Use Pydantic models to validate the input before it reaches your agent logic. For production systems, this is where you constrain amount, currency, and customer metadata.

from pydantic import BaseModel, Field

class PaymentRequest(BaseModel):
    customer_id: str = Field(..., min_length=3)
    amount: float = Field(..., gt=0)
    currency: str = Field(default="USD", min_length=3, max_length=3)
    description: str

3. •Create a CrewAI task that handles the payment workflow

In CrewAI, you typically define an agent and a task, then run the workflow through the client. The exact payment action depends on your project setup, but the pattern is consistent: create the task payload and submit it through the SDK.

from crewai import Agent, Task, Crew

payment_agent = Agent(
    role="Payments Orchestrator",
    goal="Validate and process payment-related requests safely",
    backstory="Handles structured payment operations for enterprise workflows",
)

@app.post("/payments/charge")
def charge_payment(request: PaymentRequest):
    task = Task(
        description=(
            f"Charge customer {request.customer_id} "
            f"{request.amount} {request.currency} "
            f"for {request.description}"
        ),
        expected_output="A structured payment result with status and reference id",
        agent=payment_agent,
    )

    crew = Crew(
        agents=[payment_agent],
        tasks=[task],
        verbose=False,
    )

    result = crew.kickoff()
    return {"status": "submitted", "result": str(result)}

4. •Expose a production-ready FastAPI endpoint

For real deployments, add request tracing and stable response shapes. If your CrewAI payment flow returns structured data, pass it through directly instead of returning raw strings.

from fastapi import HTTPException

@app.post("/payments/authorize")
async def authorize_payment(request: PaymentRequest):
    try:
        task = Task(
            description=f"Authorize {request.amount} {request.currency} for {request.customer_id}",
            expected_output="JSON with authorization status and transaction id",
            agent=payment_agent,
        )

        crew = Crew(agents=[payment_agent], tasks=[task])
        result = crew.kickoff()

        return {
            "customer_id": request.customer_id,
            "amount": request.amount,
            "currency": request.currency,
            "status": "ok",
            "crew_result": str(result),
        }
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc))

5. •Run FastAPI and keep the integration isolated

Use Uvicorn in development and behind a proper ASGI server in production. Keep the CrewAI logic in one module so you can swap implementations later without touching route definitions.

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

Testing the Integration

Use curl or Python requests to verify the endpoint responds correctly.

import requests

payload = {
    "customer_id": "cust_12345",
    "amount": 49.99,
    "currency": "USD",
    "description": "Monthly subscription"
}

response = requests.post("http://localhost:8000/payments/authorize", json=payload)
print(response.status_code)
print(response.json())

Expected output:

{
  "customer_id": "cust_12345",
  "amount": 49.99,
  "currency": "USD",
  "status": "ok",
  "crew_result": "...structured result from CrewAI..."
}

If everything is wired correctly, you should get 200 back and a response body containing your request data plus the CrewAI execution result.

Real-World Use Cases

• •Payment approval agents
  • •Route high-value transactions through an agent that checks policy rules before authorizing charges.
• •Invoice processing
  • •Let a CrewAI workflow extract invoice details, validate them, and trigger payment execution via FastAPI.
• •Collections automation
  • •Build an agent that identifies overdue accounts and exposes collection actions through secured API endpoints.

The main pattern here is simple: FastAPI handles transport and validation, while CrewAI handles decisioning and workflow orchestration. Keep those responsibilities separate, and you get something maintainable enough for production instead of a demo glued together with scripts.

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.