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

When you build multi-agent systems that move money, you need two things to be boringly reliable: a payment API and a durable system of record. FastAPI handles the payment-facing service layer well, while PostgreSQL gives you transactional state, auditability, and coordination between agents without inventing your own storage protocol.

This combo is useful when one agent initiates a charge, another agent verifies risk, and a third agent reconciles the result. You get a clean boundary: FastAPI for external payment operations, PostgreSQL for internal workflow state.

Prerequisites

• •Python 3.10+
• •FastAPI installed
• •Uvicorn installed
• •psycopg2-binary or asyncpg for PostgreSQL access
• •A running PostgreSQL instance
• •A payment provider account and API key
• •Basic familiarity with async/await
• •Environment variables configured:
  • •PAYMENT_API_KEY
  • •DATABASE_URL

Integration Steps

1. •

   Create the database schema for agent payment state

   Start by storing payment intents, agent IDs, and status transitions in PostgreSQL. This is what lets multiple agents coordinate without stepping on each other.

import os
import psycopg2
from psycopg2.extras import RealDictCursor

DATABASE_URL = os.environ["DATABASE_URL"]

DDL = """
CREATE TABLE IF NOT EXISTS payment_jobs (
    id SERIAL PRIMARY KEY,
    agent_id TEXT NOT NULL,
    provider_payment_id TEXT UNIQUE,
    amount_cents INTEGER NOT NULL,
    currency TEXT NOT NULL DEFAULT 'usd',
    status TEXT NOT NULL DEFAULT 'pending',
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
"""

conn = psycopg2.connect(DATABASE_URL)
conn.autocommit = True

with conn.cursor() as cur:
    cur.execute(DDL)

conn.close()

2. •

   Build a FastAPI endpoint that creates a payment intent

   In production, your app should call the provider’s SDK from inside the API route and persist the provider ID immediately after creation.

import os
import psycopg2
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# Example provider SDK; replace with your actual payments SDK.
# For Stripe-like providers this pattern is common:
# import stripe

app = FastAPI()
DATABASE_URL = os.environ["DATABASE_URL"]
PAYMENT_API_KEY = os.environ["PAYMENT_API_KEY"]

class PaymentRequest(BaseModel):
    agent_id: str
    amount_cents: int
    currency: str = "usd"

def get_db():
    return psycopg2.connect(DATABASE_URL)

@app.post("/payments/create")
def create_payment(req: PaymentRequest):
    try:
        # Example SDK call pattern:
        # stripe.api_key = PAYMENT_API_KEY
        # intent = stripe.PaymentIntent.create(
        #     amount=req.amount_cents,
        #     currency=req.currency,
        #     metadata={"agent_id": req.agent_id},
        # )

        intent = {
            "id": "pi_mock_123",
            "status": "requires_confirmation",
        }

        conn = get_db()
        with conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO payment_jobs (agent_id, provider_payment_id, amount_cents, currency, status)
                    VALUES (%s, %s, %s, %s, %s)
                    RETURNING id
                    """,
                    (req.agent_id, intent["id"], req.amount_cents, req.currency, intent["status"]),
                )
                job_id = cur.fetchone()[0]

        return {"job_id": job_id, "payment_id": intent["id"], "status": intent["status"]}

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

3. •

   Add an async workflow endpoint for multi-agent coordination

   This is where one agent can reserve funds while another verifies business rules. PostgreSQL becomes your shared state machine.

import asyncio
import asyncpg
from fastapi import BackgroundTasks

@app.on_event("startup")
async def startup():
    app.state.db = await asyncpg.create_pool(DATABASE_URL)

@app.post("/agents/{agent_id}/reserve")
async def reserve_for_agent(agent_id: str, amount_cents: int):
    async with app.state.db.acquire() as conn:
        row = await conn.fetchrow(
            """
            INSERT INTO payment_jobs (agent_id, amount_cents, currency, status)
            VALUES ($1, $2, 'usd', 'reserved')
            RETURNING id, status
            """,
            agent_id,
            amount_cents,
        )
        return dict(row)

@app.post("/agents/{agent_id}/mark-paid")
async def mark_paid(agent_id: str):
    async with app.state.db.acquire() as conn:
        row = await conn.fetchrow(
            """
            UPDATE payment_jobs
            SET status='paid', updated_at=NOW()
            WHERE agent_id=$1 AND status IN ('reserved', 'requires_confirmation')
            RETURNING id, status
            """,
            agent_id,
        )
        if not row:
            raise HTTPException(status_code=404, detail="No payable job found")
        return dict(row)

4. •

   Synchronize provider webhooks back into PostgreSQL

   Payments are not done when your API returns 200. They are done when the provider confirms them. Webhooks are how you keep Postgres aligned with real provider state.

from fastapi import Request

@app.post("/webhooks/payments")
async def payments_webhook(request: Request):
    payload = await request.json()

    # Example webhook shape; adapt to your provider.
    event_type = payload.get("type")
    payment_id = payload.get("data", {}).get("id")

    if event_type == "payment_intent.succeeded":
        async with app.state.db.acquire() as conn:
            await conn.execute(
                """
                UPDATE payment_jobs
                SET status='paid', updated_at=NOW()
                WHERE provider_payment_id=$1
                """,
                payment_id,
            )

    return {"received": True}

5. •

   Run the service and wire it to your agents

   Your agents should never write directly to the payment provider. They should call FastAPI endpoints and read/write only through the database-backed workflow.

# run with:
# uvicorn main:app --reload

import httpx

async def trigger_payment(agent_id: str):
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "http://localhost:8000/payments/create",
            json={"agent_id": agent_id, "amount_cents": 2500, "currency": "usd"},
        )
        resp.raise_for_status()
        return resp.json()

Testing the Integration

Use a simple round-trip test: create a payment job through FastAPI and verify it landed in PostgreSQL.

import psycopg2
import requests

BASE_URL = "http://localhost:8000"
DATABASE_URL = "postgresql://postgres:postgres@localhost:5432/payments"

response = requests.post(
    f"{BASE_URL}/payments/create",
    json={"agent_id": "agent-risk-01", "amount_cents": 4200, "currency": "usd"},
)
data = response.json()

conn = psycopg2.connect(DATABASE_URL)
with conn.cursor() as cur:
    cur.execute(
        "SELECT agent_id, provider_payment_id, amount_cents, status FROM payment_jobs WHERE id=%s",
        (data["job_id"],),
    )
    row = cur.fetchone()

print(data)
print(row)

Expected output:

{'job_id': 1, 'payment_id': 'pi_mock_123', 'status': 'requires_confirmation'}
('agent-risk-01', 'pi_mock_123', 4200, 'requires_confirmation')

Real-World Use Cases

• •

  Agent-driven checkout flows

  • •One agent prices the order.
  • •Another validates fraud signals.
  • •FastAPI creates the charge and PostgreSQL tracks each step.
• •

  Insurance claim disbursement

  • •A triage agent approves payout eligibility.
  • •A finance agent triggers the transfer.
  • •PostgreSQL stores claim state and payout audit trails.
• •

  Multi-agent billing orchestration

  • •Agents split charges across services or tenants.
  • •FastAPI exposes billing endpoints.
  • •PostgreSQL keeps reconciliation data for disputes and reporting.

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.