1. 为什么我要用 Codex 重写公众号机器人微信公众号机器人这个需求几乎每个做私域、做内容、做客服的团队都绕不开。手动回消息回不过来关键词回复又太死板菜单改一次要翻半天后台。我最早是用现成的第三方平台搭的功能受限、数据不在自己手里后来干脆自己写 Flask 服务但每次加功能都要查微信文档、调 XML 格式、处理签名校验一个下午就没了。这次我换了个思路把整个公众号机器人项目交给 Codex 来生成我只负责描述需求和验证结果。实测下来从零到能跑通关注自动回复、关键词匹配、自定义菜单大概两个小时。这篇文章就是把这个过程完整拆开包括 Flask 路由怎么写、菜单 JSON 怎么配、Token 校验怎么做、本地怎么调试、公众号后台怎么验证。适合谁看有 Python 基础、想自己掌控公众号后台逻辑的开发者正在用 Codex 或类似 AI 编程工具、想找一个完整实战案例的人以及被第三方平台限制、想迁移到自建服务的团队。核心检索词先明确Codex 生成 Flask 微信公众号机器人实现自动回复与自定义菜单。下面所有代码和配置都可以直接复制改掉 AppID、AppSecret、Token 就能用。整个项目结构不复杂但涉及微信回调的签名校验、XML 消息解析、access_token 缓存、菜单创建这几个关键点。我会按「先跑通再优化」的顺序来写每一步都有可验证的结果。2. TaoToken 统一 Key 接入 Codex 的前置配置Codex 本身是一个 AI 编程工具但如果你在本地或服务器上跑需要给它一个稳定的模型通道。我试过直接填各种零散的 Key管理起来很乱后来统一走 TaoToken 的 API 通道一个 Key 管所有模型调用省心不少。TaoToken 在这里的角色是提供统一的 API 入口Codex 通过它来调用模型生成代码。你不需要在多个平台之间切换也不用担心 Key 过期后到处改配置。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。具体配置分两步。第一步在 TaoToken 控制台创建一个 API Key路径是 console 页面下的 api-keys。第二步把 Key 填到 Codex 的配置里。如果你用的是 Claude Code 或类似的 coding agent配置方式略有不同但核心三件套是一样的Base URL、API Key、Model ID。我实测下来Codex 在生成 Flask 项目时最怕的是上下文断裂——生成到一半模型换了、Key 失效了代码就接不上。统一通道的好处就是整个项目生成过程中模型调用是连续的不会出现「前半段用 A 模型、后半段用 B 模型」导致的风格不一致。如果你还没配好可以先去模型对话页面测试一下通道是否正常 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。确认能正常返回后再开始下面的项目生成。对于长期要做编码和 Agent 任务的建议直接上 Coding Plan省得每次单独配 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。配置完成后你的 Codex 环境应该能正常执行codex --version并返回版本号。这一步不做后面的代码生成会频繁中断。3. 可复制的 Flask 路由与菜单 JSON 配置这一节是核心直接给可复制的配置和代码。我按文件拆开你照着建目录就行。3.1 项目结构与依赖先建目录mkdir -p wechat-bot/{app/{wechat,models,routes,utils},templates,logs} cd wechat-botrequirements.txt内容flask3.1.1 flask-sqlalchemy3.1.1 flask-cors5.0.1 requests2.32.3 python-dotenv1.1.0 lxml5.4.0 pycryptodome3.21.0 gunicorn23.0.0安装pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple3.2 环境变量 .envWECHAT_APP_IDwx1234567890abcdef WECHAT_APP_SECRETyour_app_secret_here WECHAT_TOKENyour_custom_token_2024 DATABASE_URLsqlite:///wechat_bot.db LOG_LEVELINFO注意WECHAT_TOKEN是你自己随便设的一串字符不是微信给的后面公众号后台要填一样的值。3.3 签名校验模块 app/wechat/crypto.py微信签名验证模块 import hashlib def check_signature(token: str, signature: str, timestamp: str, nonce: str) - bool: 验证微信服务器签名。 步骤token、timestamp、nonce 三个参数排序 - 拼接 - SHA1 - 对比 signature params sorted([token, timestamp, nonce]) raw_string .join(params) sha1_hash hashlib.sha1(raw_string.encode(utf-8)).hexdigest() return sha1_hash signature3.4 微信回调路由 app/routes/wechat_routes.py微信回调路由 import os import logging from flask import Blueprint, request, make_response from app.wechat.crypto import check_signature from app.wechat.handler import MessageHandler logger logging.getLogger(__name__) wechat_bp Blueprint(wechat, __name__) handler MessageHandler() wechat_bp.route(/wechat, methods[GET]) def wechat_verify(): 微信接入验证原样返回 echostr signature request.args.get(signature, ) timestamp request.args.get(timestamp, ) nonce request.args.get(nonce, ) echostr request.args.get(echostr, ) token os.getenv(WECHAT_TOKEN, ) if check_signature(token, signature, timestamp, nonce): logger.info(微信接入验证成功) return echostr logger.warning(微信接入验证失败) return Verification Failed, 403 wechat_bp.route(/wechat, methods[POST]) def wechat_message(): 接收微信消息推送5 秒内返回 XML 回复 signature request.args.get(signature, ) timestamp request.args.get(timestamp, ) nonce request.args.get(nonce, ) token os.getenv(WECHAT_TOKEN, ) if not check_signature(token, signature, timestamp, nonce): return Invalid Signature, 403 xml_data request.data.decode(utf-8) reply_xml handler.handle(xml_data) response make_response(reply_xml) response.content_type application/xml return response3.5 自定义菜单 JSON这是公众号后台创建菜单时用的 JSON可以直接复制到管理接口里{ button: [ { type: click, name: 功能, sub_button: [ {type: click, name: 帮助, key: MENU_HELP}, {type: click, name: 联系客服, key: MENU_CONTACT}, {type: click, name: 最新资讯, key: MENU_LATEST} ] }, { type: click, name: 关于, sub_button: [ {type: click, name: 关于我们, key: MENU_ABOUT}, {type: view, name: 官方网站, url: https://example.com} ] }, { type: click, name: 我的, key: MENU_MINE } ] }菜单里click类型的按钮会触发事件推送key值对应你在 handler 里配置的回复逻辑view类型直接跳转网页不需要后端处理。3.6 消息处理器核心逻辑 app/wechat/handler.py微信消息处理器 import logging from app.wechat.message import WechatMessage, WechatReply logger logging.getLogger(__name__) WELCOME_MESSAGE 欢迎关注回复「帮助」查看功能列表。 DEFAULT_REPLY 抱歉我暂时没理解你的意思。回复「帮助」看看我能做什么。 HELP_MESSAGE 功能列表\n1. 回复「帮助」- 查看功能\n2. 回复「价格」- 产品报价\n3. 回复「客服」- 转人工 class MessageHandler: def handle(self, xml_data: str) - str: try: msg WechatMessage(xml_data) logger.info(f收到消息: type{msg.msg_type}, from{msg.from_user}) if msg.msg_type text: return self._handle_text(msg) elif msg.msg_type event: return self._handle_event(msg) else: return self._reply_text(msg, DEFAULT_REPLY) except Exception as e: logger.error(f消息处理异常: {e}, exc_infoTrue) return success def _handle_text(self, msg): content msg.content.strip() if content in (帮助, help, ?): return self._reply_text(msg, HELP_MESSAGE) if 价格 in content: return self._reply_text(msg, 基础版 99 元/月专业版 299 元/月。) if 客服 in content: return self._reply_text(msg, 客服电话400-123-4567) return self._reply_text(msg, DEFAULT_REPLY) def _handle_event(self, msg): event msg.event.lower() if event subscribe: return self._reply_text(msg, WELCOME_MESSAGE) if event click: menu_replies { MENU_HELP: HELP_MESSAGE, MENU_ABOUT: 我们是一家专注技术创新的公司。, MENU_CONTACT: 客服电话400-123-4567, MENU_LATEST: 正在获取最新资讯..., } return self._reply_text(msg, menu_replies.get(msg.event_key, 你点击了菜单)) return success def _reply_text(self, msg, content): return WechatReply.text(msg.to_user, msg.from_user, content)3.7 消息解析与回复构建 app/wechat/message.py微信消息解析与构建 import time from lxml import etree class WechatMessage: def __init__(self, xml_data: str): self._data {} root etree.fromstring(xml_data.encode(utf-8)) for child in root: self._data[child.tag] child.text or property def msg_type(self): return self._data.get(MsgType, ) property def content(self): return self._data.get(Content, ) property def from_user(self): return self._data.get(FromUserName, ) property def to_user(self): return self._data.get(ToUserName, ) property def event(self): return self._data.get(Event, ) property def event_key(self): return self._data.get(EventKey, ) class WechatReply: staticmethod def text(from_user, to_user, content): return fxml ToUserName![CDATA[{to_user}]]/ToUserName FromUserName![CDATA[{from_user}]]/FromUserName CreateTime{int(time.time())}/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[{content}]]/Content /xml3.8 应用工厂与启动 app/init.pyFlask 应用工厂 from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_cors import CORS db SQLAlchemy() def create_app(): app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] sqlite:///wechat_bot.db app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db.init_app(app) CORS(app) from app.routes.wechat_routes import wechat_bp app.register_blueprint(wechat_bp) with app.app_context(): db.create_all() return apprun.pyfrom dotenv import load_dotenv load_dotenv() from app import create_app app create_app() if __name__ __main__: app.run(host0.0.0.0, port8080, debugTrue)到这里核心配置和代码就齐了。你可以直接复制建文件也可以把上面的需求描述丢给 Codex 让它生成效果一样。4. 本地调试与公众号后台验证步骤代码写完了关键是验证能不能跑通。我分本地调试和后台验证两步走。4.1 本地启动python run.py看到Running on http://0.0.0.0:8080就说明服务起来了。但微信回调需要公网地址本地 127.0.0.1 微信访问不到。解决办法是用内网穿透工具把 8080 端口映射出去拿到一个公网 URL。假设你拿到的公网地址是https://abc123.example.com那么微信后台要填的回调 URL 就是https://abc123.example.com/wechat4.2 手动模拟微信验证请求在浏览器或 curl 里模拟微信的 GET 验证请求确认签名逻辑没问题curl http://127.0.0.1:8080/wechat?signaturexxxtimestamp123nonce456echostrhello如果签名不对会返回Verification Failed。你可以写个小脚本算出正确的 signatureimport hashlib token your_custom_token_2024 timestamp 123 nonce 456 params sorted([token, timestamp, nonce]) print(hashlib.sha1(.join(params).encode()).hexdigest())把算出来的值填到 signature 参数里再请求一次应该返回hello。4.3 模拟 POST 消息推送用 curl 模拟一条文本消息curl -X POST http://127.0.0.1:8080/wechat?signaturexxxtimestamp123nonce456 \ -H Content-Type: text/xml \ -d xml ToUserName![CDATA[gh_xxx]]/ToUserName FromUserName![CDATA[oUser123]]/FromUserName CreateTime1700000000/CreateTime MsgType![CDATA[text]]/MsgType Content![CDATA[帮助]]/Content /xml正常应该返回一段 XML里面包含HELP_MESSAGE的内容。如果返回success说明消息类型没匹配上检查MsgType解析。4.4 公众号后台配置登录微信公众平台进入「设置与开发」-「基本配置」服务器地址 URL 填你的公网地址加/wechatToken 填.env里的WECHAT_TOKENEncodingAESKey 随机生成消息加解密方式选「明文模式」开发阶段方便调试。点「提交」如果本地日志出现「微信接入验证成功」就说明通了。然后扫码关注测试号发一条「帮助」看是否自动回复。4.5 创建自定义菜单菜单创建需要 access_token。你可以写个临时脚本调用import os, requests from dotenv import load_dotenv load_dotenv() app_id os.getenv(WECHAT_APP_ID) app_secret os.getenv(WECHAT_APP_SECRET) token_resp requests.get( https://api.weixin.qq.com/cgi-bin/token, params{grant_type: client_credential, appid: app_id, secret: app_secret} ).json() access_token token_resp[access_token] menu { button: [ {type: click, name: 功能, sub_button: [ {type: click, name: 帮助, key: MENU_HELP}, {type: click, name: 联系客服, key: MENU_CONTACT} ]}, {type: click, name: 关于, key: MENU_ABOUT} ] } resp requests.post( fhttps://api.weixin.qq.com/cgi-bin/menu/create?access_token{access_token}, jsonmenu ).json() print(resp)返回{errcode:0,errmsg:ok}就说明菜单创建成功。回到公众号会话窗口底部菜单应该已经更新。点击「帮助」看是否收到自动回复。4.6 验证结果对照验证项预期结果实际排查点GET 验证返回 echostr签名算法、Token 是否一致关注事件收到欢迎语subscribe 分支是否命中文本「帮助」收到功能列表关键词匹配逻辑菜单点击收到对应回复EventKey 是否匹配菜单显示底部出现自定义菜单access_token 是否有效这套流程走完一个能用的公众号机器人就上线了。5. 常见报错排查401、local proxy failed、reading choices这一节是我踩过的坑按报错类型整理。5.1 401 Unauthorized这个报错通常出现在调用微信 API 时比如创建菜单、获取用户信息。原因一般是 access_token 无效或过期。排查步骤先确认.env里的WECHAT_APP_ID和WECHAT_APP_SECRET没填错然后检查 access_token 缓存逻辑微信的 token 有效期 7200 秒我建议提前 300 秒刷新最后确认服务器时间是否准确时间偏差太大会导致签名失败。如果你用的是 TaoToken 通道调用模型生成代码401 也可能是 API Key 失效。去 console 页面重新生成一个 Key更新到配置里。5.2 local proxy failed这个报错一般出现在本地调试时Codex 或 requests 请求走了系统代理但代理不可用。解决办法是在代码里显式禁用代理import os os.environ[NO_PROXY] *或者在 requests 调用里加proxies{http: None, https: None}。注意这里说的是本地网络配置问题不涉及任何网络访问方式的选择只是让请求直连。5.3 reading choices 报错这个报错通常出现在模型返回格式异常时比如 Codex 调用模型生成代码返回的 JSON 里没有choices字段。原因可能是模型通道不稳定或者请求参数不对。排查先确认 Base URL 和 Model ID 是否匹配。如果你用的是 TaoToken 统一通道Base URL 应该是https://taotoken.net/apiModel ID 按文档填。然后检查请求体里messages格式是否正确。如果频繁出现建议换一个稳定的通道或者把请求重试逻辑加上import time def call_with_retry(fn, retries3): for i in range(retries): try: return fn() except Exception as e: if i retries - 1: raise time.sleep(2 ** i)5.4 OAuth 相关报错公众号网页授权时会出现 OAuth 报错常见的是redirect_uri参数错误。检查两点一是后台「网页授权域名」有没有配二是redirect_uri有没有做 URL encode。5.5 菜单创建失败 errcode 40016这个错误是「invalid button size」说明菜单按钮数量超了。微信规定一级菜单最多 3 个二级菜单最多 5 个。检查你的 JSON 结构。5.6 消息回复超时微信要求 5 秒内返回如果处理逻辑太重比如查数据库、调外部 API容易超时。解决办法是把耗时操作异步化先返回success再用客服消息接口主动推送。5.7 Codex 生成代码时的配置三件套如果你在 Codex 里配置模型通道记住三件套Base URL、API Key、Model ID。以 Claude Code 为例配置文件里要写全{ base_url: https://taotoken.net/api, api_key: your_taotoken_key, model: claude-sonnet-4-20250514 }如果是 Codex 的 auth.json格式类似{ api_key: your_taotoken_key, base_url: https://taotoken.net/api }Cline MCP 的配置则在 settings 里填 Base URL 和 Key。三件套缺一不可少一个就会报 401 或 reading choices。6. 接入文档与后续扩展代码跑通之后你可能会想加更多功能图文消息回复、模板消息推送、用户标签管理、消息日志统计。这些都可以在现有结构上扩展。消息日志我建议一开始就加上方便排查问题。在handler.py里加一个_log_message方法把openid、msg_type、content、received_at存到数据库。后面做数据分析、热门关键词统计都用得上。access_token 缓存也别用内存多进程部署时会冲突。建议用 Redis 或数据库存加个过期时间字段。如果你在接入过程中遇到签名验证失败、菜单创建报错、消息回复超时这些问题可以先去看接入文档里面有完整的参数说明和示例 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。需要重新生成 API Key 的话在 console 页面操作 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。长期做编码和 Agent 任务的直接上 Coding Plan 更划算 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。最后说一个实用技巧公众号后台的「接口权限」里自定义菜单和消息推送是默认开通的但模板消息、网页授权需要认证服务号才有。测试号可以先用着功能验证没问题再迁移到正式号。整个项目我建议用 git 管理.env加到.gitignore里别把 AppSecret 提交上去。部署到服务器时用 gunicorn 加 nginx微信回调走 80 或 443 端口记得配 HTTPS 证书。这套流程走下来你手里就有一个完全可控的公众号机器人了。后面想加什么功能直接改 handler 里的分支就行不用再受第三方平台限制。
