语音智能体的服务端 Webhook：验签、幂等与订阅

页面关了，事情还得继续。审计、告警、计费、跨系统联动这些事不能指望用户的浏览器一直开着。

所以分工是清楚的：客户端事件负责更新当前页面，服务端 Webhook 负责剩下那些不能丢的部分，由云端主动推给你，和用户在不在线无关。

这篇教程搭一个能直接部署的 FastAPI 接收器，四件事做到位——只接对话式 AI 引擎的事件、按原始请求体验签、用 noticeId 去重、把新事件推给内部看板。

接上之后，回调进来能正常验签入库；把同一条通知重发一次，数据库里仍然只有一条记录。

一. 架构与准备工作

声网对话式 AI 引擎
        │ POST /ncsNotify + Agora-Signature-V2
        ▼
公网 HTTPS 接收器 ── HMAC-SHA256 验签 ── productId / eventType 校验
        │
        ├── noticeId 幂等写入 SQLite
        └── SSE 推送到受保护的内部看板

你需要在控制台启用消息通知服务、拿到通知密钥，并准备一个公网可访问的 HTTPS 地址。对话式 AI 引擎的 productId 是 17；本文处理 101 加入、102 离开、103 历史、104 Token 即将过期、110 错误、111 指标、112 轮次结束这几类事件。

二. 环境变量

AGORA_NOTIFICATION_SECRET=REPLACE_WITH_NOTIFICATION_SECRET
WEBHOOK_DASHBOARD_TOKEN=REPLACE_WITH_RANDOM_DASHBOARD_TOKEN
WEBHOOK_ADMIN_TOKEN=REPLACE_WITH_RANDOM_ADMIN_TOKEN
WEBHOOKS_DB_PATH=webhooks.db

三个密钥都只放服务端。AGORA_NOTIFICATION_SECRET 要和控制台的消息通知配置一致。另外注意：没配密钥时应该直接返回错误，不能自动切到「跳过验签」模式——那等于把接口对全网敞开。

三. 完整代码

下面这段保留了接收链路最关键的四个性质：用原始字节验签、只接受产品 17、限制事件类型、按 noticeId 幂等。

import hashlib
import hmac
import json
import os
import sqlite3
import time

from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()
DB_PATH = os.getenv("WEBHOOKS_DB_PATH", "webhooks.db")
KNOWN_EVENTS = {101, 102, 103, 104, 110, 111, 112}

def connect() -> sqlite3.Connection:
    db = sqlite3.connect(DB_PATH)
    db.execute("""
        CREATE TABLE IF NOT EXISTS webhook_events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            notice_id TEXT NOT NULL UNIQUE,
            event_type INTEGER NOT NULL,
            payload_json TEXT NOT NULL,
            received_at REAL NOT NULL
        )
    """)
    db.commit()
    return db

def verify(raw: bytes, signature: str | None) -> bool:
    secret = os.getenv("AGORA_NOTIFICATION_SECRET", "")
    if not secret or not signature:
        return False
    expected = hmac.new(
        secret.encode(), raw, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(
        expected.lower(), signature.strip().lower()
    )

def save_event(db: sqlite3.Connection, payload: dict) -> bool:
    notice_id = str(payload.get("noticeId") or "").strip()
    product_id = int(payload.get("productId"))
    event_type = int(payload.get("eventType"))
    if not notice_id:
        raise ValueError("noticeId 不能为空")
    if product_id != 17 or event_type not in KNOWN_EVENTS:
        raise ValueError("不是受支持的对话式 AI 事件")
    try:
        db.execute(
            """INSERT INTO webhook_events(
                notice_id, event_type, payload_json, received_at
            ) VALUES (?, ?, ?, ?)""",
            (
                notice_id,
                event_type,
                json.dumps(payload, ensure_ascii=False),
                time.time(),
            ),
        )
        db.commit()
        return True
    except sqlite3.IntegrityError:
        return False

@app.post("/ncsNotify")
async def notify(
    request: Request,
    signature: str | None = Header(
        None, alias="Agora-Signature-V2"
    ),
):
    raw = await request.body()
    if not verify(raw, signature):
        raise HTTPException(401, "invalid signature")
    try:
        payload = json.loads(raw)
        with connect() as db:
            created = save_event(db, payload)
    except (json.JSONDecodeError, TypeError, ValueError) as exc:
        raise HTTPException(400, str(exc)) from exc
    return {"code": 0, "duplicate": not created}

配套交付的完整代码还提供了受 Bearer 保护的 /webhooks/recent、SSE /webhooks/stream，以及受管理员密钥保护的删除接口。

别把事件原文直接写进应用日志。用户标识和转写内容都要按业务要求脱敏，并设好保留期。

四. 给智能体加可追踪标签

创建智能体时打上业务标签，回调时这些标签会原样带回来，服务端就能把事件关联到具体的租户或业务会话，排查线上问题时特别省事。

agent = agent.with_labels({
    "recipe": "server-webhooks",
    "tenant": "demo",
})

标签不是鉴权信息，别往里放密钥、手机号、身份证号。

五. 运行与控制台配置

cd code/server-webhooks
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env.local
uvicorn webhook_service:app --host 0.0.0.0 --port 9100

部署到公网 HTTPS 后，在控制台的消息通知配置里填：

回调地址：https://REPLACE_WITH_PUBLIC_HOST/ncsNotify
产品：对话式 AI 引擎
通知密钥：与 AGORA_NOTIFICATION_SECRET 相同

不想等真实回调的话，本地可以自己构造一条正确签名先把链路跑通：

BODY='{"noticeId":"demo-1","productId":17,"eventType":101,"payload":{"agentId":"demo"}}'
SIGNATURE=$(BODY="$BODY" python -c \
  'import hashlib,hmac,os; print(hmac.new(os.environ["AGORA_NOTIFICATION_SECRET"].encode(),os.environ["BODY"].encode(),hashlib.sha256).hexdigest())')
curl -X POST http://127.0.0.1:9100/ncsNotify \
  -H "Agora-Signature-V2: $SIGNATURE" \
  -H 'Content-Type: application/json' \
  -d "$BODY"

六. 应该看到什么

第一次请求返回 {"code":0,"duplicate":false}；把完全相同的 noticeId 再发一次，返回 duplicate:true，数据库里仍然只有一条记录。签名错误返回 401，产品或事件类型不对返回 400。

上线前这几项要全部走通：控制台的真实回调、至少收到 101/102/110/111 四类事件、重复通知幂等、签名错误有告警、数据库不可用时能重试、标签和业务会话正确关联。

七. 故障排查

• 验签总是失败：必须用**未经解析和重排的原始请求体字节**算 HMAC。先 json.loads() 再序列化回去，字段顺序和空格都可能变，签名必然对不上。这是这类接口最高频的失败原因。
• 收到重复事件：通知系统会重试，这是正常现象，靠 noticeId 做唯一键解决，不是 bug。
• 控制台显示投递失败：检查公网证书链、回调路径、响应时延和防火墙。别填 localhost。
• 看板 SSE 没数据：接收端只推送验签通过且第一次入库的事件，重复通知不会二次推送。

八. 下一步

服务端拿到事件之后，客户端这条线也补上，一次会话才能从两头看全。

《语音智能体可观测性：状态、延迟、错误与字幕》