本文为原文前 6,000 字符的节选翻译，完整内容请查看原文。

Build a WhatsApp Coexistence Sync Monitor in Python

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.

WhatsApp 共存功能允许团队在同一个电话号码上同时使用 WhatsApp Business 应用程序和 WhatsApp Business API。当人工客服在应用程序中处理敏感或异常对话，而自动化系统通过 API 发送提醒和更新时，此功能非常有用。这也带来了一个同步问题：当人工在 Business 应用程序中发送消息时，API 端应用程序需要一种可靠的方式来获知发生了什么。whatsapp-coexistence-sync-monitor 示例是一个小型 Flask 服务，旨在弥补这一差距。

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.

它能够：在解析事件前验证 Telnyx Webhook 签名；接收由 Business 应用程序活动创建的 message.echo 事件；检查号码当前是否处于活跃的共存状态；在号码未激活时确认但跳过事件；将已接受的事件写入经过个人身份信息（PII）最小化处理的 JSONL 审计追踪中。

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. 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.

架构

该示例包含三个组件：WhatsApp Business 应用程序 | | message.echo v Flask Webhook 接收器 | +— Ed25519 签名和时间戳检查 +— 共存生命周期状态门控 +— 仅追加的 JSONL 审计日志。Webhook 路由为 POST /webhooks/whatsapp/echo。在将主体解码为 JSON 之前，服务会验证 telnyx-signature-ed25519 和 telnyx-timestamp 标头。

Verify the webhook before parsing it

The sample uses the current Telnyx Python SDK client and its webhook unwrapping helper. 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.

在解析前验证 Webhook

该示例使用了当前的 Telnyx Python SDK 客户端及其 Webhook 解包辅助工具。受限的时间戳窗口有助于拒绝重放请求。如果签名缺失、格式错误、过期或无效，则返回 False，路由将响应 401 而不处理有效负载。这里有一个 SDK 细节需要注意：unwrap 参数是 key，而不是 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. The result is cached for 60 seconds by default, so a burst of echo webhooks does not trigger the same lifecycle request repeatedly.

检查号码的共存状态

有效的 Webhook 并不自动意味着 echo 流已准备好使用。监控器还会检查两个生命周期字段：is_on_biz_app 用于确认号码是否已在 WhatsApp Business 应用程序上注册。coexistence_state 必须为 active，事件才能被写入审计追踪。在实时模式下，StateChecker 会从 WhatsApp Business 账户中检索配置的号码。结果默认缓存 60 秒，因此突发的 echo Webhook 不会重复触发相同的生命周期请求。

Acknowledge mid-sync events without logging them

If the number is not active, the monitor returns 200 but skips the audit write. 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 and cost. Unexpected values are preserved as audit signals rather than silently discarded.

在同步过程中确认但不记录事件

如果号码未激活，监控器会返回 200 但跳过审计写入。这是一个有意的处理决定，而非服务器故障。返回成功响应可以防止应用程序已经评估过的事件被重复投递 Webhook。对于活跃号码，处理程序会记录两个额外的判定结果：origin_matches_biz_app 和 cost。意外的值会被保留为审计信号，而不是被静默丢弃。

Keep the audit trail useful without storing phone numbers

The sample writes one JSON object per line. 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.

在不存储电话号码的情况下保持审计追踪的有效性

该示例每行写入一个 JSON 对象。记录中有意省略了 from、to 和货币值。审计追踪依然显示了应用程序接收到的内容以及如何评估该事件，而不会成为客户电话号码的存储库。JSONL 格式使该示例易于运行和检查。生产环境的实现可以在保留相同接口的同时，将文件替换为数据库或队列，并对消息文本应用组织自身的保留和访问策略。

Run the example

Configure the environment with placeholders or your own credentials, then start the Flask service: python app.py. 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.

运行示例

使用占位符或您自己的凭据配置环境，然后启动 Flask 服务：python app.py。健康检查端点位于 GET /health。要接收真实的 Webhook，请通过公共 HTTPS URL 公开 Flask 服务，在 Telnyx 中配置该 URL，将 DEMO_MODE 设置为 false，并提供您账户的 WhatsApp 和 Telnyx 凭据。