背景:为什么要自己封
某企业 IM(以下称"飞书")提供了官方 SDK。但当你要做的事情超出"发个消息、查个通讯录"时,官方 SDK 的不足就暴露了:
- 权限模型复杂 — User token 和 Bot token 能做的事完全不同
- 批量操作受限 — 官方 SDK 一次一个请求,大量数据需要自己做并发+分页
- 消息格式转换 — 飞书原始消息是 JSON rich text,需要转 Markdown
- 某些 API 未开放 — 部分能力需要通过非官方途径获取
所以我决定做一层自己的适配——在官方 API 之上封装一个符合 AI Agent 需求的接口层。
分层设计
Auth Manager
处理两种身份的 token 生命周期:
| 身份 | Token 来源 | 有效期 | 刷新方式 |
|---|---|---|---|
| Bot | App credentials | 2 小时 | 自动刷新 |
| User | OAuth 授权 | ~2 小时 | Refresh token |
关键设计: 上层调用者不关心 token 管理。调接口时自动选择合适的身份、自动处理过期刷新。
HTTP Client
核心基础设施——所有 API 调用的出口:
内置能力:
- 令牌桶限流 — 全局 + 每 API 独立限制
- 指数退避重试 — 429/5xx 自动重试,最多 3 次
- 自动分页 — 提供迭代器接口,调用者无感知
- 请求日志 — 每次调用记录时间、状态码、耗时
Service Layer
按业务领域封装高级接口:
| 服务 | 核心方法 | 说明 |
|---|---|---|
ContactService | search, getUser, getDepartment | 通讯录查询 |
MessageService | getHistory, send, search | 消息收发+搜索 |
CalendarService | getEvents, getFreeBusy | 日历查询 |
DocService | getContent, search | 文档读取 |
MinutesService | list, getTranscript, getSummary | 妙记数据 |
每个 Service 方法返回统一格式的 Python 对象(不是原始 JSON),并处理好分页、错误转换、数据清洗。
踩坑实录
坑 1: 消息格式地狱
飞书消息有 10+ 种 content_type:text、post(富文本)、image、file、merge_forward、share_chat、share_user、audio、media、sticker...
每种格式的 JSON 结构完全不同。要统一转为 Markdown,需要为每种类型写转换器:
# 消息类型 → Markdown 转换
converters = {
"text": lambda msg: msg["text"],
"post": convert_rich_text, # 复杂:含@/链接/图片
"merge_forward": convert_merge, # 递归:转发嵌套
"image": lambda msg: f"[图片:{msg['image_key']}]",
# ...
}最坑的是 merge_forward(合并转发) — 它里面嵌套的消息可以再嵌套合并转发,需要递归处理。
坑 2: 权限迷宫
同一个 API,Bot 和 User 身份看到的数据完全不同:
- Bot 能看群消息,但看不到你没加入的群
- User 能看自己的私聊,但 Bot 看不到任何私聊
- 某些 API 只对企业管理员开放
解法: Auth Manager 维护一个"能力矩阵"——每个操作标记用哪个身份调用。如果一种身份失败(403),尝试另一种。
坑 3: 增量的游标管理
飞书的分页游标(page_token)有有效期——不能存下来下次用。
解法: 不依赖飞书的游标。自己用时间戳做增量标记:记录"上次采集到哪个时间点",下次从那个时间开始请求。
设计决策回顾
| 决策 | 当时选择 | 现在评价 |
|---|---|---|
| 自建 vs 用官方 SDK | 自建适配层 | ✅ 正确——灵活性决定了能力上限 |
| Python 实现 | Python (同步 + aiohttp) | ✅ 与 AI agent 生态匹配 |
| 统一 Markdown 输出 | 所有消息转 Markdown | ⚠️ 部分类型转换有损 |
| 双身份模式 | Bot + User 并存 | ✅ 覆盖面足够 |
| 文件级游标 | YAML 存储游标状态 | ✅ 简单可靠 |
给做类似事情的人的建议
- 先做最小可用 — 不要一开始就想覆盖所有 API。先做"读消息+读通讯录",够用了再扩展
- 限流第一天就加 — 不加限流,你会在第一次批量操作时被封
- 消息格式做好抽象 — 不要在业务代码里解析 JSON,统一在适配层转换
- Token 刷新必须自动 — 手动管理 token 在 2 小时后你就会忘记
- 日志必须开 — 企业 API 出问题时,没日志你连问题在哪都定位不了
复盘总结:自建 API 适配层的 ROI 取决于你的使用深度。如果只是偶尔发个消息——用官方 SDK。如果要做数据管道式的批量采集——自建适配层几乎是必须的。关键是控制好范围:只封装你确实需要的 API,不要追求"完整覆盖"。