概览

ws://<当前域名>/onebot/{实例ID}?token={实例 OneBot Token}

{ "type": "event", "instance_id": "ins_xxxxxx", "data": { "post_type": "message", "message_type": "group", "group_id": 123456, "user_id": 10001, "raw_message": "你好", "sender": { "nickname": "小明" } } }

// 浏览器 / Node 侧监听平台事件流（token 走查询参数）
const token = '<前台登录令牌 napcat_token>';
const ws = new WebSocket('ws://127.0.0.1:8080/api/ws/events?instance=ins_xxxxxx&token=' + token);

ws.onmessage = (e) => {
  const packet = JSON.parse(e.data);
  if (packet.type !== 'event') return;          // hello / ping / status 忽略
  const evt = packet.data;
  if (evt.post_type === 'message' && evt.message_type === 'group') {
    console.log('群消息', evt.group_id, evt.raw_message);
  }
};

import asyncio, json
import websockets

URL = "ws://127.0.0.1:8080/onebot/ins_xxxxxx?token=<实例 OneBot Token>"

async def main():
    async with websockets.connect(URL) as ws:
        # 主动调用 OneBot 接口
        await ws.send(json.dumps({"action": "get_login_info", "params": {}, "echo": 1}))
        while True:
            msg = json.loads(await ws.recv())
            if "post_type" in msg:
                print("事件:", msg["post_type"], msg.get("raw_message", ""))

asyncio.run(main())