本文约2200字，预计需要8分钟
知更Ai（zhigengai.com）微信websocket实时接收消息接口，适合在自己的脚本、桌面工具或服务端订阅机器人收到的微信消息。接口地址为本机地址：
ws://127.0.0.1:5010/api/ws/relay
服务默认只监听 127.0.0.1，不暴露到局域网；调用方通常需要和知更Ai运行在同一台电脑上，并通过第一帧鉴权后才能开始接收推送。
接口基础信息
按知更Ai在线文档，websocket实时接收消息接口的基础信息如下：
| 项 | 值 |
|---|---|
| 监听地址 | 127.0.0.1 |
| 监听端口 | 5010 |
| WebSocket 路径 | /api/ws/relay |
| 完整地址 | ws://127.0.0.1:5010/api/ws/relay |
| 消息方向 | 服务端 → 客户端单向推送 |
| 鉴权方式 | 第一帧发送 {"type":"auth","key":"<token>"} |
| 鉴权 Token | 系统设置里查看或刷新 |
| 可用前提 | 具备对应 VIP 功能权限 |
| 数据格式 | 文本帧（UTF-8 JSON） |
这个接口属于“旁路推送”，不会影响知更Ai现有的自动回复、AI 智脑和消息路由。订阅方只是额外复制一份消息用于自己的业务，原始逻辑仍按软件默认设置运行。
调用前的准备工作
正式调用前，建议按顺序确认以下条件：
- 知更Ai客户端已经启动并登录账号。
- 微信机器人已添加到机器人管理并处于在线状态。
- 当前账号具备对应的 WebSocket 推送功能权限。
- 已在系统设置页面查看并复制
ws_token。 - 调用程序能够访问本机的
5010端口。 - 已准备好用于解析 JSON 的依赖，例如 Python 中的
websockets或 Node.js 中的ws。 - 准备好目标平台的处理逻辑，例如写入数据库、调用 AI 或对接业务系统。
如果调用方在另一台电脑上，应先确认这一点是否被允许：服务监听 127.0.0.1，因此跨机器调用需要先解决网络可达性，否则不能直接连。
鉴权握手
必须发送的第一帧
客户端建立连接后，第一帧必须是一段 UTF-8 JSON 文本帧，内容如下：
{"type": "auth",
"key": "wk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
}
其中：
type必须是"auth"，否则视为握手失败；key填写系统设置里的ws_token；- JSON 必须以文本帧发送，不能使用二进制帧。
鉴权结果
握手结果按以下规则判断：
- 成功时，服务端不会返回额外确认帧，客户端从下一条消息开始即可接收推送。
- 失败时，服务端会发送类似下面的内容后立即关闭连接：
"type": "auth_failed",
"reason": "invalid token"
}
常见的 reason 包括：
invalid token：token 不正确或已刷新；invalid JSON：鉴权帧不是合法 JSON；client closed before auth：未发送鉴权帧就关闭了连接；first frame type must be 'auth', got 'xxx'：第一帧的type字段不是auth。
Python 鉴权示例
import asyncio
import json
import websockets
WS_URL = "ws://127.0.0.1:5010/api/ws/relay"
WS_TOKEN = "wk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
async def main():
async with websockets.connect(WS_URL) as ws:
await ws.send(json.dumps({
"type": "auth",
"key": WS_TOKEN
}))
async for raw in ws:
msg = json.loads(raw)
if msg.get("type") == "auth_failed":
print("鉴权失败：", msg.get("reason"))
break
asyncio.run(main())
JavaScript 浏览器示例
const WS_URL = "ws://127.0.0.1:5010/api/ws/relay";
const WS_TOKEN = "wk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx";
const ws = new WebSocket(WS_URL);
ws.onopen = () => {
ws.send(JSON.stringify({
type: "auth",
key: WS_TOKEN
}));
};
ws.onmessage = (event) => {
const msg = JSON.parse(event.data);
if (msg.type === "auth_failed") {
console.error("鉴权失败：", msg.reason);
return;
}
console.log("收到消息：", msg);
};
需要注意的是，浏览器直接连接本机端口会受到页面来源限制。生产环境通常把鉴权后的消息转发到前端，再由前端展示。
推送消息格式
鉴权通过后，机器人每收到一条微信消息，服务端就会推送一段 JSON 文本帧：
{
"event": "on_message",
"robot_wxid": "wxid_xxxxxxxxxxxxxx",
"robot_nickname": "机器人A",
"data": {
"msglist": ,
"msgnumber": "1",
"selfwxid": "wxid_xxxxxxxxxxxxxx",
"sendorrecv": "1"
}
}
顶层字段
| 字段 | 类型 | 说明 |
|---|---|---|
event |
string | 事件类型，消息接收场景下固定为 "on_message"，后续可能扩展 "system"、"error" 等 |
robot_wxid |
string | 消息来自哪个机器人，多账号环境用它做隔离 |
robot_nickname |
string 或 null | 机器人昵称，启动时一次性从数据库缓存，未命中时为 null |
data |
object | 微信原始协议消息的封装 |
data 内部字段
| 字段 | 类型 | 说明 |
|---|---|---|
msglist |
array | 消息列表，单条或多条消息都在这里 |
msgnumber |
string | 消息条数 |
selfwxid |
string | 自身微信号标识 |
sendorrecv |
string | 收发方向 |
后续如果要使用转发任意消息接口或文本发送接口，需要先从 msglist 中取到消息内容，再判断是否触发后续动作。
使用 robot_wxid 而不是昵称
当一台电脑同时挂载多个机器人时，robot_wxid 是判断消息归属的最稳定字段。昵称可能在登录后变更或为空，因此不建议用 robot_nickname 判定消息来源。
一个最小的完整订阅脚本
下面的 Python 示例演示完整流程：建立连接 → 鉴权 → 接收推送 → 把消息内容写入本地文件。示例只负责“打日志”，实际业务中可以替换为调用 AI、写入数据库或触发其他接口。
import asyncio
import json
import websockets
WS_URL = "ws://127.0.0.1:5010/api/ws/relay"
WS_TOKEN = "wk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
async def main():
async with websockets.connect(WS_URL) as ws:
await ws.send(json.dumps({
"type": "auth",
"key": WS_TOKEN
}))
async for raw in ws:
msg = json.loads(raw)
if msg.get("type") == "auth_failed":
print("鉴权失败：", msg.get("reason"))
break
if msg.get("event") != "on_message":
continue
robot_wxid = msg.get("robot_wxid")
data = msg.get("data") or {}
messages = data.get("msglist", [])
for item in messages:
print(f"[{robot_wxid}]", item)
asyncio.run(main())
WebSocket 推送只解决“接收实时消息”的问题。如果业务上还需要回复或转发，可以配合以下本机接口：
| 用途 | 接口 |
|---|---|
| 发送新文本消息 | http://127.0.0.1:5011/api/msg/text |
| 发送图片 | http://127.0.0.1:5011/api/msg/image |
| 转发任意历史消息 | http://127.0.0.1:5011/api/forward/any |
| 调用插件 action | http://127.0.0.1:5011/api/plugin/action/{plugin_id} |
典型组合示例：WebSocket 实时收到消息 → 程序判断是否需要回复 → 调用 /api/msg/text 发送文本；如果需要直接转发已经收到的原始消息，可以先从 msglist 中提取 msgsvrid，再调用 /api/forward/any。
需要提醒的是，转发任意消息接口目前不支持语音消息。如业务包含语音，请使用其他处理方式，而不是直接通过 /api/forward/any 转发。
常见连接与运行问题
连接成功但收不到推送
按以下顺序排查：
- 端口是否正确为
5010，路径是否为/api/ws/relay。 - 第一帧是否确实发送了
authJSON。 - Token 是否来自系统设置中的最新值，没有被刷新或误填。
- 机器人是否在线，是否登录了至少一个微信账号。
- 当前账号是否具备 WebSocket 推送功能权限。
- 是否有反向代理或防火墙拦截了本机
5010端口。
频繁断开
WebSocket 长连接在网络环境变化或进程空闲时可能被中间设备关闭。建议加入心跳机制，例如客户端定时发送 ping 服务端 pong；Python 中可以使用 websockets 库的 ping_interval 参数。
跨机器调用
由于服务只监听 127.0.0.1，默认情况下另一台电脑无法连接。出现这类需求时，应先确认是否有官方提供的外部入口或插件方案，再考虑自建转发网关。不要直接把本机端口通过端口映射或反向代理暴露到公网，存在安全和合规风险。
JSON 解析失败
推送数据是 UTF-8 JSON 文本帧。如果客户端使用了二进制帧或自定义解码方式，可能会解析失败。建议：
- 确认 WebSocket 库使用文本帧模式。
- 对每一条推送都使用
try/except包裹解析逻辑，避免单条异常导致整个订阅退出。 - 对
data.msglist做空值判断，例如data.get("msglist", [])。
数据处理与合规建议
WebSocket 推送的内容包含联系人、群聊和业务消息。处理这部分数据时建议遵守以下原则：
- 只对已获得授权的对象和消息进行处理，不要把所有消息都写入日志或转发。
ws_token是接口访问凭据，应仅保存在可信的本地环境，不要写入代码仓库或前端页面。- 对外部网络开放本机
5010端口存在数据泄露风险，应保持在127.0.0.1上运行。 - 重要业务逻辑应保留人工确认环节，尤其是涉及资金、客户资料或敏感关键词的场景。
- 自动化行为应符合平台规则和当地法律法规，知更Ai提供的是本地化接口能力，使用方式由使用者负责。
接入思路建议
不同的业务场景可以采用不同的接入方式：
- 想保存所有消息做分析：建立 WebSocket → 写入数据库或文件 → 定期归档。
- 想接入 AI 自动回复：建立 WebSocket → 调用大模型 → 通过
/api/msg/text发送。 - 想在多群之间同步内容：建立 WebSocket → 判定群和关键词 → 调用
/api/forward/any。 - 想把消息推给前端页面：建立 WebSocket → 用自己的服务端转发 → 前端通过 HTTP 拉取或再次建立连接。
这些场景的关键都在于：从 msglist 中准确取出消息字段，并根据业务规则决定下一步动作。后续如果接口字段调整，应以官方文档为准。
以平台官方说明为准，使用自动化功能前请了解并遵守相关平台政策与法律法规。
常见问题
问：websocket消息推送是什么用途？
答：用于实时接收知更Ai机器人收到的微信消息，适合保存消息日志、接入 AI、触发自动化或同步到业务系统。
问：websocket发送数据时需要注意什么？
答：连接建立后必须先发送鉴权 JSON，否则连接会被关闭；之后只接收服务端推送，客户端可以通过 ping/pong 维持连接。
问：websocket服务端消息推送数据长度有限制吗？
答：单条推送是 JSON 文本帧，具体上限取决于 WebSocket 实现和消息体大小。处理大消息时应分段读取，并避免一次性把过大的对象加载进内存。
问：微信websocket实时接收消息接口地址是什么？
答：默认地址是 ws://127.0.0.1:5010/api/ws/relay，监听本机 5010 端口。
问：微信websocket实时接收消息接口需要鉴权吗？
答：需要。第一帧必须发送 {"type":"auth","key":"<token>"}，鉴权失败时服务端会返回 auth_failed 并关闭连接。
问：如何区分不同机器人的消息？
答：使用顶层字段 robot_wxid，不要使用昵称，因为昵称在缓存未命中时可能为空。
问：能否在另一台电脑调用这个接口？
答：默认监听 127.0.0.1，跨机器调用需先确认官方是否提供外部入口或插件方案，不要直接把本机端口暴露到公网。
