KOOK Bot 开发实战：用 Python 搭建一个服务器监控告警机器人

原创

PC电脑医生

发布于 2026-07-10 10:23:24

130

摘要：KOOK（原开黑啦）开放了机器人 API，开发者可以用 Python 快速构建自定义 Bot。本文从零讲起，涵盖 KOOK 开发者应用注册、Python SDK 接入、消息收发基础，最终实现一个能读取服务器 CPU 和内存使用率、在负载异常时自动向指定频道推送告警的实用 Bot。全程提供可运行代码，适合有 Python 基础的开发者。

引言

团队协作中，服务器状态监控的消息推送方案有很多选择：企业微信机器人、钉钉 Webhook、Telegram Bot、邮件告警等。但如果你和团队日常使用 KOOK 进行语音和文字沟通，把监控告警直接推送到 KOOK 频道里，能让所有人第一时间在同一平台上看到，不需要额外安装或切换工具。

KOOK 提供了一套完整的机器人 API，支持收发消息、管理频道、处理事件等。相比于企业微信和钉钉的群机器人，KOOK Bot 的事件回调机制更灵活，可以实现双向交互而不只是单向推送。

本文分三个阶段递进：先注册 Bot 跑通最基础的收发消息，再接入 psutil 实现服务器资源采集，最后部署到服务器上常驻运行。演示环境为 Ubuntu 22.04 + Python 3.10。

KOOK 机器人机制简述

在动手写代码之前，先理清 KOOK Bot 的工作方式：

KOOK 的机器人通过 WebSocket 与 KOOK 服务器保持长连接。
当 Bot 所在频道发生事件（收到消息、用户加入等），KOOK 服务器会向 Bot 推送事件数据。
Bot 也可以主动调用 HTTP API 发送消息、修改频道等。
整个通信基于 KOOK 的开放协议，官方提供了 Python SDK 简化开发。

本文使用的 khl.py 是目前社区维护最活跃的 Python SDK，封装了 WebSocket 连接和 API 调用，开发者只需关注业务逻辑。

第一步：注册 KOOK 开发者应用并获取 Token

操作步骤：

打开 KOOK 开发者中心并登录。
点击「新建应用」，填写应用名称（例如 server-monitor-bot）。
创建成功后，在应用详情页的「凭证」区域找到 Bot Token，格式为一串以数字开头的字符串，点击复制。
切换到「邀请」标签页，生成邀请链接，将 Bot 邀请到你自己的测试服务器（需有管理员权限）。

关键提醒：

Bot Token 是 Bot 的唯一身份凭证，等同于密码，绝对不要提交到公开仓库。建议存储在环境变量或 .env 文件中。
在邀请 Bot 时，权限按需勾选。本文的监控 Bot 仅需要「发送消息」和「读取消息」权限，不要勾选多余权限。

第二步：搭建 Python 开发环境

以下操作在服务器或本地开发机上执行：

# 创建虚拟环境（推荐）
python3 -m venv kook-bot-env
source kook-bot-env/bin/activate

# 安装依赖
pip install khl.py psutil python-dotenv

依赖说明：

khl.py：KOOK Python SDK，处理 WebSocket 连接和 API 调用。
psutil：跨平台的系统资源监控库，用于获取 CPU、内存、磁盘等指标。
python-dotenv：从 .env 文件加载环境变量，避免 Token 硬编码在脚本中。

在项目根目录创建 .env 文件：

BOT_TOKEN=你的Bot_Token
TARGET_CHANNEL_ID=目标频道ID
ALERT_CPU_THRESHOLD=80
ALERT_MEM_THRESHOLD=80
CHECK_INTERVAL=300

获取频道 ID 的方法：在 KOOK 客户端中，右键点击目标频道 →「复制频道 ID」，或在开发者模式中查看。

第三步：编写基础 Bot —— 跑通消息收发

先从一个最简版本开始，确保 Bot 能正确连接并响应指令。

创建 bot.py：

import os
from dotenv import load_dotenv
from khl import Bot, Message

load_dotenv()

bot = Bot(token=os.getenv("BOT_TOKEN"))

@bot.command(name="ping")
async def ping(msg: Message):
    """健康检查指令"""
    await msg.reply("pong! Bot is running.")

@bot.command(name="status")
async def quick_status(msg: Message):
    """快速查看当前服务器资源"""
    import psutil
    cpu = psutil.cpu_percent(interval=1)
    mem = psutil.virtual_memory().percent
    disk = psutil.disk_usage("/").percent
    await msg.reply(
        f"CPU: {cpu}%  |  内存: {mem}%  |  磁盘: {disk}%"
    )

if __name__ == "__main__":
    bot.run()

代码解析：

@bot.command(name="ping") 声明一个指令。当用户在频道中发送 /ping，Bot 会自动回复 pong!。
@bot.command(name="status") 是另一个指令，调用 psutil 实时采集 CPU、内存、磁盘使用率并返回。
bot.run() 启动 WebSocket 连接，Bot 开始监听事件。

运行测试：

python bot.py

在 KOOK 频道中输入 /ping 和 /status，确认 Bot 能正常回复，基础框架就跑通了。

第四步：进阶 —— 定时监控与自动告警推送

基础 Bot 只能被动响应指令，实际场景中需要 Bot 主动在资源异常时向频道推送告警。这里引入异步定时任务来实现。

更新 bot.py，在原有代码基础上增加定时检测逻辑：

import os
import asyncio
import psutil
from dotenv import load_dotenv
from khl import Bot, Message, PublicChannel

load_dotenv()

bot = Bot(token=os.getenv("BOT_TOKEN"))

CPU_THRESHOLD = int(os.getenv("ALERT_CPU_THRESHOLD", 80))
MEM_THRESHOLD = int(os.getenv("ALERT_MEM_THRESHOLD", 80))
CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", 300))
TARGET_CHANNEL_ID = os.getenv("TARGET_CHANNEL_ID")

async def monitor_loop():
    """定时采集资源指标，超阈值时推送告警到指定频道"""
    await bot.client.wait_for_startup()
    channel = await bot.client.fetch_public_channel(TARGET_CHANNEL_ID)

    alert_sent_cpu = False
    alert_sent_mem = False

    while True:
        cpu = psutil.cpu_percent(interval=1)
        mem = psutil.virtual_memory().percent

        if cpu > CPU_THRESHOLD and not alert_sent_cpu:
            await channel.send(
                f"⚠️ CPU 告警：当前使用率 {cpu}%，已超过阈值 {CPU_THRESHOLD}%"
            )
            alert_sent_cpu = True
        elif cpu <= CPU_THRESHOLD and alert_sent_cpu:
            await channel.send(f"✅ CPU 已恢复正常：{cpu}%")
            alert_sent_cpu = False

        if mem > MEM_THRESHOLD and not alert_sent_mem:
            await channel.send(
                f"⚠️ 内存告警：当前使用率 {mem}%，已超过阈值 {MEM_THRESHOLD}%"
            )
            alert_sent_mem = True
        elif mem <= MEM_THRESHOLD and alert_sent_mem:
            await channel.send(f"✅ 内存已恢复正常：{mem}%")
            alert_sent_mem = False

        await asyncio.sleep(CHECK_INTERVAL)


@bot.command(name="ping")
async def ping(msg: Message):
    await msg.reply("pong! Bot is running.")

@bot.command(name="status")
async def quick_status(msg: Message):
    cpu = psutil.cpu_percent(interval=1)
    mem = psutil.virtual_memory().percent
    disk = psutil.disk_usage("/").percent
    await msg.reply(
        f"CPU: {cpu}%  |  内存: {mem}%  |  磁盘: {disk}%"
    )

if __name__ == "__main__":
    asyncio.ensure_future(monitor_loop())
    bot.run()

设计要点：

alert_sent_cpu 和 alert_sent_mem 两个状态标志，防止告警风暴——只在指标首次超过阈值时发一条告警，恢复后发一条解除通知，不会每分钟刷屏。
CHECK_INTERVAL 控制检测频率（默认 300 秒 = 5 分钟），避免对 KOOK API 造成过大压力。
使用 fetch_public_channel 获取频道对象后直接 send，不需要等待用户指令触发。

第五步：部署到服务器并持久化运行

Bot 脚本跑通后，需要部署到服务器上常驻运行。推荐使用 systemd 管理 Bot 进程。

创建 systemd 服务文件 /etc/systemd/system/kook-monitor-bot.service：

[Unit]
Description=KOOK Server Monitor Bot
After=network.target

[Service]
Type=simple
User=your_username
WorkingDirectory=/opt/kook-monitor-bot
Environment="PATH=/opt/kook-monitor-bot/kook-bot-env/bin"
ExecStart=/opt/kook-monitor-bot/kook-bot-env/bin/python /opt/kook-monitor-bot/bot.py
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target

启用并启动服务：

sudo systemctl daemon-reload
sudo systemctl enable kook-monitor-bot
sudo systemctl start kook-monitor-bot

验证运行状态：

sudo systemctl status kook-monitor-bot

确认状态为 active (running) 后，可以在 KOOK 频道发送 /ping 验证 Bot 是否在线。

踩坑总结与扩展方向

在实际开发和部署过程中，有几点值得留意：

Token 安全：绝对不要把 BOT_TOKEN 写死在代码里。用 .env 文件或服务器环境变量管理，并在 .gitignore 中排除 .env。
频道 ID 的获取：KOOK 客户端默认不显示频道 ID，需要在「设置 → 高级设置」中开启开发者模式，之后右键频道才能看到复制 ID 的选项。
异步编程注意：khl.py 基于 asyncio，所有网络操作都是异步的。如果在 Bot 命令中调用了同步阻塞函数（如 time.sleep），会导致整个 Bot 卡住。解决方案是使用 asyncio.to_thread 将阻塞操作放到线程池中执行，或者直接使用异步版本的库。
速率限制：KOOK API 对消息发送有频率限制，过于频繁的请求（如每秒发送多条消息）可能被暂时拒绝。本文通过 CHECK_INTERVAL 和告警状态标志来控制发送频率，正常情况下不会触发限制。