【分享】使用飞书长连接模式订阅事件回调（免搭建服务器）

经验分享

出

出道即颠峰

2024-02-02 12:36·浏览量：10882

出

出道即颠峰

影刀高级开发者

发布于 2024-02-02 12:34更新于 2024-02-02 12:3610882浏览

1. 功能介绍

开发者通过集成飞书 SDK 与开放平台建立一条 WebSocket 全双工通道，当有事件回调发生时，开放平台会通过该通道向开发者发送消息。

与传统的 Webhook 模式相比，长连接模式大大降低了接入成本，将原先 1 周左右的开发周期降低到 5 分钟。具体优势如下：

测试阶段无需使用内网穿透工具，通过长连接模式在本地开发环境中即可接收事件回调；
只在建连时进行鉴权，后续事件推送均为明文数据，无需开发者再处理解密和验签逻辑；
只需保证运行环境具备访问公网能力即可，无需提供公网 IP 或域名；
无需部署防火墙和配置白名单。

2. 接入方式

接入限制：

目前仅支持企业自建应用
每个应用最多建立 50 个连接（每初始化一个 client 就是一个连接）

2.1. 步骤一：集成 SDK

如果你已经使用飞书 SDK 通过 Webhook 模式订阅事件，那几乎是零成本切换，因为所有的业务逻辑都可以复用，只需修改启动方式即可。

Python

安装 SDK

pip install lark-oapi==1.1.0b1

示例代码

import requests
import json,time
from lark_oapi import EventDispatcherHandler, ws, JSON, im, LogLevel

class FeishuConfig:
    '''飞书API的配置信息'''
    APP_ID = 'cli_xxxxxxxxxxxxxxx'
    APP_SECRET = '7Tjxxxxxxxxxxxxxxxxxxxxxxxxx'

class FeishuApi:
    '''FeishuApi类用于处理与飞书API的交互'''
    TOKEN_URL = 'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal'
    REPLY_MESSAGE_URL_TEMPLATE = 'https://open.feishu.cn/open-apis/im/v1/messages/{message_id}/reply'
    HEADERS = {'Content-Type': 'application/json; charset=utf-8'}

    def __init__(self):
        self.session = requests.Session()
        self.token = self.get_token()

    def get_token(self):
        '''获取飞书API的访问令牌'''
        data = {'app_id': FeishuConfig.APP_ID, 'app_secret': FeishuConfig.APP_SECRET}
        response = self.session.post(self.TOKEN_URL, headers=self.HEADERS, json=data)
        response.raise_for_status()
        return response.json().get('tenant_access_token')

    def reply_message(self, message_id, user_id, message):
        '''回复飞书群聊消息'''
        url = self.REPLY_MESSAGE_URL_TEMPLATE.format(message_id=message_id)
        content = f'{{"text":"<at user_id=\\"{user_id}\\"></at> {message}"}}'
        data = {"content": content, "msg_type": "text"}
        headers = {'Authorization': 'Bearer ' + self.token, **self.HEADERS}
        response = self.session.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()

def handle_p2_im_message(data: im.v1.P2ImMessageReceiveV1):
    '''处理接收到的个人或群聊消息'''
    data_dict = json.loads(JSON.marshal(data)) 
    message_id = data_dict["event"]["message"]["message_id"]
    content = eval(data.event.message.content).get("text")
    content = content.split(" ")[-1].replace('"}', '').strip()
    user_id = data_dict["event"]["sender"]["sender_id"]["user_id"]
    # time.sleep(5)
    feishu = FeishuApi()
    feishu.reply_message(message_id=message_id, user_id=user_id, message=f'已经收到您的消息啦：{content}')

def main():
    '''启动飞书长连接 WebSocket客户端'''
    event_handler = EventDispatcherHandler.builder("", "") \
        .register_p2_im_message_receive_v1(handle_p2_im_message) \
        .build()

    cli = ws.Client(FeishuConfig.APP_ID, FeishuConfig.APP_SECRET, event_handler=event_handler, log_level=LogLevel.DEBUG)
    cli.start()

if __name__ == "__main__":
    main()

步骤说明：

通过 lark.EventDispatcherHandler.builder() 初始化事件处理器（event_handler），两个参数建议填空字符串
通过 event_handler 的 register_xxxx() 方法监听不同的事件类型，上述示例中分别监听了「接收消息 v1.0」和「接收消息 v2.0」两个事件，更多事件可参考处理消息事件回调 v1.0没写
通过 lark.ws.Client() 初始化长连接客户端，必填参数为应用的 APP_ID 和 APP_SECRET，可在开发者后台获取
可选参数传入 event_handler，同时可设置日志级别
通过 cli.start() 启动客户端，如连接成功，控制台会打印 "connected to wss://xxxxx"，主线程将阻塞，直到进程结束