接收回調

希望先跑通介面和簽名回撥時，見 Demo 配置即時回撥：先建立地址，再帶地址 ID 和金鑰啟動 Demo，最後驗證地址。

本頁介紹用 Python SDK 接收事件回撥和同步回撥：校驗簽名、防止重放、按事件 ID 去重、把事件和同步回撥分派給你的處理函式、自動回應 URL 驗證，以及與 Flask、Django、FastAPI、Starlette 的整合和本地測試。

回撥的處理在 deeprespond_im.callback 中：同步程式用 CallbackHandler，非同步程式用 AsyncCallbackHandler。處理器與 Web 框架無關，接收請求頭和原始請求體，返回狀態碼、響應頭和響應體；框架適配只負責取出原始請求體和寫回響應。

要做什麼	看哪裡
先跑通一個接收端	快速開始
處理事件、交給 Celery 等佇列	處理事件回撥
在訊息傳送前稽核、拒絕或修改	處理同步回撥
多程序部署，用 Redis 防重放和去重	防重放與去重的儲存
接入 Flask、Django、FastAPI 等框架	與 Web 框架整合
對回撥做單元測試	本地測試

快速開始

以 Flask 為例，處理訊息抄送事件和發訊息前的同步回撥：

import os

from flask import Flask

from deeprespond_im.callback import Allow, CallbackHandler, Event, MemoryDedupStore, MemoryNonceStore, Reject, Verdict
from deeprespond_im.callback.events import MessageSent
from deeprespond_im.callback.flask import callback_view
from deeprespond_im.callback.hooks import MessageBeforeSend

handler = CallbackHandler(
    # 回调地址 ID → 这个地址的密钥列表（轮换期间新旧两个，新的在前）
    secrets={os.environ["IM_CALLBACK_ENDPOINT_ID"]: os.environ["IM_CALLBACK_SECRETS"].split(",")},
    nonce_store=MemoryNonceStore(),      # 单个进程；多进程、多台服务器部署时用 Redis 的实现
    dedup_store=MemoryDedupStore(),
)


@handler.on_event("message.sent")
def archive(event: Event, data: MessageSent) -> None:
    print("消息", data.message_id, data.body_raw.text if data.body_raw else None)


@handler.on_hook("message.before_send")
def check(request: MessageBeforeSend) -> Verdict:
    if "加微信" in str(request.body.get("text", "")):
        return Reject("ad", "消息中有不允许的内容")
    return Allow()


app = Flask(__name__)
app.add_url_rule("/im/callback", view_func=callback_view(handler), methods=["POST"])

在控制台把 https://你的域名/im/callback 配置為回撥地址，記下地址 ID 和簽名金鑰（見回撥地址）。儲存地址時 IM 服務傳送 URL 驗證請求，處理器自動回應，不需要另外處理。

處理器做了什麼

handler.handle(headers, body) 對每個請求依次：

1. 請求體超過 max_body_bytes（預設 2 MB）的返回 413；
2. 按請求頭 X-IM-Endpoint-Id 找到這個地址的金鑰，用每個金鑰校驗 X-IM-Signature，再檢查 X-IM-Timestamp 與本機時間相差不超過 5 分鐘（演算法見驗證回撥請求）；
3. 用隨機數儲存檢查 X-IM-Nonce 在 10 分鐘內沒有出現過（防重放）；
4. 解析請求體，核對其中的 endpoint_id 與請求頭相同；
5. 按請求的種類處理：URL 驗證回應 challenge；同步回撥呼叫對應的處理函式，返回它的結果；事件回撥逐個事件按事件 ID 去重，呼叫對應的處理函式。

情況	響應
事件全部處理成功，或是已經處理過的重複事件	204
一批中有事件處理失敗（處理函式丟擲異常），或同一個事件正在另一個請求中處理	503，IM 服務稍後重發整批；已成功的事件按事件 ID 去重，不會再次處理
處理函式丟擲 RetryLater(seconds)	503，帶 Retry-After（最長 1 小時），IM 服務按它推遲重發
同步回撥的處理函式返回了結果	200，響應體為結果
同步回撥超時、出錯、返回的結果不合規，或沒有註冊處理函式	hook_fallback 給出的結果，沒有配置時 503（IM 服務按同步回撥的失敗處理規則處理）
URL 驗證	200，{"challenge": "..."}
地址 ID 不在配置中、簽名不對、時間戳過期、隨機數重複、endpoint_id 不一致	401
請求體不是合法的回撥請求	400
金鑰提供函式、隨機數儲存或去重儲存出錯	503

校驗失敗時 SDK 記一條 ERROR 日誌（日誌器 deeprespond_im.callback），說明原因和排查方向，如“檢查金鑰配置，以及請求體是否被框架、中介軟體或代理改動過”。

必須用原始請求體

簽名按原始請求體的位元組計算。交給 handle() 的必須是未經解析的原始位元組，不能是框架解析後重新序列化的結果（如 Pydantic 模型的 model_dump_json()）。SDK 提供的框架適配已經正確處理了這一點。

建立處理器

引數	預設值	說明
secrets	必填	地址 ID 到金鑰列表的對映，如 {"1840200000000000007": ["cbs_新密钥", "cbs_旧密钥"]}；也可以是按地址 ID 返回金鑰列表的函式（找不到時返回空列表），非同步處理器中可以是協程函式。金鑰列表不能寫成單個字串（丟擲 TypeError）
nonce_store	必填	隨機數儲存，見防重放與去重的儲存
dedup_store	必填	事件去重儲存
queue	None	事件佇列：事件交給它後立即返回，見交給佇列處理。與 on_event 二選一
hook_fallback	None	同步回撥超時或出錯時返回的結果，如 Allow()；None 表示返回 503
max_concurrent_hooks	16	同時執行的同步回撥處理函式的上限，已滿時新的同步回撥直接按 hook_fallback 或 503 返回，不排隊
claim_ttl	30.0	處理一個事件的佔用時限（秒）：超過這個時間沒有處理完的，重發的請求可以再次處理它
max_body_bytes	2097152	請求體的大小上限（位元組）
clock	time.time	校驗時間戳用的時鐘，測試時可以替換

一個回撥地址的多個金鑰（輪換金鑰的寬限期內）都寫進列表，任何一個校驗通過即可。多個地址指向同一個接收端時，每個地址一項；多個應用共用一個接收端時，處理函式中可以用 current_request().app 區分應用。

處理事件回撥

用 on_event 註冊事件的處理函式，函式收到 Event 和按事件型別解析好的資料類：

from deeprespond_im.callback import CallbackHandler, Event, MemoryDedupStore, MemoryNonceStore, current_request
from deeprespond_im.callback.events import GroupMembersAdded, UserCreated

handler = CallbackHandler(
    secrets={"1840200000000000007": ["cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE"]},
    nonce_store=MemoryNonceStore(),
    dedup_store=MemoryDedupStore(),
)


@handler.on_event("user.created")
def on_user_created(event: Event, data: UserCreated) -> None:
    print(current_request().app, "新用户", data.username, "来自", data.created_via, "第", event.attempt, "次投递")


@handler.on_event("group.members_added")
def on_members_added(event: Event, data: GroupMembersAdded) -> None:
    print("群", data.group_id, "加入", [m.username for m in data.members])


@handler.on_unknown_event
def on_unknown(event: Event) -> None:
    print("SDK 还不认识的事件", event.type, event.data)

• Event 有 id、type、occurred_at、attempt（第幾次投遞）、test（控制台的測試傳送）、data（dict）和 data_raw（原文）；event.decode() 按型別返回資料類；
• 資料類在 deeprespond_im.callback.events 中，每種事件一個，類名由事件名轉換而來，如 message.sent 為 MessageSent、friend_request.received 為 FriendRequestReceived、rtc.call_ended 為 RtcCallEnded。欄位與事件列表相同，另有 origin（事件由誰引起）、truncated（超過 70 KB 被截短，這時需要按 ID 呼叫 OpenAPI 讀取）和 raw；訊息類事件另有 body_raw、ext_raw，與請求體中的原文逐位元組相同；
• 註冊時檢查事件名：寫錯的事件名在註冊時丟擲 ValueError，不會靜默失效。要處理 SDK 還不認識的新事件，用 on_unknown_event，或註冊時傳 allow_unknown=True（這時第二個引數是 dict）；沒有註冊處理函式的事件直接確認收到；
• 處理函式正常返回即成功；丟擲任何異常即失敗，這一批返回 503，IM 服務稍後重發。需要指定重發時間的，丟擲 RetryLater(seconds)；
• 同一個請求中的多個事件依次處理，不併發；同一個事件重發時（事件 ID 相同），已經處理成功的不再呼叫處理函式；
• 公共資訊：處理函式中用 current_request() 取得這個請求的 kind、request_id、app（AppKey）、endpoint_id、sent_at，線上程和非同步任務中都正確。

事件可能重複和亂序到達（見重複與亂序）：SDK 按事件 ID 去重，但業務上請按事件中的版本號、序號或時間判斷新舊。

處理函式要儘快返回：IM 服務等待響應的時間有限，訊息抄送這樣量大的事件，建議只寫入自己的佇列或資料庫，再非同步處理。

交給佇列處理

配置了 queue 的處理器不呼叫處理函式：校驗、去重後把每個事件交給佇列的 enqueue(event)，返回即視為收到，之後由你的佇列（Celery、RQ、Kafka、資料庫表等）負責處理；enqueue 丟擲異常時這一批返回 503。

from deeprespond_im.callback import CallbackHandler, Event, MemoryDedupStore, MemoryNonceStore


class CeleryQueue:
    def enqueue(self, event: Event) -> None:
        # 例如：process_im_event.delay(event.id, event.type, event.data_raw.text if event.data_raw else "{}")
        print("入队", event.id, event.type)


handler = CallbackHandler(
    secrets={"1840200000000000007": ["cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE"]},
    nonce_store=MemoryNonceStore(),
    dedup_store=MemoryDedupStore(),
    queue=CeleryQueue(),
)

非同步處理器的佇列用協程方法 async def enqueue(self, event)。

處理同步回撥

用 on_hook 註冊同步回撥的處理函式，函式收到請求的資料類，返回一個結果：

結果	說明
Allow()	放行
Reject(reason=None, message=None)	拒絕。reason 為 1 到 64 個小寫字母、數字或 _ . -；message 是給使用者看的提示，不超過 256 個字元，不能有換行等控制字元
Modify(body=..., ext=...)	修改訊息，只用於 message.before_send 和 chatroom.before_send。body 為 dict 或 RawJSON；ext 另外可以為 None（去掉 ext）；沒給的保持不變，至少給一個
Partial([RejectedTarget(username, reason, message), ...])	部分拒絕，只用於 group.before_join 的邀請和建群（joined_via 為 invite、create），被拒絕的使用者必須在請求的 targets 中

同步回撥	請求的資料類（deeprespond_im.callback.hooks）
message.before_send	MessageBeforeSend
chatroom.before_send	ChatroomBeforeSend
friend.before_add	FriendBeforeAdd
group.before_join	GroupBeforeJoin

import time

from deeprespond_im.callback import (
    Allow,
    CallbackHandler,
    MemoryDedupStore,
    MemoryNonceStore,
    Modify,
    Partial,
    RejectedTarget,
    Verdict,
)
from deeprespond_im.callback.hooks import GroupBeforeJoin, MessageBeforeSend

handler = CallbackHandler(
    secrets={"1840200000000000007": ["cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE"]},
    nonce_store=MemoryNonceStore(),
    dedup_store=MemoryDedupStore(),
    hook_fallback=Allow(),   # 处理函数超时或出错时放行
)

BLOCKED = {"blocked_user"}


@handler.on_hook("message.before_send")
def mask(request: MessageBeforeSend) -> Verdict:
    text = request.body.get("text")
    if isinstance(text, str) and "敏感词" in text:
        return Modify(body={**request.body, "text": text.replace("敏感词", "***")})
    return Allow()


@handler.on_hook("group.before_join")
def check_join(request: GroupBeforeJoin) -> Verdict:
    remaining = request.deadline - time.monotonic()   # 留给这个处理函数的时间（秒）
    print("还有", round(remaining * 1000), "毫秒")
    rejected = [RejectedTarget(username=u, reason="blocked") for u in request.targets if u in BLOCKED]
    if rejected and request.joined_via in ("invite", "create"):
        return Partial(rejected)
    return Allow()

• 時限：處理函式必須在請求中的 timeout_ms 減 50 毫秒內返回。資料類的 timeout 是 IM 服務給出的等待時間（秒），deadline 是 time.monotonic() 下的截止時刻，可以據此給資料庫查詢、外部請求設定時限。超時後處理器立即返回 hook_fallback 或 503；
• 同步處理器的處理函式在 SDK 的守護執行緒中執行：Python 不能強行結束執行緒，超時的處理函式在後臺繼續執行到結束，它的結果被丟棄並記警告日誌。所以處理函式應自己儘快結束；同時執行的數量受 max_concurrent_hooks 限制。非同步處理器中的協程處理函式超時後被取消；
• 返回前檢查結果：結果不合規（如 Reject 的 message 有換行、Modify 用在了不支援的同步回撥上）時按處理出錯處理，並記錯誤日誌；處理函式返回 None（忘了 return）同樣視為出錯，不當作放行；
• test 為真的是控制台“測試傳送”產生的請求。

防重放與去重的儲存

處理器需要兩個儲存：隨機數儲存記住 10 分鐘內見過的 X-IM-Nonce，防止請求被原樣重放；去重儲存記住處理過的事件 ID，同一個事件重發時不再處理。

部署方式	隨機數儲存	去重儲存
單個程序（開發、單程序的服務）	MemoryNonceStore()	MemoryDedupStore()
多個工作程序（gunicorn、uWSGI）、多臺伺服器	deeprespond_im.redis.RedisNonceStore(r)	deeprespond_im.redis.RedisDedupStore(r)
非同步處理器	AsyncMemoryNonceStore() 或 AsyncRedisNonceStore(r)	AsyncMemoryDedupStore() 或 AsyncRedisDedupStore(r)

記憶體儲存只在一個程序內有效：多個工作程序各有一份，起不到防重放和去重的作用。處理器啟動時如果環境變數 WEB_CONCURRENCY 大於 1 而仍用記憶體儲存，會記一條警告日誌；多臺伺服器的部署檢測不到，請自己確認。

import os

import redis

from deeprespond_im.callback import CallbackHandler
from deeprespond_im.redis import RedisDedupStore, RedisNonceStore

r = redis.Redis.from_url(os.environ["REDIS_URL"])
handler = CallbackHandler(
    secrets={os.environ["IM_CALLBACK_ENDPOINT_ID"]: os.environ["IM_CALLBACK_SECRETS"].split(",")},
    nonce_store=RedisNonceStore(r, key_prefix="drim:"),
    dedup_store=RedisDedupStore(r, key_prefix="drim:", retention=7 * 24 * 3600),
)

• Redis 實現接受你建立的 redis.Redis（非同步為 redis.asyncio.Redis）客戶端，decode_responses 開啟與否都可以。鍵為 {key_prefix}nonce:{随机数} 和 {key_prefix}event:{事件 ID}；
• retention 是事件處理完成後保留去重記錄的時長：記憶體儲存預設 24 小時，Redis 預設 7 天（失敗的投遞在控制台保留 7 天，可以在這期間手動重新投遞）；
• 佔用：開始處理一個事件時先佔用它（claim_ttl，預設 30 秒），處理成功後標記完成，失敗時釋放，讓重發的請求可以再次處理。處理中的事件又被重發到另一個程序時，那個請求返回 503，稍後再試；
• 記憶體儲存的上限：預設各 1000000 條（max_entries）。去重儲存滿了淘汰最早的記錄並記警告日誌；隨機數儲存不淘汰還沒到期的隨機數（淘汰了就能重放），滿了時新的請求返回 503；
• 也可以用其他儲存實現，提供以下方法即可：NonceStore.seen(nonce, ttl) -> bool（已存在返回 True，否則記下並返回 False），DedupStore.claim(event_id, ttl) -> (ClaimState, token | None)、complete(event_id, token)、release(event_id, token)（只在佔用令牌相同時釋放）。非同步版的方法是協程。

與 Web 框架整合

介面卡只負責取出原始請求體和請求頭、寫回響應。同步處理器和非同步處理器都可以傳入：

框架	匯入	說明
Flask	deeprespond_im.callback.flask.callback_view	用 request.get_data() 取原始請求體（帶快取，別處讀過 request.json 也不影響）。傳入非同步處理器時返回非同步檢視，需要 flask[async]
Django	deeprespond_im.callback.django.callback_view	用 request.body；檢視已標為不檢查 CSRF、只接受 POST。傳入非同步處理器時返回非同步檢視
FastAPI、Starlette	deeprespond_im.callback.fastapi.callback_endpoint（與 .starlette 相同）	用 await request.body()；端點不宣告請求體引數。傳入同步處理器時線上程池中執行，不阻塞事件迴圈

Flask

from flask import Flask

from deeprespond_im.callback import CallbackHandler, MemoryDedupStore, MemoryNonceStore
from deeprespond_im.callback.flask import callback_view

handler = CallbackHandler(
    secrets={"1840200000000000007": ["cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE"]},
    nonce_store=MemoryNonceStore(),
    dedup_store=MemoryDedupStore(),
)

app = Flask(__name__)
app.add_url_rule("/im/callback", view_func=callback_view(handler), methods=["POST"])

Django

# urls.py
from django.urls import path

from deeprespond_im.callback import CallbackHandler, MemoryDedupStore, MemoryNonceStore
from deeprespond_im.callback.django import callback_view

handler = CallbackHandler(
    secrets={"1840200000000000007": ["cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE"]},
    nonce_store=MemoryNonceStore(),
    dedup_store=MemoryDedupStore(),
)

urlpatterns = [path("im/callback", callback_view(handler))]

FastAPI

from fastapi import FastAPI

from deeprespond_im.callback import Allow, AsyncCallbackHandler, AsyncMemoryDedupStore, AsyncMemoryNonceStore, Event, Verdict
from deeprespond_im.callback.events import MessageSent
from deeprespond_im.callback.fastapi import callback_endpoint
from deeprespond_im.callback.hooks import ChatroomBeforeSend

handler = AsyncCallbackHandler(
    secrets={"1840200000000000007": ["cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE"]},
    nonce_store=AsyncMemoryNonceStore(),
    dedup_store=AsyncMemoryDedupStore(),
)


@handler.on_event("message.sent")
async def on_message(event: Event, data: MessageSent) -> None:
    print("消息", data.message_id)


@handler.on_hook("chatroom.before_send")
async def check_room_message(request: ChatroomBeforeSend) -> Verdict:
    return Allow()   # 聊天室的同步回调时限通常只有几百毫秒，处理必须快


app = FastAPI()
app.add_api_route("/im/callback", callback_endpoint(handler), methods=["POST"], include_in_schema=False)

Starlette 的寫法相同：Route("/im/callback", callback_endpoint(handler), methods=["POST"])，從 deeprespond_im.callback.starlette 匯入。

自己在 FastAPI 中寫路由的，交給處理器的必須是 await request.body() 的原始位元組，不能宣告 Pydantic 模型的請求體引數再重新序列化。

非同步處理器中也可以註冊普通函式：事件處理函式用 asyncio.to_thread 線上程中執行；同步回撥的處理函式與同步處理器一樣放進 SDK 的守護執行緒。同步處理器中不能註冊協程函式（async def），會丟擲 TypeError。

其他框架

aiohttp、Tornado、雲函式等，直接呼叫 handler.handle(headers, body)（非同步處理器為 await handler.handle(...)），把返回的 status、headers、body 寫回響應。請求頭按不區分大小寫查詢。下面是標準庫 http.server 的寫法：

from http.server import BaseHTTPRequestHandler, HTTPServer

from deeprespond_im.callback import CallbackHandler, MemoryDedupStore, MemoryNonceStore

handler = CallbackHandler(
    secrets={"1840200000000000007": ["cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE"]},
    nonce_store=MemoryNonceStore(),
    dedup_store=MemoryDedupStore(),
)


class Receiver(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        result = handler.handle(dict(self.headers.items()), body)
        self.send_response(result.status)
        for key, value in result.headers.items():
            self.send_header(key, value)
        self.send_header("Content-Length", str(len(result.body)))
        self.end_headers()
        self.wfile.write(result.body)


if __name__ == "__main__":
    HTTPServer(("0.0.0.0", 8080), Receiver).serve_forever()

只校驗簽名

自己處理分派的，可以只用校驗和解析的函式：verify(secrets, headers, body) 校驗簽名和時間戳，不通過時丟擲 SignatureError；parse(body) 解析請求體，返回 CallbackRequest（kind、events、hook、data、challenge 等）。

from collections.abc import Mapping

from deeprespond_im.callback import SignatureError, parse, verify

SECRETS = ["cbs_oSi8UFoUx4RQ8TlhuDzI06dQGntx4oJ6dOaloTrcCCE"]


def receive(headers: Mapping[str, str], body: bytes) -> int:
    try:
        verify(SECRETS, headers, body)
    except SignatureError as e:
        print("校验失败：", e.reason)   # bad_signature、bad_timestamp、stale_timestamp、bad_format
        return 401
    request = parse(body)
    # 还需要自己检查 X-IM-Nonce 是否重复，并按事件 ID 去重
    for event in request.events:
        print(event.id, event.type)
    return 204

• SignatureError 不是 DRError 的子類；它的 reason 為 bad_signature（簽名不對：金鑰錯誤或請求體被改動）、bad_timestamp（時間戳不是純數字）、stale_timestamp（與本機時間相差超過 5 分鐘）、bad_format（缺少請求頭或隨機數格式不對）；
• verify 不做隨機數的防重放，也不去重，這兩步需要你自己完成（見驗證回撥請求）；
• sign(secret, timestamp, nonce, body) 和 signature_header(secrets, timestamp, nonce, body) 可以計算簽名，用於除錯。

本地測試

deeprespond_im.testing 可以構造帶正確簽名的回撥請求，不需要 IM 服務就能測試你的處理函式：

import json

from deeprespond_im.callback import CallbackHandler, Event, MemoryDedupStore, MemoryNonceStore, Reject, Verdict
from deeprespond_im.callback.events import UserCreated
from deeprespond_im.callback.hooks import FriendBeforeAdd
from deeprespond_im.testing import event_body, hook_body, signed_request, verify_body

SECRET = "cbs_test_secret_0000000000000000000000000000000"
ENDPOINT = "1840200000000000007"


def make_handler(seen: list[str]) -> CallbackHandler:
    handler = CallbackHandler(secrets={ENDPOINT: [SECRET]}, nonce_store=MemoryNonceStore(), dedup_store=MemoryDedupStore())

    @handler.on_event("user.created")
    def on_user_created(event: Event, data: UserCreated) -> None:
        seen.append(data.username)

    @handler.on_hook("friend.before_add")
    def on_friend(request: FriendBeforeAdd) -> Verdict:
        return Reject("closed", "暂不接受好友申请")

    return handler


def test_event_is_handled_once() -> None:
    seen: list[str] = []
    handler = make_handler(seen)
    event = {
        "id": "1001", "type": "user.created", "occurred_at": "2026-10-06T08:00:00.000Z", "attempt": 1, "test": False,
        "data": {"username": "zhangsan", "nickname": "", "created_via": "openapi", "created_at": "2026-10-06T08:00:00.000Z"},
    }
    body = event_body(event, endpoint_id=ENDPOINT)
    assert handler.handle(*signed_request(SECRET, ENDPOINT, body)).status == 204
    assert handler.handle(*signed_request(SECRET, ENDPOINT, body)).status == 204   # 重发：新的随机数，同一个事件 ID
    assert seen == ["zhangsan"]


def test_hook_and_verify() -> None:
    handler = make_handler([])
    data = {"stage": "request", "from_username": "a", "to_username": "b", "message": "你好", "add_source": "search"}
    body = hook_body("friend.before_add", data, endpoint_id=ENDPOINT)
    response = handler.handle(*signed_request(SECRET, ENDPOINT, body))
    assert json.loads(response.body)["action"] == "reject"

    response = handler.handle(*signed_request(SECRET, ENDPOINT, verify_body("abc", endpoint_id=ENDPOINT)))
    assert json.loads(response.body) == {"challenge": "abc"}

函式	說明
signed_request(secret, endpoint_id, body, *, timestamp=None, nonce=None)	給請求體簽名，返回 (请求头, 请求体)。secret 可以是新舊兩個金鑰的列表；不給時間戳和隨機數時用當前時間和新的隨機數
event_body(*events, endpoint_id=..., app=...)	事件回撥的請求體。每個事件是完整的 dict，或 (type, data)（自動生成 ID 和時間）
hook_body(hook, data, *, timeout_ms=1000, test=False, endpoint_id=..., app=...)	同步回撥的請求體
verify_body(challenge, *, endpoint_id=..., app=...)	URL 驗證的請求體
SIGNATURE_VECTORS	簽名的測試向量（與服務端和其他語言的 SDK 相同）

data 按事件列表和同步回撥給出欄位，缺少的欄位解析為零值（空字串、0、False、None）。用 (type, data) 簡寫時，事件 ID 每次隨機生成，測試去重時請給出完整的 dict。

SIGNATURE_VECTORS 中的每一項有 name、secret（接收端配置的金鑰）、timestamp、nonce、body_b64（請求體的 Base64）、expected（用 secret 算出的簽名）、now（校驗時的當前時間）、valid（是否應當通過）和 reason（說明）；有的項另有 signature_header（請求頭 X-IM-Signature 中實際的值，沒有時就是 expected）、signing_secrets（傳送方簽名用的新舊金鑰）和 error（不通過時 SignatureError 的 reason）。在你的部署環境中跑一遍，可以確認框架、中介軟體和代理沒有改動請求體，也可以用來核對自己實現的校驗：

import base64

from deeprespond_im.callback import SignatureError, verify
from deeprespond_im.testing import SIGNATURE_VECTORS


def test_signature_vectors() -> None:
    for v in SIGNATURE_VECTORS:
        headers = {
            "X-IM-Timestamp": v["timestamp"],
            "X-IM-Nonce": v["nonce"],
            "X-IM-Signature": v.get("signature_header", v["expected"]),
        }
        try:
            verify([v["secret"]], headers, base64.b64decode(v["body_b64"]), now=v["now"])
            error = None
        except SignatureError as e:
            error = e.reason
        assert (error is None) == v["valid"], v["name"]
        assert error == v.get("error"), v["name"]

聯調時，也可以在控制台對回撥地址“測試傳送”，或用除錯簽名中的方法排查。

獨立頻道回撥

事件	強型別模型
rtc_channel.created	RTCChannelCreated
rtc_channel.updated	RTCChannelUpdated
rtc_channel.closing	RTCChannelClosing
rtc_channel.closed	RTCChannelClosed
rtc_channel.session_joined	RTCChannelSessionJoined
rtc_channel.session_leaving	RTCChannelSessionLeaving
rtc_channel.session_left	RTCChannelSessionLeft
rtc_channel.member_banned	RTCChannelMemberBanned
rtc_channel.member_unbanned	RTCChannelMemberUnbanned

這 9 種事件和其他事件一樣由處理器分派，同樣校驗簽名、去重和處理錯誤。欄位見頻道事件；有些欄位可能沒有，除會話相關的事件外，不要假設事件中有目標使用者。頻道事件與通話（rtc.call_*）的事件名和型別是分開的；配置變化的內部廣播不會作為回調發出。

遊戲房間回撥

遊戲房間的 7 種事件和其他事件一樣由處理器分派，同樣校驗簽名、去重和處理錯誤。大多數遊戲只需要處理 game_room.closed：它帶著結算結果和每個玩家的線上時長，用來發獎勵、記戰績、核對計費。

from deeprespond_im.callback import CallbackHandler, Event, MemoryDedupStore, MemoryNonceStore
from deeprespond_im.callback.events import GameRoomClosed, GameRoomMemberLeft

handler = CallbackHandler(secrets={ENDPOINT_ID: [SECRET]}, nonce_store=MemoryNonceStore(), dedup_store=MemoryDedupStore())


@handler.on_event("game_room.closed")
def on_room_closed(event: Event, data: GameRoomClosed) -> None:
    # data.result 是结束房间时给的结算结果；超时、全员离开等自动结束时为 None
    save_match(data.room_id, data.close_reason, data.result)
    for m in data.members:
        add_play_minutes(m.username, m.billable_minutes)


@handler.on_event("game_room.member_left")
def on_member_left(event: Event, data: GameRoomMemberLeft) -> None:
    print(data.username, "离开了", data.room_id, data.reason)

事件	資料類（deeprespond_im.callback.events）	說明
game_room.created	GameRoomCreated	遊戲房間建立成功（已分配到遊戲節點）
game_room.started	GameRoomStarted	開局（遊戲節點開始推進幀）
game_room.member_joined	GameRoomMemberJoined	玩家拿到座位
game_room.member_connected	GameRoomMemberConnected	玩家第一次連上游戲節點；頻繁，按需訂閱
game_room.member_disconnected	GameRoomMemberDisconnected	玩家與遊戲節點斷線（重連視窗內可恢復）；頻繁，按需訂閱
game_room.member_left	GameRoomMemberLeft	玩家離開（主動離開、超時、被移出、沒有按時連線）
game_room.closed	GameRoomClosed	房間結束，帶時長、結算結果和每個玩家的線上時長、計費分鐘數

• 事件可能重複或亂序到達：處理器按事件 ID 去重，業務上用事件中的 version（房間版本，每次狀態變化加一）判斷新舊。
• result 和 ext 是 dict，原文在 result_raw、ext_raw；房間自動結束（超時、全員離開）時 result 為 None。
• billable_minutes 是這一局按人向上取整的分鐘數，供你給玩家計時和核對；賬單按天彙總後取整，見計費。
• 欄位的含義見遊戲房間的回撥事件。