Webhook Data Reference¶

# Basic text message
if webhook.get_message_type_name() == "text":
    text = webhook.get_message_text()                    # "Hello!"
    message_id = webhook.message.message_id              # "wamid.abc123..."

    # Reply context (if replying to another message)
    if webhook.message.has_context():
        context = webhook.message.get_context()
        original_id = context.original_message_id        # ID of replied message

# Button or list selection
if webhook.get_message_type_name() == "interactive":
    selection = webhook.get_interactive_selection()      # "help_button"

    # Access interactive details
    interactive_type = webhook.message.interactive_type  # InteractiveType.BUTTON_REPLY
    option_title = webhook.message.selected_option_title # "Get Help"
    original_msg = webhook.message.original_message_id   # Interactive message ID

    # Check selection type
    is_button = webhook.message.is_button_reply()        # bool
    is_list = webhook.message.is_list_reply()            # bool

# Image, video, audio, document, sticker
if webhook.get_message_type_name() in ["image", "video", "audio", "document", "sticker"]:
    media_id = webhook.message.media_id                  # "media_abc123"
    file_size = webhook.message.file_size                # 1024000 (bytes)
    caption = webhook.message.caption                    # "Check this out!"

    # Download information
    download_info = webhook.message.get_download_info()  # dict with URL/headers

    # Type-specific properties
    if webhook.message.is_image():
        mime_type = webhook.message.media_type           # MediaType.IMAGE_JPEG
    elif webhook.message.is_audio():
        duration = webhook.message.duration              # 30 (seconds)
        is_voice = webhook.message.is_voice_message()    # bool

if webhook.get_message_type_name() == "location":
    lat = webhook.message.latitude                       # 40.7589
    lng = webhook.message.longitude                      # -73.9851
    address = webhook.message.address                    # "New York, NY"
    name = webhook.message.location_name                 # "Central Park"

if webhook.get_message_type_name() == "contact":
    name = webhook.message.contact_name                  # "John Doe"
    phone = webhook.message.contact_phone                # "+1234567890"
    contact_data = webhook.message.contact_data          # dict with all contact info

# No JSON parsing needed!
async def process_message(self, webhook):
    # Clean, typed access
    user_id = webhook.user.user_id              # "1234567890123"
    user_name = webhook.user.profile_name       # "John Doe" 
    message_text = webhook.get_message_text()   # "Hello there!"
    timestamp = webhook.timestamp               # datetime(2022, 1, 20, 10, 0, 45)

    # Helper methods work across all message types
    message_type = webhook.get_message_type_name()  # "text"

    # Optional: access raw data for debugging
    raw_json = webhook.get_raw_webhook_data()   # Full original JSON

# Simple type checking
message_type = webhook.get_message_type_name()

if message_type == "text":
    text = webhook.get_message_text()
elif message_type == "interactive":
    selection = webhook.get_interactive_selection()
elif message_type in ["image", "video", "audio", "document"]:
    media_id = webhook.message.media_id
    caption = webhook.message.caption
elif message_type == "location":
    lat, lng = webhook.message.latitude, webhook.message.longitude

# Handle button and list selections
if webhook.get_message_type_name() == "interactive":
    selection = webhook.get_interactive_selection()  # "help", "product_123"

    # Pattern matching on selection IDs
    if selection == "help":
        await self.messenger.send_text("Here's help!", webhook.user.user_id)
    elif selection.startswith("product_"):
        product_id = selection.replace("product_", "")
        await self.show_product_details(product_id, webhook.user.user_id)

# Check message context
if webhook.was_forwarded():
    # Handle forwarded messages differently
    await self.messenger.send_text("Thanks for forwarding that!", webhook.user.user_id)

if webhook.has_business_context():
    # User interacted with catalog or business buttons
    context = webhook.business_context
    if context.has_product_context():
        product_id = context.product_retailer_id
        # Handle product inquiry

if webhook.has_ad_referral():
    # User came from advertisement
    referral = webhook.ad_referral
    campaign_id = referral.source_id
    # Track ad conversion

# Status webhook error handling
async def process_status(self, webhook):
    if webhook.is_failed_status() and webhook.has_errors():
        error = webhook.get_primary_error()

        if error.error_code == 131051:  # Unsupported message type
            self.logger.info("User sent unsupported message - normal")
        elif "rate limit" in error.error_title.lower():
            self.logger.warning("Rate limited - implement backoff")
            # Implement retry logic
        else:
            self.logger.error(f"Message delivery failed: {error.error_title}")

# Error webhook handling  
async def process_error(self, webhook):
    if webhook.has_critical_errors():
        # 5xx server errors - escalate
        await self.notify_admin(webhook.get_error_codes())
    elif webhook.has_retryable_errors():
        # Temporary issues - retry later
        await self.schedule_retry(webhook)

# Some users send unsupported message types
if webhook.get_message_type_name() == "unsupported":
    # WhatsApp couldn't process the message type
    await self.messenger.send_text(
        "Sorry, I can't process that type of message.",
        webhook.user.user_id
    )

# Empty text messages (rare but possible)
if webhook.get_message_type_name() == "text":
    text = webhook.get_message_text()
    if not text.strip():
        # Handle empty message
        return

# Interactive selections might be None
selection = webhook.get_interactive_selection()
if not selection:
    self.logger.warning("Interactive message with no selection")
    return

# Original message might not exist anymore
if webhook.get_message_type_name() == "interactive":
    original_id = webhook.message.original_message_id
    # This message ID might be from a different conversation

# Media files have size limits and expiration
if webhook.message.is_image():
    file_size = webhook.message.file_size
    if file_size and file_size > 16 * 1024 * 1024:  # 16MB
        await self.messenger.send_text(
            "Image too large. Please send smaller file.",
            webhook.user.user_id
        )
        return

    # Media URLs expire after some time
    download_info = webhook.message.get_download_info()
    # Download immediately or store media_id for later retrieval

# Status updates might arrive out of order
async def process_status(self, webhook):
    # A "read" might arrive before "delivered" due to network timing
    status = webhook.status.value

    # Don't assume linear progression: sent → delivered → read
    # Always check the actual status value

async def process_message(self, webhook):
    # Enable in development to see raw WhatsApp JSON
    if settings.is_development:
        raw_data = webhook.get_raw_webhook_data()
        self.logger.debug(f"Raw webhook: {json.dumps(raw_data, indent=2)}")

    # Your processing logic here

# Log webhook summaries
summary = webhook.get_summary()
self.logger.info(f"Processing: {summary}")

# Output example:
# {
#   "webhook_type": "incoming_message",
#   "platform": "whatsapp", 
#   "message_type": "text",
#   "sender": "1234567890123",
#   "tenant": "123456789",
#   "timestamp": "2025-01-15T10:30:45Z"
# }

class MyApp(WappaEventHandler):
    async def process_message(self, webhook):
        # Always validate dependencies first
        if not self.validate_dependencies():
            self.logger.error("Dependencies not properly injected")
            return

        # Get dependency status for debugging
        status = self.get_dependency_status()
        self.logger.debug(f"Dependencies: {status}")