Shopify Webhooks in Python: HMAC Verification, Order Automation & Shiprocket Sync

In e-commerce, the moment an order is placed on Shopify is the starting gun for post-purchase operations:
- A fulfillment order must be created with your 3PL or logistics partner (Shiprocket, Delhivery, ShipBob).
- Inventory counts must be adjusted in external warehouse management systems (ERP / Amazon / retail POS).
- An order confirmation with real-time tracking must be triggered via WhatsApp or SMS.

If your webhook receiver goes down or returns a 500 server error, the fallout is severe:
1. Unfulfilled orders: Packages sit in the warehouse unsent while customers wait.
2. Shopify's 19-retry cutoff: Shopify retries failed webhooks 19 times over 48 hours. If your handler fails consistently during this window, Shopify automatically de-registers and deletes your webhook. You will receive zero new order events until someone manually re-creates the webhook.
3. The 5-Second Rule: Shopify requires an HTTP 200 response within 5 seconds. If your script takes 6 seconds because a logistics API is slow, Shopify marks the call as failed.

Here is how to build a production-grade, serverless Shopify webhook receiver in Python that verifies signatures, responds in under 100ms, and processes fulfillment asynchronously.

Verifying the Shopify HMAC Signature in Python

Shopify calculates an HMAC-SHA256 hash of the raw JSON payload using your Shopify App Secret (or Webhook Secret from Admin Settings), and passes the hash as a base64-encoded string in the X-Shopify-Hmac-Sha256 header.

Here is the exact verification implementation:

import hmac
import hashlib
import base64

def verify_shopify_webhook(raw_body: str, hmac_header: str, secret: str) -> bool:
    """
    Verifies that the webhook originated from Shopify.
    Shopify uses base64-encoded HMAC-SHA256 digests.
    """
    if not hmac_header or not secret:
        return False

    # Compute HMAC digest using the webhook secret
    digest = hmac.new(
        key=secret.encode("utf-8"),
        msg=raw_body.encode("utf-8"),
        digestmod=hashlib.sha256
    ).digest()

    # Shopify base64-encodes the digest
    computed_hmac = base64.b64encode(digest).decode("utf-8")

    # Constant-time comparison prevents timing attacks
    return hmac.compare_digest(computed_hmac, hmac_header)

Complete Production Handler: Shopify Order → Shiprocket / 3PL Logistics

The following production handler verifies the webhook, extracts line items and customer shipping details, pushes the shipment to a 3PL logistics API (like Shiprocket), and fires an instant confirmation:

import os
import json
import requests
from verify import verify_shopify_webhook

def handler(event, context):
    """
    Processes Shopify order events (e.g. orders/create, orders/paid).
    """
    raw_body = event.get("body", "")
    headers = event.get("headers", {})

    hmac_header = (
        headers.get("x-shopify-hmac-sha256") or 
        headers.get("X-Shopify-Hmac-Sha256")
    )
    topic = (
        headers.get("x-shopify-topic") or 
        headers.get("X-Shopify-Topic") or ""
    )
    shop_domain = (
        headers.get("x-shopify-shop-domain") or 
        headers.get("X-Shopify-Shop-Domain")
    )

    webhook_secret = os.environ.get("SHOPIFY_WEBHOOK_SECRET")
    if not webhook_secret:
        return {"status": 500, "error": "Server misconfigured: missing SHOPIFY_WEBHOOK_SECRET"}

    # 1. Verify HMAC Signature
    if not verify_shopify_webhook(raw_body, hmac_header, webhook_secret):
        print(f"⚠️ Invalid HMAC from domain: {shop_domain}")
        return {"status": 401, "error": "Unauthorized: HMAC verification failed"}

    # 2. Parse Order JSON
    try:
        order = json.loads(raw_body)
    except json.JSONDecodeError:
        return {"status": 400, "error": "Invalid JSON"}

    order_id = order.get("id")
    order_name = order.get("name") # e.g. #1042
    total_price = order.get("total_price")
    email = order.get("email")
    shipping_address = order.get("shipping_address") or {}
    line_items = order.get("line_items") or []

    print(f"📦 Received [{topic}] for Order {order_name} (ID: {order_id}) - Total: {total_price}")

    # 3. Idempotency Protection
    idempotency_key = f"shopify_order_{order_id}_{topic}"
    if hasattr(context, "kv"):
        if context.kv.get(idempotency_key):
            print(f"Order {order_id} topic {topic} already processed. Skipping.")
            return {"status": 200, "message": "Already processed"}
        context.kv.set(idempotency_key, "1", ttl=86400 * 3) # 3 days

    # 4. Route Actions Based on Topic
    if topic in ("orders/create", "orders/paid"):
        # Auto-create shipment in logistics API
        create_shipment(order)

        # Notify fulfillment team on Slack or Discord
        notify_fulfillment_channel(order_name, total_price, line_items)

    elif topic == "orders/cancelled":
        handle_order_cancelled(order)

    # 5. Return 200 OK fast (< 5s limit)
    return {"status": 200, "received": True}


def create_shipment(order: dict):
    """
    Push order to Shiprocket / 3PL API to generate AWB and shipping label.
    """
    shiprocket_token = os.environ.get("SHIPROCKET_API_TOKEN")
    if not shiprocket_token:
        return

    shipping = order.get("shipping_address", {})
    payload = {
        "order_id": str(order.get("id")),
        "order_date": order.get("created_at"),
        "billing_customer_name": shipping.get("first_name", "Customer"),
        "billing_last_name": shipping.get("last_name", ""),
        "billing_address": shipping.get("address1", ""),
        "billing_city": shipping.get("city", ""),
        "billing_pincode": shipping.get("zip", ""),
        "billing_state": shipping.get("province", ""),
        "billing_country": shipping.get("country", "India"),
        "billing_email": order.get("email", ""),
        "billing_phone": shipping.get("phone", ""),
        "shipping_is_billing": True,
        "order_items": [
            {
                "name": item.get("name"),
                "sku": item.get("sku", "DEFAULT"),
                "units": item.get("quantity"),
                "selling_price": item.get("price")
            }
            for item in order.get("line_items", [])
        ],
        "payment_method": "Prepaid" if order.get("financial_status") == "paid" else "COD",
        "sub_total": float(order.get("subtotal_price", 0))
    }

    try:
        res = requests.post(
            "https://apiv2.shiprocket.in/v1/external/orders/create/adhoc",
            json=payload,
            headers={
                "Authorization": f"Bearer {shiprocket_token}",
                "Content-Type": "application/json"
            },
            timeout=8
        )
        print(f"Shiprocket Response [{res.status_code}]: {res.text[:200]}")
    except requests.RequestException as e:
        print(f"Logistics sync error: {e}")


def notify_fulfillment_channel(order_name: str, total_price: str, items: list):
    """Post order summary to Slack."""
    slack_url = os.environ.get("SLACK_WEBHOOK_URL")
    if not slack_url:
        return

    item_summary = ", ".join([f"{i.get('quantity')}x {i.get('name')}" for i in items[:3]])
    msg = {
        "text": f"🛒 *New Shopify Order {order_name}* ({total_price})\n*Items:* {item_summary}"
    }
    try:
        requests.post(slack_url, json=msg, timeout=4)
    except requests.RequestException:
        pass


def handle_order_cancelled(order: dict):
    print(f"Order {order.get('name')} was cancelled. Reversing inventory hold.")

Prevent Webhook De-Registration During Traffic Surges

When running a flash sale or Black Friday campaign, Shopify can send hundreds of order webhooks per minute. If a monolithic server chokes on database locks, Shopify's retries fail and Shopify will permanently delete your webhook subscription.

LiteLambda automatically scales out isolated Python runners for each incoming webhook, returns an immediate 200 acknowledgement, and processes fulfillment pipelines asynchronously with automatic backoff.

Deploy your Shopify webhook on LiteLambda in 60s →

Setting Up the Webhook in Shopify Admin

You do not need to create a custom Shopify App to start receiving webhooks. You can set them up directly from the Shopify Admin settings:

Step 1: Deploy to LiteLambda

pip install litelambda-cli
litelambda login

litelambda deploy shopify-orders order_handler.py \
  --type webhook \
  --env SHOPIFY_WEBHOOK_SECRET="your_shopify_secret_here" \
  --env SHIPROCKET_API_TOKEN="ey..." \
  --env SLACK_WEBHOOK_URL="https://hooks.slack.com/services/..."

LiteLambda gives you your live HTTPS endpoint URL:

https://run.litelambda.in/w/shopify-orders/wh_live_90f84a3b

Step 2: Register in Shopify Settings

1. In your Shopify Store Admin, navigate to Settings (bottom left gear icon).
2. Click Notifications → scroll down to the Webhooks section.
3. Click Create webhook.
4. Configure the settings:
5. Event: Order creation (or Order payment).
6. Format: JSON.
7. URL: Paste your LiteLambda HTTPS endpoint URL.
8. Webhook API version: Latest (e.g. 2026-10).
9. Click Save.
10. At the bottom of the Webhooks card, locate the text:

    "All your webhooks will be signed with: XXXXXXXXXXXXXXXX"

11. Copy this string and set it as your SHOPIFY_WEBHOOK_SECRET environment variable in LiteLambda.

Step 3: Send a Test Notification

Shopify provides a Send test notification button next to your webhook. Click it to trigger an immediate test order. Open the LiteLambda dashboard to see the verified payload and fulfillment logs.

Troubleshooting Common Shopify Webhook Pitfalls

Summary

Handling e-commerce orders requires rock-solid reliability. A single missed webhook can result in customer frustration, delayed deliveries, and manual intervention.

By utilizing Python's native HMAC verification with LiteLambda's zero-maintenance serverless platform, your online store can automate order fulfillment, logistics routing, and customer notifications with total confidence.

Deploy your Shopify webhook on LiteLambda →