receiver

feishu.events.receiver

create_event_route

Python

create_event_route(dispatcher: EventDispatcher, *, path: str = '/feishu/event', encrypt_key: str | None = None, verification_token: str | None = None, seen_store: Any = _DEFAULT_SEEN_STORE, max_age_seconds: float | None = 300, now: Callable[[], float] = time) -> Route

创建处理飞书事件推送的 Starlette POST 路由。

该端点的处理流程：

1. 先读取原始请求体字节（签名校验与 AES 解密都依赖未经改动的原始字节）。
2. 当配置了 encrypt_key 且请求头包含 X-Lark-Signature 时， 经 SignatureVerifier 校验签名 并校验时间戳新鲜度（重放时间窗），记录校验结果。
3. 若请求体被 encrypt 包裹，则先行解密。
4. 处理 url_verification 握手：握手仅通过内层 verification_token 鉴权， 不要求 MAC 签名（签名豁免仅在此处生效）。
5. 其余正常事件：当配置了 encrypt_key 时，签名必须存在、时间戳在 max_age_seconds 时间窗内且校验通过；缺失签名或时间戳过期将返回 401 （防止 Webhook 注入与重放攻击绕过）。
6. 通过 seen_store 去重（默认即开启），其余事件以后台任务（BackgroundTask） 异步分发，端点立即返回 200 {}。

参数：

名称	类型	描述	默认
dispatcher	EventDispatcher	事件分发器。	必需
path	str	路由路径，默认 /feishu/event。	'/feishu/event'
encrypt_key	str | None	应用配置的 Encrypt Key；设置后启用解密与签名强校验。	None
verification_token	str | None	握手校验 Token；设置后会校验 url_verification 的内层 token。	None
seen_store	Any	事件去重存储；缺省时使用新建的 InMemorySeenStore， 从而开箱即用地提供去重/防重放保护；显式传入 None 可关闭去重， 也可传入自定义实现（如基于 Redis 的共享存储）。	_DEFAULT_SEEN_STORE
max_age_seconds	float | None	签名请求允许的最大时延（秒），默认 300；用于拒绝过期的 重放请求。设为 None 可关闭新鲜度校验（但绝不会跳过 MAC 校验）。	300
now	Callable[[], float]	返回当前 epoch 秒的可调用对象，默认 time.time；可注入以编写确定性测试。	time

返回：

类型	描述
Route	可挂载到 Starlette 应用的 [Route][starlette.routing.Route]。

飞书文档

接收事件

示例：

>>> from starlette.applications import Starlette
>>> dispatcher = EventDispatcher()
>>> route = create_event_route(dispatcher, encrypt_key="ek_secret")
>>> app = Starlette(routes=[route])

源代码位于： feishu/events/receiver.py

def create_event_route(
    dispatcher: EventDispatcher,
    *,
    path: str = "/feishu/event",
    encrypt_key: str | None = None,
    verification_token: str | None = None,
    seen_store: Any = _DEFAULT_SEEN_STORE,
    max_age_seconds: float | None = 300,
    now: Callable[[], float] = time.time,
) -> Route:
    r"""
    创建处理飞书事件推送的 Starlette POST 路由。

    该端点的处理流程：

    1. 先读取原始请求体字节（签名校验与 AES 解密都依赖未经改动的原始字节）。
    2. 当配置了 `encrypt_key` 且请求头包含 `X-Lark-Signature` 时，
       经 [SignatureVerifier][feishu.signature.SignatureVerifier] 校验签名
       **并校验时间戳新鲜度**（重放时间窗），记录校验结果。
    3. 若请求体被 `encrypt` 包裹，则先行解密。
    4. 处理 `url_verification` 握手：握手仅通过内层 `verification_token` 鉴权，
       不要求 MAC 签名（签名豁免仅在此处生效）。
    5. 其余正常事件：当配置了 `encrypt_key` 时，签名必须存在、时间戳在
       `max_age_seconds` 时间窗内且校验通过；缺失签名或时间戳过期将返回 401
       （防止 Webhook 注入与重放攻击绕过）。
    6. 通过 `seen_store` 去重（默认即开启），其余事件以后台任务（BackgroundTask）
       异步分发，端点立即返回 `200 {}`。

    Args:
        dispatcher: 事件分发器。
        path: 路由路径，默认 `/feishu/event`。
        encrypt_key: 应用配置的 Encrypt Key；设置后启用解密与签名强校验。
        verification_token: 握手校验 Token；设置后会校验 `url_verification` 的内层 token。
        seen_store: 事件去重存储；缺省时使用新建的
            [InMemorySeenStore][feishu.events.idempotency.InMemorySeenStore]，
            从而开箱即用地提供去重/防重放保护；显式传入 `None` 可关闭去重，
            也可传入自定义实现（如基于 Redis 的共享存储）。
        max_age_seconds: 签名请求允许的最大时延（秒），默认 `300`；用于拒绝过期的
            重放请求。设为 `None` 可关闭新鲜度校验（但绝不会跳过 MAC 校验）。
        now: 返回当前 epoch 秒的可调用对象，默认 [time.time][]；可注入以编写确定性测试。

    Returns:
        可挂载到 Starlette 应用的 [Route][starlette.routing.Route]。

    飞书文档:
        [接收事件](https://open.feishu.cn/document/server-docs/event-subscription-guide/event-subscription-configure-/request-url-configuration-case)

    Examples:
        >>> from starlette.applications import Starlette
        >>> dispatcher = EventDispatcher()
        >>> route = create_event_route(dispatcher, encrypt_key="ek_secret")  # doctest: +SKIP
        >>> app = Starlette(routes=[route])  # doctest: +SKIP
    """
    store = _resolve_seen_store(seen_store)
    verifier = _build_verifier(encrypt_key, max_age_seconds, now)

    async def endpoint(request: Request) -> Response:
        raw = await request.body()
        result = await _read_payload(request, raw, encrypt_key, verifier)
        if isinstance(result, Response):
            return result
        payload, sig_verified = result

        if payload.get("type") == "url_verification":
            # Handshake: authenticated by inner token, not by MAC signature.
            if verification_token is not None and not hmac.compare_digest(
                str(payload.get("token", "")), verification_token
            ):
                return JSONResponse({"msg": "token mismatch"}, status_code=401)
            return JSONResponse({"challenge": payload.get("challenge")})

        # Normal events require MAC authentication when encrypt_key is configured.
        if encrypt_key is not None and not sig_verified:
            return JSONResponse({"msg": "signature required"}, status_code=401)

        event = Event.from_payload(payload)
        if store is not None and not await claim(store, event.event_id):
            return JSONResponse({})

        return JSONResponse({}, background=BackgroundTask(dispatcher.dispatch, event))

    return Route(path, endpoint, methods=["POST"])

create_card_route

Python

create_card_route(dispatcher: EventDispatcher, *, path: str = '/feishu/card', encrypt_key: str | None = None, verification_token: str | None = None, seen_store: Any = _DEFAULT_SEEN_STORE, max_age_seconds: float | None = 300, now: Callable[[], float] = time) -> Route

创建处理飞书卡片交互回调的 Starlette POST 路由。

与 create_event_route 不同，本路由会 **同步**等待分发器执行（不使用后台任务），并将处理函数返回的 {toast, card} 字典 作为同步 JSON 响应返回，以满足飞书对卡片交互约 3 秒的响应时限。当处理函数返回 None 时， 响应为 200 {}。

安全模型与 create_event_route 一致：

• url_verification 握手仅通过内层 verification_token 鉴权（签名豁免在此生效）。
• 其余事件在配置了 encrypt_key 时，必须携带并经 SignatureVerifier 通过 X-Lark-Signature 校验，且时间戳须在 max_age_seconds 时间窗内（防重放）；缺失、过期或非法签名将返回 401，且处理函数不会被调用。

由于飞书在超时时会重试卡片回调，命中 seen_store 的重复事件会直接返回 {}，避免重复触发副作用。

参数：

名称	类型	描述	默认
dispatcher	EventDispatcher	事件分发器。	必需
path	str	路由路径，默认 /feishu/card。	'/feishu/card'
encrypt_key	str | None	应用配置的 Encrypt Key；设置后启用解密与签名强校验。	None
verification_token	str | None	握手校验 Token；设置后会校验 url_verification 的内层 token。	None
seen_store	Any	事件去重存储；缺省时使用新建的 InMemorySeenStore， 从而开箱即用地提供去重/防重放保护；显式传入 None 可关闭去重， 也可传入自定义实现。	_DEFAULT_SEEN_STORE
max_age_seconds	float | None	签名请求允许的最大时延（秒），默认 300；用于拒绝过期的 重放请求。设为 None 可关闭新鲜度校验（但绝不会跳过 MAC 校验）。	300
now	Callable[[], float]	返回当前 epoch 秒的可调用对象，默认 time.time；可注入以编写确定性测试。	time

返回：

类型	描述
Route	可挂载到 Starlette 应用的 [Route][starlette.routing.Route]。

飞书文档

卡片回传交互

示例：

>>> dispatcher = EventDispatcher()
>>> route = create_card_route(dispatcher, encrypt_key="ek_secret")

源代码位于： feishu/events/receiver.py

def create_card_route(
    dispatcher: EventDispatcher,
    *,
    path: str = "/feishu/card",
    encrypt_key: str | None = None,
    verification_token: str | None = None,
    seen_store: Any = _DEFAULT_SEEN_STORE,
    max_age_seconds: float | None = 300,
    now: Callable[[], float] = time.time,
) -> Route:
    r"""
    创建处理飞书卡片交互回调的 Starlette POST 路由。

    与 [create_event_route][feishu.events.receiver.create_event_route] 不同，本路由会
    **同步**等待分发器执行（不使用后台任务），并将处理函数返回的 `{toast, card}` 字典
    作为同步 JSON 响应返回，以满足飞书对卡片交互约 3 秒的响应时限。当处理函数返回 `None` 时，
    响应为 `200 {}`。

    安全模型与 [create_event_route][feishu.events.receiver.create_event_route] 一致：

    * `url_verification` 握手仅通过内层 `verification_token` 鉴权（签名豁免在此生效）。
    * 其余事件在配置了 `encrypt_key` 时，必须携带并经
      [SignatureVerifier][feishu.signature.SignatureVerifier] 通过 `X-Lark-Signature`
      校验，且时间戳须在 `max_age_seconds` 时间窗内（防重放）；缺失、过期或非法签名将返回
      401，且处理函数不会被调用。

    由于飞书在超时时会重试卡片回调，命中 `seen_store` 的重复事件会直接返回 `{}`，避免重复触发副作用。

    Args:
        dispatcher: 事件分发器。
        path: 路由路径，默认 `/feishu/card`。
        encrypt_key: 应用配置的 Encrypt Key；设置后启用解密与签名强校验。
        verification_token: 握手校验 Token；设置后会校验 `url_verification` 的内层 token。
        seen_store: 事件去重存储；缺省时使用新建的
            [InMemorySeenStore][feishu.events.idempotency.InMemorySeenStore]，
            从而开箱即用地提供去重/防重放保护；显式传入 `None` 可关闭去重，
            也可传入自定义实现。
        max_age_seconds: 签名请求允许的最大时延（秒），默认 `300`；用于拒绝过期的
            重放请求。设为 `None` 可关闭新鲜度校验（但绝不会跳过 MAC 校验）。
        now: 返回当前 epoch 秒的可调用对象，默认 [time.time][]；可注入以编写确定性测试。

    Returns:
        可挂载到 Starlette 应用的 [Route][starlette.routing.Route]。

    飞书文档:
        [卡片回传交互](https://open.feishu.cn/document/server-docs/im-v1/message-card/handle-card-actions)

    Examples:
        >>> dispatcher = EventDispatcher()
        >>> route = create_card_route(dispatcher, encrypt_key="ek_secret")  # doctest: +SKIP
    """
    store = _resolve_seen_store(seen_store)
    verifier = _build_verifier(encrypt_key, max_age_seconds, now)

    async def endpoint(request: Request) -> Response:
        raw = await request.body()
        result = await _read_payload(request, raw, encrypt_key, verifier)
        if isinstance(result, Response):
            return result
        payload, sig_verified = result

        if payload.get("type") == "url_verification":
            # Handshake: authenticated by inner token, not by MAC signature.
            if verification_token is not None and not hmac.compare_digest(
                str(payload.get("token", "")), verification_token
            ):
                return JSONResponse({"msg": "token mismatch"}, status_code=401)
            return JSONResponse({"challenge": payload.get("challenge")})

        # Real card events require MAC authentication when encrypt_key is configured.
        if encrypt_key is not None and not sig_verified:
            return JSONResponse({"msg": "signature required"}, status_code=401)

        event = Event.from_payload(payload)
        # Feishu retries card-action callbacks on timeout; returning {} prevents re-running side effects.
        if store is not None and not await claim(store, event.event_id):
            return JSONResponse({})

        handler_result = await dispatcher.dispatch(event)
        return JSONResponse(handler_result if handler_result is not None else {})

    return Route(path, endpoint, methods=["POST"])

create_event_app

Python

create_event_app(dispatcher: EventDispatcher, *, event_path: str = '/feishu/event', card_path: str | None = '/feishu/card', encrypt_key: str | None = None, verification_token: str | None = None, seen_store: Any = _DEFAULT_SEEN_STORE, card_seen_store: Any = _DEFAULT_SEEN_STORE, max_age_seconds: float | None = 300, now: Callable[[], float] = time) -> Starlette

返回一个可独立运行、处理飞书 Webhook 推送的 Starlette 应用。

始终在 event_path 挂载事件路由；当 card_path 不为 None 时，额外在该路径挂载卡片回调路由。 全部安全与路由逻辑分别委托给 create_event_route 与 create_card_route，因此默认即带有 签名新鲜度（防重放）时间窗与去重保护。

参数：

名称	类型	描述	默认
dispatcher	EventDispatcher	事件分发器。	必需
event_path	str	事件路由路径，默认 /feishu/event。	'/feishu/event'
card_path	str | None	卡片回调路由路径，默认 /feishu/card；为 None 时不挂载卡片路由。	'/feishu/card'
encrypt_key	str | None	应用配置的 Encrypt Key；设置后启用解密与签名强校验。	None
verification_token	str | None	握手校验 Token；设置后会校验 url_verification 的内层 token。	None
seen_store	Any	事件路由的去重存储；缺省时各路由自建 InMemorySeenStore， 显式传入 None 关闭去重，也可传入自定义实现。	_DEFAULT_SEEN_STORE
card_seen_store	Any	卡片路由的去重存储；语义同 seen_store。	_DEFAULT_SEEN_STORE
max_age_seconds	float | None	签名请求允许的最大时延（秒），默认 300；设为 None 关闭新鲜度校验。	300
now	Callable[[], float]	返回当前 epoch 秒的可调用对象，默认 time.time；可注入以编写确定性测试。	time

返回：

类型	描述
Starlette	已挂载相应路由的 [Starlette][starlette.applications.Starlette] 应用。

飞书文档

接收事件

示例：

>>> dispatcher = EventDispatcher()
>>> app = create_event_app(dispatcher, encrypt_key="ek_secret")

源代码位于： feishu/events/receiver.py

def create_event_app(
    dispatcher: EventDispatcher,
    *,
    event_path: str = "/feishu/event",
    card_path: str | None = "/feishu/card",
    encrypt_key: str | None = None,
    verification_token: str | None = None,
    seen_store: Any = _DEFAULT_SEEN_STORE,
    card_seen_store: Any = _DEFAULT_SEEN_STORE,
    max_age_seconds: float | None = 300,
    now: Callable[[], float] = time.time,
) -> Starlette:
    r"""
    返回一个可独立运行、处理飞书 Webhook 推送的 Starlette 应用。

    始终在 `event_path` 挂载事件路由；当 `card_path` 不为 `None` 时，额外在该路径挂载卡片回调路由。
    全部安全与路由逻辑分别委托给 [create_event_route][feishu.events.receiver.create_event_route]
    与 [create_card_route][feishu.events.receiver.create_card_route]，因此默认即带有
    签名新鲜度（防重放）时间窗与去重保护。

    Args:
        dispatcher: 事件分发器。
        event_path: 事件路由路径，默认 `/feishu/event`。
        card_path: 卡片回调路由路径，默认 `/feishu/card`；为 `None` 时不挂载卡片路由。
        encrypt_key: 应用配置的 Encrypt Key；设置后启用解密与签名强校验。
        verification_token: 握手校验 Token；设置后会校验 `url_verification` 的内层 token。
        seen_store: 事件路由的去重存储；缺省时各路由自建
            [InMemorySeenStore][feishu.events.idempotency.InMemorySeenStore]，
            显式传入 `None` 关闭去重，也可传入自定义实现。
        card_seen_store: 卡片路由的去重存储；语义同 `seen_store`。
        max_age_seconds: 签名请求允许的最大时延（秒），默认 `300`；设为 `None` 关闭新鲜度校验。
        now: 返回当前 epoch 秒的可调用对象，默认 [time.time][]；可注入以编写确定性测试。

    Returns:
        已挂载相应路由的 [Starlette][starlette.applications.Starlette] 应用。

    飞书文档:
        [接收事件](https://open.feishu.cn/document/server-docs/event-subscription-guide/event-subscription-configure-/request-url-configuration-case)

    Examples:
        >>> dispatcher = EventDispatcher()
        >>> app = create_event_app(dispatcher, encrypt_key="ek_secret")  # doctest: +SKIP
    """
    routes = [
        create_event_route(
            dispatcher,
            path=event_path,
            encrypt_key=encrypt_key,
            verification_token=verification_token,
            seen_store=seen_store,
            max_age_seconds=max_age_seconds,
            now=now,
        )
    ]
    if card_path is not None:
        routes.append(
            create_card_route(
                dispatcher,
                path=card_path,
                encrypt_key=encrypt_key,
                verification_token=verification_token,
                seen_store=card_seen_store,
                max_age_seconds=max_age_seconds,
                now=now,
            )
        )
    return Starlette(routes=routes)