技术支持 wechatapi.net

很多项目一开始都是从一个 main.py 起步的。
我这次做微信接 OpenClaw，也不例外。

最开始，我只是想验证：

• 微信回调能不能接到
• 消息能不能转给 OpenClaw
• 回复能不能发回微信

但做着做着我发现，真正难的不是“跑通”，而是怎么让这个单文件项目不要停留在脚本层，而是逐步长成一个可维护的入口网关。

这篇文章就想聊聊：我是怎么从一个简单脚本，慢慢把它往网关方向改的。

一、最开始的版本，其实就是一个最小中转器

最早的思路就是：

微信消息 -> callback -> openclaw agent -> 微信回复

伪代码非常简单：

@app.post("/wechat/callback")
async def handle_wechat(request: Request):
    body = await request.body()
    data = json.loads(body)
    text = extract_text(data)

    result = subprocess.run(
        ["openclaw", "agent", "--session-id", "demo", "--message", text],
        capture_output=True,
        text=True
    )

    reply_text("某个wxid", result.stdout)
    return {"status": "ok"}

这个版本的优点是：
非常容易验证。

但它的问题也很明显：

• 没有会话设计
• 没有并发设计
• 没有消息去重
• 没有白名单
• 没有群触发逻辑
• 没有错误收口
• 也没有初始化流程

它更像一个“证明这件事能做”的脚本，而不是一个“别人可以运行”的系统。

二、我后面第一个决定：先把初始化做成产品化

如果一个项目连启动都不顺，后面再多能力都白搭。

所以我后面第一步不是继续堆功能，而是把首次运行流程做成了命令行初始化，引导用户输入最关键的几项：

• WX_API_TOKEN
• PUBLIC_URL
• 群触发词
• 地区 ID

并且自动生成配置文件。

代码大概是这样：

def interactive_init():
    print("WeChat OpenClaw Gateway 首次初始化")

    api_token = input("请输入 WX_API_TOKEN: ").strip()
    public_url = input("请输入公网回调地址 PUBLIC_URL（不能为空）: ").strip()
    group_trigger = input("请输入群触发词（默认 狗子）: ").strip() or "狗子"

    cfg = DEFAULTS.copy()
    cfg.update({
        "api_token": api_token,
        "public_url": public_url,
        "group_trigger": group_trigger,
    })
    write_config(cfg)

这一步做完以后，项目的性质其实就已经变了。
它不再只是“我自己能跑”，而是开始考虑“别人怎么跑”。

三、第二个关键决定：把消息结构彻底吃透

我后来很快意识到，这种接入类项目最怕的不是没功能，而是回调解析错了还以为自己对了。

比如我后面统一做成了这个解析函数：

def parse_wechat_payload(data):
    if not isinstance(data, dict):
        return None

    type_name = data.get("TypeName")
    if type_name != "AddMsg":
        return {"event_type": type_name, "raw": data}

    wxid = str(data.get("Wxid", "") or "").strip()
    msg_data = data.get("Data", {})
    from_user = msg_data.get("FromUserName", {}).get("string", "")
    to_user = msg_data.get("ToUserName", {}).get("string", "")
    raw_content = msg_data.get("Content", {}).get("string", "")
    msg_type = msg_data.get("MsgType")

    is_self = bool(wxid and from_user == wxid)
    is_group = from_user.endswith("@chatroom") or to_user.endswith("@chatroom")

    sender_wxid = from_user
    actual_text = raw_content.strip()

    if is_group and raw_content and ":\n" in raw_content:
        possible_sender, possible_text = raw_content.split(":\n", 1)
        if possible_sender.startswith("wxid_"):
            sender_wxid = possible_sender
            actual_text = possible_text.strip()

    return {
        "event_type": "AddMsg",
        "wxid": wxid,
        "msg_type": msg_type,
        "from_user": from_user,
        "to_user": to_user,
        "sender_wxid": sender_wxid,
        "actual_text": actual_text,
        "is_group": is_group,
        "is_self": is_self,
    }

这个函数的意义，不是“把字段取出来”，而是：

• 把微信回调转换成我自己的统一消息模型
• 后面不管是微信、企微还是其他通道，都可以沿着同一层抽象继续走

这一步其实就是把脚本往“网关层”推进。

四、第三个关键决定：session 不再随便拼

如果只是私聊机器人，其实 session 逻辑很容易偷懒。
但只要一上群聊，就一定要认真设计会话模型。

我最后用了这样的 session 设计：

def build_session_id(chat_id: str, sender_wxid: str, is_group: bool, config: dict) -> str:
    def norm(s: str) -> str:
        return re.sub(r"[^a-zA-Z0-9_-]", "_", str(s or "").strip())

    if not is_group:
        return f"wechat_dm_{norm(chat_id)}"

    if config["GROUP_SESSION_MODE"] == "per_user":
        return f"wechat_group_{norm(chat_id)}_user_{norm(sender_wxid)}"

    return f"wechat_group_{norm(chat_id)}"

这一步让我对这套系统的理解变得更清楚：

• 不是“消息来了就转发”
• 而是“消息先进入会话系统，再进入 AI”

一旦这样看，很多功能就有了扩展基础。

五、第四个关键决定：并发模型不能偷懒

很多人做这类项目时，默认就一个全局队列。
我一开始也这么想，但很快意识到这样做的问题：

• 群里一热闹，全排队
• 一个慢请求拖后面所有消息
• 体感会很差

所以我做了一个折中方案：

• 不同 session 并行
• 同一 session 保证顺序
• 用哈希把 session 固定路由到某个 worker

核心是：

def shard_index_for_session(session_id: str, worker_count: int) -> int:
    h = int(hashlib.md5(session_id.encode("utf-8")).hexdigest(), 16)
    return h % worker_count

然后：

worker_queues[shard_idx].put_nowait(task)

这个设计其实特别适合入口型系统，因为它既没有把复杂度拉到消息中间件级别，又足够解决大部分常见场景。

六、第五个关键决定：先只处理文本，其他都延后

这一步看起来保守，但我现在越来越觉得这是对的。

因为微信回调类型太多了：

• 图片
• 语音
• 文件
• 名片
• 引用
• 系统通知

如果一开始全接，你会很快把自己拖进复杂度沼泽。

所以我直接在入口层挡掉了非文本消息：

if msg_type != 1:
    logger.info("📭 暂时只处理文本消息，已忽略 MsgType=%s", msg_type)
    return {"status": "ignored_msg_type"}

这让整个项目能先快速稳定下来。

七、我最后意识到，真正慢的不是网关，而是 OpenClaw CLI

这也是我做完以后最清晰的认知。

入口层能优化的，我基本都优化了：

• 初始化
• 配置
• 回调解析
• 会话设计
• 去重
• worker 分片

但延迟依旧很难做到和前端一样顺滑。
原因不在微信，而在这一步：

subprocess.run(["openclaw", "agent", "--session-id", sid, "--message", text])

只要底层是单次 CLI 调用，每条消息就都有冷启动开销。

所以我现在对这个项目的理解也变得更现实了：

单文件 main.py 可以把入口层做得很顺，但如果要继续追求低延迟，最终还是要解决 OpenClaw 的常驻化问题。

八、那为什么我还继续坚持单文件版本

因为它依然有非常强的价值：

1. 便于快速验证需求
2. 便于给客户演示
3. 便于跑通真实场景
4. 便于不断调整入口设计
5. 便于后面再抽象成更正规的服务

也就是说，它不是最终形态，但它是一个非常好的产品雏形。

九、我对这件事现在的看法

我现在越来越觉得，一个好的入口项目，未必一开始就要“工程化到极致”。
但它一定要从一开始就有这些意识：

• 初始化不能难用
• 配置不能难改
• 日志不能看不懂
• 会话不能随便拼
• 并发不能全串行
• 错误不能直接暴露给用户

只要这些意识是对的，哪怕只是一个 main.py，它也可以逐步长成一个真正的入口网关。

十、最后

这次做下来，我最大的感受是：

不是所有单文件项目都只是脚本。

如果你愿意认真对待入口层，它完全可以成为一个产品的第一版核心。

如果你也在做微信、OpenClaw、机器人入口或者类似通道层的东西，欢迎交流。