跑通"能收消息、能发回复"最小闭环只要几十行。但机器人真正上线之后，你会发现它是个需要持续照料的小进程——要巡检、要恢复、要关停。这篇从"写完能跑"到"能长期上线"讲清楚要补哪些东西。

一、启动自检：别让机器人带病上岗

进程启动时跑一遍自检：Token 能不能换、发送能不能通、回调地址能不能达。任一不过就别接真实流量，先修好。

自检代码放在 main 入口最前面，失败直接 exit，不要假装启动成功。自检失败时把具体原因打出来，别只说"启动失败"——下次排错省半小时。

二、运行巡检：定时看一眼有没有活着

进程在跑不等于机器人正常。可能 Token 过期没刷新、发送队列堵了、回调收不到了。定时巡检三件事：Token 有效期、最近有没有收到消息、发送队列积压情况。

巡检结果打日志、也可以上报监控。连续几次异常就触发告警。巡检频率不用太高——每 5 分钟一次足够。

三、异常恢复：掉线了自己爬起来

常见异常：Token 被清、网络断了几秒、进程被回收。每种情况恢复方式不同。

Token 失效——下次调用时捕获错误自动换。网络抖动——重试时加退避，别每秒狂试。进程挂了——靠守护进程（systemd、supervisor）自动拉起，不要自己写 while true。

四、优雅关停：别硬断

停机时先停接收、再把队列里剩下的消息发完、最后退出。硬断会导致队列里的消息丢失，下次启动不会自动补。

机器人生命周期对照

阶段	做什么	验证方式	常见坑
启动	四项自检	打印每项结果	带病上岗
运行	定时巡检	日志/监控	只看进程在跑
异常	分类恢复	自动重试+告警	狂试不后退避
关停	停接→排空→退出	队列清零再退	硬断丢消息

最小完整骨架

import queue, time, sys
from flask import Flask, request
import requests

app = Flask(__name__)
outbox = queue.Queue()
running = True

def startup_check(base, app_id, secret, self_wxid):
    # 四项自检：Token、发送、回调、队列
    tk = requests.post(f"{base}/auth/token",
        json={"app_id": app_id, "secret": secret}, timeout=10).json()
    if not tk.get("token"): raise RuntimeError("Token 换不下来")
    # 发给自己验证出口
    r = requests.post(f"{base}/send",
        json={"wxid": self_wxid, "content": "自检"},
        headers={"Authorization": tk["token"]}, timeout=15).json()
    if r.get("code") != 0: raise RuntimeError("发送不通: " + str(r))
    print("[启动自检] 通过")

def worker_loop(base, token_getter):
    while running:
        try:
            item = outbox.get(timeout=5)
            requests.post(f"{base}/send", json=item["payload"],
                headers={"Authorization": token_getter()}, timeout=15)
            time.sleep(3)                    # 发送间隔
        except queue.Empty:
            continue

def shutdown():
    global running
    running = False
    # 排空队列里剩下的消息
    while not outbox.empty():
        item = outbox.get_nowait()
        print("[关停] 排空", item)
    print("[关停完成]")

@app.route("/hook", methods=["POST"])
def hook():
    data = request.get_json(force=True) or {}
    outbox.put({
        "payload": {"wxid": data.get("from"), "content": "收到"},
        "event": data,
    })
    return {"code": 0}

if __name__ == "__main__":
    startup_check(BASE, APP_ID, SECRET, MY_WXID)
    # worker 和 Flask 各自守护，进程挂了靠 systemd 拉起
    import threading
    threading.Thread(target=worker_loop, daemon=True).start()
    try: app.run(port=5000)
    except KeyboardInterrupt: shutdown(); sys.exit(0)

落地建议

启动必须做自检，失败直接退出，不要带病上岗。运行定时巡检 Token、消息接收和队列积压，连续异常触发告警。异常恢复加退避，别每秒狂试；进程挂了靠守护进程拉起，不要自己写 while true。关停先停接收再排空队列，别硬断丢消息。微信侧的消息接口由 Eyun 平台这类个人微信API平台 提供，生命周期管理在自建服务实现，具体字段以平台开发文档为准。