DEV Community

Sonam Gupta
Sonam Gupta

Posted on

Build a WhatsApp Coexistence Sync Monitor in Python

#python #whatsapp #webhooks #telnyx

WhatsApp coexistence lets a team use the WhatsApp Business app and the WhatsApp Business API on the same phone number.

That is useful when human agents handle sensitive or unusual conversations in the app while automated systems send reminders and updates through the API. It also creates a synchronization problem: when a human sends a message in the Business app, the API-side application needs a trustworthy way to learn what happened.

The whatsapp-coexistence-sync-monitor example is a small Flask service that closes that gap. It:

  • verifies Telnyx webhook signatures before parsing an event
  • receives message.echo events created by Business-app activity
  • checks whether the number is currently in active coexistence state
  • acknowledges but skips events while the number is not active
  • writes accepted events to a PII-minimized JSONL audit trail

Architecture

The sample has three components:

WhatsApp Business app
        |
        | message.echo
        v
Flask webhook receiver
        |
        +-- Ed25519 signature and timestamp check
        +-- coexistence lifecycle-state gate
        +-- append-only JSONL audit log
Enter fullscreen mode Exit fullscreen mode

The webhook route is POST /webhooks/whatsapp/echo. Before the body is decoded as JSON, the service verifies the telnyx-signature-ed25519 and telnyx-timestamp headers.

Verify the webhook before parsing it

The sample uses the current Telnyx Python SDK client and its webhook unwrapping helper:

client = telnyx.Client(
    api_key=TELNYX_API_KEY,
    public_key=TELNYX_PUBLIC_KEY,
)


def verify_webhook_signature(headers: dict, raw_body: bytes) -> bool:
    signature = headers.get("telnyx-signature-ed25519")
    timestamp = headers.get("telnyx-timestamp")

    if not signature or not timestamp:
        return False

    try:
        ts = int(timestamp)
    except ValueError:
        return False

    if abs(time.time() - ts) > MAX_TIMESTAMP_SKEW_SECONDS:
        return False

    try:
        client.webhooks.unwrap(
            payload=raw_body.decode("utf-8"),
            headers={
                "telnyx-signature-ed25519": signature,
                "telnyx-timestamp": timestamp,
            },
            key=TELNYX_PUBLIC_KEY,
        )
        return True
    except Exception:
        return False
Enter fullscreen mode Exit fullscreen mode

The bounded timestamp window helps reject replayed requests. A missing, malformed, stale, or invalid signature returns False, and the route responds with 401 without processing the payload.

One SDK detail matters here: the unwrap argument is key, not public_key.

Check the number's coexistence state

A valid webhook does not automatically mean the echo stream is ready to use. The monitor also checks two lifecycle fields:

  • is_on_biz_app confirms that the number is registered on the WhatsApp Business app.
  • coexistence_state must be active before the event is written to the audit trail.

In live mode, StateChecker retrieves the configured number from the WhatsApp Business Account:

response = client.whatsapp.business_accounts.phone_numbers.list(self.waba_id)

for number in response.data:
    if number.phone_number == self.phone_number:
        return {
            "phone_number": number.phone_number,
            "is_on_biz_app": number.is_on_biz_app,
            "coexistence_state": number.coexistence_state,
        }
Enter fullscreen mode Exit fullscreen mode

The result is cached for 60 seconds by default, so a burst of echo webhooks does not trigger the same lifecycle request repeatedly. At startup, the service also cross-checks the number through:

client.whatsapp.phone_numbers.retrieve_phone_number(self.phone_number)
Enter fullscreen mode Exit fullscreen mode

DEMO_MODE=true replaces those live requests with simulated active state for local exploration. It does not disable webhook signature verification.

Acknowledge mid-sync events without logging them

If the number is not active, the monitor returns 200 but skips the audit write:

if not state_checker.is_active():
    return jsonify({
        "ok": "skipped",
        "reason": "number_not_active",
    }), 200
Enter fullscreen mode Exit fullscreen mode

This is an intentional processing decision, not a server failure. Returning a success response prevents repeated webhook delivery for an event the application has already evaluated.

For active numbers, the handler records two additional verdicts:

origin_matches_biz_app = origin == "whatsapp_business_app"

cost = message.get("cost")
if cost is None:
    no_charge = True
elif (
    isinstance(cost, dict)
    and cost.get("amount") is None
    and cost.get("currency") is None
):
    no_charge = True
else:
    no_charge = False
Enter fullscreen mode Exit fullscreen mode

Unexpected values are preserved as audit signals rather than silently discarded.

Keep the audit trail useful without storing phone numbers

The sample writes one JSON object per line:

log_entry = {
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "event_id": event_id,
    "event_type": event_type,
    "origin": origin,
    "origin_matches_biz_app": origin_matches_biz_app,
    "text": message.get("text"),
    "message_id": message.get("id"),
    "no_charge": no_charge,
}

message_logger.append(log_entry)
Enter fullscreen mode Exit fullscreen mode

The record intentionally omits from, to, and monetary values. The trail still shows what the application received and how it evaluated the event without becoming a store of customer phone numbers.

JSONL keeps the example easy to run and inspect. A production implementation can retain the same interface while replacing the file with a database or queue and applying the organization's own retention and access policies to message text.

Run the example 

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/whatsapp-coexistence-sync-monitor

python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

cp .env.example .env
Enter fullscreen mode Exit fullscreen mode

Configure the environment with placeholders or your own credentials:

TELNYX_API_KEY=your_telnyx_api_key_here
TELNYX_PUBLIC_KEY=your_telnyx_public_key_here
WABA_ID=your_whatsapp_business_account_id
WHATSAPP_NUMBER=+1555XXXXXXXX
ECHO_LOG_PATH=echo_log.jsonl
DEMO_MODE=true
STATE_CHECK_TTL_SECONDS=60
MAX_TIMESTAMP_SKEW_SECONDS=300
PORT=5000
Enter fullscreen mode Exit fullscreen mode

Then start the Flask service:

python app.py
Enter fullscreen mode Exit fullscreen mode

The health endpoint is available at GET /health. To receive real webhooks, expose the Flask service through a public HTTPS URL, configure that URL in Telnyx, set DEMO_MODE=false, and provide the WhatsApp and Telnyx credentials for your account.

Where to take it next

The example deliberately keeps persistence simple. A production sync monitor could add:

  • a database-backed audit ledger
  • alerts when an echo fails origin or cost checks
  • drift detection between manual echoes and API-side conversation state
  • per-number dashboards for lifecycle state and skipped events
  • retention controls and redaction for message content

The key pattern is small: authenticate the event, verify the channel is in a usable state, record only what the application needs, and acknowledge every event according to an explicit policy.

You can inspect the complete implementation in the Telnyx code examples repository.

Resources

Top comments (0)