Skill 清单
架构总览
飞书个人授权属于横切关注点——它不是一个独立场景,而是所有需要个人数据的场景在执行前必须经过的前置层。
分层能力汇总
| 层级 | 能力名称 | 说明 | 状态 |
|---|---|---|---|
| A — 系统能力 | check_user_auth | 检查 user_access_token 是否有效 | 🔲 待开发 |
| A — 系统能力 | refresh_user_token | 用 refresh_token 静默刷新 access_token | 🔲 待开发 |
| A — 系统能力 | send_auth_card | 向用户发送飞书交互授权卡片 | 🔲 待开发 |
| B — OAuth 流程 | build_oauth_url | 生成带 state 防 CSRF 的 OAuth 授权链接 | 🔲 待开发 |
| B — OAuth 流程 | handle_oauth_callback | 处理飞书回调,用 code 换取 token | 🔲 待开发 |
| B — OAuth 流程 | store_user_token | 加密存储 token 到 MongoDB | 🔲 待开发 |
| C — 存储 | user_tokens MongoDB 集合 | 按 open_id 索引,存储 token 和过期时间 | 🔲 待建 |
A 层 — 授权前置检查
check_user_auth(open_id) → AuthStatus
python
class AuthStatus(Enum):
VALID = "valid" # access_token 有效,直接使用
NEED_REFRESH = "refresh" # access_token 过期,refresh_token 有效
NEED_AUTH = "auth" # 未授权或 refresh_token 也过期调用规范:所有需要 user_access_token 的 Tool 在执行前必须先调用此函数,根据返回状态决定:
VALID→ 直接取access_token执行NEED_REFRESH→ 调用refresh_user_token,再执行NEED_AUTH→ 调用send_auth_card,中止当前指令执行,等待用户授权
refresh_user_token(open_id) → bool
调用飞书 POST /authen/v1/oidc/refresh_access_token,刷新成功后更新 MongoDB,返回 True;失败(refresh_token 过期)返回 False 并触发重授权流程。
send_auth_card(open_id, context: str)
向该用户发送飞书交互消息卡片,context 为触发场景描述(如「整理妙记」「查询文档」),用于卡片文案动态生成:
「需要授权才能帮你{context},点击以下按钮完成一次授权:」B 层 — OAuth 流程
build_oauth_url(open_id) → str
生成飞书 OAuth 授权链接,关键参数:
| 参数 | 值 | 说明 |
|---|---|---|
app_id | 应用 App ID | 从 .config.json 读取 |
redirect_uri | https://your-domain.com/oauth/feishu/callback | 需在飞书后台「安全设置 → 重定向 URL」注册 |
scope | minutes:minutes.basic:read minutes:minutes:readonly docx:document:readonly drive:drive:readonly task:task:read | 所有下游场景所需 scope 合集 |
state | {open_id}:{random_hex}:{timestamp} | 防 CSRF,回调时校验 |
handle_oauth_callback(code, state) → bool
注册 FastAPI 路由 GET /oauth/feishu/callback,处理飞书回调:
- 校验
state中的签名和时效(5 分钟内有效) - 解析
open_id - 调用
POST /authen/v1/oidc/access_token换取 token - 调用
store_user_token持久化 - 向用户发送「✅ 授权成功!继续执行…」消息
store_user_token(open_id, tokens: dict)
加密存储到 MongoDB user_tokens 集合,文档结构:
json
{
"open_id": "ou_xxxxxx",
"access_token": "encrypted:...",
"refresh_token": "encrypted:...",
"expires_at": "2026-03-19T14:00:00Z",
"refresh_expires_at": "2026-04-18T12:00:00Z",
"scope": "minutes:minutes.basic:read ...",
"updated_at": "2026-03-19T12:00:00Z"
}
access_token和refresh_token建议加密存储(AES-256),密钥通过环境变量注入,不存入数据库。
C 层 — Token 存储
MongoDB user_tokens 集合索引
| 字段 | 索引类型 | 说明 |
|---|---|---|
open_id | 唯一索引 | 每个用户唯一一条记录 |
expires_at | 普通索引 | 快速查询过期状态 |
refresh_expires_at | 普通索引 | 批量清理失效记录 |
与下游场景的集成方式
下游 Skill(妙记、云文档等)通过以下模式接入授权层:
python
# 每个需要 user_access_token 的 Tool 的标准写法
async def get_minute_transcript(open_id: str, minute_token: str) -> str:
status = check_user_auth(open_id)
if status == AuthStatus.NEED_REFRESH:
ok = await refresh_user_token(open_id)
if not ok:
status = AuthStatus.NEED_AUTH
if status == AuthStatus.NEED_AUTH:
await send_auth_card(open_id, context="读取妙记")
return "AUTH_REQUIRED" # 上层 agent 收到此信号后中止并等待
token = get_access_token(open_id) # 从 MongoDB 取
# ... 正常执行 API 调用各下游场景文档中引用此流程时,直接链接到本文档即可,无需重复说明授权细节。 详见:妙记会议纪要 · 授权说明 | 云文档 · 授权说明