接收回调

本页介绍用 Python SDK 接收事件回调和同步回调：校验签名、防止重放、按事件 ID 去重、把事件和同步回调分派给你的处理函数、自动回应 URL 验证，以及与 Flask、Django、FastAPI、Starlette 的集成和本地测试。

回调的处理在 deeprespond_im.callback 中：同步程序用 CallbackHandler，异步程序用 AsyncCallbackHandler。处理器与 Web 框架无关，接收请求头和原始请求体，返回状态码、响应头和响应体；框架适配只负责取出原始请求体和写回响应。

快速开始

以 Flask 为例，处理消息抄送事件和发消息前的同步回调：

python

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) 对每个请求依次：

请求体超过 max_body_bytes（默认 2 MB）的返回 413；
按请求头 X-IM-Endpoint-Id 找到这个地址的密钥，用每个密钥校验 X-IM-Signature，再检查 X-IM-Timestamp 与本机时间相差不超过 5 分钟（算法见验证回调请求）；
用随机数存储检查 X-IM-Nonce 在 10 分钟内没有出现过（防重放）；
解析请求体，核对其中的 endpoint_id 与请求头相同；
按请求的种类处理：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 和按事件类型解析好的数据类：

python

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。

python

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`

python

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 而仍用内存存储，会记一条警告日志；多台服务器的部署检测不到，请自己确认。

python

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

python

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

python

# 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

python

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 的写法：

python

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 等）。

python

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 服务就能测试你的处理函数：

python

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）。在你的部署环境中跑一遍，可以确认框架、中间件和代理没有改动请求体，也可以用来核对自己实现的校验：

python

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`

这九种事件可通过现有分派器处理，沿用签名验证、去重和错误处理。字段见频道事件；部分字段可缺省，不应假设非会话事件包含目标用户。频道与原通话的事件名和模型分开，配置内部广播不作为回调。

接收回调 ​

快速开始 ​

处理器做了什么 ​

创建处理器 ​

处理事件回调 ​

交给队列处理 ​

处理同步回调 ​

防重放与去重的存储 ​

与 Web 框架集成 ​

Flask ​

Django ​

FastAPI ​

其他框架 ​

只校验签名 ​

本地测试 ​

独立频道回调 ​

接收回调

快速开始

处理器做了什么

创建处理器

处理事件回调

交给队列处理

处理同步回调

防重放与去重的存储

与 Web 框架集成

Flask

Django

FastAPI

其他框架

只校验签名

本地测试

独立频道回调