FastAPI Integration

pip install stableops fastapi uvicorn python-dotenv

STABLEOPS_API_KEY=sk_sandbox_...
STABLEOPS_WEBHOOK_SECRET=whsec_...
STABLEOPS_API_URL=https://api.stableops.dev

import os

from stableops import StableOps

stableops = StableOps(
    api_key=os.environ["STABLEOPS_API_KEY"],
    environment="sandbox",
    base_url=os.getenv("STABLEOPS_API_URL", "https://api.stableops.dev"),
)

from datetime import datetime, timedelta, timezone

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

from lib.stableops import stableops

router = APIRouter()

class CreateOrderRequest(BaseModel):
    merchant_order_id: str
    amount: str

@router.post("/orders")
def create_order(input: CreateOrderRequest):
    try:
        # Auto-expire after 30 minutes; the order moves to `expired` and the address is released.
        expires_at = (datetime.now(timezone.utc) + timedelta(minutes=30)).isoformat()
        order = stableops.payment_orders.create(
            merchant_order_id=input.merchant_order_id,
            amount=input.amount,
            settlement_asset="USDC",
            accepted_assets=[{"chain": "base", "asset": "USDC"}],
            expires_at=expires_at,
        )
        return {
            "id": order.id,
            "status": order.status,
            "amount": order.amount,
            "payment_instructions": [
                instruction.model_dump() for instruction in order.payment_instructions
            ],
        }
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc

import json
import os

from fastapi import APIRouter, Header, Request, Response
from stableops.webhooks import SIGNATURE_HEADER, verify_webhook_signature

router = APIRouter()

@router.post("/webhooks/stableops")
async def stableops_webhook(
    request: Request,
    x_product_signature: str | None = Header(default=None, alias=SIGNATURE_HEADER),
):
    raw_body = await request.body()
    result = verify_webhook_signature(
        body=raw_body,
        header=x_product_signature,
        secret=os.environ["STABLEOPS_WEBHOOK_SECRET"],
    )

    if not result.valid:
        return Response(f"invalid signature: {result.reason}", status_code=400)

    event = json.loads(raw_body)
    if event["type"] == "payment.finalized":
        payment_order_id = event["data"]["payment_order_id"]
        # Fulfill after deduping X-Event-Id in your database.

    return Response("ok")