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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。确认能正常返回后再开始下面的项目生成。
对于长期要做编码和 Agent 任务的,建议直接上 Coding Plan,省得每次单独配: https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。
配置完成后,你的 Codex 环境应该能正常执行codex --version并返回版本号。这一步不做,后面的代码生成会频繁中断。
3. 可复制的 Flask 路由与菜单 JSON 配置
这一节是核心,直接给可复制的配置和代码。我按文件拆开,你照着建目录就行。
3.1 项目结构与依赖
先建目录:
mkdir -p wechat-bot/{app/{wechat,models,routes,utils},templates,logs} cd wechat-botrequirements.txt内容:
flask==3.1.1 flask-sqlalchemy==3.1.1 flask-cors==5.0.1 requests==2.32.3 python-dotenv==1.1.0 lxml==5.4.0 pycryptodome==3.21.0 gunicorn==23.0.0安装:
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple3.2 环境变量 .env
WECHAT_APP_ID=wx1234567890abcdef WECHAT_APP_SECRET=your_app_secret_here WECHAT_TOKEN=your_custom_token_2024 DATABASE_URL=sqlite:///wechat_bot.db LOG_LEVEL=INFO注意: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_info=True) 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 f"""<xml> <ToUserName><![CDATA[{to_user}]]></ToUserName> <FromUserName><![CDATA[{from_user}]]></FromUserName> <CreateTime>{int(time.time())}</CreateTime> <MsgType><![CDATA[text]]></MsgType> <Content><![CDATA[{content}]]></Content> </xml>"""3.8 应用工厂与启动 app/init.py
"""Flask 应用工厂""" 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.py:
from dotenv import load_dotenv load_dotenv() from app import create_app app = create_app() if __name__ == '__main__': app.run(host='0.0.0.0', port=8080, debug=True)到这里,核心配置和代码就齐了。你可以直接复制建文件,也可以把上面的需求描述丢给 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?signature=xxx×tamp=123&nonce=456&echostr=hello"如果签名不对,会返回Verification Failed。你可以写个小脚本算出正确的 signature:
import 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?signature=xxx×tamp=123&nonce=456" \ -H "Content-Type: text/xml" \ -d '<xml> <ToUserName><![CDATA[gh_xxx]]></ToUserName> <FromUserName><![CDATA[oUser123]]></FromUserName> <CreateTime>1700000000</CreateTime> <MsgType><![CDATA[text]]></MsgType> <Content><![CDATA[帮助]]></Content> </xml>'正常应该返回一段 XML,里面包含HELP_MESSAGE的内容。如果返回success,说明消息类型没匹配上,检查MsgType解析。
4.4 公众号后台配置
登录微信公众平台,进入「设置与开发」->「基本配置」:
服务器地址 URL 填你的公网地址加/wechat,Token 填.env里的WECHAT_TOKEN,EncodingAESKey 随机生成,消息加解密方式选「明文模式」(开发阶段方便调试)。
点「提交」,如果本地日志出现「微信接入验证成功」,就说明通了。然后扫码关注测试号,发一条「帮助」,看是否自动回复。
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( f'https://api.weixin.qq.com/cgi-bin/menu/create?access_token={access_token}', json=menu ).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/api,Model ID 按文档填。然后检查请求体里messages格式是否正确。
如果频繁出现,建议换一个稳定的通道,或者把请求重试逻辑加上:
import time def call_with_retry(fn, retries=3): 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_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
需要重新生成 API Key 的话,在 console 页面操作: https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。
长期做编码和 Agent 任务的,直接上 Coding Plan 更划算: https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。
最后说一个实用技巧:公众号后台的「接口权限」里,自定义菜单和消息推送是默认开通的,但模板消息、网页授权需要认证服务号才有。测试号可以先用着,功能验证没问题再迁移到正式号。
整个项目我建议用 git 管理,.env加到.gitignore里,别把 AppSecret 提交上去。部署到服务器时用 gunicorn 加 nginx,微信回调走 80 或 443 端口,记得配 HTTPS 证书。
这套流程走下来,你手里就有一个完全可控的公众号机器人了。后面想加什么功能,直接改 handler 里的分支就行,不用再受第三方平台限制。