Receiving callbacks

To get the UI and signed callbacks working first, see Configure live callbacks in the demo: create the endpoint, start the demo with the endpoint ID and secret, then verify the endpoint.

This page covers receiving event callbacks and sync callbacks with the Python SDK: verifying signatures, preventing replay, de-duplicating by event ID, dispatching events and sync callbacks to your handlers, answering URL verification automatically, plus integration with Flask, Django, FastAPI and Starlette, and local testing.

Callback handling lives in deeprespond_im.callback: sync programs use CallbackHandler and async programs AsyncCallbackHandler. The handler is independent of web frameworks: it takes the request headers and raw body and returns the status code, response headers and body; framework adapters only extract the raw body and write the response back.

You want to	See
Get a receiver working first	Quickstart
Handle events, hand them to a queue such as Celery	Handle event callbacks
Review, reject or modify messages before they're sent	Handle sync callbacks
Deploy several processes with Redis for replay prevention and de-duplication	Replay and dedup storage
Integrate with Flask, Django, FastAPI and other frameworks	Web framework integration
Unit test callbacks	Local testing

Quickstart

Using Flask, handle message copy events and the before-send sync callback:

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"])

In the console, configure https://your-domain/im/callback as the callback address and note the endpoint ID and signing secret (see Callback addresses). When you save the address, the IM service sends a URL verification request, which the handler answers automatically with nothing else needed.

What the handler does

For each request, handler.handle(headers, body) does this in order:

1. Returns 413 if the body is larger than max_body_bytes (2 MB by default).
2. Finds the endpoint's secrets by the X-IM-Endpoint-Id header, verifies X-IM-Signature with each secret, then checks that X-IM-Timestamp is within 5 minutes of the local time (for the algorithm, see Verifying callback requests).
3. Checks with the nonce store that X-IM-Nonce hasn't appeared in the last 10 minutes (replay prevention).
4. Parses the body and checks that its endpoint_id matches the header.
5. Handles the request by kind: URL verification answers the challenge; sync callbacks call the matching handler and return its result; event callbacks de-duplicate event by event by event ID and call the matching handlers.

Case	Response
Every event handled, or a duplicate event already handled	204
An event in the batch failed (its handler raised), or the same event is being handled in another request	503; the IM service resends the whole batch later, and events that succeeded are de-duplicated by event ID and not handled again
A handler raised RetryLater(seconds)	503 with Retry-After (up to 1 hour), and the IM service delays the resend accordingly
A sync callback handler returned a result	200, with the result as the body
A sync callback timed out, errored, returned an invalid result, or has no handler	The result from hook_fallback, or 503 if not configured (the IM service then follows the sync callback failure handling rules)
URL verification	200, {"challenge": "..."}
Endpoint ID not configured, wrong signature, expired timestamp, repeated nonce, or mismatched endpoint_id	401
The body isn't a valid callback request	400
The secrets function, nonce store or dedup store errored	503

When verification fails, the SDK logs an ERROR (logger deeprespond_im.callback) with the reason and where to look, such as "check the secret config, and whether a framework, middleware or proxy changed the request body".

Use the raw request body

Signatures are computed over the raw bytes of the request body. What you pass to handle() must be the unparsed raw bytes, not something a framework parsed and re-serialized (such as a Pydantic model's model_dump_json()). The SDK's framework adapters already get this right.

Create the handler

Parameter	Default	Description
secrets	Required	A map from endpoint ID to a list of secrets, such as {"1840200000000000007": ["cbs_new", "cbs_old"]}; or a function returning the list of secrets for an endpoint ID (an empty list when not found), which can be a coroutine function in the async handler. The list of secrets can't be a single string (raises TypeError)
nonce_store	Required	Nonce store; see Replay and dedup storage
dedup_store	Required	Event de-duplication store
queue	None	Event queue: events are handed to it and the handler returns immediately; see Handing events to a queue. Use instead of on_event
hook_fallback	None	Result to return when a sync callback times out or errors, such as Allow(); None means return 503
max_concurrent_hooks	16	Maximum sync callback handlers running at once; when full, new sync callbacks return hook_fallback or 503 straight away without queueing
claim_ttl	30.0	How long handling one event holds it (seconds): events not finished within this time can be handled again by a resent request
max_body_bytes	2097152	Request body size limit (bytes)
clock	time.time	Clock used to check timestamps, replaceable in tests

Put all of an endpoint's secrets (during the grace period of secret rotation) in its list; any one passing is enough. When several endpoints point to the same receiver, give each its own entry; when several apps share one receiver, tell apps apart in handlers with current_request().app.

Handle event callbacks

Register event handlers with on_event; the function receives the Event and the data class parsed for that event type:

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 has id, type, occurred_at, attempt (which delivery this is), test (a test send from the console), data (dict) and data_raw (the original text); event.decode() returns the data class by type.
• Data classes are in deeprespond_im.callback.events, one per event, named from the event name: message.sent is MessageSent, friend_request.received is FriendRequestReceived, and rtc.call_ended is RtcCallEnded. Fields match the event list, plus origin (who caused the event), truncated (truncated for being over 70 KB, in which case read it through OpenAPI by ID) and raw; message events also have body_raw and ext_raw, byte for byte as in the request body.
• Event names are checked at registration: a misspelled event name raises ValueError when registering, rather than silently never firing. To handle new events the SDK doesn't know yet, use on_unknown_event, or pass allow_unknown=True when registering (the second argument is then a dict); events without a handler are acknowledged directly.
• Returning normally means success; raising anything means failure, the batch returns 503, and the IM service resends later. To set when to resend, raise RetryLater(seconds).
• Several events in one request are handled one after another, not concurrently; when an event is resent (same event ID), handlers that already succeeded aren't called again.
• Common information: in a handler, current_request() gives this request's kind, request_id, app (AppKey), endpoint_id and sent_at, correct in both threads and async tasks.

Events may arrive duplicated and out of order (see Duplicates and ordering): the SDK de-duplicates by event ID, but in your business logic tell newer from older by the version numbers, seq or times in the events.

Handlers must return quickly: the IM service only waits a limited time for the response. For high-volume events such as message copies, just write them to your own queue or database and process them asynchronously.

Handing events to a queue

A handler configured with queue doesn't call event handlers: after verifying and de-duplicating, it hands each event to the queue's enqueue(event), and returning counts as received; from then on your queue (Celery, RQ, Kafka, a database table and so on) takes care of processing. If enqueue raises, the batch returns 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(),
)

The async handler's queue uses the coroutine method async def enqueue(self, event).

Handle sync callbacks

Register sync callback handlers with on_hook; the function receives the request's data class and returns a result:

Result	Description
Allow()	Allow
Reject(reason=None, message=None)	Reject. reason is 1 to 64 lowercase letters, digits or _ . -; message is a hint for the user, up to 256 characters, with no control characters such as newlines
Modify(body=..., ext=...)	Modify the message, only for message.before_send and chatroom.before_send. body is a dict or RawJSON; ext can also be None (remove ext); anything not given stays unchanged, and at least one must be given
Partial([RejectedTarget(username, reason, message), ...])	Reject some people, only for invitations and group creation in group.before_join (joined_via invite or create); rejected users must be in the request's targets

Sync callback	Request data class (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()

• Time limit: handlers must return within the request's timeout_ms minus 50 milliseconds. The data class's timeout is the wait the IM service gave (seconds), and deadline is the deadline on time.monotonic(), which you can use to set time limits on database queries and outgoing requests. After a timeout the handler returns hook_fallback or 503 immediately.
• Handlers in the sync handler run in SDK daemon threads: Python can't kill threads, so a timed-out handler keeps running in the background until it finishes, and its result is discarded with a warning log. Handlers should therefore finish quickly on their own; how many run at once is limited by max_concurrent_hooks. In the async handler, coroutine handlers are cancelled on timeout.
• Results are checked before returning: an invalid result (such as a Reject message with a newline, or Modify on a sync callback that doesn't support it) is treated as an error and logged; a handler returning None (a forgotten return) is also an error, not treated as allow.
• Requests with test true come from "Send test" in the console.

Replay and dedup storage

The handler needs two stores: the nonce store remembers the X-IM-Nonce values seen in the last 10 minutes, so requests can't be replayed as is; the dedup store remembers handled event IDs, so a resent event isn't handled again.

Deployment	Nonce store	Dedup store
A single process (development, single-process services)	MemoryNonceStore()	MemoryDedupStore()
Several worker processes (gunicorn, uWSGI), several servers	deeprespond_im.redis.RedisNonceStore(r)	deeprespond_im.redis.RedisDedupStore(r)
Async handler	AsyncMemoryNonceStore() or AsyncRedisNonceStore(r)	AsyncMemoryDedupStore() or AsyncRedisDedupStore(r)

In-memory stores only work within one process: with several worker processes each has its own copy, which can't prevent replay or de-duplicate. If the environment variable WEB_CONCURRENCY is greater than 1 at startup while in-memory stores are used, the handler logs a warning; multi-server deployments can't be detected, so check that yourself.

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),
)

• The Redis implementations take the redis.Redis (or redis.asyncio.Redis for async) client you create, with decode_responses on or off. Keys are {key_prefix}nonce:{nonce} and {key_prefix}event:{event ID}.
• retention is how long a de-duplication record is kept after an event is handled: 24 hours by default for in-memory stores and 7 days for Redis (failed deliveries are kept in the console for 7 days, during which you can redeliver them by hand).
• Claims: handling an event first claims it (claim_ttl, 30 seconds by default), marks it done on success, and releases it on failure so a resent request can handle it again. If an event being handled is resent to another process, that request returns 503 and is retried later.
• In-memory store limits: 1000000 entries each by default (max_entries). The dedup store evicts the oldest records when full and logs a warning; the nonce store never evicts unexpired nonces (eviction would allow replay), so when full new requests get 503.
• You can use other storage too, providing these methods: NonceStore.seen(nonce, ttl) -> bool (returns True if seen, otherwise records it and returns False), and DedupStore.claim(event_id, ttl) -> (ClaimState, token | None), complete(event_id, token) and release(event_id, token) (only releases when the claim token matches). The async versions' methods are coroutines.

Web framework integration

Adapters only extract the raw body and headers and write the response back. Both sync and async handlers can be passed in:

Framework	Import	Notes
Flask	deeprespond_im.callback.flask.callback_view	Uses request.get_data() for the raw body (cached, so reading request.json elsewhere doesn't matter). With an async handler it returns an async view, which needs flask[async]
Django	deeprespond_im.callback.django.callback_view	Uses request.body; the view is exempt from CSRF checks and only accepts POST. With an async handler it returns an async view
FastAPI, Starlette	deeprespond_im.callback.fastapi.callback_endpoint (same as .starlette)	Uses await request.body(); the endpoint declares no body parameter. A sync handler runs in the thread pool without blocking the event loop

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 works the same way: Route("/im/callback", callback_endpoint(handler), methods=["POST"]), imported from deeprespond_im.callback.starlette.

If you write the route in FastAPI yourself, pass the handler the raw bytes of await request.body(); don't declare a Pydantic model body parameter and re-serialize it.

You can also register plain functions with the async handler: event handlers run in a thread through asyncio.to_thread, and sync callback handlers go into SDK daemon threads as with the sync handler. Coroutine functions (async def) can't be registered with the sync handler and raise TypeError.

Other frameworks

For aiohttp, Tornado, cloud functions and the like, call handler.handle(headers, body) directly (await handler.handle(...) for the async handler) and write the returned status, headers and body into the response. Headers are looked up case-insensitively. Here it is with the standard library's 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()

Verifying signatures only

If you handle dispatching yourself, you can use just the verification and parsing functions: verify(secrets, headers, body) verifies the signature and timestamp, raising SignatureError on failure; parse(body) parses the body into a CallbackRequest (kind, events, hook, data, challenge and so on).

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 isn't a subclass of DRError; its reason is bad_signature (wrong signature: wrong secret or a changed body), bad_timestamp (the timestamp isn't all digits), stale_timestamp (more than 5 minutes from local time) or bad_format (a missing header or a badly formatted nonce).
• verify doesn't prevent replay with nonces or de-duplicate; you do those two steps yourself (see Verifying callback requests).
• sign(secret, timestamp, nonce, body) and signature_header(secrets, timestamp, nonce, body) compute signatures, for debugging.

Local testing

deeprespond_im.testing builds correctly signed callback requests, so you can test your handlers without the IM service:

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"}

Function	Description
signed_request(secret, endpoint_id, body, *, timestamp=None, nonce=None)	Signs the body and returns (headers, body). secret can be a list of new and old secrets; without a timestamp and nonce it uses the current time and a new nonce
event_body(*events, endpoint_id=..., app=...)	An event callback body. Each event is a complete dict or (type, data) (ID and time generated automatically)
hook_body(hook, data, *, timeout_ms=1000, test=False, endpoint_id=..., app=...)	A sync callback body
verify_body(challenge, *, endpoint_id=..., app=...)	A URL verification body
SIGNATURE_VECTORS	Signature test vectors (the same as the server and the other SDKs)

Give data fields as in the event list and Sync callbacks; missing fields parse as zero values (empty string, 0, False, None). With the (type, data) shorthand the event ID is random each time, so give a complete dict when testing de-duplication.

Each entry in SIGNATURE_VECTORS has name, secret (the secret configured on the receiver), timestamp, nonce, body_b64 (the body in Base64), expected (the signature computed with secret), now (the current time at verification), valid (whether it should pass) and reason (a description); some also have signature_header (the actual X-IM-Signature value, otherwise expected), signing_secrets (the new and old secrets the sender signed with) and error (the SignatureError reason when it fails). Running them in your deployment confirms that frameworks, middleware and proxies don't change the body, and checks your own verification code:

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"]

When testing end to end, you can also use "Send test" on the callback endpoint in the console, or the methods in Debugging signatures.

Standalone channel callbacks

Event	Type
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

These 9 events are dispatched by the handler like any other, with the same signature verification, de-duplication and error handling. For fields, see Channel events; some fields may be absent, and apart from session events, don't assume an event includes a target user. Channel events have separate names and types from call events (rtc.call_*), and internal broadcasts of config changes aren't sent as callbacks.

Game room callbacks

The 7 game room events are dispatched like any other event, with the same signature checks, deduplication and error handling. Most games only need game_room.closed: it carries the result and every player's online time, for rewards, match history and billing reconciliation.

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)

Event	Data class (deeprespond_im.callback.events)	Description
game_room.created	GameRoomCreated	A room was created (assigned to a game node)
game_room.started	GameRoomStarted	The match started (the game node began advancing frames)
game_room.member_joined	GameRoomMemberJoined	A player got a seat
game_room.member_connected	GameRoomMemberConnected	A player connected to the game node for the first time; frequent, subscribe only if needed
game_room.member_disconnected	GameRoomMemberDisconnected	A player disconnected from the game node (can recover within the reconnect window); frequent, subscribe only if needed
game_room.member_left	GameRoomMemberLeft	A player left (left, timed out, was removed or did not connect in time)
game_room.closed	GameRoomClosed	The room ended, with duration, result, and each player's online time and billable minutes

• Events can arrive more than once or out of order: the handler deduplicates by event ID; in your own logic use version (the room version, incremented on every state change) to tell newer from older.
• result and ext are dicts, with the raw text in result_raw and ext_raw; when the room closes automatically (timeout, everyone left) result is None.
• billable_minutes is this match's minutes per player, rounded up, for your own timing and reconciliation; invoices sum per day and then round, see Billing.
• Field meanings are in game room callback events.