近一年 LLM 的发展，给 AI 行业创造了更多的技术想象空间。越来越多优秀的应用方案井喷出现，从最开始的 Cline、Copilot，到现在的 Cursor、Trae，从集成环境到 CLI 的 Codex、Claude Code，从 MCP、Agent 到最近热度居高不下的Skills。最近又掀起了一波 Clawdbot（已改名 Moltbot）的热潮，网上甚至出现了屯 Mac mini 的情况；在剪辑软件方面，Remotion 可以编程式生成视频，也被拿来和剪映对比。底层逻辑其实都差不多，离不开 LLM + MCP + Prompt。

本文主要是想做和 Clawdbot 一样的事情，不同的是：我们用飞书的能力，通过和机器人对话维护上下文，指挥电脑干活。这样的好处是：可以在群聊里多人共享一个上下文，同时也不需要对外暴露公网 IP。底层的 AI 驱动我们选 Claude Code（也可以扩展成别的 Agent）。

核心思路：飞书机器人负责通信，Claude Code 当计算机「大脑」指挥操作。

飞书机器人

我们的消息通信核心就是：通过飞书机器人收发消息。这样就不用自己从零开发一套 Chatbot 了。下面带着大家一步步创建飞书应用机器人，并开通需要的聊天权限。

飞书后台应用

进入飞书开发者后台，点击企业自建应用。

createApp

这里我们可以填一些应用的基本信息：

点创建后，进到应用里，在凭证与基础信息能看到 App ID 和 App Secret，后面用这个应用都要靠这两个。

添加机器人

在应用能力里点添加应用能力，给应用加一个机器人。机器人下面还能做很多自定义菜单之类的，我们这儿用不到，保持默认就行。

长连接测试

在开发配置里的事件与回调，把订阅方式选成「使用长连接接收事件」。飞书要求我们先连上、确认没问题，才能保存这个模式。那我们就先把连接跑通（不用发布应用），代码参考飞书开发文档：

import lark_oapi as lark
client = lark.ws.Client(
    "App ID",
    "App Secret",
)
client.start()

跑起来后如果看到 connected to wss:xxxx，就说明长连接好了，这时候去事件配置里就可以保存「长连接」这种订阅方式了。

机器人权限

在事件配置中，为机器人添加事件：

addEvent

我们只要订阅 接收消息 这个事件就可以了。其他的比如「用户和机器人的会话首次被创建」，可以用来做第一次进会话的提示，按需加就行。加完事件后，会弹出来一个「确认添加权限」，点一下给机器人加上聊天权限，这个权限是免审的。

应用发布

到这儿飞书这边的权限、能力都齐了，可以去发正式版了：

  1. 点版本管理与发布；
  2. 创建版本、保存；
  3. 确认发布；

createAppVersion

发完之后，大家就都能在飞书里搜到这个机器人了（没发布前只有你自己能看到）。

本地 Python 服务

在本地跑一个长连接监听服务，用来听飞书机器人的消息，把对话内容转发给 Claude Code，再把 Claude Code 的回复发回给发消息的人。

持久化

为什么要做持久化呢？

因为 Claude Code 是靠 session_id 来区分一次对话的。如果我们每次对话都新建一个会话，Claude Code 就不知道之前的聊天上下文，相当于没有记忆。所以我们的做法是：用飞书里机器人会话的 chat_id 当上下文的 key，没有就建一个上下文，有就用当前的。用 SQLite 存这个映射的话，可以这样写：

"""
SQLite 存储 chat_id -> session_id 映射
"""
import sqlite3
from pathlib import Path

DB_PATH = Path(__file__).parent.parent.parent / "data" / "sessions.db"


def _get_conn():
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH))
    conn.execute("""
        CREATE TABLE IF NOT EXISTS sessions (
            chat_id TEXT PRIMARY KEY,
            session_id TEXT
        )
    """)
    conn.commit()
    return conn


def get_session(chat_id: str) -> str | None:
    """获取 session_id"""
    conn = _get_conn()
    cursor = conn.execute("SELECT session_id FROM sessions WHERE chat_id = ?", (chat_id,))
    row = cursor.fetchone()
    conn.close()
    return row[0] if row else None


def save_session(chat_id: str, session_id: str):
    """保存 session_id"""
    conn = _get_conn()
    conn.execute("""
        INSERT OR REPLACE INTO sessions (chat_id, session_id)
        VALUES (?, ?)
    """, (chat_id, session_id))
    conn.commit()
    conn.close()

用的时候先用 chat_id 去查有没有对应的 session_id；没有的话，第一次聊完 Claude Code 会返回它创建并维护的 session_id，我们把它存进库就行了。

上下文消息队列

上面的聊天逻辑会碰到一个问题：因为处理是异步的，如果 Claude Code 还没处理完上一条，你又发了一条，就会出问题，比如本来持久化好的 session 被重新创建了。我们肯定是希望消息一条一条按顺序处理，这样上下文才连贯。

对此我们给每个 chat_id 建一个消息队列，一条一条处理；队列空了就把这个队列收掉。要注意多线程下的资源竞争，用锁保护一下。代码如下：

def _process_chat_queue(chat_id: str, message_id: str, chat_type: str, queue: Queue):
    """
    处理指定 chat_id 的消息队列（FIFO）
    同一 chat_id 串行处理，不同 chat_id 可并行
    """
    while True:
        # 在锁内检查并获取消息，确保线程安全
        with _queue_lock:
            if queue.empty():
                # 队列空了，销毁并退出
                _active_queues.pop(chat_id, None)
                return
            text = queue.get_nowait()

        try:
            reply = chat_with_claude(chat_id, text)
            if chat_type == "group":
                reply_message(message_id, reply)
            else:
                send_message(chat_id, reply)
            logger.info(f"回复: {reply[:100]}...")
        except Exception as e:
            logger.error(f"处理失败 [{chat_id[:8]}...]: {e}")


def enqueue_message(chat_id: str, message_id: str, text: str, chat_type: str):
    """
    将消息加入队列，无队列则创建
    """
    with _queue_lock:
        if chat_id in _active_queues:
            # 已有队列，直接加入
            _active_queues[chat_id].put(text)
        else:
            # 创建新队列并启动 worker
            queue = Queue()
            queue.put(text)
            _active_queues[chat_id] = queue
            threading.Thread(
                target=_process_chat_queue,
                args=(chat_id, message_id, chat_type, queue),
                daemon=True
            ).start()

流程图：

context-queue-flowchart

Claude Code 执行

Claude Code 执行指令我们通过官方 SDK claude_agent_adk 来完成。需要注意的是要给它开执行权限，比如：

这些权限已经比较高了，用的时候自己注意安全。

消息回复与发送

消息的回复和发送我们用飞书官方 SDK，发送消息 和 回复消息 的文档都有。如果要支持图片、视频、富文本这类多模态消息，可以对着飞书开发文档自己扩展一下。

测试结果

群聊

groupChat

私聊

畅想 / TODO

目前聊天还是以文字为主，复杂的视频、图片、语音等都还没接，机器人也只能回文本。后面可以慢慢加上：

仓库地址

远程 Claude Code 服务仓库，GitHub 主页还有别的项目可以看看。

个人网站

QQ交流 群：523219063