How to Integrate FastAPI for payments with PostgreSQL for production AI

FastAPI gives you a clean way to expose payment endpoints, while PostgreSQL gives you durable storage for invoices, payment states, audit logs, and agent memory. Put them together and you get a production-grade backend for AI systems that need to accept payments, track entitlements, and persist transaction history without losing consistency.

For AI agents, this matters because payment flows are stateful. The agent can initiate a charge, poll status, reconcile webhook events, and store the full lifecycle in PostgreSQL for retries, reporting, and compliance.

Prerequisites

• •Python 3.11+
• •FastAPI installed
• •Uvicorn installed
• •PostgreSQL 14+
• •psycopg or asyncpg installed
• •A payment provider SDK or API credentials
• •A .env file with:
  • •DATABASE_URL
  • •PAYMENT_API_KEY
  • •PAYMENT_WEBHOOK_SECRET

Install the core packages:

pip install fastapi uvicorn psycopg[binary] python-dotenv requests

Integration Steps

1. •Create the database schema for payment state

You want a table that tracks each payment request from your AI system. Keep the external provider ID, internal status, amount, currency, and timestamps.

import psycopg
from psycopg.rows import dict_row

DATABASE_URL = "postgresql://postgres:postgres@localhost:5432/ai_payments"

DDL = """
CREATE TABLE IF NOT EXISTS payments (
    id BIGSERIAL PRIMARY KEY,
    user_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()
);
"""

with psycopg.connect(DATABASE_URL) as conn:
    with conn.cursor() as cur:
        cur.execute(DDL)
    conn.commit()

2. •Build a small PostgreSQL repository layer

Do not scatter SQL across your FastAPI routes. Put persistence behind functions so your agent workflow can reuse them for web requests, background jobs, and webhook handlers.

import psycopg
from psycopg.rows import dict_row

class PaymentRepository:
    def __init__(self, dsn: str):
        self.dsn = dsn

    def create_payment(self, user_id: str, amount_cents: int, currency: str = "usd"):
        sql = """
        INSERT INTO payments (user_id, amount_cents, currency)
        VALUES (%s, %s, %s)
        RETURNING id, user_id, amount_cents, currency, status;
        """
        with psycopg.connect(self.dsn) as conn:
            with conn.cursor(row_factory=dict_row) as cur:
                cur.execute(sql, (user_id, amount_cents, currency))
                row = cur.fetchone()
            conn.commit()
            return row

    def update_status(self, provider_payment_id: str, status: str):
        sql = """
        UPDATE payments
        SET status = %s,
            updated_at = NOW()
        WHERE provider_payment_id = %s
        RETURNING id, status;
        """
        with psycopg.connect(self.dsn) as conn:
            with conn.cursor(row_factory=dict_row) as cur:
                cur.execute(sql, (status, provider_payment_id))
                row = cur.fetchone()
            conn.commit()
            return row

3. •Expose payment creation through FastAPI

Here is the API layer. The route accepts a request from your AI agent or frontend and calls both the repository and the payment provider.

from fastapi import FastAPI
from pydantic import BaseModel
import os
import requests

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

class CreatePaymentRequest(BaseModel):
    user_id: str
    amount_cents: int
    currency: str = "usd"

@app.post("/payments")
def create_payment(payload: CreatePaymentRequest):
    db_payment = repo.create_payment(
        user_id=payload.user_id,
        amount_cents=payload.amount_cents,
        currency=payload.currency,
    )

    # Example payment-provider call using HTTP API.
    # Replace URL/body with your actual SDK or gateway.
    response = requests.post(
        "https://api.paymentprovider.com/v1/payments",
        headers={"Authorization": f"Bearer {os.environ['PAYMENT_API_KEY']}"},
        json={
            "amount": payload.amount_cents,
            "currency": payload.currency,
            "metadata": {"internal_payment_id": db_payment["id"]},
        },
        timeout=30,
    )
    response.raise_for_status()
    provider_data = response.json()

    return {
        "payment_id": db_payment["id"],
        "provider_payment_id": provider_data["id"],
        "status": provider_data["status"],
    }

4. •Persist webhook updates from the payment provider

Production payment systems do not trust only synchronous responses. Webhooks are where final truth usually lands. Store the provider ID and update your local record when the event arrives.

from fastapi import Request, HTTPException

@app.post("/webhooks/payments")
async def payment_webhook(request: Request):
    raw_body = await request.body()
    signature = request.headers.get("X-Signature")

    # Verify signature using your provider's webhook method here.
    # Example placeholder:
    if not signature:
      raise HTTPException(status_code=400, detail="Missing signature")

    event = await request.json()

    if event["type"] == "payment.succeeded":
        provider_payment_id = event["data"]["id"]
        repo.update_status(provider_payment_id=provider_payment_id, status="succeeded")

    if event["type"] == "payment.failed":
        provider_payment_id = event["data"]["id"]
        repo.update_status(provider_payment_id=provider_payment_id, status="failed")

    return {"ok": True}

5. •Wire it into an AI agent workflow

Your agent should not “just pay.” It should create a pending record first, trigger payment creation second, then wait for webhook confirmation before granting access or continuing execution.

def charge_for_agent_action(user_id: str):
    pending = repo.create_payment(user_id=user_id, amount_cents=5000)

    # In a real system this would call the FastAPI endpoint or shared service layer.
    print(f"Created internal payment record: {pending['id']}")

    # After webhook confirmation:
    # repo.update_status(provider_payment_id="pay_123", status="succeeded")

Testing the Integration

Run FastAPI:

uvicorn app:app --reload --port 8000

Then verify database writes and API behavior with a direct request:

import requests

resp = requests.post(
    "http://localhost:8000/payments",
    json={
        "user_id": "user_42",
        "amount_cents": 2500,
        "currency": "usd"
    },
)

print(resp.status_code)
print(resp.json())

Expected output:

200
{
  'payment_id': 1,
  'provider_payment_id': 'pay_abc123',
  'status': 'pending'
}

If you then query PostgreSQL:

import psycopg

with psycopg.connect("postgresql://postgres:postgres@localhost:5432/ai_payments") as conn:
    with conn.cursor() as cur:
      cur.execute("SELECT id, user_id, amount_cents, status FROM payments ORDER BY id DESC LIMIT 1;")
      print(cur.fetchone())

You should see the persisted record in pending state until the webhook updates it.

Real-World Use Cases

• •

  AI billing gates
  Charge before executing expensive model calls or tool actions. Store the charge state in PostgreSQL so retries do not double-bill users.

• •

  Subscription and entitlement tracking
  Let an agent check whether a customer has paid for premium workflows before granting access to higher-cost inference paths.

• •

  Payment reconciliation pipelines
  Use FastAPI to ingest webhooks and PostgreSQL to maintain an audit trail for finance teams, support staff, and automated dispute handling.

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.