飞书 Skill 开发全指南：从零构建高效机器人服务

3次阅读

共计 2821 个字符，预计需要花费 8 分钟才能阅读完成。

在企业协作场景中，飞书机器人 (Skill) 的开发常遇到以下典型问题：

认证流程复杂：
OAuth2.0 需要处理多步跳转
企业自建应用需配置多重校验
临时授权码有效期仅 10 分钟
事件处理低效：
缺少官方 SDK 的事件去重机制
相同事件可能重复触发业务逻辑
手动实现幂等性增加开发成本
API 稳定性挑战：
消息推送接口 QPS 限制严格
突发流量易触发 429 状态码
历史消息获取存在时间窗口限制

协议开销：
Webhook 只需处理飞书主动推送
轮询需维护定时任务和状态存储
实时性表现：
Webhook 平均延迟 <500ms
轮询间隔最低 1s 仍存在延迟
飞书 Skill 特殊优势：
支持事件批量推送(up to 100 条 / 次)
内置消息加密通道
自动重试失败的消息投递

import hmac
import hashlib
from flask import request, jsonify

# 飞书提供的验证令牌
VERIFICATION_TOKEN = "your_verification_token"

def verify_signature():
    timestamp = request.headers.get('X-Lark-Request-Timestamp')
    nonce = request.headers.get('X-Lark-Request-Nonce')
    signature = request.headers.get('X-Lark-Signature')

    # 拼接验证内容
    verify_content = f"{timestamp}{nonce}{VERIFICATION_TOKEN}".encode('utf-8')

    # 计算 HMAC-SHA256
    computed_signature = hmac.new(VERIFICATION_TOKEN.encode('utf-8'),
        verify_content,
        hashlib.sha256
    ).hexdigest()

    if computed_signature != signature:
        raise PermissionError("Signature verification failed")

@app.route('/webhook', methods=['POST'])
def handle_event():
    try:
        verify_signature()
        event_data = request.json

        # 处理挑战请求
        if 'challenge' in event_data:
            return jsonify({"challenge": event_data['challenge']})

        # 实际业务处理
        process_event(event_data)
        return jsonify({"code": 0})

    except Exception as e:
        logging.error(f"Event processing failed: {str(e)}")
        return jsonify({"code": 1, "msg": str(e)}), 500

布局原则：
主标题不超过 20 个汉字
行动按钮不超过 3 个
卡片宽度固定为 300px

交互元素：

{
  "config": {"wide_screen_mode": true},
  "elements": [{
    "tag": "div",
    "text": {
      "content": "请选择处理方式",
      "tag": "plain_text"
    },
    "extra": {
      "tag": "select",
      "options": [{"value": "1", "text": "通过审批"},
        {"value": "2", "text": "拒绝审批"}
      ],
      "placeholder": "请选择"
    }
  }]
}

基于 Redis 实现去重：

def is_duplicate_event(event_id):
    redis_key = f"lark_event:{event_id}"
    if redis_client.get(redis_key):
        return True
    redis_client.setex(redis_key, 3600, "1")  # 1 小时过期
    return False

消息时序控制：
使用事件中的 event_id + create_time 组合去重
对于审批类事件增加 token 字段校验

# 存储用户对话状态
REDIS_USER_SESSION_PREFIX = "lark:session:"

def update_dialog_context(user_id, context):
    key = f"{REDIS_USER_SESSION_PREFIX}{user_id}"
    redis_client.hmset(key, context)
    redis_client.expire(key, 1800)  # 30 分钟超时

# 获取历史上下文
def get_dialog_context(user_id):
    key = f"{REDIS_USER_SESSION_PREFIX}{user_id}"
    return redis_client.hgetall(key)

指数退避算法实现：

def call_with_retry(api_func, max_retries=3):
    base_delay = 0.5
    for attempt in range(max_retries):
        try:
            return api_func()
        except LarkRateLimitError as e:
            delay = base_delay * (2 ** attempt)
            time.sleep(delay + random.uniform(0, 0.1))
    raise Exception("Max retries exceeded")

重要消息队列化：
使用 RabbitMQ 延迟队列处理通知消息
非实时消息采用批量发送模式

必填材料清单：
企业营业执照扫描件
隐私政策链接
权限使用说明文档
高频驳回原因：
权限范围描述不具体
缺少用户授权示意图
未说明数据存储期限

开发工具配置：

# 启用调试代理
ngrok http 5000 --subdomain=yourdomain

# 查看原始事件
curl -X POST \
  -H "X-Lark-Request-Timestamp: 1629781234" \
  -H "X-Lark-Request-Nonce: abc123" \
  -H "X-Lark-Signature: xxxx" \
  -d @test_event.json \
  http://localhost:5000/webhook