How to Integrate FastAPI for payments with PostgreSQL for AI agents

FastAPI for payments gives you a clean HTTP layer for charging cards, creating payment intents, and exposing webhook endpoints. PostgreSQL gives your AI agent system durable state: user profiles, payment status, invoices, retries, and audit trails.

Put them together and you get an agent that can take an action, charge for it, persist the transaction, and recover safely when a webhook arrives later.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL instance
• •A FastAPI app configured for payment endpoints
• •pip installed packages:
  • •fastapi
  • •uvicorn
  • •psycopg2-binary
  • •sqlalchemy
  • •pydantic
  • •a payment provider SDK such as stripe if your “FastAPI for payments” layer wraps Stripe
• •Environment variables set:
  • •DATABASE_URL
  • •STRIPE_API_KEY or your payment provider secret
  • •STRIPE_WEBHOOK_SECRET
• •A table design for:
  • •users
  • •payment_intents
  • •payment_events

Integration Steps

1. •

   Create the PostgreSQL schema

   Start with a minimal schema that tracks the payment lifecycle. You want the database to be the source of truth, not the webhook payload.

   from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text
   from sqlalchemy.orm import declarative_base, sessionmaker
   from sqlalchemy.sql import func
   import os
   
   DATABASE_URL = os.getenv("DATABASE_URL")
   engine = create_engine(DATABASE_URL)
   SessionLocal = sessionmaker(bind=engine)
   Base = declarative_base()
   
   class PaymentIntent(Base):
       __tablename__ = "payment_intents"
   
       id = Column(Integer, primary_key=True)
       user_id = Column(String(64), nullable=False)
       provider_payment_id = Column(String(255), unique=True, nullable=False)
       status = Column(String(32), nullable=False)
       amount_cents = Column(Integer, nullable=False)
       currency = Column(String(8), nullable=False, default="usd")
       raw_payload = Column(Text, nullable=True)
       created_at = Column(DateTime(timezone=True), server_default=func.now())
   
   Base.metadata.create_all(bind=engine)
   

2. •

   Expose a FastAPI endpoint to create a payment intent

   This is where your AI agent asks for payment before continuing an expensive workflow. If you use Stripe under the hood, the actual API call is stripe.PaymentIntent.create(...).

   from fastapi import FastAPI, HTTPException
   from pydantic import BaseModel
   import stripe
   import os
   
   app = FastAPI()
   stripe.api_key = os.getenv("STRIPE_API_KEY")
   
   class CreatePaymentRequest(BaseModel):
       user_id: str
       amount_cents: int
       currency: str = "usd"
   
   @app.post("/payments/create")
   def create_payment(req: CreatePaymentRequest):
       try:
           intent = stripe.PaymentIntent.create(
               amount=req.amount_cents,
               currency=req.currency,
               metadata={"user_id": req.user_id},
           )
           return {
               "payment_intent_id": intent["id"],
               "client_secret": intent["client_secret"],
               "status": intent["status"],
           }
       except stripe.error.StripeError as e:
           raise HTTPException(status_code=400, detail=str(e))
   

3. •

   Persist the payment record in PostgreSQL

   After creating the provider-side intent, store it locally. This gives you idempotency and lets your agent resume after failure.

   from sqlalchemy.orm import Session
   
   def save_payment_intent(db: Session, user_id: str, intent: dict, amount_cents: int):
       row = PaymentIntent(
           user_id=user_id,
           provider_payment_id=intent["id"],
           status=intent["status"],
           amount_cents=amount_cents,
           currency=intent.get("currency", "usd"),
           raw_payload=str(intent),
       )
       db.add(row)
       db.commit()
       db.refresh(row)
       return row
   
   @app.post("/payments/create-and-store")
   def create_and_store(req: CreatePaymentRequest):
       db = SessionLocal()
       try:
           intent = stripe.PaymentIntent.create(
               amount=req.amount_cents,
               currency=req.currency,
               metadata={"user_id": req.user_id},
           )
           row = save_payment_intent(db, req.user_id, intent.to_dict(), req.amount_cents)
           return {"db_id": row.id, "provider_payment_id": row.provider_payment_id}
       finally:
           db.close()
   

4. •

   Handle webhooks and update PostgreSQL

   Webhooks are how you keep the database aligned with the real payment state. Verify the signature first using stripe.Webhook.construct_event(...), then update your row.

   from fastapi import Request
   from sqlalchemy import select
   
   WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET")
   
   @app.post("/webhooks/payments")
   async def payments_webhook(request: Request):
       payload = await request.body()
       sig_header = request.headers.get("stripe-signature")
   
       try:
           event = stripe.Webhook.construct_event(
               payload=payload,
               sig_header=sig_header,
               secret=WEBHOOK_SECRET,
           )
       except Exception as e:
           raise HTTPException(status_code=400, detail=f"Webhook error: {e}")
   
       if event["type"] == "payment_intent.succeeded":
           intent_obj = event["data"]["object"]
           db = SessionLocal()
           try:
               stmt = select(PaymentIntent).where(
                   PaymentIntent.provider_payment_id == intent_obj["id"]
               )
               row = db.execute(stmt).scalar_one_or_none()
               if row:
                   row.status = "succeeded"
                   db.commit()
           finally:
               db.close()
   
       return {"received": True}
   

5. •

   Let your AI agent read payment state before acting

   Your agent should query PostgreSQL before generating paid output or unlocking premium actions. This keeps policy decisions inside your system instead of buried in prompt logic.

    from sqlalchemy import text
   
    def get_payment_status(user_id: str) -> str | None:
        db = SessionLocal()
        try:
            result = db.execute(
                text("""
                    SELECT status
                    FROM payment_intents
                    WHERE user_id = :user_id
                    ORDER BY created_at DESC
                    LIMIT 1
                """),
                {"user_id": user_id},
            ).fetchone()
   
            return result[0] if result else None
        finally:
            db.close()
   
    def can_agent_continue(user_id: str) -> bool:
        status = get_payment_status(user_id)
        return status == "succeeded"
   

Testing the Integration

Run the app:

uvicorn main:app --reload

Then test with a simple request:

import requests

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

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

Expected output:

200
{
  'payment_intent_id': 'pi_12345',
  'client_secret': 'pi_12345_secret_abc',
  'status': 'requires_payment_method'
}

After webhook delivery or manual DB update, verify PostgreSQL contains the final state:

db_status = get_payment_status("user_123")
print(db_status)

Expected output:

succeeded

Real-World Use Cases

• •

  Paid agent actions
  Charge before running expensive workflows like document summarization, KYC checks, or report generation.

• •

  Subscription-backed copilots
  Store subscription status in PostgreSQL and let agents enforce access rules before answering premium requests.

• •

  Usage-based billing
  Track each tool call or token-heavy task in PostgreSQL and send aggregated charges through FastAPI endpoints on schedule.

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.