How to Build a WhatsApp AI Customer Support Bot with Business API Integration

whatsapp-support-bot/
├── app/
│   ├── __init__.py
│   ├── webhook_security.py
│   ├── parser_dispatcher.py
│   └── main.py
├── tests/
│   └── test_whatsapp_bot.py
└── requirements.txt

import hmac
import hashlib

def verify_meta_webhook_challenge(
    mode: str | None,
    token: str | None,
    challenge: str | None,
    expected_verify_token: str,
) -> str:
    """Validates Meta's GET /webhook subscription verification challenge."""
    if mode == "subscribe" and token and hmac.compare_digest(token, expected_verify_token):
        if challenge is None:
            raise ValueError("Missing hub.challenge parameter.")
        return challenge
    raise PermissionError("Webhook verification token mismatch.")

def verify_meta_hmac_signature(
    raw_body: bytes,
    signature_header: str | None,
    app_secret: str,
) -> bool:
    """Verifies X-Hub-Signature-256 header using constant-time HMAC-SHA256 comparison."""
    if not signature_header or not signature_header.startswith("sha256="):
        return False
    expected_sig = hmac.new(
        key=app_secret.encode("utf-8"),
        msg=raw_body,
        digestmod=hashlib.sha256,
    ).hexdigest()
    received_sig = signature_header.split("sha256=", 1)[1]
    return hmac.compare_digest(expected_sig, received_sig)

from typing import Dict, Any, Optional, Set
from pydantic import BaseModel

class InboundWhatsAppMessage(BaseModel):
    wamid: str
    sender_phone: str
    timestamp_unix: int
    message_type: str
    text_body: Optional[str] = None
    media_id: Optional[str] = None

def extract_inbound_message(
    webhook_payload: Dict[str, Any],
    processed_wamids: Set[str],
) -> Optional[InboundWhatsAppMessage]:
    """Extracts a new user message while ignoring delivery/read status updates and duplicate wamids."""
    try:
        entries = webhook_payload.get("entry", [])
        if not entries:
            return None
        changes = entries[0].get("changes", [])
        if not changes:
            return None
        value = changes[0].get("value", {})

        # Ignore outbound delivery/read status receipts
        if "messages" not in value or not value["messages"]:
            return None

        msg = value["messages"][0]
        wamid = msg.get("id", "")
        if not wamid or wamid in processed_wamids:
            return None

        processed_wamids.add(wamid)
        msg_type = msg.get("type", "unknown")
        text_body = msg.get("text", {}).get("body") if msg_type == "text" else None
        media_id = msg.get(msg_type, {}).get("id") if msg_type in ("audio", "document", "image") else None

        return InboundWhatsAppMessage(
            wamid=wamid,
            sender_phone=msg.get("from", ""),
            timestamp_unix=int(msg.get("timestamp", 0)),
            message_type=msg_type,
            text_body=text_body,
            media_id=media_id,
        )
    except (KeyError, IndexError, ValueError):
        return None

def build_outbound_whatsapp_payload(
    recipient_phone: str,
    reply_text: str,
    last_inbound_unix: int,
    current_unix: int,
    fallback_template_name: str = "support_followup_update",
) -> Dict[str, Any]:
    """Builds a free-form text payload if within 24 hours (86,400s), else switches to a Template message."""
    elapsed_seconds = current_unix - last_inbound_unix
    if elapsed_seconds < 86400:
        return {
            "messaging_product": "whatsapp",
            "recipient_type": "individual",
            "to": recipient_phone,
            "type": "text",
            "text": {"preview_url": False, "body": reply_text[:4096]},
        }
    return {
        "messaging_product": "whatsapp",
        "to": recipient_phone,
        "type": "template",
        "template": {
            "name": fallback_template_name,
            "language": {"code": "en_US"},
            "components": [
                {
                    "type": "body",
                    "parameters": [{"type": "text", "text": reply_text[:1024]}],
                }
            ],
        },
    }

import hmac
import hashlib
import pytest
from app.webhook_security import verify_meta_webhook_challenge, verify_meta_hmac_signature
from app.parser_dispatcher import extract_inbound_message, build_outbound_whatsapp_payload

def test_meta_hmac_signature_validation():
    secret = "my_meta_app_secret"
    body = b'{"object":"whatsapp_business_account"}'
    valid_sig = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()

    assert verify_meta_hmac_signature(body, valid_sig, secret) is True
    assert verify_meta_hmac_signature(body, "sha256=deadbeef", secret) is False

def test_webhook_deduplication_and_status_filtering():
    seen = set()
    status_payload = {"entry": [{"changes": [{"value": {"statuses": [{"id": "wamid.1", "status": "delivered"}]}}]}]}
    assert extract_inbound_message(status_payload, seen) is None

    msg_payload = {
        "entry": [{
            "changes": [{
                "value": {
                    "messages": [{
                        "id": "wamid.abc123",
                        "from": "919876543210",
                        "timestamp": "1761800000",
                        "type": "text",
                        "text": {"body": "Where is my order?"},
                    }]
                }
            }]
        }]
    }
    first = extract_inbound_message(msg_payload, seen)
    assert first is not None
    assert first.wamid == "wamid.abc123"
    # Second delivery of same wamid must be ignored
    assert extract_inbound_message(msg_payload, seen) is None

def test_24_hour_window_switches_to_template():
    within_24h = build_outbound_whatsapp_payload("919876543210", "Your order shipped!", 1000, 1000 + 3600)
    assert within_24h["type"] == "text"

    after_24h = build_outbound_whatsapp_payload("919876543210", "Your order shipped!", 1000, 1000 + 90000)
    assert after_24h["type"] == "template"