用 Python 做微信机器人，适合快速验证规则和接大模型。个人号侧一般是 HTTP 通道：你发请求登录和发送，对方把新消息 POST 到你的 webhook。这篇用 FastAPI 写最小可运行结构，Flask 同理。

目录建议

app/
  channel.py    # 调登录、发文本
  webhook.py    # 收回调
  bot.py        # 规则
  store.py      # msgid 去重、会话
  queue.py      # 异步消费

不要把规则写在 webhook 函数里。Python 项目后期加 AI，最容易把 30 秒推理塞进回调，然后对方超时重推，客户收到两条。

回调：验 token、去重、入队

from fastapi import FastAPI, Header, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/hooks/wechat/message")
async def wechat_message(request: Request, x_token: str | None = Header(default=None)):
    if x_token != settings.callback_token:
        raise HTTPException(status_code=401, detail="bad token")
    raw = await request.body()
    event = parse_event(raw)
    if event is None:
        return JSONResponse({"ok": True})
    if store.seen(event.account_id, event.msg_id):
        return JSONResponse({"ok": True})
    store.save_raw(event.account_id, event.msg_id, raw)
    await queue.put(event)
    return JSONResponse({"ok": True})

生产把 queue 换成 Redis。本地可以用 asyncio.Queue 先跑通。

解析时至少取出：实例 ID、msgid、是否群聊、发送方、文本。字段名按通道文档映射，不要假设一定叫 content。

发送客户端

import httpx

class WeChatChannel:
    def __init__(self, base_url: str, token: str):
        self._client = httpx.Client(base_url=base_url, timeout=8.0, headers={"Authorization": f"Bearer {token}"})

    def send_text(self, account_id: str, to_wxid: str, text: str, request_id: str) -> dict:
        r = self._client.post("/message/text", json={
            "accountId": account_id,
            "to": to_wxid,
            "content": text,
            "requestId": request_id,
        })
        r.raise_for_status()
        return r.json()

路径是示意。封装的价值是：业务只调用 send_text，换环境只改 base_url。

发送前查在线。离线不要抛成普通 HTTP 500，要让任务进 blocked，等上线再决定补不补发。

规则：先同步可测

def reply(event, session) -> str | None:
    if session.human_taken or event.is_group:
        return None
    text = (event.text or "").strip()
    if not text:
        return "请发文字，或回复「人工」"
    if text == "人工":
        session.take_human()
        return "已转接，请稍候"
    if "地址" in text:
        return "门店地址：请换成你们真实地址"
    return None  # None 走兜底或转人工

用纯函数方便 pytest：构造 event / session 即可测，不必起 FastAPI。

兜底不要让大模型直接承诺库存。模型若要用，放在规则之后，并且超时（例如 8 秒）就转人工。

消费者

async def worker():
    while True:
        event = await queue.get()
        try:
            session = store.session(event.account_id, event.from_id)
            text = reply(event, session)
            if text:
                channel.send_text(
                    event.account_id,
                    event.from_id,
                    text,
                    request_id=event.msg_id,
                )
        except Exception:
            log.exception("handle failed msgid=%s", event.msg_id)
        finally:
            queue.task_done()

request_id 用 msgid 可以，但出站失败重试时要换新的业务 ID，避免通道侧当成重复请求丢掉。更稳是用你们库里的自增 ID。

Flask 差异

Flask 回调同样：读 raw body → 入队 → 马上 return ("", 200)。用 threading 或 Celery 消费。不要在视图里直接 requests.post 发微信，高峰会拖垮 worker。

联调清单

1. 通道在线

2. pytest 覆盖去重、人工停机、群消息不回

3. 用一份真实回调 JSON 打 webhook，确认入库

4. 看发送日志里的 request_id 和错误码

图片、语音第一版只回引导语，不要卡在下载素材。

小结

Python 微信机器人开发要利用语言擅长的部分：规则快、实验快，同时用队列把慢操作从 webhook 里拿出去。FastAPI / Flask 只负责收事件；通道客户端只负责 HTTP；bot.py 保持可单测。私聊文本闭环稳定后，再加 AI 和多账号。