本文含工具体验分享

做海外私域这些年，WhatsApp Business API 的坑我基本踩了个遍。从申请被拒、webhook验签失败，到回调消息丢失、模板消息状态追踪断链——每一步都有值得单独写一篇的经验。今天就把核心链路捋一遍，按 Business API 开通 → 消息回调搭建 → 模板发送 → 状态追踪这条线,串一套可落地的 Python 方案。

一、Business API 申请流程与前期准备

开通 WhatsApp Business API 不是注册一个 App 就完事，有几个前置关卡需要提前准备。

必备条件：

• 一个通过 Meta 企业认证的 Business Manager 账号
• 一个已注册的 Meta 应用（App Type 选 Business）
• 一个已绑定的 WhatsApp Business Account（WABA）
• 一个已验证的 Business 电话号码（未被其他 WhatsApp 账号占用）

实测申请步骤：

1. 登录 developers.facebook.com，创建或选择已有应用，在 Add Product 中添加 WhatsApp
2. 在 WhatsApp 面板中创建 WABA，选择 WABA 名称和时区
3. 添加收信电话号码——这里要注意，号码必须能接收国际短信或语音验证码，国内手机号实测成功率不高，建议用境外实体号或虚拟号（部分虚拟号也会被拒）
4. 完成号码验证后，WABA 会进入沙盒模式（Sandbox），此时只能给管理后台添加的测试号码发消息

我踩过的坑：第一次申请时，直接用国内手机号注册的 Facebook 账号创建的 App，WABA 一直卡在"待审核"。后来发现是 Business Manager 的企业认证没有完成——Meta 要求营业执照信息与 BM 注册信息一致。换了个已完成认证的 BM 后一天就过了。

拿到 API 访问权限后，在 WhatsApp > API Setup 面板获取 临时访问令牌（Temporary Access Token），有效期 24 小时，正式环境需要换成永久令牌（Permanent Token）或走系统用户（System User）的 Token。

二、消息回调 Webhook 验证

WhatsApp 的消息回调使用 Webhook 机制，Meta 服务器会向你的回调地址发送 POST 请求。配置 Webhook 时需要先过一轮验证：

1. 你的服务器收到一个 GET 请求，参数包含 hub.mode、hub.challenge、hub.verify_token
2. 如果 hub.verify_token 与你预设的 token 一致，返回 hub.challenge 的值
3. Meta 收到正确响应后，Webhook 配置完成

下面是 Python Flask 实现的验证端点：

"""
WhatsApp Business API Webhook 验证与消息回调接收
依赖: pip install flask requests
"""

import os
import json
import logging
from flask import Flask, request, jsonify

# 日志配置
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("whatsapp_webhook")

app = Flask(__name__)

# 配置项 —— 从环境变量读取，避免硬编码
VERIFY_TOKEN = os.environ.get("WHATSAPP_VERIFY_TOKEN", "my_custom_verify_token_2024")
ACCESS_TOKEN = os.environ.get("WHATSAPP_ACCESS_TOKEN", "")
PHONE_NUMBER_ID = os.environ.get("WHATSAPP_PHONE_NUMBER_ID", "")


@app.route("/webhook", methods=["GET", "POST"])
def whatsapp_webhook():
    """WhatsApp 消息回调入口"""
    if request.method == "GET":
        # Meta 发送的 Webhook 验证请求
        mode = request.args.get("hub.mode")
        challenge = request.args.get("hub.challenge")
        verify_token = request.args.get("hub.verify_token")

        logger.info(f"收到验证请求: mode={mode}, token={verify_token}")

        if mode == "subscribe" and verify_token == VERIFY_TOKEN:
            logger.info("Webhook 验证成功")
            return challenge, 200
        else:
            logger.warning("Webhook 验证失败: token 不匹配")
            return "Forbidden", 403

    if request.method == "POST":
        # 接收消息回调
        payload = request.get_json()
        logger.info(f"收到消息回调: {json.dumps(payload, indent=2, ensure_ascii=False)}")

        # 处理每种消息变更入口
        if "entry" not in payload:
            return "OK", 200

        for entry in payload["entry"]:
            for change in entry.get("changes", []):
                if change.get("field") == "messages":
                    # 消息回调
                    handle_incoming_message(change.get("value", {}))
                elif change.get("field") == "message_template_status_update":
                    # 模板消息状态回调
                    handle_template_status_update(change.get("value", {}))

        return "OK", 200


def handle_incoming_message(value: dict) -> None:
    """处理收到的用户消息"""
    messages = value.get("messages", [])
    contacts = value.get("contacts", [])

    if not messages:
        return

    for msg in messages:
        sender = msg.get("from", "")
        msg_type = msg.get("type", "unknown")

        logger.info(f"收到消息 from={sender} type={msg_type}")

        # 根据消息类型分发处理
        if msg_type == "text":
            text_body = msg.get("text", {}).get("body", "")
            logger.info(f"文本内容: {text_body}")

        elif msg_type == "image":
            image_id = msg.get("image", {}).get("id", "")
            caption = msg.get("image", {}).get("caption", "")
            logger.info(f"收到图片 id={image_id} caption={caption}")

        elif msg_type == "interactive":
            # 交互式消息（按钮/列表回复）
            interactive_data = msg.get("interactive", {})
            logger.info(f"交互式消息: {json.dumps(interactive_data)}")


def handle_template_status_update(value: dict) -> None:
    """处理模板消息发送状态更新"""
    statuses = value.get("message_template_status_updates", [])

    for status in statuses:
        msg_id = status.get("message_id", "")
        biz_status = status.get("event", "")
        reason = status.get("reason", {}).get("description", "")

        logger.info(f"模板状态更新: msg_id={msg_id} status={biz_status} reason={reason}")

        # 典型状态: sent / delivered / read / failed
        update_send_status(msg_id, biz_status, reason)


def update_send_status(msg_id: str, status: str, reason: str) -> None:
    """更新本地数据库中的消息发送状态（示例用日志代替DB写入）"""
    # 实际项目中这里会写入数据库 update whatsapp_messages set status=... where msg_id=...
    valid_statuses = {"sent", "delivered", "read", "failed", "deleted"}
    if status in valid_statuses:
        logger.info(f"状态更新: [{status}] {msg_id}")
    else:
        logger.warning(f"未知状态类型: {status}")


if __name__ == "__main__":
    # 开发环境使用 flask run，生产环境配合 gunicorn
    app.run(host="0.0.0.0", port=5000, debug=False)

注意事项：

• Webhook 回调地址必须是 HTTPS 公网地址，本地开发可以用 ngrok 做穿透 ngrok http 5000
• Meta 的回调有 20 秒超时限制，handle_incoming_message 内的耗时操作（如下载媒体文件、调 AI 接口）必须异步处理，否则 Meta 会认为回调失败并重试
• 返回状态码必须为 200，否则 Meta 会持续重发消息

三、通过 API 发送模板消息

沙盒模式下先给测试号码发一条消息确认连通性。正式使用前需要提交通知模板（Notification Template）并通过审核。

"""
通过 WhatsApp Business API 发送模板消息
"""

import requests

WHATSAPP_API_BASE = "https://graph.facebook.com/v19.0"


def send_template_message(to_phone: str, template_name: str,
                          language_code: str = "zh_CN",
                          components: list = None) -> dict:
    """
    发送 WhatsApp 模板消息

    Args:
        to_phone: 接收者手机号（含国际区号，如 8613800138000）
        template_name: 已审核通过的模板名称
        language_code: 模板语言代码
        components: 可选，header/body 参数列表
    """
    url = f"{WHATSAPP_API_BASE}/{PHONE_NUMBER_ID}/messages"

    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }

    body = {
        "messaging_product": "whatsapp",
        "to": to_phone,
        "type": "template",
        "template": {
            "name": template_name,
            "language": {"code": language_code},
        },
    }

    # 如果模板有变量参数，填充到 body 区域
    if components:
        body["template"]["components"] = components

    logger.info(f"发送模板消息 to={to_phone} template={template_name}")

    try:
        resp = requests.post(url, headers=headers, json=body, timeout=15)
        result = resp.json()
        logger.info(f"发送结果: {json.dumps(result, ensure_ascii=False)}")

        if resp.status_code == 200:
            return {"success": True, "message_id": result["messages"][0]["id"]}
        else:
            error_msg = result.get("error", {}).get("message", "未知错误")
            return {"success": False, "error": error_msg}

    except requests.RequestException as e:
        logger.error(f"API 请求异常: {e}")
        return {"success": False, "error": str(e)}


# 使用示例
if __name__ == "__main__":
    result = send_template_message(
        to_phone="8613800138000",
        template_name="order_confirmation",
        language_code="zh_CN",
        components=[
            {
                "type": "body",
                "parameters": [
                    {"type": "text", "text": "#20240701"},
                    {"type": "text", "text": "3-5个工作日"},
                ],
            }
        ],
    )
    print(f"发送结果: {result}")

关于模板审核：模板审核周期通常 1-24 小时，名称不能含 emoji，body 参数用 {{1}}、{{2}} 占位。审核被拒的常见原因我后面单独整理了一篇——简单说就是营销味太重的模板容易挂，尽量走"实用通知"路线。

四、发送状态追踪的完整链路

消息发出后，WhatsApp 会通过 Webhook 回调五种核心状态：

状态	含义	触发时机
sent	已送达 WhatsApp 服务器	消息被 Meta 接收并排队
delivered	已送达用户设备	用户手机收到消息
read	用户已读	用户点开对话（需开启已读回执）
failed	发送失败	号码无效、网络问题等
deleted	消息被删除	用户或系统删除消息

我们可以利用回调中的 wamid（WhatsApp Message ID）来关联发送记录和送达状态。实际项目中，一般会在 send_template_message 成功后，将返回的 messages[0].id 存入数据库，再在 update_send_status 中根据这个 ID 更新状态。

下面是状态追踪的增强版代码片段：

import sqlite3
from datetime import datetime, timezone

# 简易 SQLite 消息状态表
def init_message_db(db_path: str = "whatsapp_messages.db"):
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS message_status (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            wamid TEXT UNIQUE NOT NULL,            -- WhatsApp 返回的消息ID
            recipient TEXT NOT NULL,                -- 接收者手机号
            template_name TEXT DEFAULT '',          -- 模板名称
            status TEXT DEFAULT 'pending',          -- pending/sent/delivered/read/failed
            sent_at TEXT,                           -- 发送时间
            delivered_at TEXT,                      -- 送达时间
            read_at TEXT,                           -- 已读时间
            error_reason TEXT DEFAULT '',           -- 失败原因
            created_at TEXT DEFAULT (datetime('now'))
        )
    """)
    conn.commit()
    return conn


def record_send_result(conn, wamid: str, recipient: str, template_name: str) -> None:
    """发送成功后记录初始状态"""
    cur = conn.cursor()
    cur.execute(
        """INSERT OR REPLACE INTO message_status (wamid, recipient, template_name, status, sent_at)
           VALUES (?, ?, ?, 'sent', ?)""",
        (wamid, recipient, template_name, datetime.now(timezone.utc).isoformat()),
    )
    conn.commit()


def update_status_by_wamid(conn, wamid: str, new_status: str, reason: str = "") -> None:
    """根据 wamid 更新消息状态"""
    now = datetime.now(timezone.utc).isoformat()
    cur = conn.cursor()

    status_col_map = {
        "delivered": "delivered_at",
        "read": "read_at",
    }

    if new_status in status_col_map:
        col = status_col_map[new_status]
        cur.execute(
            f"UPDATE message_status SET status=?, {col}=?, error_reason=? WHERE wamid=?",
            (new_status, now, reason, wamid),
        )
    else:
        cur.execute(
            "UPDATE message_status SET status=?, error_reason=? WHERE wamid=?",
            (new_status, reason, wamid),
        )
    conn.commit()
    logger.info(f"状态同步: {wamid} -> {new_status}")

五、方案选型对比

维度	自建 Business API	WADesk 消息通道	原生 WhatsApp Business App
接入复杂度	高，需自建 Webhook、Token 管理	低，可视化配置回调地址	低，无 API
群发支持	单条调 API，需自建队列	内置群发引擎，支持调度	仅广播列表（受限）
回调可靠性	需自己处理重试和异常	自带重试+消息队列保障	不支持回调
模板管理	手动提审、追踪	模板库+批量提审	无模板消息
适用场景	中大团队，有开发能力	中小外贸团队，上手快	个体户/微商

自建适合对数据自主性要求高、有专职开发人员的团队。如果团队只一两个人，从零搭这套链路成本确实不小。WADesk 这类工具把 Webhook 验证、Token 轮换、回调重试封装成了配置项，实测部署时间从两三天压缩到半天以内。不过核心的消息服务和数据仍然走 Meta 官方 API，不用担心合规问题。

六、注意事项与避坑建议

1. Token 有效期：临时 Token 只有 24 小时，正式环境务必使用系统用户 Token 或长效 Token。建议用 cron 定时刷新并记录到配置中心。
2. 回调幂等性：Meta 在高负载时可能重复推送同一条消息，务必在业务逻辑中按 wamid 做去重。
3. 号码质量：Business API 对号码质量有严格限制，如果大量发送后用户举报/拉黑，号码会被限制甚至封禁。建议配合客户意向标签，只向明确 opt-in 的用户发消息。
4. 速率限制：单个号码每秒最多 20 条消息，大量群发需自建流量整形队列。

下一步建议先搭好 Webhook 验证和消息接收的基础架子，确认能正常收到回调后再接入模板发送和状态追踪——这样问题排查起来最简单。下一篇文章我会把 Webhook 服务的高可用架构和高并发优化展开讲。