How to Build a WhatsApp Order Tracking Bot for E-Commerce

Customer sends a WhatsApp message to your store's number
  -> Chatmaid posts a signed message.received webhook to your server (or n8n)
  -> your code finds the order number in the text
  -> it looks the order up in Shopify, WooCommerce or your database
  -> it checks the order belongs to the sender and writes the reply
  -> POST /v1/messages/send delivers the reply from your WhatsApp number

pip install flask requests phonenumbers

order-bot/
├── chatmaid.py             # send messages, verify Chatmaid webhooks
├── parsing.py              # find the order number in a message
├── lookup_shopify.py       # or lookup_woocommerce.py, or lookup_sqlite.py
├── ownership.py            # check the order belongs to the sender
├── replies.py              # turn an order into a WhatsApp reply
├── handoff.py              # returns, problems and human takeover
├── notifications.py        # proactive Shopify notifications
└── app.py                  # the Flask server

export CHATMAID_API_KEY="sk_live_xxxxxxxxxxxxxxxxxxxxxxxx"
export CHATMAID_SENDER="+15551234567"
export CHATMAID_WEBHOOK_SECRET="whsec_xxxxxxxxxxxxxxxxxxxxxxxx"
export SUPPORT_GROUP_ID="120363043211234567@g.us"
export SHOPIFY_STORE="your-store"
export SHOPIFY_ACCESS_TOKEN="shpat_xxxxxxxxxxxxxxxxxxxxxxxx"
export SHOPIFY_WEBHOOK_SECRET="your_shopify_app_client_secret"

import hashlib
import hmac
import os
import time

import requests

API_BASE = "https://developers-api.chatmaid.net/v1"
API_KEY = os.environ["CHATMAID_API_KEY"]
SENDER = os.environ["CHATMAID_SENDER"]  # your connected number, e.g. +15551234567
WEBHOOK_SECRET = os.environ["CHATMAID_WEBHOOK_SECRET"]
TOLERANCE_SECONDS = 15 * 60  # retried deliveries keep the original timestamp


def send_whatsapp(to: str, text: str, idempotency_key: str) -> dict:
    """Send a text from your number. `to` is an E.164 number or a group ID."""
    response = requests.post(
        f"{API_BASE}/messages/send",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "fromPhoneId": SENDER,
            "to": to,
            "content": text,
            "idempotencyKey": idempotency_key[:64],
        },
        timeout=30,
    )
    if not response.ok:
        raise RuntimeError(
            f"Chatmaid send failed ({response.status_code}): {response.text[:300]}"
        )
    return response.json()["data"]


def is_valid_signature(raw_body: bytes, header: str) -> bool:
    """Check the X-Chatmaid-Signature header, which looks like t=<unix time>,v1=<hex>."""
    try:
        parts = dict(item.strip().split("=", 1) for item in header.split(","))
        timestamp = int(parts["t"])
        provided = parts["v1"]
    except (KeyError, ValueError):
        return False
    if abs(time.time() - timestamp) > TOLERANCE_SECONDS:
        return False
    signed = parts["t"].encode() + b"." + raw_body
    expected = hmac.new(WEBHOOK_SECRET.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(provided.encode(), expected.encode())

import re

ORDER_PATTERNS = [
    r"#\s*(\d{3,10})\b",  # "#4892"
    r"\border\s*(?:number|no\.?)?\s*:?\s*(\d{3,10})\b",  # "order 4892", "order number: 4892"
    r"\b(\d{4,10})\b",  # a bare number, as a last resort
]


def extract_order_number(text: str) -> str | None:
    for pattern in ORDER_PATTERNS:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            return match.group(1)
    return None

import os

import requests

SHOP = os.environ["SHOPIFY_STORE"]  # the part before .myshopify.com
TOKEN = os.environ["SHOPIFY_ACCESS_TOKEN"]
API_VERSION = "2026-10"

QUERY = """
query FindOrder($search: String!) {
  orders(first: 5, query: $search) {
    nodes {
      name
      phone
      cancelledAt
      shippingAddress { phone countryCodeV2 }
      lineItems(first: 3) { nodes { title quantity } }
      fulfillments(first: 10) {
        createdAt
        displayStatus
        estimatedDeliveryAt
        trackingInfo(first: 1) { company number url }
      }
    }
  }
}
"""

# Shopify's fulfillment display status -> the bot's status
STATUS = {
    "DELIVERED": "delivered",
    "PICKED_UP": "delivered",
    "OUT_FOR_DELIVERY": "out_for_delivery",
    "IN_TRANSIT": "in_transit",
    "CARRIER_PICKED_UP": "in_transit",
    "READY_FOR_PICKUP": "ready_for_pickup",
    "ATTEMPTED_DELIVERY": "problem",
    "DELAYED": "problem",
    "FAILURE": "problem",
    "NOT_DELIVERED": "problem",
}


def find_order(order_number: str) -> dict | None:
    response = requests.post(
        f"https://{SHOP}.myshopify.com/admin/api/{API_VERSION}/graphql.json",
        headers={"X-Shopify-Access-Token": TOKEN},
        json={"query": QUERY, "variables": {"search": f'name:"#{order_number}"'}},
        timeout=15,
    )
    response.raise_for_status()
    payload = response.json()
    if not payload.get("data"):
        raise RuntimeError(f"Shopify error: {payload.get('errors')}")

    # Keep only an exact match, never a near match from the search
    wanted = f"#{order_number}"
    order = next(
        (o for o in payload["data"]["orders"]["nodes"] if o["name"] == wanted), None
    )
    if order is None:
        return None

    shipments = [
        f for f in order["fulfillments"]
        if f["displayStatus"] not in ("CANCELED", "LABEL_VOIDED")
    ]
    latest = max(shipments, key=lambda f: f["createdAt"], default=None)
    tracking = (latest["trackingInfo"] or [{}])[0] if latest else {}
    address = order["shippingAddress"] or {}

    if order["cancelledAt"]:
        status = "cancelled"
    elif latest is None:
        status = "processing"
    else:
        status = STATUS.get(latest["displayStatus"], "shipped")

    return {
        "number": order["name"],
        "status": status,
        "items": [f"{i['title']} x{i['quantity']}" for i in order["lineItems"]["nodes"]],
        "carrier": tracking.get("company"),
        "tracking_number": tracking.get("number"),
        "tracking_url": tracking.get("url"),
        "eta": latest.get("estimatedDeliveryAt") if latest else None,
        "phones": [p for p in (address.get("phone"), order["phone"]) if p],
        "country": address.get("countryCodeV2"),
    }

import os

import requests

STORE_URL = os.environ["WOO_STORE_URL"]  # e.g. https://your-store.com
AUTH = (os.environ["WOO_CONSUMER_KEY"], os.environ["WOO_CONSUMER_SECRET"])

# WooCommerce order status -> the bot's status
STATUS = {
    "pending": "awaiting_payment",
    "on-hold": "awaiting_payment",
    "failed": "awaiting_payment",
    "processing": "processing",
    "completed": "shipped",
    "cancelled": "cancelled",
    "refunded": "refunded",
}


def find_order(order_number: str) -> dict | None:
    response = requests.get(
        f"{STORE_URL}/wp-json/wc/v3/orders/{order_number}",
        auth=AUTH,
        timeout=15,
    )
    if response.status_code == 404:
        return None
    response.raise_for_status()
    order = response.json()
    billing = order.get("billing") or {}
    return {
        "number": f"#{order['number']}",
        "status": STATUS.get(order["status"], order["status"]),
        "items": [f"{i['name']} x{i['quantity']}" for i in order["line_items"][:3]],
        "carrier": None,  # WooCommerce core has no tracking fields, see below
        "tracking_number": None,
        "tracking_url": None,
        "eta": None,
        "phones": [billing["phone"]] if billing.get("phone") else [],
        "country": billing.get("country") or None,
    }

CREATE TABLE orders (
  number TEXT PRIMARY KEY,
  status TEXT NOT NULL,   -- processing, shipped, in_transit, delivered, ...
  items TEXT,             -- "Linen shirt x2;Canvas tote x1"
  carrier TEXT,
  tracking_number TEXT,
  tracking_url TEXT,
  eta TEXT,               -- ISO date, e.g. 2026-10-24
  phone TEXT,             -- as the customer typed it at checkout
  country TEXT            -- two-letter country code, e.g. US
);

import os
import sqlite3
from contextlib import closing

DB_PATH = os.environ.get("ORDERS_DB", "orders.db")


def find_order(order_number: str) -> dict | None:
    with closing(sqlite3.connect(DB_PATH)) as conn:
        conn.row_factory = sqlite3.Row
        row = conn.execute(
            "SELECT * FROM orders WHERE number = ?", (order_number,)
        ).fetchone()
    if row is None:
        return None
    return {
        "number": f"#{row['number']}",
        "status": row["status"],
        "items": row["items"].split(";") if row["items"] else [],
        "carrier": row["carrier"],
        "tracking_number": row["tracking_number"],
        "tracking_url": row["tracking_url"],
        "eta": row["eta"],
        "phones": [row["phone"]] if row["phone"] else [],
        "country": row["country"],
    }

import re

import phonenumbers


def to_e164(raw: str | None, country: str | None) -> str | None:
    """Turn a phone number as typed at checkout into E.164, or None."""
    if not raw:
        return None
    try:
        number = phonenumbers.parse(raw, country or None)
    except phonenumbers.NumberParseException:
        return None
    if not phonenumbers.is_possible_number(number):
        return None
    return phonenumbers.format_number(number, phonenumbers.PhoneNumberFormat.E164)


def belongs_to(order: dict, whatsapp_number: str) -> bool:
    """True when one of the order's phone numbers is the sender's WhatsApp number."""
    sender = re.sub(r"\D", "", whatsapp_number)
    for raw in order["phones"]:
        e164 = to_e164(raw, order["country"])
        if e164 and e164.lstrip("+") == sender:
            return True
    return False

TEMPLATES = {
    "processing": "Order {number} is confirmed and being prepared. We'll message you here as soon as it ships.",
    "awaiting_payment": "Order {number} is waiting for payment. Once it's confirmed, we'll start preparing it.",
    "shipped": "Order {number} has shipped.{tracking}{eta}",
    "in_transit": "Order {number} is on its way.{tracking}{eta}",
    "out_for_delivery": "Order {number} is out for delivery today.{tracking}",
    "ready_for_pickup": "Order {number} is ready for pickup.",
    "delivered": (
        "Order {number} was delivered. If it hasn't reached you or something is wrong, "
        "reply *PROBLEM* and someone from our team will help."
    ),
    "problem": (
        "The carrier reports a delivery issue with order {number}.{tracking}\n\n"
        "Reply *PROBLEM* and someone from our team will sort it out with you."
    ),
    "cancelled": "Order {number} was cancelled. Questions about a refund? Reply *REFUND*.",
    "refunded": "Order {number} was refunded. Questions? Reply *AGENT* to reach our team.",
}
FALLBACK = "Order {number} status: {status}. Reply *AGENT* to reach our team."


def format_order_reply(order: dict) -> str:
    tracking = ""
    if order.get("tracking_number"):
        tracking = (
            f"\n\nCarrier: {order.get('carrier') or 'see link'}"
            f"\nTracking number: {order['tracking_number']}"
        )
        if order.get("tracking_url"):
            tracking += f"\nTrack it: {order['tracking_url']}"
    eta = f"\nExpected by: {order['eta'][:10]}" if order.get("eta") else ""

    template = TEMPLATES.get(order["status"], FALLBACK)
    text = template.format(
        number=order["number"], status=order["status"], tracking=tracking, eta=eta
    )
    if order["items"]:
        text += "\n\nItems: " + ", ".join(order["items"][:3])
    return text

Order #1042 is on its way.

Carrier: UPS
Tracking number: 1Z999AA10123456784
Track it: https://www.ups.com/track?tracknum=1Z999AA10123456784
Expected by: 2026-10-24

Items: Linen shirt x2, Canvas tote x1

import os
import re
import time

from chatmaid import send_whatsapp

SUPPORT_GROUP = os.environ["SUPPORT_GROUP_ID"]  # e.g. 120363043211234567@g.us
PAUSE_SECONDS = 2 * 60 * 60

COMMANDS = {"AGENT", "PROBLEM", "REFUND"}
RETURN_WORDS = re.compile(r"\b(return|exchange|refund|damaged|defective|wrong item)\b", re.I)

_paused_until: dict[str, float] = {}  # customer number -> unix time


def wants_handoff(text: str) -> bool:
    return text.strip().strip("*").upper() in COMMANDS or bool(RETURN_WORDS.search(text))


def pause_bot(customer: str) -> None:
    _paused_until[customer] = time.time() + PAUSE_SECONDS


def bot_is_paused(customer: str) -> bool:
    return _paused_until.get(customer, 0) > time.time()


def hand_off(customer: str, text: str, order_number: str | None, message_id: str) -> str:
    case_id = message_id[-6:].upper()
    send_whatsapp(
        SUPPORT_GROUP,
        f"Case {case_id} from {customer}\n"
        f"Order: {order_number or 'not given'}\n"
        f"Message: {text}",
        idempotency_key=f"case-{message_id}",
    )
    pause_bot(customer)
    return (
        f"Thanks. I've passed this to our team (case {case_id}). "
        "A person will reply here as soon as they can."
    )

import json
import threading

from flask import Flask, abort, request

from chatmaid import is_valid_signature, send_whatsapp
from handoff import bot_is_paused, hand_off, pause_bot, wants_handoff
from lookup_shopify import find_order  # or lookup_woocommerce / lookup_sqlite
from ownership import belongs_to
from parsing import extract_order_number
from replies import format_order_reply

app = Flask(__name__)
seen_messages: set[str] = set()

ASK_FOR_NUMBER = (
    "Hi! I can check your order for you. Please send your order number. "
    "It's in your confirmation email, for example #1042."
)
NOT_FOUND = (
    "I couldn't find order #{number} for this WhatsApp number. "
    "Please check the number, or reply *AGENT* to reach our team."
)
UNAVAILABLE = (
    "Sorry, I can't check orders right now. "
    "Please try again in a few minutes, or reply *AGENT* to reach our team."
)


@app.post("/webhooks/chatmaid")
def chatmaid_webhook():
    raw_body = request.get_data()
    if not is_valid_signature(raw_body, request.headers.get("X-Chatmaid-Signature", "")):
        abort(401)
    event = json.loads(raw_body)
    threading.Thread(target=handle_event, args=(event,), daemon=True).start()
    return "", 200


def handle_event(event: dict) -> None:
    data = event.get("data") or {}
    if data.get("isGroup"):
        return  # only answer direct chats
    if event.get("event") == "message.outgoing":
        pause_bot(data["to"])  # a person replied from the phone, so step back
        return
    if event.get("event") != "message.received" or data["messageId"] in seen_messages:
        return
    seen_messages.add(data["messageId"])
    if bot_is_paused(data["from"]):
        return

    try:
        reply = build_reply(data)
    except Exception:
        app.logger.exception("Could not build a reply")
        reply = UNAVAILABLE
    sent = send_whatsapp(data["from"], reply, idempotency_key=f"reply-{data['messageId']}")
    print("reply sent:", sent["id"], sent["status"])


def build_reply(data: dict) -> str:
    customer = data["from"]
    text = (data.get("content") or "").strip()
    order_number = extract_order_number(text)

    if wants_handoff(text):
        return hand_off(customer, text, order_number, data["messageId"])
    if order_number is None:
        return ASK_FOR_NUMBER

    order = find_order(order_number)
    if order is None or not belongs_to(order, customer):
        return NOT_FOUND.format(number=order_number)
    return format_order_reply(order)


if __name__ == "__main__":
    app.run(port=3000)

import base64
import hashlib
import hmac
import json
import os
import threading

from flask import Blueprint, abort, request

from chatmaid import send_whatsapp
from ownership import to_e164

SHOPIFY_WEBHOOK_SECRET = os.environ["SHOPIFY_WEBHOOK_SECRET"]

shopify_bp = Blueprint("shopify", __name__)


def verified_shopify_json() -> dict:
    raw_body = request.get_data()
    digest = hmac.new(SHOPIFY_WEBHOOK_SECRET.encode(), raw_body, hashlib.sha256).digest()
    header = request.headers.get("X-Shopify-Hmac-SHA256", "")
    if not hmac.compare_digest(base64.b64encode(digest), header.encode()):
        abort(401)
    return json.loads(raw_body)


def send_in_background(to: str | None, text: str, key: str) -> None:
    if to:  # skip orders without a usable phone number
        threading.Thread(target=send_whatsapp, args=(to, text, key), daemon=True).start()


@shopify_bp.post("/webhooks/shopify/orders-create")
def order_created():
    order = verified_shopify_json()
    address = order.get("shipping_address") or {}
    phone = to_e164(address.get("phone") or order.get("phone"), address.get("country_code"))
    items = ", ".join(f"{i['title']} x{i['quantity']}" for i in order["line_items"][:3])
    send_in_background(
        phone,
        f"Thanks for your order! {order['name']} is confirmed.\n"
        f"Items: {items}\n\n"
        f"We'll message you here when it ships. Send {order['name']} anytime to check on it.",
        f"shopify-order-{order['id']}",
    )
    return "", 200


@shopify_bp.post("/webhooks/shopify/fulfillments-create")
def fulfillment_created():
    shipment = verified_shopify_json()
    address = shipment.get("destination") or {}
    phone = to_e164(address.get("phone"), address.get("country_code"))
    order_name = shipment["name"].rsplit(".", 1)[0]  # "#1042.1" -> "#1042"

    text = f"Good news: your order {order_name} has shipped."
    if shipment.get("tracking_number"):
        text += (
            f"\n\nCarrier: {shipment.get('tracking_company') or 'see link'}"
            f"\nTracking number: {shipment['tracking_number']}"
        )
    if shipment.get("tracking_url"):
        text += f"\nTrack it: {shipment['tracking_url']}"
    text += f"\n\nSend {order_name} here anytime for an update."

    send_in_background(phone, text, f"shopify-fulfillment-{shipment['id']}")
    return "", 200

from notifications import shopify_bp

app.register_blueprint(shopify_bp)

python -c "
import sqlite3
conn = sqlite3.connect('orders.db')
conn.execute('CREATE TABLE IF NOT EXISTS orders (number TEXT PRIMARY KEY, status TEXT NOT NULL, items TEXT, carrier TEXT, tracking_number TEXT, tracking_url TEXT, eta TEXT, phone TEXT, country TEXT)')
conn.execute('INSERT OR REPLACE INTO orders VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', ('1042', 'in_transit', 'Linen shirt x2;Canvas tote x1', 'UPS', '1Z999AA10123456784', 'https://www.ups.com/track?tracknum=1Z999AA10123456784', '2026-10-24', '(555) 765-4321', 'US'))
conn.commit()
"

SECRET="whsec_xxxxxxxxxxxxxxxxxxxxxxxx"  # the same value as CHATMAID_WEBHOOK_SECRET on the server
BODY='{"event":"message.received","timestamp":"2026-10-22T12:00:00.000Z","data":{"messageId":"inmsg_localtest0001","from":"+15557654321","to":"+15551234567","content":"Where is order #1042?","type":"text","isGroup":false,"groupId":null,"receivedAt":"2026-10-22T12:00:00.000Z"}}'
T=$(date +%s)
SIG=$(printf '%s' "$T.$BODY" | openssl dgst -sha256 -hmac "$SECRET" | sed 's/^.* //')

curl -X POST http://localhost:3000/webhooks/chatmaid \
  -H "Content-Type: application/json" \
  -H "X-Chatmaid-Event: message.received" \
  -H "X-Chatmaid-Signature: t=$T,v1=$SIG" \
  -d "$BODY"

const data = $input.first().json.body.data;
const text = data.content || '';
const match =
  text.match(/#\s*(\d{3,10})\b/) ||
  text.match(/\border\s*(?:number|no\.?)?\s*:?\s*(\d{3,10})\b/i) ||
  text.match(/\b(\d{4,10})\b/);

return [{ json: { ...data, orderNumber: match ? match[1] : null } }];

{
  "query": "query FindOrder($search: String!) { orders(first: 5, query: $search) { nodes { name phone cancelledAt shippingAddress { phone } fulfillments(first: 10) { createdAt displayStatus trackingInfo(first: 1) { company number url } } } } }",
  "variables": { "search": {{ JSON.stringify('name:"#' + $json.orderNumber + '"') }} }
}

const msg = $('Find Order Number').first().json;
const wanted = '#' + msg.orderNumber;
const orders = $input.first().json.data?.orders?.nodes || [];
const order = orders.find((o) => o.name === wanted);

const sender = msg.from.replace(/\D/g, '');
const phones = order ? [order.phone, order.shippingAddress?.phone].filter(Boolean) : [];
const owns = phones.some((p) => {
  const digits = p.replace(/\D/g, '').replace(/^0+/, '');
  return digits.length >= 7 && sender.endsWith(digits);
});

let reply;
if (!order || !owns) {
  reply = `I couldn't find order ${wanted} for this WhatsApp number. Please check the number, or reply *AGENT* to reach our team.`;
} else {
  const shipments = order.fulfillments
    .filter((f) => !['CANCELED', 'LABEL_VOIDED'].includes(f.displayStatus))
    .sort((a, b) => a.createdAt.localeCompare(b.createdAt));
  const latest = shipments[shipments.length - 1];
  const t = latest?.trackingInfo?.[0];
  const tracking = t?.number
    ? `\n\nCarrier: ${t.company || 'see link'}\nTracking number: ${t.number}` + (t.url ? `\nTrack it: ${t.url}` : '')
    : '';

  if (order.cancelledAt) {
    reply = `Order ${wanted} was cancelled. Questions about a refund? Reply *AGENT*.`;
  } else if (!latest) {
    reply = `Order ${wanted} is confirmed and being prepared. We'll message you here as soon as it ships.`;
  } else if (latest.displayStatus === 'DELIVERED') {
    reply = `Order ${wanted} was delivered. If something is wrong, reply *AGENT* and someone from our team will help.`;
  } else {
    reply = `Order ${wanted} has shipped.${tracking}`;
  }
}

return [{ json: { reply } }];