← Back to list

Part 3: WhatsApp Webhooks — Receiving Messages and Processing Events in Real Time

Introduction

LOKESH KUMAR · 2026-06-24 18:31 · 0 claps · 11.5 min read
#whatsapp-webhook #python #artificial-intelligence #meta-api #chatbot-development
Open on Medium ↗
Wiki topics: AI · AI · General

Part 3: WhatsApp Webhooks — Receiving Messages and Processing Events in Real Time

Introduction

In Part 2, we covered every outgoing message type available on the WhatsApp Cloud API. You now know how to send text, media, interactive buttons, lists, flows, and templates.

But a chatbot that can only send messages is useless. You need to receive messages too.

That is what webhooks do. They are the backbone of every WhatsApp chatbot. Every incoming customer message, every message status update, every account alert — they all arrive at your server as webhook events.

In this article, we will build a complete production-grade webhook handler in Python using Flask. By the end, you will have a working server that receives, validates, and processes every type of WhatsApp webhook event.

How Webhooks Work

The standard request-response cycle breaks down for messaging applications. You cannot poll the WhatsApp API every few seconds asking “did anyone send a message?” — that would be slow, expensive, and against Meta’s usage policies.

Webhooks solve this with an event-driven approach:

Customer sends WhatsApp message
             │
             ▼
WhatsApp Servers receive message
             │
             ▼
Meta sends HTTP POST to your webhook URL
             │
             ▼
Your server receives and processes the event
             │
             ▼
Your server sends a reply via Cloud API
             │
             ▼
Customer receives your response

Your server must:

  1. Be accessible over public HTTPS
  2. Respond to verification GET requests from Meta
  3. Process incoming POST events quickly (within 5 seconds)
  4. Return HTTP 200 to acknowledge every event

Project Setup

Let us build this step by step.

Install Dependencies

pip install flask requests python-dotenv

Project Structure

whatsapp_webhook/
├── app.py
├── webhook_handler.py
├── whatsapp_client.py
├── .env
└── requirements.txt

Environment Variables (.env)

WHATSAPP_ACCESS_TOKEN=your_access_token_here
WHATSAPP_PHONE_NUMBER_ID=your_phone_number_id_here
WHATSAPP_WEBHOOK_VERIFY_TOKEN=your_custom_verify_token_here
WHATSAPP_APP_SECRET=your_app_secret_here

Step 1 — Webhook Verification

When you register a webhook URL in the Meta App Dashboard, Meta sends a verification GET request to confirm you own the endpoint.

Your server must respond correctly or the webhook registration will fail.

Meta sends:

GET https://your-server.com/webhook
  ?hub.mode=subscribe
  &hub.verify_token=YOUR_VERIFY_TOKEN
  &hub.challenge=RANDOM_STRING_FROM_META

Your server must respond with:

  • HTTP status 200
  • The hub.challenge value as the response body

Step 1 — Flask Verification Handler

# app.py
from flask import Flask, request, abort, jsonify
import os
from dotenv import load_dotenv
from webhook_handler import process_webhook_event
load_dotenv()
app = Flask(__name__)
VERIFY_TOKEN = os.environ.get("WHATSAPP_WEBHOOK_VERIFY_TOKEN")

@app.route("/webhook", methods=["GET"])
def verify_webhook():
    """
    Handle Meta webhook verification challenge.
    Called once when you register your webhook URL.
    """
    mode = request.args.get("hub.mode")
    token = request.args.get("hub.verify_token")
    challenge = request.args.get("hub.challenge")
    if mode == "subscribe" and token == VERIFY_TOKEN:
        print("Webhook verified successfully")
        return challenge, 200
    else:
        print(f"Webhook verification failed. Token mismatch.")
        abort(403)

@app.route("/webhook", methods=["POST"])
def receive_webhook():
    """
    Receive and process all incoming webhook events.
    Must return 200 quickly - process events asynchronously
    for production workloads.
    """
    data = request.get_json()

    if data:
        process_webhook_event(data)

    return jsonify({"status": "ok"}), 200

if __name__ == "__main__":
    app.run(debug=True, port=5000)

Step 2 — Validating Webhook Signatures

Meta signs every webhook POST request using your app secret. Validating this signature ensures the request actually came from Meta and not from an attacker.

# security.py
import hmac
import hashlib
from flask import request, abort

def validate_webhook_signature(app_secret: str) -> bool:
    """
    Validate that the incoming webhook request is from Meta.
    Meta signs requests using HMAC-SHA256 with your app secret.
    """
    signature_header = request.headers.get("X-Hub-Signature-256", "")

    if not signature_header:
        return False

    # Remove the 'sha256=' prefix
    expected_signature = signature_header.replace("sha256=", "")

    # Compute HMAC
    body = request.get_data()
    computed_signature = hmac.new(
        key=app_secret.encode("utf-8"),
        msg=body,
        digestmod=hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(computed_signature, expected_signature)

Updated webhook POST handler with signature validation:

import os
from security import validate_webhook_signature
APP_SECRET = os.environ.get("WHATSAPP_APP_SECRET")
@app.route("/webhook", methods=["POST"])
def receive_webhook():
    if not validate_webhook_signature(APP_SECRET):
        print("Invalid webhook signature - request rejected")
        abort(403)

    data = request.get_json()

    if data:
        process_webhook_event(data)

    return jsonify({"status": "ok"}), 200

Step 3 — Understanding the Webhook Payload Structure

Every webhook payload from Meta follows this structure:

{
  "object": "whatsapp_business_account",
  "entry": [
    {
      "id": "<WABA_ID>",
      "changes": [
        {
          "value": {
            "messaging_product": "whatsapp",
            "metadata": {
              "display_phone_number": "15550783881",
              "phone_number_id": "106540352242922"
            },
            "contacts": [ ... ],
            "messages": [ ... ],
            "statuses": [ ... ]
          },
          "field": "messages"
        }
      ]
    }
  ]
}

The field property tells you what type of event this is:

Step 4 — Processing Incoming Messages

Different message types arrive with different payload structures. Your handler needs to route each type correctly.

# webhook_handler.py
from whatsapp_client import WhatsAppClient
client = WhatsAppClient()

def process_webhook_event(data: dict):
    """
    Main entry point for all incoming webhook events.
    Routes to appropriate handler based on event type.
    """

    if data.get("object") != "whatsapp_business_account":
        return

    entries = data.get("entry", [])

    for entry in entries:
        changes = entry.get("changes", [])

        for change in changes:
            field = change.get("field")
            value = change.get("value", {})

            if field == "messages":
                handle_messages_field(value)
            elif field == "message_template_status_update":
                handle_template_status_update(value)
            elif field == "account_alerts":
                handle_account_alert(value)
            elif field == "phone_number_quality_update":
                handle_quality_update(value)

def handle_messages_field(value: dict):
    """
    Handle the 'messages' webhook field.
    Can contain incoming messages or status updates.
    """

    # Process incoming messages
    messages = value.get("messages", [])
    contacts = value.get("contacts", [])
    metadata = value.get("metadata", {})

    for message in messages:
        sender_phone = message.get("from")
        message_id = message.get("id")
        message_type = message.get("type")

        # Get sender name from contacts
        sender_name = ""
        for contact in contacts:
            if contact.get("wa_id") == sender_phone:
                sender_name = contact.get("profile", {}).get("name", "")
                break

        print(f"Message from {sender_name} ({sender_phone}): type={message_type}")

        route_incoming_message(
            message=message,
            sender_phone=sender_phone,
            sender_name=sender_name,
            message_id=message_id,
            message_type=message_type
        )

    # Process status updates
    statuses = value.get("statuses", [])
    for status in statuses:
        handle_message_status(status)

def route_incoming_message(
    message: dict,
    sender_phone: str,
    sender_name: str,
    message_id: str,
    message_type: str
):
    """
    Route incoming message to the correct handler
    based on message type.
    """

    handlers = {
        "text": handle_text_message,
        "image": handle_media_message,
        "video": handle_media_message,
        "audio": handle_media_message,
        "document": handle_document_message,
        "location": handle_location_message,
        "interactive": handle_interactive_message,
        "button": handle_button_message,
        "sticker": handle_sticker_message,
        "reaction": handle_reaction_message,
        "contacts": handle_contact_message,
    }

    handler = handlers.get(message_type, handle_unknown_message)

    handler(
        message=message,
        sender_phone=sender_phone,
        sender_name=sender_name,
        message_id=message_id
    )

Step 5 — Individual Message Type Handlers

Text Message Handler

def handle_text_message(
    message: dict,
    sender_phone: str,
    sender_name: str,
    message_id: str
):
    """Handle incoming text messages."""

    text_body = message.get("text", {}).get("body", "")

    print(f"Text message: '{text_body}' from {sender_phone}")

    # Route based on message content
    text_lower = text_body.lower().strip()

    if text_lower in ["hi", "hello", "hey", "start"]:
        send_welcome_message(sender_phone, sender_name)

    elif text_lower in ["help", "menu", "options"]:
        send_main_menu(sender_phone)

    elif text_lower in ["track", "order status"]:
        client.send_text(
            to=sender_phone,
            message="Please enter your order number (e.g., ORD-12345):"
        )

    elif text_lower.startswith("ord-"):
        handle_order_lookup(sender_phone, text_body)

    else:
        # Default fallback
        client.send_text(
            to=sender_phone,
            message=(
                f"Thanks {sender_name}! I received your message.\n\n"
                "Type *menu* to see what I can help you with."
            )
        )
def send_welcome_message(phone: str, name: str):
    """Send initial greeting with menu buttons."""

    client.send_buttons(
        to=phone,
        body_text=(
            f"Hello {name}! 👋 Welcome to our support channel.\n\n"
            "I'm here to help you with orders, returns, and more. "
            "What would you like to do today?"
        ),
        buttons=[
            {"id": "btn_track_order", "title": "Track My Order"},
            {"id": "btn_returns", "title": "Returns & Refunds"},
            {"id": "btn_speak_agent", "title": "Speak to Agent"}
        ]
    )

def send_main_menu(phone: str):
    """Send a full list menu."""

    sections = [
        {
            "title": "Orders",
            "rows": [
                {
                    "id": "track_order",
                    "title": "Track My Order",
                    "description": "Get real-time delivery updates"
                },
                {
                    "id": "cancel_order",
                    "title": "Cancel an Order",
                    "description": "Cancel before it ships"
                }
            ]
        },
        {
            "title": "Support",
            "rows": [
                {
                    "id": "returns",
                    "title": "Returns & Refunds",
                    "description": "Start a return or check refund status"
                },
                {
                    "id": "speak_agent",
                    "title": "Speak to Agent",
                    "description": "Connect with a human agent"
                }
            ]
        }
    ]

    client.send_list(
        to=phone,
        header_text="Support Menu",
        body_text="Choose a category to get started:",
        footer_text="We respond in under 5 minutes",
        button_label="View Options",
        sections=sections
    )

def handle_order_lookup(phone: str, order_number: str):
    """Look up order status and respond."""

    # In a real app, query your database or order management system here
    client.send_text(
        to=phone,
        message=(
            f"Order {order_number.upper()} Status:\n\n"
            "📦 Status: Shipped\n"
            "🚚 Carrier: FedEx\n"
            "📅 Expected Delivery: December 25, 2024\n\n"
            "Type *menu* to return to the main menu."
        )
    )

Interactive Message Handler

When users tap buttons or list items, the message type is interactive.

def handle_interactive_message(
    message: dict,
    sender_phone: str,
    sender_name: str,
    message_id: str
):
    """Handle interactive message responses (button taps, list selections)."""

    interactive = message.get("interactive", {})
    interactive_type = interactive.get("type")

    if interactive_type == "button_reply":
        # User tapped a quick reply button
        button_reply = interactive.get("button_reply", {})
        button_id = button_reply.get("id")
        button_title = button_reply.get("title")

        print(f"Button tapped: id={button_id}, title={button_title}")

        handle_button_reply(sender_phone, sender_name, button_id)

    elif interactive_type == "list_reply":
        # User selected from a list
        list_reply = interactive.get("list_reply", {})
        row_id = list_reply.get("id")
        row_title = list_reply.get("title")

        print(f"List item selected: id={row_id}, title={row_title}")

        handle_list_selection(sender_phone, sender_name, row_id)
def handle_button_reply(phone: str, name: str, button_id: str):
    """Route button responses to appropriate flows."""

    if button_id == "btn_track_order":
        client.send_text(
            to=phone,
            message="Please enter your order number (e.g., ORD-12345):"
        )

    elif button_id == "btn_returns":
        client.send_buttons(
            to=phone,
            body_text="What would you like to do with your return?",
            buttons=[
                {"id": "start_return", "title": "Start a Return"},
                {"id": "check_refund", "title": "Check Refund Status"},
                {"id": "back_menu", "title": "Back to Menu"}
            ]
        )

    elif button_id == "btn_speak_agent":
        client.send_text(
            to=phone,
            message=(
                "Connecting you with a live agent now. ⏳\n\n"
                "Average wait time: 3 minutes\n\n"
                "You can also reach us at support@example.com"
            )
        )

    elif button_id == "back_menu":
        send_main_menu(phone)

def handle_list_selection(phone: str, name: str, row_id: str):
    """Route list selection responses."""

    responses = {
        "track_order": "Please enter your order number (e.g., ORD-12345):",
        "cancel_order": (
            "To cancel an order, please provide your order number. "
            "Note: Orders can only be cancelled before they ship."
        ),
        "returns": "Let me pull up our returns process for you...",
        "speak_agent": "Connecting you with an agent now. Please hold..."
    }

    response_text = responses.get(
        row_id,
        "I received your selection. How can I help further?"
    )

    client.send_text(to=phone, message=response_text)

Media Message Handler

def handle_media_message(
    message: dict,
    sender_phone: str,
    sender_name: str,
    message_id: str
):
    """
    Handle incoming image, video, or audio messages.
    Download and store media within 7 days of receipt.
    """

    message_type = message.get("type")
    media_data = message.get(message_type, {})

    media_id = media_data.get("id")
    media_mime_type = media_data.get("mime_type")
    caption = media_data.get("caption", "")

    print(f"Media received: type={message_type}, id={media_id}")

    # IMPORTANT: Download media within 7 days
    # In production, trigger async download immediately
    download_media_async(media_id, media_mime_type)

    client.send_text(
        to=sender_phone,
        message=(
            f"Thanks {sender_name}! I received your {message_type}. "
            "Our team will review it and get back to you shortly."
        )
    )
def download_media_async(media_id: str, mime_type: str):
    """
    Download media from Meta servers.
    IMPORTANT: Media IDs expire after 7 days (as of October 2025).
    Always download and store immediately.
    """

    url = f"https://graph.facebook.com/v19.0/{media_id}"
    headers = {"Authorization": f"Bearer {os.environ.get('WHATSAPP_ACCESS_TOKEN')}"}

    # Step 1: Get the media download URL
    response = requests.get(url, headers=headers)
    media_info = response.json()
    download_url = media_info.get("url")

    if not download_url:
        print(f"Could not get download URL for media {media_id}")
        return

    # Step 2: Download the media file
    media_response = requests.get(download_url, headers=headers)

    if media_response.status_code == 200:
        # Step 3: Save to your storage system
        # In production: save to S3, GCS, or your storage service
        filename = f"media_{media_id}"
        print(f"Downloaded media: {filename} ({len(media_response.content)} bytes)")

        # For local testing, save to disk
        with open(f"downloads/{filename}", "wb") as f:
            f.write(media_response.content)

Location Message Handler

def handle_location_message(
    message: dict,
    sender_phone: str,
    sender_name: str,
    message_id: str
):
    """Handle incoming location shares."""

    location = message.get("location", {})
    latitude = location.get("latitude")
    longitude = location.get("longitude")
    name = location.get("name", "")
    address = location.get("address", "")

    print(f"Location received: {latitude}, {longitude}")

    # In production, use lat/long to find nearest store,
    # calculate delivery zones, or trigger logistics workflows

    client.send_text(
        to=sender_phone,
        message=(
            f"Got your location! 📍\n\n"
            f"Address: {address or 'Location received'}\n\n"
            "Our delivery team will use this to route your order. "
            "You will receive a confirmation shortly."
        )
    )

Document Message Handler

def handle_document_message(
    message: dict,
    sender_phone: str,
    sender_name: str,
    message_id: str
):
    """Handle incoming document uploads."""

    document = message.get("document", {})
    media_id = document.get("id")
    filename = document.get("filename", "document")
    mime_type = document.get("mime_type", "")

    print(f"Document received: {filename} (id={media_id})")

    # Download and store the document
    download_media_async(media_id, mime_type)

    client.send_text(
        to=sender_phone,
        message=(
            f"Thanks! I received your document: {filename}\n\n"
            "Our team will review it within 1 business day."
        )
    )

Step 6 — Processing Message Status Updates

Status updates tell you what happened to messages you sent.

def handle_message_status(status: dict):
    """
    Process outgoing message status updates.

    Status values:
    - sent: message left Meta's servers
    - delivered: message reached customer's device
    - read: customer opened the message
    - failed: delivery failed
    """

    message_id = status.get("id")
    status_value = status.get("status")
    timestamp = status.get("timestamp")
    recipient_id = status.get("recipient_id")

    print(f"Message {message_id} to {recipient_id}: {status_value}")

    if status_value == "sent":
        # Message left Meta servers
        # Update your database: message sent at timestamp
        pass

    elif status_value == "delivered":
        # Message reached customer device
        # Update your database: delivered at timestamp
        pass

    elif status_value == "read":
        # Customer read the message
        # Update your database: read at timestamp
        pass

    elif status_value == "failed":
        # Delivery failed
        errors = status.get("errors", [])
        for error in errors:
            error_code = error.get("code")
            error_message = error.get("message")
            print(f"Delivery failed: {error_code} — {error_message}")

        handle_failed_message(message_id, recipient_id, errors)
def handle_failed_message(message_id: str, recipient_id: str, errors: list):
    """Handle failed message delivery."""

    # Common error codes:
    # 131026 - Recipient phone number not on WhatsApp
    # 131047 - Business account outside the 24-hour window
    # 131021 - Recipient opted out

    for error in errors:
        code = error.get("code")

        if code == 131026:
            print(f"{recipient_id} is not on WhatsApp")
        elif code == 131047:
            print(f"24-hour window expired for {recipient_id}")
        elif code == 131021:
            print(f"{recipient_id} has opted out")
        else:
            print(f"Unknown error {code} for {recipient_id}")

Step 7 — Template and Account Event Handlers

def handle_template_status_update(value: dict):
    """Handle template approval and rejection events."""

    event = value.get("event")
    message_template_id = value.get("message_template_id")
    message_template_name = value.get("message_template_name")
    reason = value.get("reason", "")

    print(f"Template event: {event} for template '{message_template_name}'")

    if event == "APPROVED":
        print(f"Template '{message_template_name}' approved and ready to use")
        # Notify your team or trigger automated campaign

    elif event == "REJECTED":
        print(
            f"Template '{message_template_name}' rejected. "
            f"Reason: {reason}"
        )
        # Alert your template management team

    elif event == "DISABLED":
        print(
            f"Template '{message_template_name}' disabled due to "
            "negative customer feedback"
        )
        # Stop any scheduled sends using this template
def handle_account_alert(value: dict):
    """Handle account-level alerts."""

    alert_type = value.get("alert_type")

    print(f"Account alert: {alert_type}")

    if alert_type == "PAYMENT_ISSUE":
        # Billing issue - resolve to avoid service interruption
        pass

    elif alert_type == "PROFILE_PICTURE_LOST":
        # Profile picture removed from business account
        # Re-upload your business profile picture
        pass

def handle_quality_update(value: dict):
    """Handle phone number quality rating changes."""

    phone_number = value.get("display_phone_number")
    current_limit = value.get("current_limit")
    new_limit = value.get("new_limit")

    print(
        f"Quality update for {phone_number}: "
        f"{current_limit} → {new_limit}"
    )

    if new_limit == "RESTRICTED":
        # Number is now restricted
        # Pause campaigns, alert operations team
        print("ALERT: Phone number restricted - pausing outbound campaigns")

    elif new_limit == "FLAGGED":
        # Quality dropped - cannot upgrade tier
        print("WARNING: Phone number flagged - review template quality")

Step 8 — Complete webhook_handler.py

Here is the complete consolidated file:

# webhook_handler.py
import os
import requests
from whatsapp_client import WhatsAppClient
client = WhatsAppClient()

def process_webhook_event(data: dict):
    if data.get("object") != "whatsapp_business_account":
        return

    for entry in data.get("entry", []):
        for change in entry.get("changes", []):
            field = change.get("field")
            value = change.get("value", {})

            if field == "messages":
                handle_messages_field(value)
            elif field == "message_template_status_update":
                handle_template_status_update(value)
            elif field == "account_alerts":
                handle_account_alert(value)
            elif field == "phone_number_quality_update":
                handle_quality_update(value)

def handle_messages_field(value: dict):
    messages = value.get("messages", [])
    contacts = value.get("contacts", [])

    for message in messages:
        sender_phone = message.get("from")
        message_id = message.get("id")
        message_type = message.get("type")

        sender_name = next(
            (
                c.get("profile", {}).get("name", "")
                for c in contacts
                if c.get("wa_id") == sender_phone
            ),
            ""
        )

        route_incoming_message(
            message=message,
            sender_phone=sender_phone,
            sender_name=sender_name,
            message_id=message_id,
            message_type=message_type
        )

    for status in value.get("statuses", []):
        handle_message_status(status)

def route_incoming_message(
    message, sender_phone, sender_name, message_id, message_type
):
    handlers = {
        "text": handle_text_message,
        "image": handle_media_message,
        "video": handle_media_message,
        "audio": handle_media_message,
        "document": handle_document_message,
        "location": handle_location_message,
        "interactive": handle_interactive_message,
        "sticker": handle_sticker_message,
        "reaction": handle_reaction_message,
    }

    handler = handlers.get(message_type, handle_unknown_message)
    handler(
        message=message,
        sender_phone=sender_phone,
        sender_name=sender_name,
        message_id=message_id
    )

def handle_text_message(message, sender_phone, sender_name, message_id):
    text_body = message.get("text", {}).get("body", "").lower().strip()

    if text_body in ["hi", "hello", "hey", "start"]:
        client.send_buttons(
            to=sender_phone,
            body_text=f"Hello {sender_name}! How can we help you?",
            buttons=[
                {"id": "btn_track_order", "title": "Track My Order"},
                {"id": "btn_returns", "title": "Returns & Refunds"},
                {"id": "btn_speak_agent", "title": "Speak to Agent"}
            ]
        )
    else:
        client.send_text(
            to=sender_phone,
            message=f"Got your message! Type *menu* for options."
        )

def handle_interactive_message(
    message, sender_phone, sender_name, message_id
):
    interactive = message.get("interactive", {})
    interactive_type = interactive.get("type")

    if interactive_type == "button_reply":
        button_id = interactive.get("button_reply", {}).get("id")
        handle_button_reply(sender_phone, sender_name, button_id)

    elif interactive_type == "list_reply":
        row_id = interactive.get("list_reply", {}).get("id")
        handle_list_selection(sender_phone, sender_name, row_id)

def handle_button_reply(phone, name, button_id):
    if button_id == "btn_track_order":
        client.send_text(phone, "Enter your order number (e.g., ORD-12345):")
    elif button_id == "btn_returns":
        client.send_text(phone, "Our returns team will assist you shortly.")
    elif button_id == "btn_speak_agent":
        client.send_text(phone, "Connecting you to an agent now... ⏳")

def handle_list_selection(phone, name, row_id):
    client.send_text(phone, f"Processing your selection: {row_id}")

def handle_media_message(message, sender_phone, sender_name, message_id):
    message_type = message.get("type")
    client.send_text(
        sender_phone,
        f"Thanks! Received your {message_type}. We will review it."
    )

def handle_document_message(
    message, sender_phone, sender_name, message_id
):
    filename = message.get("document", {}).get("filename", "document")
    client.send_text(
        sender_phone,
        f"Received your document: {filename}. We will review it."
    )

def handle_location_message(
    message, sender_phone, sender_name, message_id
):
    location = message.get("location", {})
    client.send_text(
        sender_phone,
        "Got your location! Our team will use this for your delivery."
    )

def handle_sticker_message(
    message, sender_phone, sender_name, message_id
):
    client.send_text(sender_phone, "Nice sticker! 😄 How can I help?")

def handle_reaction_message(
    message, sender_phone, sender_name, message_id
):
    print(f"Reaction received from {sender_phone}")

def handle_unknown_message(
    message, sender_phone, sender_name, message_id
):
    client.send_text(
        sender_phone,
        "I received your message. Type *menu* for options."
    )

def handle_message_status(status: dict):
    message_id = status.get("id")
    status_value = status.get("status")
    print(f"Status update: {message_id} → {status_value}")

def handle_template_status_update(value: dict):
    event = value.get("event")
    name = value.get("message_template_name")
    print(f"Template '{name}': {event}")

def handle_account_alert(value: dict):
    alert_type = value.get("alert_type")
    print(f"Account alert: {alert_type}")

def handle_quality_update(value: dict):
    phone = value.get("display_phone_number")
    new_limit = value.get("new_limit")
    print(f"Quality update for {phone}: {new_limit}")

Step 9 — Testing Webhooks Locally with ngrok

# Install ngrok
# https://ngrok.com/download
# Start your Flask app
python app.py
# In a separate terminal, start ngrok
ngrok http 5000
# ngrok gives you a public URL like:
# https://abc123.ngrok.io
# Use this as your webhook callback URL in the Meta App Dashboard:
# https://abc123.ngrok.io/webhook

Running the Complete Server

# Start the Flask webhook server
python app.py
# Output:
# * Running on http://127.0.0.1:5000
# * Debug mode: on

What is Coming in Part 4

In Part 4, we will cover:

  • Message templates in depth — creating, managing, and monitoring
  • The 24-hour conversation window explained
  • Template pacing and what it means for campaigns
  • Creating templates via API
  • Authentication (OTP) templates
  • Building a complete template management system in Python

If this article helped you, please clap and follow for Part 4.


메타데이터
post_id
de3f43d1d36c
slug
part-3-whatsapp-webhooks-receiving-messages-and-processing-events-in-real-time-de3f43d1d36c
url
https://medium.com/@ls.lokesh.sheo/part-3-whatsapp-webhooks-receiving-messages-and-processing-events-in-real-time-de3f43d1d36c
canonical_url
https://medium.com/@ls.lokesh.sheo/part-3-whatsapp-webhooks-receiving-messages-and-processing-events-in-real-time-de3f43d1d36c
author_url
https://medium.com/@ls.lokesh.sheo
status
ok
fetched_at
2026-07-19 06:38:24