Send WhatsApp Messages with Python: Complete Tutorial

Step-by-step tutorial for sending WhatsApp messages with Python using the SendSeven API. Covers text, template, and media messages, HMAC webhook verification, error handling with retry logic, and a production-ready WhatsAppClient class.

WhatsApp has a 98% open rate (Mobilesquared) and responses arrive in under 90 seconds on average. If you're building a notification service, a support bot, or an e-commerce alert system, WhatsApp is the channel your users will actually read. This tutorial shows you exactly how to wire it up from Python — no fluff, just working code.

We'll use the WhatsApp Business API via SendSeven's REST API. SendSeven is a Meta Business Partner, which means you get official API access without managing your own Meta App infrastructure. The full platform also includes a unified inbox, campaigns, and AI automation.

Quick Answer: 5-Line Working Example

If you just need to send a WhatsApp message right now and read the details later:

import requests, os
from dotenv import load_dotenv
load_dotenv()

response = requests.post(
    "https://api.sendseven.com/api/v1/messages",
    headers={"Authorization": f"Bearer {os.getenv('S7_API_KEY')}"},
    json={"contact_id": "CONTACT_ID_HERE", "text": "Hello from Python!", "channel_id": "CHANNEL_ID_HERE"}
)
print(response.json())

That's the complete minimal example. The rest of this tutorial makes it production-grade.

Prerequisites

Before you start, you need:

• Python 3.8+ — f-strings and dataclasses are used throughout
• A SendSeven account — create one and grab your API key (starts with s7_api_)
• A WhatsApp channel configured — follow the WhatsApp setup guide to connect your number
• At least one contact in your SendSeven inbox (the API uses contact_id, not raw phone numbers)

Install the two dependencies:

pip install requests python-dotenv

Create a .env file in your project root:

S7_API_KEY=s7_api_your_key_here
S7_BASE_URL=https://api.sendseven.com/api/v1
S7_CHANNEL_ID=your_whatsapp_channel_id
S7_WEBHOOK_SECRET=whsec_your_secret_here

Add .env to .gitignore immediately. Never hardcode your API key in source files.

Architecture Overview

Before writing code, understand what actually happens when you send a message:

Your Python App
    |
    | POST /api/v1/messages
    |
    v
SendSeven API  <-- You authenticate here with your Bearer token
    |
    | Routes message via official WhatsApp Business API
    |
    v
Meta (Facebook) Infrastructure
    |
    | Delivers via WhatsApp network
    |
    v
End User's WhatsApp

A few rules that affect how you write your code:

• 24-hour service window: After a user sends you a message, you have 24 hours to reply with any content — including free-form text. These are "service messages" and they are free and unlimited since November 2024.
• Outside the window: You must use an approved template message. Templates are pre-approved message structures with variable placeholders.
• Per-message pricing: WhatsApp uses per-message pricing (since July 2025). Marketing messages cost more than utility or authentication messages. Service messages (replies within 24h) are free.
• contact_id, not phone number: The API operates on SendSeven contact IDs, not raw E.164 phone numbers. Contacts are created automatically when users message you, or you can import them.

Step 1: Send a Text Message

This is the core pattern. Use it when you're responding within a 24-hour service window.

import os
import requests
from dotenv import load_dotenv

load_dotenv()

BASE_URL = os.getenv("S7_BASE_URL", "https://api.sendseven.com/api/v1")
API_KEY = os.getenv("S7_API_KEY")
CHANNEL_ID = os.getenv("S7_CHANNEL_ID")

def send_text_message(contact_id: str, text: str) -> dict:
    """Send a plain text WhatsApp message to a contact.

    Args:
        contact_id: SendSeven contact ID (not a phone number)
        text: Message text. Max 4096 characters.

    Returns:
        Parsed JSON response from the API.

    Raises:
        requests.HTTPError: On 4xx/5xx responses.
    """
    url = f"{BASE_URL}/messages"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "contact_id": contact_id,
        "text": text,
        "channel_id": CHANNEL_ID,
    }

    response = requests.post(url, headers=headers, json=payload, timeout=10)
    response.raise_for_status()  # Raises HTTPError for 4xx/5xx
    return response.json()


if __name__ == "__main__":
    result = send_text_message(
        contact_id="cnt_01h9xyz",
        text="Your order #4821 has shipped! It'll arrive by Thursday."
    )
    print(f"Message sent. ID: {result.get('id')}")

A successful response looks like this:

{
  "id": "msg_01j4xyz",
  "status": "sent",
  "contact_id": "cnt_01h9xyz",
  "channel": "whatsapp",
  "created_at": "2026-02-27T10:15:30Z"
}

Important: status: sent means the API accepted the message. Actual delivery to the device triggers a message.sent webhook event. If delivery fails (number not on WhatsApp, user blocked you), you'll see a message.failed event.

Step 2: Send a Template Message

Templates are required when you initiate a conversation (no prior message from the user) or when the 24-hour service window has closed. They go through Meta's approval process — usually minutes via automated review, up to 24 hours if flagged for manual review.

First, create and approve your template in the SendSeven dashboard. Then send it like this:

import os
import requests
from dotenv import load_dotenv

load_dotenv()

BASE_URL = os.getenv("S7_BASE_URL", "https://api.sendseven.com/api/v1")
API_KEY = os.getenv("S7_API_KEY")
CHANNEL_ID = os.getenv("S7_CHANNEL_ID")


def send_template_message(
    contact_id: str,
    template_id: str,
    language: str,
    body_variables: list[str] | None = None,
    button_parameters: list[dict] | None = None,
) -> dict:
    """Send an approved WhatsApp template message.

    Args:
        contact_id: SendSeven contact ID.
        template_id: Template ID from SendSeven dashboard.
        language: Language code, e.g. 'en', 'de', 'es'.
        body_variables: List of strings to fill {{1}}, {{2}}, etc. in the template body.
        button_parameters: List of button parameter objects for dynamic URL/quick reply buttons.

    Returns:
        Parsed JSON response.
    """
    url = f"{BASE_URL}/whatsapp/templates/{template_id}/send"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "contact_id": contact_id,
        "channel_id": CHANNEL_ID,
        "language": language,
        "variables": {
            "body": body_variables or [],
            "buttons": button_parameters or [],
        },
    }

    response = requests.post(url, headers=headers, json=payload, timeout=10)
    response.raise_for_status()
    return response.json()


if __name__ == "__main__":
    # Example: order_shipped template with body {{1}} = order number, {{2}} = delivery date
    # and a dynamic URL button where {{1}} = tracking token
    result = send_template_message(
        contact_id="cnt_01h9xyz",
        template_id="tpl_order_shipped",
        language="en",
        body_variables=["#4821", "Thursday, Feb 27"],
        button_parameters=[
            {
                "index": 0,  # First button
                "parameters": [{"type": "text", "text": "TRK-8821-XYZ"}]
            }
        ],
    )
    print(f"Template sent. ID: {result.get('id')}")

About template variables:

• body is an array of strings that fill {{1}}, {{2}}, etc. in your template body text, in order.
• buttons is an array of button parameter objects. Each has an index (0-based) and a parameters array.
• If your template has no variables, pass empty arrays: "variables": {"body": [], "buttons": []}

Template approval reminder: Templates are reviewed per language. An English template approval does not automatically approve a German version. Submit each language variant separately. See the interactive messages guide for advanced template types including carousels and catalog messages.

Step 3: Send Media (Images & Documents)

Attach media to a message by adding an attachments array to the request body. The media must be hosted at a publicly accessible URL — WhatsApp's servers will fetch it directly.

import os
import requests
from dotenv import load_dotenv

load_dotenv()

BASE_URL = os.getenv("S7_BASE_URL", "https://api.sendseven.com/api/v1")
API_KEY = os.getenv("S7_API_KEY")
CHANNEL_ID = os.getenv("S7_CHANNEL_ID")


def send_image_message(contact_id: str, image_url: str, caption: str = "") -> dict:
    """Send an image to a WhatsApp contact.

    Args:
        contact_id: SendSeven contact ID.
        image_url: Public URL to the image (JPEG, PNG, WebP, GIF). Max 5MB.
        caption: Optional caption shown below the image. Max 1024 characters.

    Returns:
        Parsed JSON response.
    """
    url = f"{BASE_URL}/messages"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "contact_id": contact_id,
        "channel_id": CHANNEL_ID,
        "text": caption,
        "attachments": [
            {
                "type": "image",
                "url": image_url,
            }
        ],
    }

    response = requests.post(url, headers=headers, json=payload, timeout=10)
    response.raise_for_status()
    return response.json()


def send_document_message(
    contact_id: str, document_url: str, filename: str, caption: str = ""
) -> dict:
    """Send a document (PDF, DOCX, XLSX, etc.) to a WhatsApp contact.

    Args:
        contact_id: SendSeven contact ID.
        document_url: Public URL to the document. Max 100MB.
        filename: Display name of the file (shown in WhatsApp).
        caption: Optional caption. Max 1024 characters.

    Returns:
        Parsed JSON response.
    """
    url = f"{BASE_URL}/messages"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "contact_id": contact_id,
        "channel_id": CHANNEL_ID,
        "text": caption,
        "attachments": [
            {
                "type": "document",
                "url": document_url,
                "filename": filename,
            }
        ],
    }

    response = requests.post(url, headers=headers, json=payload, timeout=10)
    response.raise_for_status()
    return response.json()


if __name__ == "__main__":
    # Send an order confirmation PDF
    send_document_message(
        contact_id="cnt_01h9xyz",
        document_url="https://cdn.yourapp.com/invoices/INV-4821.pdf",
        filename="Invoice-4821.pdf",
        caption="Here's your invoice for order #4821.",
    )

    # Send a product image
    send_image_message(
        contact_id="cnt_01h9xyz",
        image_url="https://cdn.yourapp.com/products/running-shoe-red.jpg",
        caption="Your new running shoes are on their way!",
    )

Media requirements:

• Images: JPEG, PNG, WebP, GIF (static only). Max 5 MB. HTTPS URL required.
• Documents: PDF, DOCX, XLSX, PPTX, and others. Max 100 MB. HTTPS URL required.
• Video: MP4, 3GP. Max 16 MB.
• Audio: MP3, OGG, AAC. Max 16 MB.
• The URL must be publicly reachable by Meta's servers. Signed S3 URLs with short expiry windows often cause silent delivery failures — use long-lived or public URLs.

Step 4: Error Handling & Retry Logic

The API returns standard HTTP status codes. Some errors are permanent (wrong contact ID), others are transient (rate limit, 500). Only retry transient errors.

Error Code Reference

Here's a production-grade retry wrapper:

import os
import time
import random
import logging
from typing import Callable, TypeVar

import requests
from requests import Response
from dotenv import load_dotenv

load_dotenv()

logger = logging.getLogger(__name__)

T = TypeVar("T")

RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
MAX_RETRIES = 3
BASE_DELAY = 1.0  # seconds
MAX_DELAY = 60.0  # seconds


def with_retry(func: Callable[[], Response]) -> Response:
    """Execute an HTTP call with exponential backoff retry logic.

    Retries only on transient errors (429, 5xx). Raises immediately on
    permanent errors (4xx except 429).

    Args:
        func: A zero-argument callable that returns a requests.Response.

    Returns:
        The successful Response object.

    Raises:
        requests.HTTPError: After all retries are exhausted.
    """
    last_exception: Exception | None = None

    for attempt in range(MAX_RETRIES + 1):
        try:
            response = func()

            if response.status_code not in RETRYABLE_STATUS_CODES:
                response.raise_for_status()
                return response

            # Handle rate limiting: respect Retry-After header if present
            if response.status_code == 429:
                retry_after = float(response.headers.get("Retry-After", BASE_DELAY))
                wait = min(retry_after, MAX_DELAY)
                logger.warning("Rate limited. Waiting %.1fs before retry %d/%d.", wait, attempt + 1, MAX_RETRIES)
                time.sleep(wait)
            else:
                # Exponential backoff with jitter for 5xx errors
                wait = min(BASE_DELAY * (2 ** attempt) + random.uniform(0, 1), MAX_DELAY)
                logger.warning(
                    "HTTP %d on attempt %d/%d. Retrying in %.1fs.",
                    response.status_code, attempt + 1, MAX_RETRIES, wait
                )
                time.sleep(wait)

            last_exception = requests.HTTPError(response=response)

        except requests.ConnectionError as e:
            wait = min(BASE_DELAY * (2 ** attempt) + random.uniform(0, 1), MAX_DELAY)
            logger.warning("Connection error on attempt %d/%d: %s. Retrying in %.1fs.", attempt + 1, MAX_RETRIES, e, wait)
            time.sleep(wait)
            last_exception = e

    raise last_exception


def send_with_retry(contact_id: str, text: str) -> dict:
    """Send a text message with automatic retry on transient failures."""
    base_url = os.getenv("S7_BASE_URL", "https://api.sendseven.com/api/v1")
    api_key = os.getenv("S7_API_KEY")
    channel_id = os.getenv("S7_CHANNEL_ID")

    def make_request() -> Response:
        return requests.post(
            f"{base_url}/messages",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"contact_id": contact_id, "text": text, "channel_id": channel_id},
            timeout=10,
        )

    response = with_retry(make_request)
    return response.json()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    result = send_with_retry(
        contact_id="cnt_01h9xyz",
        text="Test message with retry logic."
    )
    print(result)

Rate limits: The WhatsApp Business API allows up to 80 messages per second (MPS) on standard tiers. If you're sending bulk notifications, implement a token bucket or semaphore to stay within your limit. When using WhatsApp Coexistence (App + API on the same number), throughput drops to 20 MPS.

Step 5: Receive Webhooks with Flask

Webhooks are how SendSeven notifies your server about incoming messages, delivery status changes, and conversation events in real time. Without webhooks, you'd have to poll the API constantly — inefficient and slow.

For a deep dive on webhook architecture, see the webhooks explainer. The webhook setup guide walks through the dashboard configuration.

How Webhook Verification Works

Every incoming webhook includes an X-Webhook-Signature header with the format sha256=<hex_digest>. SendSeven computes this by running HMAC-SHA256 over the raw request body using your webhook secret (starts with whsec_). You verify the same computation on your end before processing the payload. This prevents attackers from sending fake events to your endpoint.

import os
import hmac
import hashlib
import logging
import json
from flask import Flask, request, jsonify, abort
from dotenv import load_dotenv

load_dotenv()

app = Flask(__name__)
logger = logging.getLogger(__name__)

WEBHOOK_SECRET = os.getenv("S7_WEBHOOK_SECRET", "")  # starts with whsec_
API_KEY = os.getenv("S7_API_KEY")
BASE_URL = os.getenv("S7_BASE_URL", "https://api.sendseven.com/api/v1")
CHANNEL_ID = os.getenv("S7_CHANNEL_ID")


def verify_signature(raw_body: bytes, signature_header: str) -> bool:
    """Verify the HMAC-SHA256 webhook signature.

    Uses hmac.compare_digest for timing-safe comparison to prevent
    timing-based side-channel attacks.

    Args:
        raw_body: Raw request body bytes (before any parsing).
        signature_header: Value of X-Webhook-Signature header, format 'sha256='.

    Returns:
        True if the signature is valid, False otherwise.
    """
    if not signature_header.startswith("sha256="):
        return False

    expected_sig = signature_header[len("sha256="):]
    computed_sig = hmac.new(
        WEBHOOK_SECRET.encode("utf-8"),
        raw_body,
        hashlib.sha256
    ).hexdigest()

    # Timing-safe comparison prevents timing attacks
    return hmac.compare_digest(computed_sig, expected_sig)


def auto_reply(contact_id: str, text: str) -> None:
    """Send an automated reply to a contact (within 24-hour service window)."""
    import requests as req
    try:
        req.post(
            f"{BASE_URL}/messages",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"contact_id": contact_id, "text": text, "channel_id": CHANNEL_ID},
            timeout=5,
        )
    except Exception as e:
        logger.error("Auto-reply failed: %s", e)


@app.route("/webhook", methods=["POST"])
def handle_webhook():
    """Receive and process SendSeven webhook events."""
    # Step 1: Verify signature BEFORE parsing body
    raw_body = request.get_data()
    sig_header = request.headers.get("X-Webhook-Signature", "")

    if not verify_signature(raw_body, sig_header):
        logger.warning("Webhook signature verification failed. IP: %s", request.remote_addr)
        abort(401)

    # Step 2: Parse the event
    try:
        event = json.loads(raw_body)
    except json.JSONDecodeError:
        abort(400)

    event_type = event.get("event")
    event_id = event.get("id")
    data = event.get("data", {})

    logger.info("Received event %s (id=%s)", event_type, event_id)

    # Step 3: Route by event type
    if event_type == "message.received":
        _handle_inbound_message(data)
    elif event_type == "conversation.closed":
        _handle_conversation_closed(data)
    # Add more handlers as needed:
    # elif event_type == "message.sent": ...
    # elif event_type == "contact.created": ...

    # Step 4: Acknowledge receipt (must respond 200 within 10 seconds)
    # If your processing is slow, acknowledge first and process async
    return jsonify({"received": True}), 200


def _handle_inbound_message(data: dict) -> None:
    """Process an inbound message event."""
    contact_id = data.get("contact_id")
    content = data.get("content", "")
    direction = data.get("direction")  # 'inbound' or 'outbound'
    channel = data.get("channel")  # 'whatsapp', 'telegram', etc.
    message_type = data.get("message_type")  # 'text', 'image', 'document', etc.

    if direction != "inbound" or channel != "whatsapp":
        return  # Only process inbound WhatsApp messages

    logger.info("Inbound message from contact %s: %s", contact_id, content[:100])

    # Simple keyword-based auto-reply (customize this for your bot logic)
    content_lower = content.lower().strip() if content else ""

    if "order status" in content_lower or "track" in content_lower:
        auto_reply(
            contact_id=contact_id,
            text="I can help with your order! Please share your order number and I'll look it up."
        )
    elif "pricing" in content_lower or "price" in content_lower or "cost" in content_lower:
        auto_reply(
            contact_id=contact_id,
            text="Our Pay as you go plan has a €9/channel/month minimum spend, fully consumed by your usage. For full pricing, visit sendseven.com/pricing or I can connect you with our sales team."
        )
    elif content_lower in ("hi", "hello", "hey"):
        auto_reply(
            contact_id=contact_id,
            text=f"Hi! I'm the SendSeven support bot. How can I help you today?\n\nReply with:\n- ORDER STATUS to track an order\n- PRICING for plan info\n- HUMAN to speak with a person"
        )
    elif "human" in content_lower or "agent" in content_lower or "support" in content_lower:
        # In a real app, you'd assign the conversation to a human agent here
        auto_reply(
            contact_id=contact_id,
            text="Connecting you with a support agent now. Typical response time is under 5 minutes during business hours."
        )


def _handle_conversation_closed(data: dict) -> None:
    """Handle a conversation closed event."""
    conversation_id = data.get("conversation_id")
    contact_id = data.get("contact_id")
    logger.info("Conversation %s closed for contact %s", conversation_id, contact_id)
    # Trigger post-conversation workflows here:
    # - Send satisfaction survey
    # - Update CRM record
    # - Trigger follow-up sequence


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    # In production, use gunicorn or uvicorn, not the Flask dev server
    app.run(host="0.0.0.0", port=8080, debug=False)

Event payload structure: Every webhook event has this shape:

{
  "id": "evt_01j4abc",           # Unique event ID (use for deduplication)
  "event": "message.received",   # Event type
  "created_at": "2026-02-27T10:15:30Z",
  "tenant_id": "tnt_01h2xyz",
  "data": {
    "message_id": "msg_01j4xyz",
    "conversation_id": "conv_01h8xyz",
    "contact_id": "cnt_01h9xyz",
    "channel": "whatsapp",
    "content": "Hello, I need help with my order",
    "message_type": "text",       # 'text', 'image', 'document', 'audio', 'video'
    "direction": "inbound"        # 'inbound' (from user) or 'outbound' (from you)
  }
}

Available event types: message.received, message.sent, conversation.created, conversation.closed, contact.created, contact.updated

Retry schedule: If your endpoint returns a non-200 response or times out (>10 seconds), SendSeven retries at: 1 minute, 5 minutes, 30 minutes, 2 hours, 8 hours. After 20 consecutive failures, the webhook endpoint is circuit-broken and you'll receive an alert.

Production tip: Acknowledge the webhook immediately (return 200), then process the event asynchronously using a task queue (Celery, RQ, or AWS SQS). This prevents processing delays from causing timeouts and retries.

Complete WhatsAppClient Class

Here's a production-ready client that wraps everything above into a clean interface:

import os
import hmac
import hashlib
import logging
import time
import random
from typing import Any

import requests
from requests import Session, Response
from dotenv import load_dotenv

load_dotenv()

logger = logging.getLogger(__name__)


class WhatsAppClient:
    """Production-ready WhatsApp messaging client for SendSeven API.

    Handles authentication, retries, and webhook verification.

    Usage:
        client = WhatsAppClient()
        client.send_text(contact_id="cnt_01h9xyz", text="Hello!")
    """

    BASE_URL = os.getenv("S7_BASE_URL", "https://api.sendseven.com/api/v1")
    MAX_RETRIES = 3
    BASE_DELAY = 1.0
    MAX_DELAY = 60.0
    RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}

    def __init__(
        self,
        api_key: str | None = None,
        channel_id: str | None = None,
        webhook_secret: str | None = None,
    ):
        """Initialize the client.

        Args:
            api_key: SendSeven API key. Defaults to S7_API_KEY env var.
            channel_id: WhatsApp channel ID. Defaults to S7_CHANNEL_ID env var.
            webhook_secret: Webhook secret for signature verification.
                            Defaults to S7_WEBHOOK_SECRET env var.
        """
        self.api_key = api_key or os.getenv("S7_API_KEY")
        self.channel_id = channel_id or os.getenv("S7_CHANNEL_ID")
        self.webhook_secret = webhook_secret or os.getenv("S7_WEBHOOK_SECRET", "")

        if not self.api_key:
            raise ValueError("API key is required. Set S7_API_KEY env var or pass api_key.")
        if not self.channel_id:
            raise ValueError("Channel ID is required. Set S7_CHANNEL_ID env var or pass channel_id.")

        self._session = Session()
        self._session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        })

    def send_text(self, contact_id: str, text: str) -> dict:
        """Send a plain text message within the 24-hour service window."""
        return self._post("/messages", {
            "contact_id": contact_id,
            "text": text,
            "channel_id": self.channel_id,
        })

    def send_template(
        self,
        contact_id: str,
        template_id: str,
        language: str = "en",
        body_variables: list[str] | None = None,
        button_parameters: list[dict] | None = None,
    ) -> dict:
        """Send an approved template message (works outside the 24-hour window)."""
        return self._post(f"/whatsapp/templates/{template_id}/send", {
            "contact_id": contact_id,
            "channel_id": self.channel_id,
            "language": language,
            "variables": {
                "body": body_variables or [],
                "buttons": button_parameters or [],
            },
        })

    def send_image(self, contact_id: str, image_url: str, caption: str = "") -> dict:
        """Send an image message (JPEG/PNG/WebP/GIF, max 5MB)."""
        return self._post("/messages", {
            "contact_id": contact_id,
            "channel_id": self.channel_id,
            "text": caption,
            "attachments": [{"type": "image", "url": image_url}],
        })

    def send_document(
        self, contact_id: str, document_url: str, filename: str, caption: str = ""
    ) -> dict:
        """Send a document (PDF/DOCX/etc., max 100MB)."""
        return self._post("/messages", {
            "contact_id": contact_id,
            "channel_id": self.channel_id,
            "text": caption,
            "attachments": [{"type": "document", "url": document_url, "filename": filename}],
        })

    @staticmethod
    def verify_webhook_signature(raw_body: bytes, signature_header: str, secret: str) -> bool:
        """Verify the HMAC-SHA256 webhook signature.

        Call this before processing any webhook payload.

        Args:
            raw_body: Raw request body bytes.
            signature_header: X-Webhook-Signature header value ('sha256=').
            secret: Your webhook secret (starts with 'whsec_').

        Returns:
            True if valid, False otherwise.
        """
        if not signature_header.startswith("sha256="):
            return False
        expected = signature_header[len("sha256="):]
        computed = hmac.new(
            secret.encode("utf-8"),
            raw_body,
            hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(computed, expected)

    def _post(self, path: str, payload: dict[str, Any]) -> dict:
        """Make a POST request with retry logic."""
        url = f"{self.BASE_URL}{path}"
        last_exception: Exception | None = None

        for attempt in range(self.MAX_RETRIES + 1):
            try:
                response: Response = self._session.post(url, json=payload, timeout=10)

                if response.status_code not in self.RETRYABLE_STATUS_CODES:
                    response.raise_for_status()
                    return response.json()

                if response.status_code == 429:
                    wait = float(response.headers.get("Retry-After", self.BASE_DELAY))
                    wait = min(wait, self.MAX_DELAY)
                else:
                    wait = min(
                        self.BASE_DELAY * (2 ** attempt) + random.uniform(0, 1),
                        self.MAX_DELAY
                    )

                logger.warning(
                    "HTTP %d on attempt %d/%d. Retrying in %.1fs.",
                    response.status_code, attempt + 1, self.MAX_RETRIES, wait
                )
                time.sleep(wait)
                last_exception = requests.HTTPError(response=response)

            except requests.ConnectionError as e:
                wait = min(self.BASE_DELAY * (2 ** attempt) + random.uniform(0, 1), self.MAX_DELAY)
                logger.warning("Connection error attempt %d/%d: %s", attempt + 1, self.MAX_RETRIES, e)
                time.sleep(wait)
                last_exception = e

        raise last_exception


# Usage example
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    client = WhatsAppClient()

    # Send text (within 24-hour service window — free)
    client.send_text(
        contact_id="cnt_01h9xyz",
        text="Your appointment is confirmed for tomorrow at 10:00 AM."
    )

    # Send template (outside window or proactive outreach)
    client.send_template(
        contact_id="cnt_01h9xyz",
        template_id="tpl_appointment_reminder",
        language="en",
        body_variables=["10:00 AM", "Thursday, Feb 27"],
    )

    # Send invoice PDF
    client.send_document(
        contact_id="cnt_01h9xyz",
        document_url="https://cdn.yourapp.com/invoices/INV-4821.pdf",
        filename="Invoice-4821.pdf",
        caption="Here's your invoice.",
    )

Troubleshooting

Next Steps

You've got the fundamentals down. Here's where to go from here:

• Node.js version — Same tutorial, JavaScript/TypeScript. Good if you're working in a mixed stack.
• Interactive messages — Buttons, lists, carousels, and product catalogs. The next level of WhatsApp UX.
• AI features — Connect your SendSeven inbox to an AI model. Handle tier-1 support automatically.
• Team inbox setup — Route inbound conversations to agents. Set up assignment and routing rules.
• Webhook setup guide — Configure endpoints, test with ngrok, monitor event logs.
• WhatsApp API complete guide — Deep dive into API limits, message categories, phone number verification, and compliance.

FAQ

Can I use async Python (httpx) instead of requests?

Yes. Replace requests with httpx for async support. The API endpoints and payloads are identical. Example:

import httpx
import asyncio
import os
from dotenv import load_dotenv

load_dotenv()

async def send_text_async(contact_id: str, text: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{os.getenv('S7_BASE_URL')}/messages",
            headers={"Authorization": f"Bearer {os.getenv('S7_API_KEY')}"},
            json={"contact_id": contact_id, "text": text, "channel_id": os.getenv('S7_CHANNEL_ID')},
            timeout=10.0,
        )
        response.raise_for_status()
        return response.json()

asyncio.run(send_text_async("cnt_01h9xyz", "Async hello!"))

How do I handle opt-outs?

When a user sends "STOP", "UNSUBSCRIBE", or a similar opt-out keyword, WhatsApp blocks further outbound messages to that number. Your webhook will receive a contact.updated event with the opt-out status. Store this in your database and check it before sending any future messages — attempting to message an opted-out user generates a 422 error from the API and counts against your quality rating.

What are the rate limits?

The WhatsApp Business API supports up to 80 messages per second (MPS) on standard business tiers. Check your account's current tier in the dashboard (API Only: EUR 9/channel/month; Full Suite: Pay as you go €9/channel/month minimum spend, Scale EUR 199/month, Business EUR 499/month). See full pricing details. When using WhatsApp Coexistence (running the WhatsApp Business App alongside the API on the same number), throughput is capped at 20 MPS.

Do I need a separate WhatsApp number for the API?

Not necessarily. WhatsApp Coexistence (introduced in 2025) allows you to run the WhatsApp Business App and the WhatsApp Business API on the same phone number simultaneously. You get real-time sync of contacts and up to 6 months of chat history between both. This is ideal if your team also uses the WhatsApp Business App for manual replies. Requires Business App version 2.24.17 or higher.

Are service messages really free?

Yes. Since November 2024, all service messages — replies sent within the 24-hour window after a customer contacts you first — are free and unlimited. Marketing, utility, and authentication messages are charged per message at Meta's rates (which vary by recipient country). See Meta's pricing page for current rates. This is a significant change from the previous conversation-based pricing model (deprecated July 2025).

How long does template approval take?

Meta's automated (AI) review usually approves templates within minutes. If a template gets flagged for manual review, it can take up to 24 hours. Templates are rejected most commonly for: policy violations (e.g., promotional content in a utility template), misleading variable placeholders, or restricted content categories (gambling, pharmaceuticals without verification). Always match your template's content to its declared category.

Can I send to users who haven't messaged me first?

Yes, with a template message. You can initiate a conversation with any contact who has opted in to receive WhatsApp communications from your business. Use a marketing or utility template for the outreach. The contact must have explicitly opted in — sending unsolicited messages is a policy violation that can get your number banned. Store opt-in timestamps and keep records in case Meta requests proof of consent.

How do I test without sending real WhatsApp messages?

Use the SendSeven sandbox environment during development. Sandbox requests are validated against the API schema but do not route to Meta or deliver to actual devices. You can also test webhook parsing locally using ngrok to expose your local Flask server — see the webhook setup guide for a step-by-step ngrok walkthrough.