4.2 KiB
你要的 IM实现逻辑清单（按这个做就能稳定接）：
入站推送到 OpenClaw 调用：POST /wechat_abot/webhook Body：直接发你现在 message.py 的原始 JSON Header：X-Wechat-Abot-Signature: sha256=<hmac_sha256(raw_body, webhookSecret)> 超时建议：3~5s；失败重试（指数退避） 签名实现（必须） 算法：HMAC-SHA256 输入：HTTP原始body字节（不要重排JSON） secret：和 channels.wechat_abot.webhookSecret 一致 防重放：建议带时间戳并限制窗口（可选增强） 出站发送接口（你IM提供） POST /messages/send Header：Authorization: Bearer 入参：{ to, text, replyToMessageId? } 返回：{ messageId } 错误码要清晰（401/403/429/5xx） 幂等与去重 入站按 MsgId/NewMsgId 去重（你侧做一次，插件侧也做一次更稳） 出站按你生成的 requestId/messageId 防重复发送 群聊处理 保持原始 Content.string（sender:\n正文）格式即可 插件会自动提取真实发送者并清理前缀 群里是否触发可在你侧先做过滤（如必须@机器人） 可靠性 出站接口支持重试（网络错误/502/503） 429 限流要返回 retry_after 全链路打日志：eventId、sender、conversationId、status、耗时 安全 webhook 只允许你的 IM 源IP（可选） token/secret 不写死代码，走环境变量 关闭调试日志里的敏感字段（token、签名原文）
可以，给你一份 Python 直连接口参考逻辑（入站推 OpenClaw + 出站发送接口），你按自己项目替换发送实现就行。
import os import json import hmac import hashlib import requests from flask import Flask, request, jsonify
app = Flask(name)
===== 配置 =====
OPENCLAW_WEBHOOK_URL = os.getenv("OPENCLAW_WEBHOOK_URL", "http://127.0.0.1:18789/wechat_abot/webhook") OPENCLAW_WEBHOOK_SECRET = os.getenv("OPENCLAW_WEBHOOK_SECRET", "change_me")
IM_BOT_TOKEN = os.getenv("IM_BOT_TOKEN", "change_me") # 给 OpenClaw 插件鉴权用
你自己的 IM SDK/client 在这里初始化
im_client = YourIMClient(...)
def sign_sha256(raw_body: bytes, secret: str) -> str: return hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
def verify_bearer(auth_header: str, expected_token: str) -> bool: if not auth_header or not auth_header.startswith("Bearer "): return False token = auth_header[7:].strip() return hmac.compare_digest(token, expected_token)
1) 你 IM 收到消息后的入口（示例）
@app.post("/im/inbound") def im_inbound():
这里 body 就是你 message.py 能解析的原始 JSON
raw = request.get_data() # 原始字节，签名必须用它 if not raw: return jsonify({"error": "empty body"}), 400
直接转发给 OpenClaw 插件 webhook
sig = sign_sha256(raw, OPENCLAW_WEBHOOK_SECRET) headers = { "Content-Type": "application/json", "X-Wechat-Abot-Signature": f"sha256={sig}", }
try: resp = requests.post( OPENCLAW_WEBHOOK_URL, data=raw, headers=headers, timeout=5 )
建议你自己加重试策略（429/5xx）
return jsonify({ "ok": resp.ok, "status": resp.status_code, "body": resp.json() if "application/json" in resp.headers.get("Content-Type", "") else resp.text }), 200 except requests.RequestException as e: return jsonify({"ok": False, "error": str(e)}), 502
2) 给 OpenClaw 插件调用的发送接口
@app.post("/api/messages/send") def api_messages_send():
鉴权：对应 channels.wechat_abot.botToken
auth = request.headers.get("Authorization", "") if not verify_bearer(auth, IM_BOT_TOKEN): return jsonify({"error": "unauthorized"}), 401
data = request.get_json(silent=True) or {} to = data.get("to") text = data.get("text", "") reply_to = data.get("replyToMessageId")
if not to or not text: return jsonify({"error": "missing to/text"}), 400
TODO: 这里换成你真实 IM 发送逻辑
result = im_client.send_text(to=to, text=text, reply_to=reply_to)
message_id = result["message_id"]
message_id = f"out_{hashlib.md5((to + text).encode()).hexdigest()[:12]}" # demo return jsonify({"messageId": message_id}), 200
if name == "main": app.run(host="0.0.0.0", port=8088, debug=False) 你只需要改两处：
/im/inbound：挂到你现有消息接收流程里 /api/messages/send：替换成你真实 IM 发消息函数