Python FastAPI Guide: Creating Instant UPI Payment Webhooks with HMAC Verification

When building modern fintech applications, AI commerce agents, or Python backend services, handling payment events with sub-second latency and military-grade cryptographic verification is non-negotiable.

FastAPI has become the framework of choice for modern Python developers due to its native asynchronous performance, strict Pydantic type safety, and lightweight footprint. Here is the complete production guide to building a secure, high-throughput UPI webhook receiver in Python.

---

Why FastAPI is Ideal for Payment Event Listeners

Direct Answer: FastAPI leverages Python’s native asynchronous event loop (ASGI) to process thousands of incoming payment notifications concurrently with minimal memory usage. Its async request streaming allows raw bytes to be cryptographically verified against HMAC signatures before passing validated payloads into background database tasks.

During high-traffic events (such as flash sales or product launches), receiving 500 simultaneous bank confirmation callbacks will choke traditional synchronous WSGI frameworks (like standard Flask or Django). FastAPI processes each webhook in non-blocking I/O cycles, keeping response times under 20 milliseconds.

---

The Cryptographic Handshake: HMAC-SHA256

To verify that an incoming HTTP POST request actually originated from VyaparGateway (and not an attacker probing your API):

1. The gateway concatenates the exact raw request body bytes.
2. It hashes the payload using your secret key via the HMAC-SHA256 algorithm:

   Signature = HMAC_SHA256(Raw_Body_Bytes, Webhook_Secret_Key)

3. The resulting hexadecimal digest is sent in the X-Webhook-Signature header.
4. Your server computes the identical hash locally and performs a constant-time string comparison using hmac.compare_digest() to eliminate timing attack vulnerabilities.

---

Complete FastAPI Webhook Handler (main.py)

Here is the complete, production-ready implementation:

import hmac
import hashlib
import os
from fastapi import FastAPI, Request, HTTPException, Header, status
from pydantic import BaseModel
from typing import Optional

app = FastAPI(title="UPI Webhook Service")

WEBHOOK_SECRET = os.getenv("VYAPAR_WEBHOOK_SECRET", "whsec_live_example_key_9942")

class PaymentWebhookPayload(BaseModel):
    status: str
    client_txn_id: str
    order_id: str
    amount: float
    utr: Optional[str] = None
    customer_mobile: Optional[str] = None
    created_at: Optional[str] = None

def verify_signature(raw_body: bytes, header_signature: str, secret: str) -> bool:
    """Computes HMAC-SHA256 hash and executes constant-time verification."""
    computed_digest = hmac.new(
        key=secret.encode("utf-8"),
        msg=raw_body,
        digestmod=hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(computed_digest, header_signature)

@app.post("/api/v1/webhook/payment", status_code=status.HTTP_200_OK)
async def handle_payment_webhook(
    request: Request,
    x_webhook_signature: Optional[str] = Header(None)
):
    if not x_webhook_signature:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Missing X-Webhook-Signature header"
        )

    # 1. Capture RAW BYTES before parsing (crucial for exact hash matching)
    raw_body = await request.body()

    # 2. Cryptographic signature check
    if not verify_signature(raw_body, x_webhook_signature, WEBHOOK_SECRET):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid HMAC signature. Rejecting forged event."
        )

    # 3. Parse validated JSON payload into Pydantic model
    payload_data = await request.json()
    event = PaymentWebhookPayload(**payload_data)

    # 4. Process payment status
    if event.status == "SUCCESS":
        # Execute business logic (e.g. mark paid in database, notify warehouse)
        print(f"Verified payment of INR {event.amount} for Order {event.client_txn_id} (UTR: {event.utr})")
    elif event.status == "FAILED":
        print(f"Payment failed for Order {event.client_txn_id}")

    # 5. Always return 200 OK fast to acknowledge receipt
    return {"status": "acknowledged", "order_id": event.order_id}

---

Handling Idempotency and Preventing Duplicate Credits

In distributed systems, networks retry failed requests. Your webhook listener will occasionally receive the exact same webhook twice.

To avoid fulfilling an order twice:

1. Extract the utr (Bank Unique Transaction Reference) or client_txn_id.
2. Check your PostgreSQL database using an atomic transaction:

   INSERT INTO payment_events (utr, order_id, amount, status)
   VALUES (:utr, :order_id, :amount, 'PROCESSED')
   ON CONFLICT (utr) DO NOTHING;

3. If zero rows are inserted, acknowledge with 200 OK immediately without re-triggering inventory or shipping actions.

---

Testing Webhooks Locally with cURL and ngrok

To simulate an authentic webhook locally during development:

# Generate valid HMAC signature in Python CLI:
python3 -c '
import hmac, hashlib
secret = "whsec_live_example_key_9942"
body = b"{\"status\":\"SUCCESS\",\"client_txn_id\":\"ORD_491\",\"order_id\":\"VG_9918\",\"amount\":1499.0}"
sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
print("Signature:", sig)
'

Then dispatch the cURL command:

curl -X POST http://localhost:8000/api/v1/webhook/payment \
  -H "Content-Type: application/json" \
  -H "X-Webhook-Signature: <GENERATED_SIG>" \
  -d '{"status":"SUCCESS","client_txn_id":"ORD_491","order_id":"VG_9918","amount":1499.0}'

Test signature validation visually using our free Webhook Signature Generator Tool.