project/
├── larkx/                  # 产品代码
│   ├── cli.py              # lark 命令: auth/send/messages/history/api/download/listen…
│   ├── client.py           # WS 长连接 + /im/gateway/ 网关 + 通用 api() + 拉历史
│   ├── auth.py             # 凭证管理 + 扫码登录(QrLogin)
│   ├── config.py           # 配置(.env + 环境变量)
│   ├── media.py            # 图片/文件等资源下载
│   ├── proto/              # pb2 定义、25 种消息解码、请求构造、通用网关、id 生成
│   ├── storage/            # 消息存储(SQLAlchemy: SQLite 默认,MySQL 可切)
│   └── agent/              # 本地 coding agent 接入(Claude Code;触发规则;命令控制)
├── main.py                 # 产品入口: python main.py = lark listen --agent
├── dev/proto_pipeline/     # 维护者工具: 协议更新后重新提取 proto/cmd 的管线
├── static/resource/        # 图片资源
├── .env.example            # 配置样例
└── requirements.txt        # 项目依赖

import asyncio
from larkx import LarkAuth, LarkClient

client = LarkClient(LarkAuth())                  # 自动读取 ~/.larkx/credentials.json

async def on_message(msg):
    print(msg["chat_id"], msg["from_id"], msg["content"])
    # msg 已按类型解码(文本/图片/卡片/文件…),含 scope/anchor 话题边界

asyncio.run(client.connect_websocket(on_message))  # 自动 ACK + 30s 心跳 + 全类型解码

# 发消息 / 调任意接口
client.send_msg("你好", chat_id="<chat_id>")
client.api("chats.PullChatsByIdsRequest", {"chatIds": ["<chat_id>"]})

pip install -e .
# 或: pip install -r requirements.txt && python -m larkx.cli --help

LARKX_HOME=~/.larkx                                    # 数据目录
LARKX_STORAGE_URL=                                     # 留空=SQLite(LARKX_HOME/messages.db);MySQL 填 mysql+pymysql://<user>:<password>@<host>:3306/<db>
LARKX_CONTEXT_SCOPE=anchor                             # agent 上下文边界: anchor(会话+话题) | chat(会话) | global(全部共享)
LARKX_AGENT_BACKEND=claude                             # claude | none
LARKX_SYSTEM_PROMPT=                                   # 数字人系统提示词,留空用内置默认
LARKX_MARK_READ=true                                   # 收到消息自动标已读(清理未读角标),false 关闭
LARKX_TRIGGER_PREFIX=                                  # 只处理以此前缀开头的消息,如 /run;留空=全部处理

# 1. 登录(二选一)
lark auth qr                                     # 扫码登录(推荐): 终端出二维码,飞书 App 扫一下即可
lark auth import --cookie "<粘贴你的完整 cookie>"  # 或手动: F12 复制任意请求的完整 Cookie 头
lark auth check                                  # 校验凭证是否过期

# 2. 启动(WS 收消息 + 接入本地 AI 自动回复)
python main.py                                   # 等价于 lark listen --agent
# 凭证缺失/失效时会弹出交互菜单(扫码/粘贴 cookie),不用手动处理

<message>
  <chat id="7627..." name="项目A群" type="GROUP" anchor=""/>
  <sender id="7314..." name="张三"/>
  <at_me>true</at_me>
  <time>2026-07-26 10:00:00</time>
  <content>方案看一下</content>
</message>

pip install pytest
python -m pytest tests/    # 33 个离线测试,全自包含(合成帧,不依赖抓包)