news 2026/10/9 15:51:26

Codex 100个真实案例 - 用AI做微信公众号机器人(自动回复+菜单)

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
Codex 100个真实案例 - 用AI做微信公众号机器人(自动回复+菜单)

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-bot

requirements.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/simple

3.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 == signature

3.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 response

3.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 app

run.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/wechat

4.2 手动模拟微信验证请求

在浏览器或 curl 里模拟微信的 GET 验证请求,确认签名逻辑没问题:

curl "http://127.0.0.1:8080/wechat?signature=xxx&timestamp=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&timestamp=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 里的分支就行,不用再受第三方平台限制。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/10/9 15:51:18

基于YOLOv8的地下管廊积水渗漏检测系统实战指南

简介&#xff1a;这是一套基于YOLOv8的智慧城市地下管廊积水渗漏检测系统&#xff0c;专为计算机视觉、深度学习方向的毕业设计或课程设计打造&#xff0c;源代码经过实际运行测试&#xff0c;功能稳定&#xff0c;涵盖完整数据集、可视化交互界面与部署文档。系统可自动输出核…

作者头像 李华
网站建设 2026/10/9 15:46:47

PRM-DUL 实战:Oracle 数据文件误删后的紧急恢复与跨版本迁移

简介&#xff1a;PRM-DUL Oracle数据库恢复工具v4.1是一款面向DBA与数据库运维人员的企业级数据救援软件&#xff0c;专为Oracle数据库在异常宕机、文件损坏或误删等场景下的数据抢救而设计。它可运行于AIX、HPUX、SOLARIS、Linux及Windows等多种操作平台&#xff0c;并兼容Ora…

作者头像 李华
网站建设 2026/10/9 15:45:00

多商户多仓库SaaS进销存源码:数据隔离与库存并发实践

简介&#xff1a;这是一套面向企业信息化与进销存SaaS开发场景的多商户ERP管理系统源码包。系统支持总公司—子公司—门店三级组织架构&#xff0c;各企业数据完全隔离&#xff0c;同门店多仓库共享基础数据但单据隔离&#xff0c;不同门店的仓库也支持调拨&#xff0c;总公司与…

作者头像 李华
网站建设 2026/10/9 15:42:45

MFC连接MySQL数据库:ODBC配置与增删改查实战指南

简介&#xff1a;面向需要在MFC应用中集成MySQL数据库的C开发者&#xff0c;这份压缩包以ODBC方式打通数据库连接链路&#xff0c;围绕驱动安装、系统DSN创建、CDatabase/CRecordset封装、SQL执行与结果集遍历展开&#xff0c;并给出异常处理与事务管理思路&#xff0c;适合初学…

作者头像 李华
网站建设 2026/10/9 15:42:40

开源BOM管理软件:用集中式数据库替代Excel物料清单

简介&#xff1a;这套开源物料清单管理工具是一份完整的C#桌面应用源码&#xff0c;面向电子制造企业的研发与采购人员&#xff0c;解决多用户协同维护元器件清单、跟踪版本变更等管理难题。得益于与Ciiva电子元件搜索接口的深度集成&#xff0c;系统能够在一个集中式数据库中统…

作者头像 李华