L1 拿了 token，L2 会发消息，L3 会发卡片——到这里机器人还只是个"喇叭"，只会说，不会听。

群里 @它问一句，它没反应；点了卡片按钮，它不知道。让机器人"听得见"的机制叫事件订阅：用户在飞书里的动作（发消息、点按钮、进群）由飞书服务器推送给你的服务，你处理完再用 L2/L3 的函数回复。

这一课把整条链路打通：收消息 → 喂大模型 → 回复。学完这课，一个能听会说的飞书 AI 机器人就闭环了。

〇、先看全景：一条 @ 的完整旅程

群里 @机器人 说"帮我翻译你好"，到它回复，中间发生了六步：

步骤	谁在动	发生了什么	你要做的
①	用户	群里 @机器人 说话	无，测试而已
②	飞书服务器	把 im.message.receive_v1 事件推到你的回调 URL	后台注册回调 URL + 订阅该事件
③	你的服务	接推送：验证、解密、解析出消息文本	本课代码：challenge 应答 + 事件解析
④	你的服务	文本 + 系统提示词喂大模型	call_llm（任意模型 API）
⑤	你的服务 → 飞书	调 L2 的 send_text 把回复发回群	复用 L2 代码
⑥	用户	群里看到 AI 回复	完

下面讲的每个机制，都服务这张表的某一步：第二节=②的入场券（URL验证）、第三节=③的读信、第四节=③的解密、第五节=③④⑤全部代码、第六节=③的保命细节。先存这张地图，再看细节。

一、两种订阅方式：长连接 vs HTTP 回调

	长连接（WebSocket）	HTTP 回调（Webhook）
公网地址	不需要	需要公网 HTTPS URL
部署难度	低（SDK 一行启动）	中（要配服务器/内网穿透）
适合	本地开发、单机机器人	生产环境、多实例
官方支持	lark-oapi SDK（Python/Go/Java/Node）	任意 HTTP 服务

本地开发直接用长连接，不用买服务器不用 ngrok：

# pip install lark-oapi
import lark_oapi as lark

def on_message(data: lark.im.P2MessageReceiveV1) -> None:
    print("收到消息:", data.event.message.content)
    # 在这里调大模型 + L2 的 send_text 回复

handler = lark.EventDispatcherHandler.builder("", "") \
    .register_p2_im_message_receive_v1(on_message) \
    .build()

cli = lark.ws.Client(
    lark.ClientOption.builder()
        .app_id("cli_xxx")           # 你的 App ID
        .app_secret("xxx")           # 你的 App Secret
        .build(),
    event_handler=handler,
    log_level=lark.LogLevel.INFO,
)
cli.start()   # 阻塞运行，保持长连接

生产环境要 HTTP 回调，往下看——回调模式有三个必须处理的机制：URL 验证、事件解析、去重。

二、URL 验证：challenge 必须原样返回

在开放平台后台填你的回调 URL 时，飞书会立刻发一个验证请求确认这个地址是你的：

{
  "challenge": "ajls384kdjx98XX",
  "token": "你的 Verification Token",
  "type": "url_verification"
}

你的服务必须把 challenge 原样返回：

{ "challenge": "ajls384kdjx98XX" }

应答不对，URL 配置保存不了。这是新手卡住的第一关——配置 URL 时你的服务必须已经在运行，否则验证请求没人接。

如果开启了 Encrypt Key，验证请求也是加密的（{"encrypt": "base64串"}），要先解密再取 challenge（解密见第四节）。

三、事件结构 v2.0：header 定类型，event 装内容

验证通过后，真实事件长这样（以「接收消息」im.message.receive_v1 为例）：

{
  "schema": "2.0",
  "header": {
    "event_id": "f7984f25108f8137722bb63cee927e66",
    "token": "Verification Token",
    "create_time": "1603977298000000",
    "event_type": "im.message.receive_v1",
    "tenant_key": "xxx",
    "app_id": "cli_xxx"
  },
  "event": {
    "sender": { "sender_id": { "open_id": "ou_xxx" } },
    "message": {
      "message_id": "om_xxx",
      "chat_id": "oc_xxx",
      "chat_type": "group",
      "message_type": "text",
      "content": "{\"text\":\"@_user_1 帮我翻译一下\"}"
    }
  }
}

读事件的三个要点：

1. header.event_type 决定分发——按它路由到不同处理函数（收消息、点卡片按钮、进群是不同 event_type）
2. event.message.message_id 用来去重——官方明确提醒：特殊情况下会收到重复推送，幂等要用 message_id 去重，不要依赖 event_id（见第六节坑 2）
3. message.content 又是一层 JSON 字符串——L2 的双重序列化，收消息时反过来要双重解析；群里 @机器人 时文本里带 @_user_1 占位符，要清掉再喂给大模型
4. sender.sender_type 区分发送者——官方文档：取值是 user 或 bot，防机器人互聊死循环就靠它（坑 3）

常见事件类型速查（header.event_type 的值，决定你的服务要不要处理）：

event_type	什么时候推	典型用途
im.message.receive_v1	用户发消息给机器人/被@	AI 问答机器人（本课主角）
im.message.message_read_v1	消息被已读	已读回执统计
卡片回传交互事件	用户点卡片按钮（L3 的 value 模式）	审批、确认类交互
机器人菜单事件	用户点机器人菜单	菜单触发的功能入口

前两个是官方 SDK 明确支持的事件（OnP2MessageReceiveV1 / OnP2MessageReadV1）；后两类事件名以你后台「事件订阅」页面列出的为准，完整清单查官方「事件列表」文档——别照抄任何文章里的事件名，后台能搜到的才是真的。

所需权限：后台「权限管理」开通 im:message（读取用户发给机器人的单聊消息）+ 群里被 @ 的消息权限，并在「事件与回调」里添加 im.message.receive_v1 事件。权限不开、事件不加，服务写得再对也收不到推送。

四、Encrypt Key：开了加密就要解密

后台配置了 Encrypt Key 后，飞书推的所有请求 body 都变成：

{ "encrypt": "base64编码的密文" }

解密算法（官方规范）：AES-256-CBC，密钥 = SHA256(Encrypt Key)，IV = 密文前 16 字节，明文去 PKCS7 padding：

# pip install pycryptodome
import base64
import hashlib
import json
from Crypto.Cipher import AES


def decrypt_event(encrypt_b64: str, encrypt_key: str) -> dict:
    key = hashlib.sha256(encrypt_key.encode("utf-8")).digest()
    raw = base64.b64decode(encrypt_b64)
    iv, data = raw[:16], raw[16:]
    plain = AES.new(key, AES.MODE_CBC, iv).decrypt(data)
    plain = plain[: -plain[-1]]  # 去 PKCS7 padding
    return json.loads(plain.decode("utf-8"))

没配 Encrypt Key 就是明文推送，本地调试可以先不开；生产环境建议开——不然拿到你的回调 URL 就能伪造事件。

五、完整代码：一个不依赖框架的回调服务

Python 标准库 http.server 实现（除解密外零依赖），把 challenge 验证、事件解析、去重、AI 回复全串起来，发送复用 L2 的 send_text：

import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

from l1_token import FeishuToken   # L1 的 token 管理器
from l2_send import send_text      # L2 的发送函数（send_text 单独存文件）

import os
ft = FeishuToken(os.environ["FEISHU_APP_ID"], os.environ["FEISHU_APP_SECRET"])

# 去重：生产环境换 Redis。event_id 挡重推，message_id 挡重复消息（官方建议以 message_id 做幂等）
_seen_events = set()
_seen_msgs = set()
_seen_lock = threading.Lock()

SYSTEM_PROMPT = "你是群里的 AI 助手，回答简洁、准确，不确定就说不确定。"


def call_llm(user_text: str) -> str:
    """接你的大模型。示例用任意 OpenAI 兼容接口，替换 base_url/model/key 即可。"""
    import urllib.request
    req = urllib.request.Request(
        os.environ.get("LLM_BASE_URL", "https://api.example.com/v1") + "/chat/completions",
        data=json.dumps({
            "model": os.environ.get("LLM_MODEL", "your-model"),
            "messages": [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_text},
            ],
        }).encode("utf-8"),
        headers={
            "Content-Type": "application/json",
            "Authorization": "Bearer " + os.environ["LLM_API_KEY"],
        },
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=60) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    return data["choices"][0]["message"]["content"]


def handle_message(event: dict) -> None:
    # 防循环：机器人(bot)发的消息不处理（坑 3；官方：sender_type 取值 user 或 bot）
    sender = event.get("sender") or {}
    if sender.get("sender_type") == "bot":
        return
    msg = event.get("message", {})
    if msg.get("message_type") != "text":
        return  # 只处理文本，图片/文件先跳过
    # message_id 第二层去重（官方：幂等以 message_id 为准，不依赖 event_id）
    mid = msg.get("message_id")
    if mid:
        with _seen_lock:
            if mid in _seen_msgs:
                return
            _seen_msgs.add(mid)
    content = json.loads(msg.get("content", "{}"))   # content 是 JSON 字符串
    text = (content.get("text") or "").replace("@_user_1", "").strip()
    if not text:
        return

    # 异步处理：先应答 200，AI 生成慢也别阻塞飞书
    def work():
        try:
            answer = call_llm(text)
        except Exception as e:
            answer = f"AI 服务暂时不可用：{e}"
        # 超长截断（L3 讲过：文本 150KB，但群消息太长没人看，按 2000 字截）
        send_text(ft.auth_header(), msg["chat_id"], answer[:2000],
                  receive_id_type="chat_id")
    threading.Thread(target=work, daemon=True).start()


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(length) or b"{}")

        # 开了 Encrypt Key 时先解密（decrypt_event 见第四节）
        if "encrypt" in body:
            from l4_decrypt import decrypt_event
            body = decrypt_event(body["encrypt"], os.environ["FEISHU_ENCRYPT_KEY"])

        # ① URL 验证请求：原样返回 challenge
        if body.get("type") == "url_verification":
            self._reply({"challenge": body.get("challenge")})
            return

        # ② 事件去重（第一层）：同一 event_id 只处理一次（飞书失败会重推）
        #    官方提醒：幂等以 message_id 为准（第二层去重在 handle_message 里）
        event_id = (body.get("header") or {}).get("event_id")
        if event_id:
            with _seen_lock:
                if event_id in _seen_events:
                    self._reply({})     # 重复事件：应答但不处理
                    return
                _seen_events.add(event_id)

        # ③ 先应答 200（官方要求快速响应，超时飞书会重推），再异步处理
        self._reply({})

        event_type = (body.get("header") or {}).get("event_type")
        if event_type == "im.message.receive_v1":
            handle_message(body.get("event") or {})

    def _reply(self, obj):
        data = json.dumps(obj).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, *args):
        pass  # 静默访问日志


if __name__ == "__main__":
    print("回调服务启动: http://0.0.0.0:3000/feishu/callback")
    HTTPServer(("0.0.0.0", 3000), Handler).serve_forever()

从零跑通的 7 步（照做即可，每步都有明确的"成功标志"）：

步骤	操作	成功标志
1	本地启动上面的回调服务（设好 3 个环境变量）	终端打印"回调服务启动"
2	ngrok / 内网穿透把 3000 端口暴露成公网 HTTPS 地址	拿到 https://xxx.ngrok-free.app 类地址
3	开放平台后台「事件与回调」→ 请求地址填第 2 步 URL → 保存	后台提示验证通过（challenge 应答成功）
4	「权限管理」开通 im:message → 创建版本并发布	版本状态"已发布"
5	「事件与回调」→ 添加事件 im.message.receive_v1	事件出现在已订阅列表
6	把机器人拉进测试群（群设置→群机器人→添加自建应用）	群里能看到机器人
7	群里 @机器人 说句话	收到它的 AI 回复，服务日志有事件记录

第 3 步失败基本都是"服务没在跑"或"URL 不是 HTTPS"；第 7 步没反应先查第 4 步版本发布没有、第 5 步事件加了没有。

六、三个必踩的坑

坑 1：响应慢导致事件重复推送

官方要求回调快速返回 HTTP 200（超时约 3 秒，以官方文档为准），否则视为失败会重试推送。大模型生成动辄十几秒——绝不能同步等 AI 回复完才应答。上面代码的做法：立刻回 200，AI 处理放线程里异步做。

坑 2：不去重，用户收到重复回复

飞书在没及时收到 200 时会重推事件，特殊情况还会重复推送同一条消息。官方明确提醒：幂等请用 message_id 去重，不要依赖 event_id。所以代码里做了两层：do_POST 里用 event_id 挡同一次重推，handle_message 里用 message_id 挡同一条消息（真正防重复回复的是后者）。内存 set 只是演示，生产环境用 Redis（SETNX message_id EX 3600）——多实例部署时内存 set 各存各的，挡不住。

坑 3：机器人和自己聊出无限循环

如果群里有两个机器人、或者你的处理逻辑会触发自己的消息事件，就可能"A 回复 → 触发 B → B 回复 → 触发 A"死循环。防御：handle_message 里检查 sender.sender_type，是 bot 发的消息直接跳过；另外 L2 讲的 uuid 幂等在回复时也带上。

结论：事件订阅三条命——快速应答 200、event_id 去重、跳过机器人消息。

七、一段可以直接复制的机器人提示词

上面的 SYSTEM_PROMPT 只是占位。机器人的人设和能力，全靠系统提示词定义。给你一段站内真实条目——提示岛 id=5226「中英翻译助手」，把它放进 call_llm 的 system 位，@机器人 就变成一个群翻译官：

Act as a Chinese to English Translation Assistant. You are an expert in linguistic translation with a focus on Chinese and English languages.

Your task is to translate the provided Chinese text into English.

You will:
- Ensure the translation maintains the original meaning and context.
- Use appropriate vocabulary and grammar.

Rules:
- Always consider cultural nuances and context.
- Deliver a fluent and natural English translation.

Example:
- Input: "你好，世界！"
- Output: "Hello, world!"

Variables:
- ${input} - The Chinese text to be translated.

这段提示词的结构值得抄：角色定义（Act as…）→ 任务（Your task）→ 执行要求（You will）→ 规则（Rules）→ 示例（Example，一对输入输出）→ 变量（Variables）。带 Example 的提示词在机器人场景尤其重要——群里用户输入五花八门，一个示例能锚定输出格式，比十句描述管用。

用法：把用户 @机器人 说的话填进 ${input}，回复直接是译文。换成客服、周报、SQL 生成等人设，同样套这个六段结构。

结论：机器人提示词用「角色→任务→要求→规则→示例→变量」六段式，示例是稳定输出的锚。

这一课做完，你手上有什么

• 长连接 20 行代码（本地开发免公网）+ HTTP 回调完整服务（生产用）
• challenge 验证、v2.0 事件解析、Encrypt Key 解密三套机制
• 异步应答 + event_id 去重 + 防循环三个保命实践
• 一段六段式机器人系统提示词模板

到这里，L1-L4 拼起来就是一个完整的"能听会说"飞书 AI 机器人：token（L1）→ 发消息（L2）→ 卡片（L3）→ 事件订阅（L4）。

下一课（L5）讲多维表格 × AI：用 bitable API 读写表格数据，让 AI 批量打标签、做摘要——从"聊天机器人"升级到"业务数据处理"。

---

附：提示岛是什么

本文提示词来自提示岛（prompts.ssjintaitong.com），AI 提示词与 Skill 库，实测收录 10908 条，其中文字提示词 3483 条（完整全文、一键复制），Skill/插件 5298 条，23 个分类。免登录、免注册。

飞书相关入口：

• 飞书 AI 自动化专题（146 条）：prompts.ssjintaitong.com/topic/feishu
• 本文提示词原文：prompts.ssjintaitong.com/prompts/5226
• 本系列 L1-L3 教程条目：同专题内，搜「飞书对接 AI」
• 文字提示词总库：prompts.ssjintaitong.com/prompts?type=text

站内两个实用功能：Skill 质量评分（5298 个插件条目按四维度打 A-D 档，报告在 /quality-report）；每条附原始出处（GitHub 链接 + 许可证）。搜索技巧：用场景词搜，搜「飞书」有 132 条，搜「翻译」有 27 条提示词。