Plugipay
ForjioHuudisIdentityPlugipayPaymentsStorlaunchE-commerceFulkrumaFulfilmentRiplloMarketingLinkSnapShort URLsPawpadoGPU streaming
Plugipay Docs
⌘K

Webhooks

Plugipay signs every outbound webhook with HMAC-SHA256. The Python SDK ships a single helper — verify_webhook — that validates the signature, checks freshness, and returns the parsed event envelope. You wire it into Flask, FastAPI, Django, or any other framework that exposes the raw request body.

from plugipay import verify_webhook, PlugipaySignatureError

event = verify_webhook(raw_body, signature_header, secret)
# event.type      → 'plugipay.invoice.paid.v1'
# event.id        → 'evt_01HX…'
# event.object    → dict of the resource that triggered the event

The signature recipe itself — what verify_webhook is doing under the hood — is documented at API → Webhooks. This page covers framework integration and the gotchas every team hits at least once.

The helper

verify_webhook(
    raw_body: bytes | str,
    signature_header: str | None,
    secret: str,
    *,
    tolerance_sec: int = 300,
    now: float | None = None,
) -> WebhookEvent
Argument What it expects
raw_body The unparsed request body, as bytes (preferred) or str. Not the parsed JSON.
signature_header The value of the X-Plugipay-Signature header, or None if absent.
secret The webhook endpoint's signing secret — returned once when you create the endpoint.
tolerance_sec How old a signature can be before it's rejected. Default 300 seconds (5 minutes).
now Inject a clock for tests (seconds since epoch). Production code never sets this.

On success returns a WebhookEvent. On any failure raises PlugipaySignatureError with one of four codes:

  • signature_missing — header absent or empty, or secret is empty.
  • signature_malformed — header doesn't have t= and v1= parts.
  • signature_stale — timestamp older than tolerance_sec.
  • signature_invalid — HMAC mismatch, or body isn't valid UTF-8 JSON.

The raw body trap

The single most common bug in webhook handlers across all languages: you cannot re-serialise the JSON. The signature is computed over the exact bytes Plugipay sent. If your framework parses the body into a dict and you json.dumps() it back, whitespace and key order will differ and the signature will never match.

Always read the body before any JSON parser touches it.

Framework Get raw bytes
Flask request.get_data() — bytes, before request.get_json()
FastAPI await request.body() — bytes
Django request.body — bytes
Starlette await request.body()
aiohttp await request.read()
Tornado self.request.body
Sanic request.body

Flask

import os
from flask import Flask, request, abort
from plugipay import verify_webhook, PlugipaySignatureError

app = Flask(__name__)
SECRET = os.environ["PLUGIPAY_WEBHOOK_SECRET"]


@app.post("/webhooks/plugipay")
def plugipay_webhook():
    raw = request.get_data()
    sig = request.headers.get("X-Plugipay-Signature", "")

    try:
        event = verify_webhook(raw, sig, SECRET)
    except PlugipaySignatureError:
        abort(400)

    if event.type == "plugipay.invoice.paid.v1":
        handle_invoice_paid(event.object)
    elif event.type == "plugipay.checkout_session.succeeded.v1":
        handle_checkout_succeeded(event.object)

    return "", 204

Return any 2xx to acknowledge. Plugipay treats non-2xx as a delivery failure and retries with exponential backoff. See API → Webhooks → Retries for the full retry contract.

FastAPI

import os
from fastapi import FastAPI, Request, HTTPException, status
from plugipay import verify_webhook, PlugipaySignatureError

app = FastAPI()
SECRET = os.environ["PLUGIPAY_WEBHOOK_SECRET"]


@app.post("/webhooks/plugipay", status_code=status.HTTP_204_NO_CONTENT)
async def plugipay_webhook(request: Request):
    raw = await request.body()
    sig = request.headers.get("x-plugipay-signature", "")

    try:
        event = verify_webhook(raw, sig, SECRET)
    except PlugipaySignatureError as exc:
        raise HTTPException(status_code=400, detail=exc.code)

    # Hand off to a sync handler — verify_webhook is sync-only and fast.
    await handle_event(event)
    return None


async def handle_event(event):
    if event.type == "plugipay.invoice.paid.v1":
        invoice = event.object
        await record_payment(invoice["id"], invoice["total"])

Don't declare the function with a Pydantic model parameter. FastAPI will eagerly parse the body into the model and you'll lose the original bytes — signature verification will fail. Take a Request and call await request.body() first.

Django

# views.py
import os
from django.http import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from plugipay import verify_webhook, PlugipaySignatureError

SECRET = os.environ["PLUGIPAY_WEBHOOK_SECRET"]


@csrf_exempt
@require_POST
def plugipay_webhook(request):
    raw = request.body  # bytes
    sig = request.headers.get("X-Plugipay-Signature", "")

    try:
        event = verify_webhook(raw, sig, SECRET)
    except PlugipaySignatureError:
        return HttpResponseBadRequest()

    if event.type == "plugipay.checkout_session.succeeded.v1":
        session = event.object
        # …
    return HttpResponse(status=204)

csrf_exempt is mandatory — Plugipay isn't a browser, it doesn't have your CSRF token. The HMAC signature is the authentication.

The WebhookEvent object

event.id          # 'evt_01HX…'
event.type        # 'plugipay.invoice.paid.v1' — see event catalog
event.account_id  # 'acc_01HX…' — the merchant workspace
event.occurred_at # ISO 8601 timestamp
event.object      # dict — the resource that triggered the event
event.raw         # full envelope dict — useful for forwarding/storage

event.object is whatever payload makes sense for the event:

  • plugipay.customer.created.v1 → the customer dict
  • plugipay.invoice.paid.v1 → the invoice dict
  • plugipay.checkout_session.succeeded.v1 → the checkout session dict
  • plugipay.subscription.cancelled.v1 → the subscription dict
  • plugipay.refund.succeeded.v1 → the refund dict
  • plugipay.payout.paid.v1 → the payout dict

The full event catalog is at API → Webhooks → Events. Versioned suffixes (.v1) mean adding a field is non-breaking; renaming or removing a field bumps to .v2 and emits both for a deprecation window.

Switching on event type

For more than a handful of event types, a dispatch dict reads better than a chain of ifs:

def handle_invoice_paid(invoice): ...
def handle_subscription_cancelled(sub): ...
def handle_refund_succeeded(refund): ...

HANDLERS = {
    "plugipay.invoice.paid.v1": handle_invoice_paid,
    "plugipay.subscription.cancelled.v1": handle_subscription_cancelled,
    "plugipay.refund.succeeded.v1": handle_refund_succeeded,
}

def dispatch(event):
    handler = HANDLERS.get(event.type)
    if handler is None:
        # Unknown event — log and ack so Plugipay stops retrying.
        log.info("plugipay event unhandled: %s", event.type)
        return
    handler(event.object)

Idempotency

Plugipay may deliver the same event more than once — it's "at least once" delivery, not "exactly once". event.id is stable, so the canonical pattern is:

def dispatch(event):
    if PaymentEvent.objects.filter(event_id=event.id).exists():
        return  # already processed; ignore
    PaymentEvent.objects.create(event_id=event.id, type=event.type, raw=event.raw)
    HANDLERS[event.type](event.object)

Or, inside a transaction, with a unique constraint on event_id:

try:
    with transaction.atomic():
        PaymentEvent.objects.create(event_id=event.id, ...)
        handler(event.object)
except IntegrityError:
    pass  # duplicate delivery

Either pattern works; pick what matches your stack.

Replay protection

verify_webhook rejects signatures whose timestamp is more than 300 seconds old (configurable via tolerance_sec). That's enough margin for normal clock skew (Plugipay's clocks are NTP-synced) and small enough to block replay attacks — a captured webhook signature is useless five minutes later.

If you see signature_stale errors in production, your server clock is likely off. Run timedatectl status (Linux) or check System Settings → Date & Time (macOS). Most cloud platforms (Heroku, Fly, Cloud Run, Render) handle clock sync for you.

Testing your handler

Use the SDK's helper with a fixed clock:

import hashlib, hmac, json, time
from plugipay import verify_webhook

SECRET = "whsec_test_xxx"
body = json.dumps({"id": "evt_01HX", "type": "plugipay.invoice.paid.v1",
                   "data": {"object": {"id": "inv_01H"}}}).encode()
t = int(time.time())
sig = hmac.new(SECRET.encode(), f"{t}.".encode() + body, hashlib.sha256).hexdigest()
header = f"t={t},v1={sig}"

event = verify_webhook(body, header, SECRET, now=t)
assert event.type == "plugipay.invoice.paid.v1"

In production code, the simpler approach is to forward a real webhook from the portal:

  1. Mint a webhook endpoint in Settings → Webhooks pointing at an ngrok / cloudflared tunnel.
  2. Trigger an event from test mode (create a checkout session, simulate success).
  3. Watch your handler receive it.

The portal also has a "Replay" button on every event in the delivery log — click to re-send to your endpoint without re-doing the upstream flow.

Next

  • API → Webhooks — the full signature recipe, event catalog, retry contract.
  • Errors — the PlugipaySignatureError cases in detail.
  • Portal → Webhooks — mint endpoints, watch the delivery log, replay.
Plugipay — Payments that don't tax your success