FeishuBot — 飞书 Bot 开发框架

基于 lark-oapi SDK 的 FeishuChannel 构建的装饰器风格飞书机器人框架。

特性

• 装饰器注册 — @bot.on_message() / @bot.on_event() / @bot.use()
• 命令路由 — @cmds.command("ping") 自动解析 /ping 等命令
• 正则匹配 — @cmds.match(r"^(hi|hey)\b") 灵活匹配消息
• 洋葱中间件 — @bot.use() AOP 风格中间件链，支持前置/后置逻辑
• Koa 风格 handler 链 — handler 可通过 await next() 传递给下一个处理器
• 卡片构建器 — Card().header().add_text().add_button().build() 流式构建消息卡片
• 错误处理 — handler 异常自动回复用户 + 转发到 @bot.on_event("error")
• 定时任务 — @bot.scheduler.cron("*/5 * * * *") cron 表达式调度
• 插件系统 — plugins/ 目录下的 .py 文件自动发现和加载
• 插件热重载 — hot_reload=True 文件变更自动重载，开发无需重启
• 细粒度过滤 — @bot.on_message(sender_id=..., filter=...) 灵活匹配
• 扩展机制 — bot.install() 安装可选扩展

快速开始

1. 环境准备

# 克隆项目
git clone <repo-url> && cd FeishuBot

# 安装依赖（需要 uv）
uv sync

2. 配置飞书应用

在飞书开放平台创建应用，获取 APP_ID 和 APP_SECRET，写入 .env：

APPID=cli_xxxxxxxxxxxxx
APPSEC=xxxxxxxxxxxxxxxxxxxxxxxx

确保应用已开启以下能力：

• 机器人能力（Bot）
• 事件订阅 → 消息接收（im.message.receive_v1）
• 事件订阅方式选择「使用长连接接收」

3. 启动

uv run python main.py

项目结构

FeishuBot/
├── main.py                  # 入口：初始化 Bot → 加载插件 → 启动
├── .env                     # APPID / APPSEC 配置
├── pyproject.toml
├── wings_bot/               # 框架核心
│   ├── __init__.py          # 统一导出
│   ├── app.py               # FeishuBot 主类
│   ├── card.py              # 卡片构建器
│   ├── context.py           # MessageContext 消息上下文
│   ├── middleware.py         # 洋葱模型中间件链
│   ├── scheduler.py          # 定时任务调度器
│   ├── plugin.py             # 插件自动发现与加载
│   └── ext/
│       ├── __init__.py
│       └── commands.py       # 命令路由 + 正则匹配扩展
└── plugins/                  # 插件目录（自动加载）
    ├── __init__.py
    ├── echo.py               # 示例：消息回显 + 命令
    └── hello.py              # 示例：正则路由 + 定时任务 + 事件

---

核心概念

FeishuBot

框架入口类，持有飞书连接、处理器注册表和调度器。

from wings_bot import FeishuBot

bot = FeishuBot(
    app_id="cli_xxx",
    app_secret="xxx",
    error_message="处理消息时发生错误",  # 可选，handler 异常时的用户提示
    # 以下参数透传给 FeishuChannel，按需配置
    # log_level=lark.LogLevel.DEBUG,
)

属性	类型	说明
bot.channel	FeishuChannel	底层 SDK 实例，可直接调用 send()、stream() 等高级 API
bot.scheduler	Scheduler	定时任务调度器

MessageContext

所有消息处理器和中间件接收的统一上下文对象。

async def handler(ctx: MessageContext):
    # 只读属性
    ctx.text            # str   — 消息纯文本
    ctx.sender_id       # str   — 发送者 open_id
    ctx.sender_name     # str   — 发送者显示名称
    ctx.chat_id         # str   — 当前聊天 chat_id
    ctx.chat_type       # str   — "p2p" 或 "group"
    ctx.message_type    # str   — "text" / "image" / "interactive" 等
    ctx.message_id      # str   — 消息 ID
    ctx.is_mentioned    # bool  — Bot 是否被 @
    ctx.mentions        # list  — @提及 列表
    ctx.reply_to_id     # str|None — 如果是回复消息，被回复的消息 ID
    ctx.parent_message  # ReplyRef|None — 被回复的消息引用
    ctx.content         # MessageContent — 归一化内容对象
    ctx.raw             # InboundMessage — 原始 SDK 对象

    # 操作方法
    await ctx.reply("你好")                          # 回复纯文本
    await ctx.reply("**粗体**", msg_type="markdown") # 回复 markdown
    await ctx.reply({"text": "自定义内容"})           # 回复自定义内容
    await ctx.reply_card(card)                        # 回复卡片（传 Card.build() 或 dict）
    await ctx.send(chat_id, "主动发消息")             # 向指定聊天发送

    # 卡片交互（仅在 cardAction 回调中可用）
    ctx.event              # CardActionEvent | None — 卡片交互事件
    await ctx.update_card(card)                   # 更新当前卡片

    # 共享状态
    ctx.state           # dict  — 中间件和处理器之间传递数据

消息类型判断

ctx.content 是归一化的 discriminated union，通过 kind 字段判断：

if ctx.content.kind == "text":
    print(ctx.content.text)
elif ctx.content.kind == "image":
    print(ctx.content.image_key)
elif ctx.content.kind == "file":
    print(ctx.content.file_key, ctx.content.file_name)
elif ctx.content.kind == "interactive":
    print(ctx.content.card)  # 卡片 JSON

支持的 kind 值：text、post、image、file、audio、media、sticker、interactive、share_chat、share_user、system、location、folder、hongbao、general_calendar、share_calendar_event、video_chat、calendar、vote、todo、merge_forward、unknown。

---

消息处理

@bot.on_message()

注册消息处理器，支持过滤条件和优先级。

@bot.on_message()
async def handle_all(ctx):
    """处理所有消息。"""
    await ctx.reply(f"收到: {ctx.text}")

过滤条件

# 仅处理私聊文本消息
@bot.on_message(chat_type="p2p", message_type="text")
async def dm_text(ctx):
    await ctx.reply("这是私聊文本")

# 仅处理群聊中被 @ 的消息
@bot.on_message(chat_type="group", mentioned=True)
async def group_mention(ctx):
    await ctx.reply("你在群里 @ 了我")

# 仅处理指定用户的消息
@bot.on_message(sender_id="ou_abc123")
async def from_admin(ctx):
    await ctx.reply("管理员你好")

# 仅处理指定聊天的消息
@bot.on_message(chat_id="oc_xyz789")
async def specific_group(ctx):
    await ctx.reply("特定群消息")

# 自定义过滤函数
@bot.on_message(filter=lambda ctx: "urgent" in ctx.text)
async def urgent_only(ctx):
    await ctx.reply("紧急消息！")

# 组合过滤
@bot.on_message(chat_type="group", sender_id="ou_abc", message_type="image")
async def admin_images(ctx):
    await ctx.reply("收到管理员图片")

参数	类型	说明
chat_type	str	过滤聊天类型："p2p" 或 "group"
message_type	str	过滤消息类型："text"、"image" 等
mentioned	bool	过滤是否被 @：True = 被 @，False = 未被 @
sender_id	str	过滤发送者 open_id
chat_id	str	过滤聊天 chat_id
filter	Callable	自定义过滤函数，接收 MessageContext 返回 bool
priority	int	优先级，数值越大越先执行，默认 10

处理流程

1. 所有注册的 handler 按 priority 降序排列
2. 匹配的 handler 组成一条调用链（Koa 风格）
3. handler 签名决定行为：
   • async def h(ctx, next) — 接受 next，自行决定是否 await next() 传递
   • async def h(ctx) — 不接受 next，执行后链自动终止
4. 不调用 await next() 即可阻止后续 handler 执行

@bot.on_message(priority=100)
async def auth(ctx, next):
    if ctx.sender_id not in ALLOWED:
        await ctx.reply("无权限")
        return          # 不调 next，链终止
    await next()        # 传递给下一个 handler

@bot.on_message()
async def echo(ctx):
    await ctx.reply(ctx.text)  # 无 next 参数，链终止

---

命令路由

通过 CommandsExtension 实现，以指定前缀解析文本消息为命令。

安装

from wings_bot.ext.commands import CommandsExtension

cmds = CommandsExtension(prefix="/")
bot.install(cmds)

@cmds.command()

@cmds.command("echo", description="回复你发送的内容")
async def echo(ctx, args: list[str]):
    await ctx.reply(" ".join(args))

用户发送 /echo hello world 时，args = ["hello", "world"]。

别名

@cmds.command("help", aliases=["h", "?"], description="显示帮助")
async def help_cmd(ctx, args):
    await ctx.reply(cmds.generate_help())

/help、/h、/? 都会触发同一个处理器。

自动帮助

cmds.generate_help()
# 输出：
# 📋 可用命令：
#
#   /echo — 回复你发送的内容
#   /help (别名: /h, /?) — 显示帮助

@cmds.match()

正则表达式匹配，按注册顺序尝试，第一个匹配即执行。

@cmds.match(r"^(hi|hey|hello)\b")
async def greet(ctx, match: re.Match):
    await ctx.reply(f"{match.group(1)}！有什么可以帮你的？")

@cmds.match(r"^(时间|几点|what\s*time)")
async def show_time(ctx, match: re.Match):
    from datetime import datetime
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    await ctx.reply(f"现在是 {now}")

处理器签名：async def handler(ctx: MessageContext, match: re.Match)。

匹配优先级

安装 CommandsExtension 后，文本消息的处理顺序：

1. 命令匹配 — 文本以 / 开头，尝试匹配已注册命令
2. 正则匹配 — 按注册顺序尝试所有正则表达式
3. 穿透 — 都未命中时，消息继续流入普通的 @bot.on_message() 处理器

---

中间件

@bot.use()

洋葱模型中间件，按注册顺序包裹 handler。

@bot.use()
async def logging_mw(ctx, next):
    print(f"→ 收到: {ctx.text}")
    await next()               # 调用下一个中间件或最终 handler
    print(f"← 处理完成")

中断链

不调用 await next() 即可阻止后续处理：

@bot.use()
async def auth_mw(ctx, next):
    if ctx.sender_id not in ALLOWED_USERS:
        await ctx.reply("你没有权限使用此机器人")
        return                 # 不调用 next()，链中断
    await next()

共享状态

中间件可以通过 ctx.state 传递数据给下游：

@bot.use()
async def user_info_mw(ctx, next):
    ctx.state["user_role"] = get_user_role(ctx.sender_id)
    ctx.state["start_time"] = time.time()
    await next()

@bot.on_message()
async def handler(ctx):
    role = ctx.state.get("user_role", "guest")
    await ctx.reply(f"你的角色是: {role}")

执行顺序

注册了 3 个中间件 + 1 个 handler 时：

mw_1 前置 → mw_2 前置 → mw_3 前置 → handler
                                        ↓
mw_1 后置 ← mw_2 后置 ← mw_3 后置 ← ──┘

---

卡片构建器

通过 Card 类流式构建飞书消息卡片 JSON。

基本用法

from wings_bot.card import Card

card = (
    Card()
    .header(title="通知", subtitle="系统消息", template="blue")
    .add_text("你的订单已发货")
    .add_hr()
    .add_button("查看详情", variant="primary", url="https://example.com")
    .add_button("取消订单", variant="danger", value={"action": "cancel"})
    .build()
)
await ctx.reply_card(card)

可用方法

方法	说明
.config(wide_screen, enable_forward)	卡片配置
.header(title, subtitle?, template?)	设置标题头，template 可选颜色：blue/green/orange/red/violet 等
.add_text(text)	添加纯文本元素
.add_markdown(content)	添加 Markdown 元素
.add_hr()	添加分割线
.add_img(img_key, alt?)	添加图片
.add_button(text, variant?, url?, value?, on_click?)	添加按钮，连续调用自动合并到同一行，on_click 绑定点击回调
.add_column_set(columns)	添加多列布局（传 raw dict）
.build()	构建并返回 dict

按钮自动分组

连续 add_button 调用自动合并到同一个 action 行：

Card().add_button("A", variant="primary").add_button("B").build()
# → elements: [{"tag": "action", "actions": [btn_a, btn_b]}]

插入其他元素（如 add_hr()）会断开分组，后续按钮开始新的一行。

按钮回调（on_click）

add_button 支持 on_click 参数，绑定点击回调。回调按 message_id 隔离，不同卡片的同名按钮不会串。

from wings_bot.card import Card

async def on_confirm(ctx):
    # ctx.event → CardActionEvent（action.tag, action.value, message_id, chat_id, operator）
    await ctx.update_card(
        Card().header(title="已确认", template="green").add_text("操作完成").build()
    )

async def on_cancel(ctx):
    await ctx.update_card(
        Card().header(title="已取消", template="red").add_text("已取消").build()
    )

card = (
    Card()
    .header(title="确认操作", template="orange")
    .add_text("是否确认？")
    .add_button("确认", variant="primary", on_click=on_confirm)
    .add_button("取消", variant="danger", on_click=on_cancel)
)
await ctx.reply_card(card)  # 发送后自动绑定，无需手动处理 message_id

回调中可用的 ctx 方法：

方法	说明
ctx.event	CardActionEvent（含 action.value, message_id, operator 等）
ctx.chat_id	卡片所在聊天
ctx.sender_id	点击者 open_id
await ctx.update_card(card)	更新当前卡片（接受 Card 实例或 dict）
await ctx.reply(text)	在当前聊天发消息

手动处理 cardAction（不使用 on_click）

仍可通过 @bot.on_event("cardAction") 统一处理所有卡片事件：

@bot.on_event("cardAction")
async def on_card(event):
    if event.action.value and event.action.value.get("action") == "cancel":
        await bot.channel.update_card(event.message_id, new_card_dict)

---

错误处理

内置行为

当 handler 抛出异常时，框架自动：

1. 回复用户一条错误消息（可配置）
2. 将异常转发给 @bot.on_event("error") 处理器
3. 记录日志，bot 继续运行不会崩溃

自定义错误消息

bot = FeishuBot(
    app_id="...",
    app_secret="...",
    error_message="抱歉，处理你的消息时出了点问题，请稍后重试。",
)

监听错误事件

@bot.on_event("error")
async def on_error(exc):
    # 接收所有错误：handler 异常 + 发送失败
    logger.exception("Bot 错误", exc_info=exc)
    # 也可以发到监控群
    await bot.channel.send(monitor_chat_id, {"text": f"错误: {exc}"})

---

定时任务

@bot.scheduler.cron()

# 同步函数
@bot.scheduler.cron("*/5 * * * *")
def heartbeat():
    print("Bot is alive")

# 异步函数（通过 FeishuChannel.schedule() 执行）
@bot.scheduler.cron("0 9 * * 1-5")
async def morning_report():
    await bot.channel.send(chat_id, {"text": "早上好！这是今日报告。"})

Cron 表达式格式

5 字段标准格式：分 时 日 月 周

表达式	说明
*/5 * * * *	每 5 分钟
0 * * * *	每小时整点
0 9 * * 1-5	工作日早上 9 点
30 14 * * *	每天 14:30
0 0 1 * *	每月 1 日零点

调度行为

• 后台守护线程，对齐整分钟触发
• 协程任务通过 FeishuChannel.schedule() 在事件循环中执行
• 同步任务直接在后台线程中执行
• 调度器随 bot.run() 自动启动，随 bot.stop() 自动停止

---

事件处理

@bot.on_event()

注册非消息事件的处理器。支持的事件类型：

事件名	说明	handler 参数
"message"	收到消息（通常用 @bot.on_message() 代替）	InboundMessage
"cardAction"	卡片交互（按钮点击等）	CardActionEvent
"reaction"	表情回应	ReactionEvent
"botAdded"	Bot 被添加到群	BotAddedEvent
"botLeave"	Bot 被移出群	BotLeaveEvent
"messageRead"	消息已读	MessageReadEvent
"comment"	文档评论	Comment event
"raw"	原始事件	原始 payload
"error"	错误事件	Exception
"reconnecting"	WS 重连中	无参数
"reconnected"	WS 已重连	无参数

@bot.on_event("botAdded")
async def on_added(event):
    print(f"Bot 被添加到群: {event.chat_id}")

@bot.on_event("cardAction")
async def on_card(event):
    action = event.action
    print(f"卡片按钮: tag={action.tag}, value={action.value}")

@bot.on_event("reaction")
async def on_reaction(event):
    if event.action == "added":
        print(f"用户 {event.operator.open_id} 添加了 {event.emoji_type}")

@bot.on_event("error")
async def on_error(exc):
    logger.exception("Bot 发生错误", exc_info=exc)

---

插件系统

约定

• 每个插件是 plugins/ 目录下的一个 .py 文件
• 文件必须定义 register(bot, **kwargs) 函数
• 以 _ 开头的文件会被跳过（如 __init__.py）

编写插件

# plugins/weather.py

def register(bot, cmds=None):
    """天气插件。"""

    if cmds is not None:
        @cmds.command("weather", description="查询天气")
        async def weather(ctx, args):
            city = " ".join(args) if args else "北京"
            # ... 查询天气 API ...
            await ctx.reply(f"{city}: 晴，25°C")

        @cmds.match(r"今天.*天气")
        async def weather_nlp(ctx, match):
            await ctx.reply("今天天气不错！")

    @bot.scheduler.cron("0 8 * * *")
    async def daily_weather():
        await bot.channel.send(
            monitor_chat_id,
            {"text": "今日天气：晴，25°C"}
        )

自动加载

# main.py 中
from wings_bot import load_plugins

load_plugins(bot, cmds=cmds)
# 扫描 plugins/ 目录，按文件名字典序加载

禁用插件

load_plugins(bot, cmds=cmds, disabled=["weather", "debug"])

热重载（开发模式）

load_plugins(bot, cmds=cmds, hot_reload=True)

启用后，修改 plugins/ 目录下的 .py 文件并保存，框架会自动：

1. 移除旧插件注册的 handlers 和 middlewares
2. 重新加载模块并调用 register()
3. 无需重启 bot

注意事项：

• 定时任务（@bot.scheduler.cron）不支持热重载
• 0.5s 去抖，防编辑器多次保存触发
• 依赖 watchdog 库

传递额外参数

load_plugins() 的 **kwargs 会透传给每个插件的 register() 函数：

# main.py
load_plugins(bot, cmds=cmds, db=db_connection)

# plugins/data.py
def register(bot, cmds=None, db=None):
    # db 是从 main.py 传过来的
    ...

---

完整示例

main.py

import logging
import os

from dotenv import load_dotenv

from wings_bot import FeishuBot, load_plugins
from wings_bot.ext.commands import CommandsExtension

load_dotenv()
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
)

bot = FeishuBot(
    app_id=os.environ["APPID"],
    app_secret=os.environ["APPSEC"],
    error_message="处理消息时发生错误，请稍后重试。",
)

cmds = CommandsExtension(prefix="/")
bot.install(cmds)

load_plugins(bot, cmds=cmds)

if __name__ == "__main__":
    bot.run()

plugins/echo.py

import time
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from wings_bot.context import MessageContext

def register(bot, cmds=None):
    @bot.use()
    async def logging_mw(ctx: MessageContext, next):
        start = time.time()
        print(f"→ [{ctx.chat_type}] {ctx.sender_name}: {ctx.text}")
        await next()
        print(f"← ({time.time() - start:.3f}s)")

    @bot.on_message(message_type="text")
    async def echo_text(ctx: MessageContext):
        await ctx.reply(ctx.text)

    if cmds is not None:
        @cmds.command("ping")
        async def ping(ctx: MessageContext, args):
            await ctx.reply("Pong!")

        @cmds.command("help")
        async def help_cmd(ctx: MessageContext, args):
            await ctx.reply(cmds.generate_help())

        @cmds.match(r"^(hi|hello)\b")
        async def greet(ctx: MessageContext, match):
            await ctx.reply("你好！")

---

高级用法

直接使用 FeishuChannel

bot.channel 暴露了完整的 FeishuChannel API：

# 发送 markdown
await bot.channel.send(chat_id, {"markdown": "**粗体** 和 [链接](https://...)"})

# 用 Card 构建器发送卡片
from wings_bot.card import Card
card = Card().header(title="标题", template="green").add_markdown("卡片内容").build()
await bot.channel.send(chat_id, card)

# 流式输出（打字机效果）
await bot.channel.stream(chat_id, {"markdown": my_async_generator})

# 下载图片
data = await bot.channel.download_resource(file_key, "image")

# 编辑消息
await bot.channel.edit_message(message_id, "新内容")

自定义扩展

任何实现了 _install(bot) 方法的对象都可以作为扩展安装：

class MyExtension:
    def __init__(self, config):
        self.config = config

    def _install(self, bot):
        @bot.on_message(priority=50)
        async def my_handler(ctx):
            # 自定义处理逻辑
            pass

bot.install(MyExtension(config={}))

自定义 FeishuChannel 参数

FeishuBot 的 **channel_kwargs 透传给 FeishuChannel：

from lark_oapi.core.enum import LogLevel

bot = FeishuBot(
    app_id="...",
    app_secret="...",
    log_level=LogLevel.DEBUG,
    domain="https://open.larksuite.com",  # 海外版 Lark
)

---

事件流全景

用户发送消息
    ↓
飞书服务器推送事件
    ↓
WebSocket 长连接接收
    ↓
FeishuChannel 自动处理
    ├── 消息解析（19 种类型 → InboundMessage）
    ├── 去重（WS 重传保护）
    ├── 策略检查（黑白名单、@门控）
    └── 安全管道（批处理、串行队列）
    ↓
FeishuBot._on_message(msg)
    ↓
构建 MessageContext(ctx)
    ↓
全局中间件链（洋葱模型）
    ↓
处理器链（Koa 风格，按 priority 降序）
    ├── priority=100 → CommandsExtension（命令/正则）
    │   └── await next() → 穿透到下一个 handler
    ├── priority=20  → 群聊 @处理器
    ├── priority=10  → 通用处理器（默认）
    └── handler 可通过 await next() 传递
    ↓
handler 执行 → ctx.reply() / ctx.send()
    ↓
FeishuChannel.send() → 飞书 API → 消息送达

异常时：
    → 回复用户错误消息
    → 转发到 @bot.on_event("error") 处理器
    → 记录日志，bot 继续运行