WhatsApp Library

Pywce decouples the engine-processing logic and the core whatsapp module allowing you to use them independently.

In this section, we will show how to use it as just another whatsapp client library

The example below uses Fast Api

Import & setup dependencies

To get started, we will import our beloved dependencies and setup a basic fastapi app

From the config.py we created in the last section, we will only use the whatsapp instance

import uvicorn
from fastapi import FastAPI, Request, Response, Query, Depends

import pywce

from .config import whatsapp

app = FastAPI()

# ...

Webhook Challenge

To be able to receive notifications (webhooks) when a user chats with your chatbot, Meta requires that you complete & pass their verification challenge.

Pywce comes with "utility" package to simplify some of the common tasks when working with WhatsApp Cloud Api

# ~ skipped imports

@app.get("/chatbot/webhook")
def verifier(request: Request) -> Response:
    """Verify WhatsApp webhook callback URL challenge."""
    params = request.query_params
    mode, token, challenge = params.get("hub.mode"), params.get("hub.verify_token"), params.get("hub.challenge")

    if whatsapp.util.webhook_challenge(mode, challenge, token):
        return Response(content=challenge, status_code=200)

    raise HTTPException(status_code=403, detail="Forbidden")

Handle Webhook payload

The business logic will be handled by this function - it takes whatsapp webhook data and chunks it in a way that is easier for you to work with.

The engine has utility methods to process some of the repititive tasks like verifying if sent message was a success, check if webhook payload is a valid message payload and much more

# ~ as before ~

@app.post("/chatbot/webhook")
async def process_webhook(request: Request) -> Response:
    """
    Handle incoming webhook events from WhatsApp and process them
    """
    payload = await request.json()
    headers = dict(request.headers)

    if whatsapp.util.verify_webhook_payload(payload, headers):
        if whatsapp.util.is_valid_webhook_message(payload):
            # returns [WaUser] model
            user = whatsapp.util.get_wa_user(payload)
            logger.info(f"Current whatsapp user: {user}")

            # returns [ResponseStructure] model
            response = whatsapp.util.get_response_structure(payload)
            logger.info(f"Webhook response structure: {response}")

            match response.typ:
                case pywce.MessageTypeEnum.TEXT:
                    result = await whatsapp.send_message(
                        recipient_id=user.wa_id,
                        message=f"You said: {response.body.get('body')}"
                    )

                # TODO: implement other types and process them accordingly

                case _:
                    result = await whatsapp.send_message(
                        recipient_id=user.wa_id,
                        message=f"Received whatsapp message type as: {response.typ}"
                    )

            if whatsapp.util.was_request_successful(user.wa_id, result):
                return Response(content="Ack", status_code=200)

    return Response(content="Err", status_code=400)

Library Models

The engine defines some models you will likely work with more.

The whatsapp.util.get_wa_user(payload) call returns a WaUser object defined as below

class WaUser:
    # default whatsapp account user name
    name: str = None
    # whatsapp user mobile number
    wa_id: str = None
    # current message id
    msg_id: str = None
    # webhook timestamp
    timestamp: str = None

The ResponseStructure response processes each webhook, determine the message type and its response payload

class ResponseStructure:
    body: Any = None
    typ: MessageTypeEnum = MessageTypeEnum.UNKNOWN

For example, text messages usually have a response payload as below (Only showing the critical part)

"messages": [
    {
        "from": "16315551234",
        "id": "wamid.ABGGFlCGg0cvAgo-sJQh43L5Pe4W...",
        "timestamp": "1603059201",
        "text": {
            "body": "Hello this is a message body"
        },
        "type": "text"
    }
]

The ResponseStructure will be as below

class ResponseStructure:
    body: Any = {"body": "Hello this is a message body"}
    typ: MessageTypeEnum = MessageTypeEnum.TEXT

Let's check an image message

"messages": [
        {
            "from": "15226273733",
            "id": "wamid.8127ygd7d83...",
            "timestamp": "17727273",
            "type": "image",
            "image": {
                "caption": "This is a caption",
                "mime_type": "image/jpeg",
                "sha256": "81d3bd8a8db4868c9520ed47186e8b7c5789e61ff79f7f834be6950b808a90d3",
                "id": "2754859441498128"
            }
        }
]

The ResponseStructure will be as below

class ResponseStructure:
    body: Any = {
        "caption": "This is a caption",
        "mime_type": "image/jpeg",
        "sha256": "81d3bd8a8db4868c9520ed47186e8b7c5789e61ff79f7f834be6950b808a90d3",
        "id": "2754859441498128"
    }
    typ: MessageTypeEnum = MessageTypeEnum.IMAGE

I hope you get the idea of how the model will be like.

Happy chatbot development 😀

Performance tip

WhatsApp is a high traffic channel, instead of processing each message as it comes > process it > sends message back > acknowledge the whatsapp webhook.

If using, FastApi, you can take advantage of Background Tasks or any event queueing mechanism to ack webhook while processing the webhook in the background.

This is because, if your sync logic fails, and whatsapp does not get a success acknowledgement of the webhook, it uses exponential retry strategy thereby seeing same webhook being retried to be delivered until whatsappe exhausts its retries. This may lead to duplicate message processing if your logic does not handle this well.