From feishu-bot-dev
Guides Feishu bot development in zero-code (automation workflows), full-code (open platform APIs), and AI MCP Server modes. Covers SDKs, message card builder, opdev CLI, triggers, and best practices.
npx claudepluginhub yrzhe/claude-skills --plugin feishu-bot-devThis skill is limited to using the following tools:
飞书机器人开发全栈指南,支持零代码、全代码、AI集成三种模式。
Sends messages and interactive cards to Feishu (飞书) and Lark channels via webhooks or Bot API. Use for rich-text announcements, marketing updates, and team notifications.
Manages Feishu (Lark) calendars: lists and searches calendars, checks schedules, syncs events, creates tasks with dates, and sets up shared calendars via Node.js scripts.
Operates WeCom (Enterprise WeChat) via CLI for contacts, todos, meetings, messages, schedules, documents, and smart sheets. Provides 12 AI agent skills for terminal or agent integration.
Share bugs, ideas, or general feedback.
飞书机器人开发全栈指南,支持零代码、全代码、AI集成三种模式。
| 模式 | 适用场景 | 技术要求 | 入口 |
|---|---|---|---|
| 飞书机器人助手 | 定时提醒、消息推送、自动化流程 | 无需编程 | botbuilder.feishu.cn |
| 飞书开放平台 API | 复杂业务逻辑、系统深度集成 | 需要编程 | open.feishu.cn |
| MCP Server 集成 | AI Agent 协作、智能助手 | 需要编程 | MCP 文档 |
| 工具 | 用途 | 链接 |
|---|---|---|
| 消息卡片搭建工具 | 可视化设计卡片消息,告别手写JSON | Card Builder |
| API Explorer | 在线调试API,一键生成代码 | API Explorer |
| AI Playground | 自然语言生成代码 | AI Playground |
| opdev CLI | 命令行开发工具 | npm i -g @anthropic/opdev |
系统提供多种预设模板:
| 模板类型 | 典型场景 |
|---|---|
| 消息助手 | 行政通知、财务通知 |
| 群智能助手 | 入群欢迎、自动回复 |
| 新人入职 | 入职引导、信息收集 |
| 定时提醒 | 周报填写、任务催办 |
| 值班通知 | 排班提醒、签到管理 |
| 数据推送 | 系统告警、数据同步 |
触发器:流程的起点,定义何时启动
操作节点:触发后执行的动作
| 触发器 | 说明 | 示例场景 |
|---|---|---|
| 定时任务 | 按时间和频率触发 | 每天9点发送周报提醒 |
| 工作表变化 | 新增/修改/删除记录 | 新员工入职自动欢迎 |
| 机器人会话首次创建 | 用户首次打开机器人 | 发送使用指南 |
| 单聊机器人消息 | 用户私聊关键词触发 | 关键词自助查询 |
| @机器人群聊消息 | 群内@机器人触发 | 问题反馈汇总 |
| 触发器 | 说明 |
|---|---|
| 新员工入职 | 账号激活或入职特定部门 |
| 多维表格内容变更 | 满足条件的记录变更 |
| 用户进群 | 用户加入指定群组 |
| 收到新审批 | 审批状态变更 |
| 触发器 | 说明 |
|---|---|
| Webhook 触发 | 接收外部 HTTP 请求 |
| GitLab 事件 | commit/MR/Issue 变动 |
| Sentry 事件 | 新错误产生 |
| Grafana 告警 | 监控报警触发 |
| Gmail 新邮件 | 收到新邮件时触发 |
## 消息内容支持的格式
| 样式 | 语法 | 效果 |
|------|------|------|
| 加粗 | `**文字**` | **粗体** |
| 斜体 | `*文字*` | *斜体* |
| 删除线 | `~~文字~~` | ~~删除线~~ |
| 颜色 | `<font color='green'>文字</font>` | 绿色文字 |
| @指定人 | `<at id=open_id></at>` | @用户名 |
| @所有人 | `<at user_id="all"></at>` | @所有人 |
| 超链接 | `[文字](https://url)` | 可点击链接 |
注意事项:
# 测试 Webhook (macOS)
curl -X POST -H "Content-Type: application/json" \
-d '{"msg_type":"text","content":{"id":"1","name":"Tom"}}' \
[Webhook地址]
# 测试 Webhook (Windows)
curl -X POST -H "Content-Type: application/json" \
-d "{\"msg_type\":\"text\",\"content\":{\"text\":\"test\"}}" \
[Webhook地址]
| 问题 | 解决方案 |
|---|---|
| 无法接收消息 | 检查是否在应用可用范围内 |
| 流程运行失败 | 查看运行日志排查原因 |
| 群组发送失败 | 确保群设置允许所有成员发言 |
飞书 API 支持两种访问凭证:
| 凭证类型 | 适用场景 | 获取方式 |
|---|---|---|
tenant_access_token | 应用身份调用 API | App ID + App Secret |
user_access_token | 用户身份调用 API | OAuth 2.0 授权 |
import requests
def get_tenant_access_token(app_id: str, app_secret: str) -> str:
"""获取 tenant_access_token"""
url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
payload = {
"app_id": app_id,
"app_secret": app_secret
}
response = requests.post(url, json=payload)
data = response.json()
if data.get("code") == 0:
return data["tenant_access_token"]
else:
raise Exception(f"获取 token 失败: {data}")
# 使用示例
token = get_tenant_access_token("cli_xxx", "xxx")
// Node.js 版本
const axios = require('axios');
async function getTenantAccessToken(appId, appSecret) {
const response = await axios.post(
'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal',
{ app_id: appId, app_secret: appSecret }
);
if (response.data.code === 0) {
return response.data.tenant_access_token;
}
throw new Error(`获取 token 失败: ${JSON.stringify(response.data)}`);
}
飞书提供多语言官方 SDK(截至 2026 年最新版本):
# Python
pip install lark-oapi
# Node.js (v1.56.1)
npm install @larksuiteoapi/node-sdk
# Go (v3)
go get github.com/larksuite/oapi-sdk-go/v3
# Java
# Maven: com.larksuite.oapi:oapi-sdk
飞书官方 MCP Server 支持 AI Agent 与飞书深度协作:
# 安装 MCP Server
npm install @larksuiteoapi/lark-mcp
支持的能力:
配置示例(Claude Desktop / Cursor):
{
"mcpServers": {
"feishu": {
"command": "npx",
"args": ["@larksuiteoapi/lark-mcp"],
"env": {
"FEISHU_APP_ID": "your_app_id",
"FEISHU_APP_SECRET": "your_app_secret"
}
}
}
}
常用 MCP 工具:
| 工具名 | 功能 |
|---|---|
send_text_message | 发送文本消息 |
send_card_message | 发送卡片消息 |
create_document | 创建文档 |
read_document | 读取文档内容 |
create_calendar_event | 创建日程 |
search_bitable_records | 查询多维表格 |
强烈推荐:使用 消息卡片搭建工具 可视化设计卡片!
核心优势:
使用流程:
template_id# 使用模板发送卡片(推荐)
def send_card_by_template(receive_id: str, template_id: str, template_variable: dict):
"""使用卡片模板发送消息"""
content = {
"type": "template",
"data": {
"template_id": template_id,
"template_variable": template_variable
}
}
return send_message(receive_id, "interactive", content)
# 使用示例
send_card_by_template(
receive_id="ou_xxx",
template_id="AAqkxxxxxxxx",
template_variable={
"title": "周报提醒",
"content": "请在今天下班前提交周报"
}
)
API 端点:POST /open-apis/im/v1/messages
import lark_oapi as lark
from lark_oapi.api.im.v1 import *
# 创建客户端
client = lark.Client.builder() \
.app_id("cli_xxx") \
.app_secret("xxx") \
.build()
# 发送文本消息
def send_text_message(receive_id: str, text: str, receive_id_type: str = "open_id"):
"""
发送文本消息
Args:
receive_id: 接收者ID (open_id/user_id/union_id/email/chat_id)
text: 消息内容
receive_id_type: ID类型
"""
request = CreateMessageRequest.builder() \
.receive_id_type(receive_id_type) \
.request_body(CreateMessageRequestBody.builder()
.receive_id(receive_id)
.msg_type("text")
.content('{"text": "' + text + '"}')
.build()) \
.build()
response = client.im.v1.message.create(request)
if not response.success():
raise Exception(f"发送失败: {response.msg}")
return response.data.message_id
# 发送富文本消息
def send_rich_text_message(receive_id: str, title: str, content_list: list):
"""
发送富文本消息
Args:
receive_id: 接收者ID
title: 消息标题
content_list: 富文本内容列表
"""
import json
content = {
"zh_cn": {
"title": title,
"content": content_list
}
}
request = CreateMessageRequest.builder() \
.receive_id_type("open_id") \
.request_body(CreateMessageRequestBody.builder()
.receive_id(receive_id)
.msg_type("post")
.content(json.dumps(content))
.build()) \
.build()
response = client.im.v1.message.create(request)
return response.data.message_id
# 使用示例
send_text_message("ou_xxx", "Hello, 飞书!")
# 富文本示例
content = [
[
{"tag": "text", "text": "项目进度更新:"},
{"tag": "a", "text": "查看详情", "href": "https://example.com"}
],
[
{"tag": "at", "user_id": "ou_xxx"},
{"tag": "text", "text": " 请确认"}
]
]
send_rich_text_message("ou_xxx", "项目周报", content)
const lark = require('@larksuiteoapi/node-sdk');
// 创建客户端
const client = new lark.Client({
appId: 'cli_xxx',
appSecret: 'xxx',
});
// 发送文本消息
async function sendTextMessage(receiveId, text, receiveIdType = 'open_id') {
const response = await client.im.message.create({
params: { receive_id_type: receiveIdType },
data: {
receive_id: receiveId,
msg_type: 'text',
content: JSON.stringify({ text }),
},
});
if (response.code !== 0) {
throw new Error(`发送失败: ${response.msg}`);
}
return response.data.message_id;
}
// 发送卡片消息
async function sendCardMessage(receiveId, card) {
const response = await client.im.message.create({
params: { receive_id_type: 'open_id' },
data: {
receive_id: receiveId,
msg_type: 'interactive',
content: JSON.stringify(card),
},
});
return response.data.message_id;
}
// 卡片消息示例
const card = {
"config": { "wide_screen_mode": true },
"header": {
"title": { "tag": "plain_text", "content": "项目通知" },
"template": "blue"
},
"elements": [
{
"tag": "div",
"text": { "tag": "lark_md", "content": "**任务完成**\n项目A已完成部署" }
},
{
"tag": "action",
"actions": [
{
"tag": "button",
"text": { "tag": "plain_text", "content": "查看详情" },
"type": "primary",
"url": "https://example.com"
}
]
}
]
};
sendCardMessage('ou_xxx', card);
from lark_oapi.api.im.v1 import *
def create_chat(name: str, description: str, owner_id: str, user_id_list: list):
"""创建群组"""
request = CreateChatRequest.builder() \
.user_id_type("open_id") \
.request_body(CreateChatRequestBody.builder()
.name(name)
.description(description)
.owner_id(owner_id)
.user_id_list(user_id_list)
.chat_mode("group")
.chat_type("private")
.build()) \
.build()
response = client.im.v1.chat.create(request)
return response.data.chat_id
# 使用示例
chat_id = create_chat(
name="项目讨论群",
description="项目A进度讨论",
owner_id="ou_xxx",
user_id_list=["ou_aaa", "ou_bbb", "ou_ccc"]
)
def add_chat_members(chat_id: str, member_ids: list):
"""添加群成员"""
request = CreateChatMembersRequest.builder() \
.chat_id(chat_id) \
.member_id_type("open_id") \
.request_body(CreateChatMembersRequestBody.builder()
.id_list(member_ids)
.build()) \
.build()
response = client.im.v1.chat_members.create(request)
return response.success()
from lark_oapi.api.docx.v1 import *
def create_document(title: str, folder_token: str = None):
"""创建飞书文档"""
request = CreateDocumentRequest.builder() \
.request_body(CreateDocumentRequestBody.builder()
.title(title)
.folder_token(folder_token)
.build()) \
.build()
response = client.docx.v1.document.create(request)
return {
"document_id": response.data.document.document_id,
"title": response.data.document.title
}
# 使用示例
doc = create_document("周会纪要 2024-01-15")
print(f"文档ID: {doc['document_id']}")
def create_document_block(document_id: str, block_type: str, content: dict):
"""向文档添加内容块"""
request = CreateDocumentBlockRequest.builder() \
.document_id(document_id) \
.request_body(CreateDocumentBlockRequestBody.builder()
.block_type(block_type)
.text_run(TextRun.builder()
.content(content.get("text", ""))
.build())
.build()) \
.build()
response = client.docx.v1.document_block.create(request)
return response.data.block
from lark_oapi.api.bitable.v1 import *
def list_bitable_records(app_token: str, table_id: str, page_size: int = 20):
"""获取多维表格记录"""
request = ListAppTableRecordRequest.builder() \
.app_token(app_token) \
.table_id(table_id) \
.page_size(page_size) \
.build()
response = client.bitable.v1.app_table_record.list(request)
return response.data.items
def create_bitable_record(app_token: str, table_id: str, fields: dict):
"""新增多维表格记录"""
request = CreateAppTableRecordRequest.builder() \
.app_token(app_token) \
.table_id(table_id) \
.request_body(AppTableRecord.builder()
.fields(fields)
.build()) \
.build()
response = client.bitable.v1.app_table_record.create(request)
return response.data.record
# 使用示例
record = create_bitable_record(
app_token="bascnxxx",
table_id="tblxxx",
fields={
"姓名": "张三",
"部门": "技术部",
"入职日期": "2024-01-15"
}
)
from lark_oapi.api.calendar.v4 import *
def create_calendar_event(
summary: str,
start_time: int,
end_time: int,
attendee_ids: list = None
):
"""创建日程"""
attendees = None
if attendee_ids:
attendees = [
CalendarEventAttendee.builder()
.type("user")
.user_id(uid)
.build()
for uid in attendee_ids
]
request = CreateCalendarEventRequest.builder() \
.calendar_id("primary") \
.request_body(CalendarEvent.builder()
.summary(summary)
.start_time(TimeInfo.builder()
.timestamp(str(start_time))
.build())
.end_time(TimeInfo.builder()
.timestamp(str(end_time))
.build())
.attendees(attendees)
.build()) \
.build()
response = client.calendar.v4.calendar_event.create(request)
return response.data.event
# 使用示例
import time
start = int(time.time()) + 3600 # 1小时后
end = start + 3600 # 持续1小时
event = create_calendar_event(
summary="项目评审会议",
start_time=start,
end_time=end,
attendee_ids=["ou_xxx", "ou_yyy"]
)
from flask import Flask, request
import lark_oapi as lark
from lark_oapi.adapter.flask import *
app = Flask(__name__)
# 创建事件处理器
def handle_message_receive(data):
"""处理接收消息事件"""
message = data.event.message
content = message.content
sender_id = message.sender.sender_id.open_id
print(f"收到消息: {content}, 发送者: {sender_id}")
# 自动回复
# send_text_message(sender_id, "收到您的消息")
handler = lark.EventDispatcherHandler.builder("", "") \
.register_p2_im_message_receive_v1(handle_message_receive) \
.build()
@app.route("/webhook/event", methods=["POST"])
def event_handler():
return handler.do(parse_req())
if __name__ == "__main__":
app.run(port=8080)
| 事件 | 说明 |
|---|---|
im.message.receive_v1 | 接收消息 |
im.chat.member.user.added_v1 | 用户入群 |
im.chat.member.user.deleted_v1 | 用户退群 |
contact.user.created_v3 | 新员工入职 |
approval.approval.updated_v4 | 审批状态变更 |
| 权限 | 说明 |
|---|---|
im:message | 获取与发送单聊、群聊消息 |
im:chat | 获取与更新群组信息 |
contact:user.base:readonly | 获取用户基本信息 |
docs:doc | 读写云文档 |
bitable:app | 读写多维表格 |
calendar:calendar | 读写日历 |
def safe_api_call(func, *args, **kwargs):
"""安全的 API 调用封装"""
try:
response = func(*args, **kwargs)
if hasattr(response, 'success') and not response.success():
print(f"API 错误: code={response.code}, msg={response.msg}")
return None
return response.data
except Exception as e:
print(f"调用异常: {e}")
return None
import time
class TokenManager:
"""Token 管理器,支持自动刷新"""
def __init__(self, app_id, app_secret):
self.app_id = app_id
self.app_secret = app_secret
self._token = None
self._expire_time = 0
def get_token(self):
# 提前5分钟刷新
if time.time() > self._expire_time - 300:
self._refresh_token()
return self._token
def _refresh_token(self):
token = get_tenant_access_token(self.app_id, self.app_secret)
self._token = token
self._expire_time = time.time() + 7200 # 2小时有效期
import time
from functools import wraps
def rate_limit(max_calls, period):
"""限流装饰器"""
calls = []
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
calls[:] = [c for c in calls if c > now - period]
if len(calls) >= max_calls:
sleep_time = period - (now - calls[0])
time.sleep(sleep_time)
calls.append(time.time())
return func(*args, **kwargs)
return wrapper
return decorator
# 使用示例:每秒最多5次调用
@rate_limit(max_calls=5, period=1)
def send_message(receive_id, text):
# ...
pass
| 错误码 | 说明 | 解决方案 |
|---|---|---|
| 99991663 | tenant token invalid | 检查 App ID/Secret,重新获取 token |
| 99991668 | user token invalid | 用户授权已失效,需重新授权 |
| 99991400 | permission denied | 检查是否申请了所需权限 |
| 99991401 | app not activated | 应用未发布或未被管理员审核通过 |
| 230001 | message send failed | 检查 receive_id 是否正确 |
飞书提供在线调试工具:API Explorer
功能:
飞书 AI 助手支持自然语言生成代码:AI Playground
支持语言:
加入官方开发者社区获取技术支持。