AI伙伴·星瞳计划(一)——鸿蒙初判,基于飞书的对话机器人

那么怎么建立长连接呢？配置方式下面写着“无需注册公网域名或配置加密策略，仅需使用 官方 SDK^[2] 启动长连接飞书客户端，并确保连接成功后，即可开启该模式“看的人一头雾水，我来解释一下吧（举个栗子）：你需要选择一台服务器（按照咱们的偏好，就选择Rocky linux9吧），然后运行一个脚本，让这个脚本去请求跟这个应用建立长连接，这时候，飞书后台（类似我们应用她妈）就会告诉这个应用，外面有人叫你，接下来这个应用就可以选择建立长连接了；

飞书官方给出了详尽且丰富的代码范例：服务端 SDK 概述 - 服务端 API - 开发文档 - 飞书开放平台^[3]

但我们想从头自己一点点搞，就像养育一个孩子，我们想看着她一点点长大！

首先，先实现一些最简单的功能，我们希望先建立连接，然后能在服务器端收到用户发来的信息：

准备工作

准备一台Rocky linux 服务器，调通外网，关闭selinux和防火墙（测试阶段，生产不建议），安装python环境；（你可以参考往期配置zabbix时的文档预备初始环境）

创建文件

mkdir -p /opt/Astra
vim /opt/Astra/lark_bot.py

import lark_oapi as lark              # 导入飞书官方工具箱：这是和飞书沟通的基础
from lark_oapi.ws import Client as WSClient  # 导入“长连接”设备：让机器人像电话一样时刻在线
from lark_oapi.api.im.v1 import * # 导入“消息模块”：专门负责发短信、收短信的功能
import json                           # 导入“翻译官”：把飞书发的复杂数据转换成人类能读懂的文字

# --- 【第一部分：机器人身份证】 ---
# 这两行是机器人在飞书世界的唯一通行证，证明它是“星瞳”本人
APP_ID = "xxxxxxxxxxx"#在你建立的应用界面获取
APP_SECRET = "xxxxxxxxxxxxxxxxxxxxxxxxxx"#在你建立的应用界面获取

# --- 【第二部分：大脑反应区（收到消息怎么办？）】 ---
defdo_p2_im_message_receive_v1(data):
"""
    这是一个“触发器”。只要飞书有人给机器人发消息，这个“大脑”就会自动运转。
    """
# 1. 记下说话人的“身份证号”
# 这个 user_id (OpenID) 是唯一的，以后机器人想单独找这个人，全靠这个号
    user_id = data.event.sender.sender_id.open_id

# 2. 拆解信息
# 飞书发来的是一个大包裹，我们直接从里面拿出“消息(message)”和“内容(content)”
    message = data.event.message
    content_dict = json.loads(message.content) # 把包裹里的信件翻译成中文
    text = content_dict.get("text", "").strip() # 拿到对方说的具体那句话
    chat_id = message.chat_id                   # 记下是在哪个窗口（私聊或群聊）说的

# 在服务器的黑窗口里打印出来，方便运维人员观察
    print(f"📡 [收到消息] 内容: {text} | 发送人: {user_id} | 会话ID: {chat_id}")

# --- 【自动化动作：根据话语做反应】 ---

# 如果对方说了 "ping"（不论大小写）
if text.lower() == "ping":
# 机器人立刻回一句 "pong!"
        send_simple_reply(chat_id, "pong! 星瞳系统连接正常。")
        print("🤖 已发送自动回复: pong")

# 如果对方的话里包含了“巡检”两个字
elif"巡检"in text:
# 机器人先礼貌回应，然后准备去后台干活
        send_simple_reply(chat_id, "收到指令！正在启动 Zabbix 集群自动巡检，请稍候...")
# 这里就是未来存放“自动化巡检脚本”的地方

# --- 【第三部分：嘴巴功能（怎么回话？）】 ---
defsend_simple_reply(chat_id, content_text):
"""
    这个函数专门负责“把话传回去”。
    """
# 填写发信单：告诉飞书要把什么话(content)发到哪个窗口(receive_id)
    req = CreateMessageRequest.builder() \
        .receive_id_type("chat_id") \
        .request_body(CreateMessageRequestBody.builder() \
            .receive_id(chat_id) \
            .msg_type("text") \
            .content(json.dumps({"text": content_text})) \
            .build()) \
        .build()

# 拿起电话拨号发送
    resp = api_client.im.v1.message.create(req)
ifnot resp.success():
        print(f"❌ 发送失败: {resp.msg}")

# --- 【第四部分：总调度室（开机运行）】 ---

# 1. 准备一个“主动发信器”：让机器人具备主动说话的能力
api_client = lark.Client.builder().app_id(APP_ID).app_secret(APP_SECRET).build()

# 2. 挂上“助听器”：告诉系统，有人发消息就去执行上面的“大脑反应区”函数
event_handler = lark.EventDispatcherHandler.builder("", "") \
    .register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \
    .build()

if __name__ == "__main__":
# 3. 正式拨通飞书的“24小时语音热线”
    client = WSClient(
        APP_ID, 
        APP_SECRET, 
        event_handler=event_handler,      # 带着大脑上班
        log_level=lark.LogLevel.INFO      # 只记录重要的事情
    )

    print("🚀 [星瞳服务] 正在建立长连接...")
    client.start()  # 程序开始运行，除非断网或断电，否则不会停止

安装依赖模块：

pip install lark_oapi

在脚本目录下，启用lark_bot.py

python3 lark_bot.py

配置长连接

这时候我们该去创建长连接了，登录到飞书开发者后台的应用界面，在事件与回调的事件配置界面，将订阅方式更改为长连接，这时候你发现可以保存了：

在事件配置下方，添加事件：接收消息v2.0

之后，在权限管理界面，再开通相关的权限（主要是消息的读取和发送）：

好的，完成配置后保存并发布应用。

消息测试

应用发布成功后，我们去后台搜索发布的应用，私聊测试一下，看是否给我们回应：

成功了，”乒乓暗号“对接成功，说明我们的长连接已经成功建立，星瞳可以接收并回复消息了，不过别忙着高兴，此时的她，如同数字植物人，她只能机械的回复这些内容，不能思考；

聪明的你早已经想到了，把星瞳接入大模型，让她具备思考能力，可以和我们流程对话！没错，这就是下期我们要改造的——给她装上大脑！

参考资料

本文使用 Notion Nice 排版一键生成