BotZone

多平台机器人管理 SDK — 飞书 + 微信 iLink + 企业微信 Webhook，供 AI Agent / 后端服务使用。

零外部依赖，Node.js ≥ 20。

npm install botzone

架构

botzone/src/
├── index.js                   # 统一导出
├── core/                      # 通用基础设施
│   ├── api-client.js          HTTP 基类（AbortController 超时、重试、日志）
│   ├── errors.js              错误体系（Token/RateLimit/Permission/Network）
│   ├── logger.js              调试日志
│   ├── bot-manager.js         多 Bot 管理（工厂注入、标签、广播）
│   ├── poller.js              轮询引擎（Session 恢复、降级策略）
│   ├── state-store.js         通用 JSON 文件持久化 + cursor 管理
│   ├── config-cache.js        TTL 内存缓存
│   └── message-store.js       消息持久化（按平台/bot/会话 分文件）
├── feishu/                    # 飞书平台
│   ├── feishu-api.js          FeishuApiClient + 错误映射
│   ├── feishu-bot.js          FeishuBot（收发/卡片/文件/@/撤回/延时/历史）
│   ├── feishu-token.js        tenant_access_token 管理
│   ├── feishu-message.js      消息构造器（text/post/card/image/file）
│   └── feishu-uploader.js     图片/文件上传
└── wechat/                    # 微信平台
    ├── wechat-api.js          WechatApiClient（iLink Bot API）
    ├── wechat-bot.js          WechatBot（收发/媒体/光标轮询/typing）
    ├── wechat-media.js        AES-128-ECB + CDN 上传/下载管线
    ├── wechat-message.js      微信消息 item 构造器
    ├── wechat-robot.js        企业微信 Webhook 发送器
    └── wechat-token.js        微信 access_token 管理

快速开始

import { BotManager, FeishuBot, FeishuApiClient, FeishuTokenStore } from 'botzone'

const api = new FeishuApiClient()
const tokenStore = new FeishuTokenStore(api)
const mgr = new BotManager({
  factory: (name, info) => new FeishuBot({ name, appId: info.app_id, appSecret: info.app_secret, apiClient: api, tokenStore })
})
mgr.loadConfig()  // 从 ~/.botzone/bots.json 加载

const bot = mgr.getBot('mybot')
await bot.sendToChat('oc_xxx', '你好')

API 参考

一、Core（通用基础设施）

ApiClient

通用 HTTP 客户端，平台差异通过子类 override _headers / _buildError 实现。

import { ApiClient } from 'botzone'

构造函数

new ApiClient(opts?)

| 参数 | 类型 | 默认 | 说明 | |------|------|------|------| | opts.baseUrl | string | '' | API 基础 URL | | opts.debug | boolean | false | 开启调试 | | opts.maxRetries | number | 3 | 最大重试次数 | | opts.retryDelay | number | 1000 | 初始退避延迟 ms | | opts.timeout | number | 30000 | 请求超时 ms（AbortController） | | opts.logger | Logger | null | 日志实例 |

方法

request(method, path, { token?, body?, params?, headers?, signal? }) → Promise<object|string>
get(path, opts?)     → Promise<object|string>
post(path, opts?)    → Promise<object|string>
patch(path, opts?)   → Promise<object|string>
delete(path, opts?)  → Promise<object|string>

• method — 'GET'|'POST'|'PATCH'|'DELETE'
• path — API 路径（如 '/im/v1/messages'）
• opts.token — Bearer token
• opts.body — 请求体（自动 JSON.stringify）
• opts.params — URL 查询参数
• opts.headers — 额外请求头
• opts.signal — 外部 AbortSignal（与内部超时合并）

特性：

• 自动重试：5xx / 429，指数退避 delay * 2^n
• 超时控制：AbortController，默认 30s
• 子类可 override _headers(token)、_buildError(resp)、_onTimeout(method, path)

Errors

import { BotZoneError, TokenError, RateLimitError, PermissionError, NetworkError } from 'botzone'

所有错误继承自 BotZoneError（继承 Error）：

| 属性 | 类型 | 说明 | |------|------|------| | code | number | 错误码 | | message | string | 错误消息 | | hint | string\|null | 排查建议 | | apiResponse | object\|null | 原始 API 响应 | | cause | any\|null | 原始错误 |

try { ... }
catch (err) {
  if (err instanceof TokenError)    { /* Token 失效 */ }
  if (err instanceof RateLimitError){ /* 限流 */ }
  if (err instanceof PermissionError){ /* 权限不足 */ }
  if (err instanceof NetworkError)  { /* 网络错误 */ }
}

BotManager

多 Bot 生命周期管理 + 配置持久化 + 批量操作。

import { BotManager } from 'botzone'

构造函数

new BotManager({ factory, configPath? })

| 参数 | 类型 | 默认 | 说明 | |------|------|------|------| | factory | (name, info) => bot | 必填 | Bot 工厂函数 | | configPath | string | ~/.botzone/bots.json | 配置文件路径 |

配置

loadConfig()           → this           从文件加载所有 Bot，返回 this（链式）
saveConfig()           → void           持久化当前配置 + 缓存到文件

Bot 管理

addBot(name, appId, appSecret, tags?) → void       注册新 Bot
removeBot(name)                       → void       移除 Bot
getBot(name)                          → bot        获取单个 Bot（不存在则抛异常）
listBots()                            → Array      列出所有 Bot [{name, appId, tags}]

标签

tagBot(name, tag)      → void          打标签
untagBot(name, tag)    → void          去标签
getBotsByTag(tag)      → Array<bot>    按标签获取 Bot 列表

链接

getCreateBotLink()      → string      飞书创建应用链接
getAuthLink(name, scope)→ string      为某 Bot 生成授权链接
getAuthLinks(scope)     → object      为所有 Bot 批量生成授权链接

联系人

refreshContacts()  → Promise<object>  从 API 拉取所有 Bot 的群聊+用户+私聊映射
getContactsMap()   → object           从缓存读取（不调 API）

批量操作

broadcastToChat(chatId, content, opts?)     → Promise<Array>  所有 Bot → 同一群聊
broadcastToTag(tag, chatId, content, opts?)  → Promise<Array>  按标签 → 群聊
broadcastPrivate(userOpenId, content)        → Promise<Array>  所有 Bot → 私聊
healthCheckAll()                             → Promise<Array>  所有 Bot 健康检查
forwardImage(srcBot, dstBot, messageId, key) → Promise<object> 跨 Bot 转发图片

Poller

轮询接收引擎，支持群聊 / 私聊 / 全量三种模式。

import { Poller } from 'botzone'

构造函数

new Poller(bot, opts?)

| 参数 | 类型 | 默认 | 说明 | |------|------|------|------| | bot | object | 必填 | Bot 实例（duck-typing） | | opts.interval | number | 3000 | 轮询间隔 ms | | opts.dedupWindowMs | number | 300000 | 去重窗口 ms | | opts.store | MessageStore | null | 传入后自动保存每条消息 | | opts.platform | string | 'feishu' | 平台标识（消息分目录用） | | opts.onSessionExpired | (bot, err) => Promise<boolean> | null | Session 过期回调（返回 true 表示已恢复） | | opts.sessionCooldown | number | 60000 | Session 过期冷却 ms |

方法

fetchOnce(containerId)                         → Promise<Array>   一次性拉取（不去重）
poll(containerId, { signal?, onMessage? })     → AsyncGenerator   持续轮询
pollChat(chatId, opts?)                        → AsyncGenerator   poll 别名
pollPrivate(userOpenId, opts?)                 → AsyncGenerator   P2P 轮询（自动激活通道）
pollAll({ signal?, onMessage? })               → AsyncGenerator   全量轮询（所有群+私聊）

pollAll 产出的每条消息附加 _source 字段：

{ bot: 'botname', chatName: '群名', chatId: 'oc_xxx', mode: 'group'|'p2p' }

错误处理：

• TokenError → 触发 Session 恢复（60s 冷却 → 回调重载 token → 继续）
• 其他错误 → yield { _error: message }，不中断轮询

StateStore

通用 JSON 文件持久化。

import { StateStore } from 'botzone'

new StateStore(stateDir?)      默认 ~/.botzone/

loadJSON(filename)             → any|null          读取 JSON
saveJSON(filename, data)       → void              原子写入 JSON（tmp+rename, chmod 0600）
loadText(filename)             → string|null        读取纯文本
saveText(filename, text)       → void              写入纯文本
remove(filename)               → void              删除文件
listFiles()                    → Array<string>     列出文件
loadCursor(filename?)          → string            读取游标（默认 cursor.txt）
saveCursor(cursor, filename?)  → void              写入游标
clearCursor(filename?)         → void              删除游标
subStore(subdir)               → StateStore        创建子目录 Store

const base = new StateStore()                      // ~/.botzone/
const wechat = base.subStore('wechat')             // ~/.botzone/wechat/
wechat.saveCursor('xxx')                           // ~/.botzone/wechat/cursor.txt

ConfigCache

TTL 内存缓存。

import { ConfigCache } from 'botzone'

const cache = new ConfigCache(ttlMs?)   // 默认 24h
cache.get(key)           → any|null
cache.set(key, value, ttlMs?)
cache.getOrSet(key, fn, ttlMs?)  → Promise<any>   缓存命中返回，否则调用 fn 并缓存
cache.has(key)           → boolean
cache.delete(key)        → void
cache.clear()            → void
cache.size               → number

MessageStore

消息持久化存储，按 平台/bot/会话 分文件。

import { MessageStore } from 'botzone'

const store = new MessageStore(baseDir?)  // 默认 ~/.botzone/messages/

方法

saveMessages(platform, botName, chatType, chatId, chatName?, msgs) → number
loadMessages(platform, botName, chatType, chatId) → {messages, count, updatedAt}|null
listChats(platform, botName)    → Array<{chatType, chatId, file, count}>
listBots(platform)              → Array<string>
stats(platform)                 → {bots: {[name]: {chats, messages}}, total}

目录结构：~/.botzone/messages/{feishu|wechat}/{botName}/{chatType}_{chatId}.json

每条消息标准化格式：

{
  "messageId": "om_xxx",
  "msgType": "text",
  "content": "...",
  "sender": { "id": "...", "idType": "..." },
  "mentions": [],
  "rootId": null,
  "createTime": 1717000000000
}

Logger

调试日志，可开关，输出到 stderr，[BOTZONE] 前缀，500 字符截断。

import { Logger } from 'botzone'

const log = new Logger(enabled?)   // 默认 false
log.enable() / log.disable()
log.enabled             → boolean
log.request(method, url, reqBody, status, respBody, duration)
log.error(method, url, error, retry?)
log.info(msg)
log.warn(msg)

二、Feishu（飞书平台）

FeishuApiClient

import { FeishuApiClient, fromApiResponse } from 'botzone'

const api = new FeishuApiClient(opts?)  // baseUrl 默认 https://open.feishu.cn/open-apis

继承 ApiClient，override _buildError 调用 fromApiResponse 分类错误。

fromApiResponse(apiResp) → BotZoneError 子类

| 飞书 code | 映射错误 | |-----------|---------| | 99991663, 99991664, 99991665 | TokenError | | 99991400, 230001 | RateLimitError | | 230027 | PermissionError | | 其他 | BotZoneError |

FeishuBot

单个飞书机器人的全部能力。需要先通过 BotManager 或手动提供凭据创建。

import { FeishuBot } from 'botzone'

const bot = new FeishuBot({ name, appId, appSecret, apiClient, tokenStore? })

授权

getAuthLink(scope) → string   生成飞书授权链接

联系人

getUsers()  → Promise<Array<{name, openId, unionId}>>    获取用户列表（自动缓存）
getChats()  → Promise<Array<{name, chatId, mode}>>       获取群聊列表（自动缓存）

消息发送

sendToChat(chatId, content, opts?)    → Promise<{messageId, chatId}>
sendPrivate(userOpenId, content, opts?) → Promise<{messageId, chatId}>  自动缓存 P2P chatId
reply(messageId, content, opts?)       → Promise<{messageId}>
sendImage(targetId, imageKey, targetType?)  → Promise<{messageId, chatId}>
sendFile(targetId, fileKey, targetType?)    → Promise<{messageId, chatId}>
sendCard(targetId, cardObj, targetType?)    → Promise<{messageId, chatId}>
sendLocalImage(targetId, imagePath, targetType?) → Promise<{messageId}>
sendLocalFile(targetId, filePath, fileType?, targetType?) → Promise<{messageId}>
sendDelayed(chatId, content, opts?)  → {taskId, promise, cancel}  返回取消函数

| opts 字段 | 类型 | 说明 | |-----------|------|------| | atUserIds | string[] | @ 的用户 open_id | | atNames | string[] | @ 的用户显示名 | | msgType | string | 消息类型，默认 'text' | | delayMs | number | 延时毫秒（sendDelayed），默认 1000 |

消息接收

receiveMessages(chatId, opts?)        → Promise<Array>    拉取群聊消息（自动过滤自己的消息）
receivePrivateMessages(opts?)          → Promise<object>   拉取所有私聊消息
receiveMentions(chatId, opts?)         → Promise<Array>    只拉 @ 自己的消息
fetchHistory(chatId, opts?)            → Promise<{items, nextPageToken, hasMore}>  分页拉取
fetchAllHistory(chatId, opts?)         → Promise<Array>    自动翻页拉全量

| opts 字段 | 类型 | 默认 | 说明 | |-----------|------|------|------| | pageSize | number | 20 | 每页条数 | | seenIds | Set<string> | 新 Set | 去重集合 | | sortType | string | ByCreateTimeDesc | 排序 | | startTime | string\|Date | — | 起始时间 | | endTime | string\|Date | — | 结束时间 | | pageToken | string | — | 分页游标 | | maxPages | number | 50(fetchAllHistory) | 最大翻页数 |

消息对象格式：

{
  messageId: 'om_xxx',
  chatId: 'oc_xxx',
  rootId: null,           // 回复的消息 ID
  parentId: null,
  msgType: 'text',
  content: '{"text":"hello"}',  // 原始 JSON 字符串
  sender: { id: 'ou_xxx', idType: 'open_id', senderType: 'user' },
  mentions: [{ key: '...', id: 'ou_xxx', name: '张三', idType: 'open_id' }],
  createTime: 1717000000000,
  updateTime: 1717000000000,
}

资源下载

downloadResource(messageId, fileKey, type, savePath?) → Promise<{buffer, contentType, savedPath, size}>

消息管理

recall(messageId)        → Promise<true>       撤回消息
syncAndSave(store, opts?)→ Promise<{group, p2p}>  同步所有会话到 MessageStore
healthCheck()            → Promise<{ok, tokenValid, expireIn?, token?}>

FeishuTokenStore

飞书 tenant_access_token 缓存与自动刷新。

import { FeishuTokenStore } from 'botzone'

const store = new FeishuTokenStore(apiClient)

getToken(appId, appSecret)      → Promise<string>       获取有效 token（自动刷新）
refreshToken(appId, appSecret)  → Promise<string>       强制刷新
getTokenInfo(appId)             → {token, expiresAt}|null  查询缓存
clear(appId) / clearAll()      → void                   清除缓存

MessageBuilder（飞书消息构造器）

import { text, atTag, post, card, image, file, MessageBuilder } from 'botzone'

text(text, { atUserIds?, atNames? }?)  → string    '{"text":"..."}'
atTag(userId, name?)                   → string    '<at user_id="..">@name</at>'
post({ title?, content })              → string    富文本消息
card(cardObj)                          → string    交互式卡片
image(imageKey)                        → string    '{"image_key":"..."}'
file(fileKey)                          → string    '{"file_key":"..."}'

// @ 某人的消息
const msg = text('看消息', { atUserIds: ['ou_xxx'], atNames: ['张三'] })

// 卡片
const card = card({
  header: { title: '确认', template: 'red' },
  elements: [{ tag: 'button', text: { content: '确认' }, type: 'primary' }]
})
await bot.sendCard('oc_xxx', card)

Uploader（飞书上传）

import { uploadImage, uploadFile, Uploader } from 'botzone'

uploadImage(bot, filePath, { onProgress? }?)  → Promise<{imageKey, size}>
uploadFile(bot, filePath, fileType?, { onProgress? }?) → Promise<{fileKey, size}>

支持进度回调：onProgress({ state: 'uploading'|'done', loaded, total })

三、WeChat（微信平台）

WechatApiClient

微信 iLink Bot API 客户端。

import { WechatApiClient, SESSION_TIMEOUT_ERRCODE } from 'botzone'

const api = new WechatApiClient({ baseUrl?, token?, routeTag? })

继承 ApiClient，自动添加 AuthorizationType: ilink_bot_token、X-WECHAT-UIN、Content-Length。

常量

SESSION_TIMEOUT_ERRCODE = -14

方法

getBotQrcode(botType?)           → Promise<object>    获取登录二维码
getQrcodeStatus(qrcode, timeout?)→ Promise<object>    轮询扫码状态（长轮询，超时返回 {status:'wait'}）
getUpdates(buf?, timeout?)       → Promise<object>    长轮询拉取消息
getUploadUrl({filekey, media_type, to_user_id, rawsize, rawfilemd5, filesize, aeskey}) → Promise<object>
sendMessage({toUserId, items, contextToken?})   → Promise<{clientId, response}>
sendTextMessage({toUserId, text, contextToken?})→ Promise<{clientId, response}>
getConfig(ilinkUserId, contextToken?)           → Promise<object>
sendTyping({ilinkUserId, typingTicket, status?})→ Promise<object>

二维码状态枚举： | status | 含义 | |--------|------| | wait | 等待扫码 | | scaned | 已扫码待确认 | | scaned_but_redirect | 需重定向到其他 IDC | | confirmed | 登录成功 | | expired | 二维码过期 |

WechatBot

微信 iLink Bot 高层封装。

import { WechatBot } from 'botzone'

const bot = new WechatBot({ name, apiClient, store? })

| 参数 | 类型 | 说明 | |------|------|------| | name | string | Bot 名称 | | apiClient | WechatApiClient | API 客户端实例 | | store | StateStore | 可选，用于持久化 cursor |

消息发送

sendText(toUserId, text, contextToken?)    → Promise<{clientId, response}>
sendImage(toUserId, filePath, ctx?)         → Promise<{clientId, response}>   上传+发送
sendFile(toUserId, filePath, fileName?, ctx?) → Promise<{clientId, response}>
sendVideo(toUserId, filePath, ctx?)         → Promise<{clientId, response}>
sendMessage(toUserId, items, contextToken?) → Promise<{clientId, response}>   自定义 item_list

消息接收

receiveMessages(opts?)  → Promise<Array>       一次性拉取
poll(opts?)             → AsyncGenerator       持续长轮询（服务端建议超时动态调整）

| opts | 类型 | 默认 | 说明 | |------|------|------|------| | timeout | number | 35 | 单次轮询超时 s | | store | StateStore | 构造传入的 | cursor 持久化 |

消息对象格式：

{
  messageId: '...',         // MessageStore 标准字段
  msgType: 'text',           // text|image|voice|file|video
  content: '...',            // 文本内容或 JSON items
  sender: { id: '...', idType: 'wechat_user' },
  mentions: [],
  rootId: null,
  createTime: 1717000000000,
  // 微信扩展字段
  fromUserId: '...',        // 发送者 ID
  toUserId: '...',
  contextToken: '...',      // 回复时所需的上下文 token
  messageType: 1,           // 1=inbound, 2=outbound
  messageState: 2,          // 0=new, 1=generating, 2=finished
  items: [...],             // extractMediaItems 结果
  raw: {...},               // 原始 API 响应
}

消息工具

extractText(msg)          → string                  提取纯文本
extractMediaItems(msg)    → Array<{type, ...}>     提取结构化媒体项
downloadMedia(msgItem, saveDir?) → Promise<{filePath, fileName, size}>  下载+解密媒体

其他

getConfig(ilinkUserId, contextToken?)  → Promise<object>
sendTyping(ilinkUserId, typingTicket, status?) → Promise<object>

WechatRobot

企业微信 Webhook 发送器。

import { WechatRobot } from 'botzone'

const robot = new WechatRobot({ key: 'webhook-key' })

await robot.sendMes('text', '你好')           → Promise<boolean>
await robot.sendMes('markdown', '# 标题')     → Promise<boolean>
await robot.sendCustomMsg({ text, status?, name?, number? }) → Promise<string>

WechatTokenStore

微信 access_token 管理（client_credential 模式）。

import { WechatTokenStore } from 'botzone'

const store = new WechatTokenStore(apiClient)

getToken(appId, appSecret)      → Promise<string>
refreshToken(appId, appSecret)  → Promise<string>
clear(appId) / clearAll()      → void

wechat-media（媒体管线）

AES-128-ECB 加解密 + 微信 CDN 上传/下载管线。

import { encryptAesEcb, decryptAesEcb, uploadMediaToCDN, downloadMediaFromCDN, detectExtension } from 'botzone'

常量

UploadMediaType  = { IMAGE:1, VIDEO:2, FILE:3, VOICE:4 }
MessageItemType  = { TEXT:1, IMAGE:2, VOICE:3, FILE:4, VIDEO:5 }

AES

encryptAesEcb(plaintext, key)   → Buffer        AES-128-ECB 加密
decryptAesEcb(ciphertext, key)  → Buffer        AES-128-ECB 解密
aesEcbPaddedSize(plaintextSize) → number        PKCS7 填充后大小
parseAesKey(aesKeyBase64, label)→ Buffer         解析密钥（兼容 16/32 字节）

CDN

uploadMediaToCDN(api, {filePath, mediaType, toUserId, fileName?}) → Promise<UploadedFileInfo>
downloadMediaFromCDN(encryptQueryParam, aesKeyBase64, label, fullUrl?) → Promise<Buffer>
uploadToCDN(cdnUrl, ciphertext, label) → Promise<string>
buildCdnUploadUrl(uploadParam, filekey) → string
buildCdnDownloadUrl(encryptQueryParam)  → string

文件工具

detectExtension(buf)             → string|null   根据 magic bytes 检测扩展名 (jpg/png/gif/webp/bmp/pdf/mp4)
guessExtension(fallback, buf)    → string         优先 magic bytes → fallback → .bin
inboundMediaDir()                → string         临时媒体目录
cleanupOldFiles(dir, maxAgeMs?)  → void           清理过期文件

WechatMessageBuilder

import { WechatMessageBuilder } from 'botzone'

WechatMessageBuilder.text('hello')  // → { type: 1, text_item: { text: 'hello' } }

配置文件格式

~/.botzone/bots.json：

{
  "bots": {
    "bot名": {
      "app_id": "cli_xxx",
      "app_secret": "xxx",
      "tags": ["prod"],
      "p2p_chats": { "ou_xxx": "oc_xxx" },
      "group_chats": { "群名": "oc_xxx" },
      "users": { "ou_xxx": { "name": null, "unionId": "on_xxx" } }
    }
  }
}

~/.botzone/wechat/account.json（微信登录态）：

{
  "token": "xxx",
  "account_id": "[email protected]",
  "user_id": "[email protected]",
  "base_url": "https://ilinkai.weixin.qq.com",
  "updated_at": "2026-05-31T..."
}

完整导入清单

import {
  // Core
  ApiClient, BotManager, Poller, Logger, StateStore,
  ConfigCache, MessageStore,
  BotZoneError, TokenError, RateLimitError, PermissionError, NetworkError,

  // Feishu
  FeishuApiClient, fromApiResponse, FeishuBot, FeishuTokenStore,
  MessageBuilder, text, atTag, post, card, image, file,
  Uploader, uploadImage, uploadFile,

  // WeChat
  WechatApiClient, SESSION_TIMEOUT_ERRCODE,
  WechatBot, WechatTokenStore, WechatMessageBuilder, WechatRobot,
  UploadMediaType, MessageItemType,
  encryptAesEcb, decryptAesEcb, aesEcbPaddedSize,
  buildCdnUploadUrl, buildCdnDownloadUrl, parseAesKey,
  downloadMediaFromCDN, uploadToCDN, uploadMediaToCDN,
  inboundMediaDir, cleanupOldFiles, detectExtension, guessExtension,
} from 'botzone'