前言
做私域运营或自动化机器人,绕不开一个问题:怎么用程序控制微信发消息?官方公众号 API 有明确限制,个人号场景则完全没有官方接入方式。很多开发者第一次摸这块时容易踩坑——环境搭不起来、协议逆向风险大、一不小心触发风控。
本文聚焦最直接的路径:用 Python 通过 HTTP 接口对接微信,完整跑通"登录 → 获取好友 → 发文本消息"这一最基础链路,让你对整个机制有清晰认知,再也不用靠猜。代码全部可运行,按步骤做下来你会有一份可复用的基础模块。
一、先搞清楚:微信 API 有几种技术路线?
在动手写代码之前,先把技术全景理清楚,省得走弯路。
1.1 官方开放平台(公众号 / 小程序 / 企业微信)
腾讯官方提供了合规的 API,但适用场景有限:
| 类型 | 场景 | 核心限制 |
|---|---|---|
| 公众号 API | 订阅号/服务号与用户对话 | 用户必须先关注;无法主动发给任意微信号 |
| 企业微信 API | 企业内部沟通 | 双方必须同属一个组织 |
| 小程序消息 | 用户完成特定行为后的通知 | 严格的模板限制,不能随意发文本 |
如果你的需求落在这些范围内,优先走官方路线,稳定、合规、有文档。
1.2 Hook / 注入方案
在 Windows 客户端进程里注入 DLL,拦截消息收发。技术门槛高,依赖微信客户端版本,每次微信更新都可能失效,且存在一定法律风险,不推荐生产环境使用。另外,这类方案通常要求机器安装特定版本的微信 PC 端,还要应对反注入检测,维护成本极高。
1.3 托管式 HTTP API(本文使用的方案)
另一种思路是:由第三方平台负责维护微信的底层连接,对外暴露统一的 REST API,你只需调 HTTP 接口,完全不用关心协议细节。开发体验接近调普通 Web 服务,Python 几十行就能跑起来。
这类方案通常基于开源框架(如 gewechat)封装,市面上有现成的托管服务可直接接入,比自己搭底层省力得多。WechatApi(https://wechatapi.net)提供扫码登录、消息收发、好友与群管理等 REST 接口,HTTP 调用即可,文档齐全,接口风格统一,下文示例按此风格编写——具体接口路径和字段名以你所用平台的官方文档为准。
二、环境准备
2.1 Python 依赖
只需要 requests 库,几乎所有环境都已内置:
bashpip install requests
Python 版本建议 3.8+,示例使用了 f-string 和 dataclass,低版本需要小改动。
2.2 获取接入凭证
不同平台流程略有差异,通常需要:
- 注册账号,创建应用
- 扫码让一个微信账号上线(API 托管你的微信连接)
- 平台会给你
Token(鉴权凭证)和appId(该设备/账号的唯一标识)
安全提醒:Token 等同于账号密钥,不要硬编码进源码,建议放环境变量或配置文件。
关于 appId 和 Token 的区别需要说清楚:appId 标识的是某一个微信账号的上线实例,一个 Token 下面可以挂多个 appId;发消息时每次都要带上 appId,不然平台不知道你要用哪个账号发出去。这一点是很多新手第一次调接口时搞混的地方,调用返回鉴权失败,往往不是 Token 本身错了,而是 appId 遗漏或填错了。
2.3 项目结构
wechat_bot/
├── config.py # 配置集中管理
├── client.py # HTTP 封装层
├── demo_send.py # 本文主体示例
└── callback.py # 接收消息用(后面章节)
三、封装一个轻量 HTTP 客户端
先写一个通用请求层,避免后面每个接口都重复写 header 和错误处理。
python# config.py
# 所有占位符需替换为你的真实凭证,凭证在平台文档/控制台获取
BASE = "https://你的接口域名" # 注册后在官方文档获取
TOKEN = "你的Token"
APPID = "你的appId"
HEADERS = {"token": TOKEN} # 鉴权字段名以官方文档为准
python# client.py
import requests
from config import BASE, HEADERS
def post(path: str, body: dict) -> dict:
"""
统一 POST 封装。
所有接口均为 POST + JSON body,鉴权 token 放请求头。
返回 {"ret": 200, "msg": "操作成功", "data": {...}}
"""
url = f"{BASE}{path}"
try:
resp = requests.post(url, json=body, headers=HEADERS, timeout=10)
resp.raise_for_status()
result = resp.json()
except requests.exceptions.Timeout:
return {"ret": -1, "msg": "请求超时", "data": None}
except requests.exceptions.RequestException as e:
return {"ret": -1, "msg": str(e), "data": None}
if result.get("ret") != 200:
print(f"[WARN] 接口返回非200: {result}")
return result
def ok(result: dict) -> bool:
"""快捷判断是否成功"""
return result.get("ret") == 200
这个封装做了三件事:统一拼接 URL、注入鉴权 Header、捕获网络层异常。后续所有功能模块都复用它,不会分散错误处理逻辑。
有一个细节值得注意:timeout=10 是请求整体超时,不是连接超时。如果你的服务器和 API 节点距离较远,偶尔会有网络抖动,建议把连接超时和读取超时分开设置,写成 timeout=(5, 15),分别代表"建立连接最多等 5 秒"和"等待响应最多等 15 秒",这样能更精准地区分到底是连不上还是服务端慢响应。
四、登录上线:让微信账号保持在线
托管式 API 的第一步是让微信账号通过扫码登录并保持在线状态。通常分两步:获取二维码 → 轮询确认登录结果。
python# demo_login.py
import time
from client import post, ok
def get_qrcode():
"""获取登录二维码,返回二维码图片 URL 或 base64"""
result = post("/login/getLoginQrCode", {})
if ok(result):
data = result["data"]
print("请用微信扫描以下二维码登录:")
print(data.get("qrImgBase64") or data.get("qrUrl")) # 字段名以文档为准
else:
print("获取二维码失败:", result["msg"])
return result
def wait_for_login(max_wait: int = 120):
"""轮询登录状态,最多等待 max_wait 秒"""
from config import APPID
deadline = time.time() + max_wait
while time.time() < deadline:
result = post("/login/checkLogin", {"appId": APPID})
if ok(result):
status = result["data"].get("status")
if status == 2: # 具体状态码以官方文档为准
print("登录成功!")
return True
elif status == 1:
print("已扫码,等待确认...")
time.sleep(3)
print("登录超时,请重试")
return False
if __name__ == "__main__":
get_qrcode()
wait_for_login()
注意:status 字段的取值含义(如"已扫码"/"已确认"/"已过期")以你所用平台的官方文档为准,上面的 1/2 只是示意。
登录这步有一个常见坑:二维码有有效期,通常是 3 分钟左右。如果用户扫码慢或者网络不好,拿到的二维码可能已经过期,轮询 checkLogin 会一直返回"等待扫码"的状态,让人以为是代码问题。遇到这种情况,应当重新调 getLoginQrCode 刷新二维码,而不是无限等下去。可以在轮询逻辑里加一个二维码刷新策略:超过 90 秒仍未扫码,自动重新拉取二维码并展示。
账号登录之后,平台会维持这个连接,直到你主动注销或者账号被踢下线。如果服务是 7×24 小时运行的,建议额外维护一个心跳检测协程,定时调 checkOnline 接口,确认账号仍然在线,掉线后自动触发重新扫码流程,不然半夜账号悄悄下线、早上才发现消息全部堆积,损失就大了。
五、核心功能:发送文本消息
登录完成后,发消息只需要三个参数:appId、对方微信 ID(toWxid)和消息内容。
python# demo_send.py
from config import APPID
from client import post, ok
def send_text(to_wxid: str, content: str) -> bool:
"""
发送文本消息。
toWxid 为对方微信号,可从联系人列表接口获取,
也可以是群聊的 chatroom_id(通常以 @chatroom 结尾)。
"""
body = {
"appId": APPID,
"toWxid": to_wxid,
"content": content,
}
result = post("/message/postText", body)
if ok(result):
print(f"[OK] 消息已发送 -> {to_wxid}")
return True
else:
print(f"[FAIL] 发送失败: {result['msg']}")
return False
if __name__ == "__main__":
# 替换为真实的微信 ID
TARGET = "wxid_xxxxxxxxxxxxxx"
send_text(TARGET, "Hello from Python!这是我的第一条自动消息。")
运行后如果返回 [OK] 消息已发送,说明整条链路已经打通。
关于 toWxid 字段,有几点需要说明。首先,微信号和 wxid 是两回事:用户设置的微信号(如 "zhangsan123")是给人看的,wxid 才是系统内部的唯一标识(格式通常是 "wxid_" 开头的一串字符)。接口里填的是 wxid,而不是用户自定义的微信号。如果你手上只有微信号,需要先调 search 接口搜索一下,拿到对应的 wxid 再发消息。
其次,发给群聊时,toWxid 填的是群的 chatroom ID,格式通常是一串数字后面跟 @chatroom,不是任何一个成员的 wxid。如果想在群里 @ 某人,部分平台的文本消息接口还支持一个额外的 ats 参数,填入要 @ 的成员 wxid 列表,消息内容里相应位置加 @昵称,组合使用即可实现群 @ 效果。
六、进阶:发图片、文件、链接卡片
文本只是开始,实际场景往往需要更丰富的消息类型。接口风格基本一致,只是 body 字段不同。
6.1 发送图片
pythondef send_image(to_wxid: str, img_url: str) -> bool:
"""
发送图片消息。
img_url 为图片的公网可访问 URL。
注意:批量发图建议先上传一次获取 fileId,
再用转发接口(forwardImage),避免重复上传。
具体字段以官方文档为准。
"""
body = {
"appId": APPID,
"toWxid": to_wxid,
"imgUrl": img_url,
}
result = post("/message/postImage", body)
return ok(result)
6.2 发送链接卡片
链接卡片适合推送文章、产品页等,点击率通常高于纯文本。
pythondef send_link(to_wxid: str, title: str, desc: str, link_url: str, thumb_url: str = "") -> bool:
"""
发送链接卡片消息。
thumbUrl 为卡片左侧缩略图 URL,可选。
具体字段名以官方文档为准。
"""
body = {
"appId": APPID,
"toWxid": to_wxid,
"title": title,
"desc": desc,
"linkUrl": link_url,
"thumbUrl": thumb_url,
}
result = post("/message/postLink", body)
return ok(result)
6.3 各消息类型对比
| 消息类型 | 接口路径(示例) | 主要参数 | 适用场景 |
|---|---|---|---|
| 文本 | /message/postText | content | 通知、问答 |
| 图片 | /message/postImage | imgUrl | 海报、截图 |
| 文件 | /message/postFile | fileUrl, fileName | 报告、资料 |
| 语音 | /message/postVoice | voiceUrl | 语音播报 |
| 链接卡片 | /message/postLink | title, desc, linkUrl | 文章推送 |
以上接口路径均为示例,实际路径和字段名以官方文档为准。
关于图片发送还有一个重要细节:如果你要把同一张图发给多个人,不要每次都传图片 URL 让平台重复下载上传,这样不仅慢,还会消耗不必要的流量和配额。正确的做法是第一次发完之后,从返回的 data 里拿到 fileId(部分平台叫 newMsgId 或其他字段),之后对其他收件人改用 forwardImage 接口转发,传 fileId 即可,速度快很多。文件类消息也遵循同样的策略。
七、接收消息:配置回调地址
主动发消息解决了"推"的问题,但机器人通常还需要"收"——也就是响应用户输入。托管式 API 的做法是回调(Webhook):平台把收到的消息 POST 到你预先设定的服务器地址。
7.1 设置回调地址
pythondef set_callback(callback_url: str) -> bool:
"""
向平台注册你的回调地址。
callback_url 必须是公网可访问的 HTTPS 地址。
本地开发可用 ngrok 做内网穿透。
具体参数以官方文档为准。
"""
body = {
"appId": APPID,
"callbackUrl": callback_url,
}
result = post("/login/setCallback", body)
return ok(result)
7.2 用 Flask 搭一个极简回调服务
python# callback.py
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route("/wechat/callback", methods=["POST"])
def on_message():
"""
平台把消息推送到这里。
消息结构示例(以实际文档为准):
{
"appId": "xxx",
"fromWxid": "wxid_xxx",
"toWxid": "wxid_yyy",
"type": 1, // 1=文本, 3=图片 等,具体以文档为准
"content": "你好",
"msgId": "xxxxxxxx",
"createTime": 1718000000
}
注意:主动发出的消息不会触发回调,只有收到的消息才会。
"""
data = request.get_json(silent=True) or {}
from_wxid = data.get("fromWxid", "")
msg_type = data.get("type", 0)
content = data.get("content", "")
print(f"收到消息 | 来自: {from_wxid} | 类型: {msg_type} | 内容: {content}")
# 简单的回显逻辑:收到文本就原路回复
if msg_type == 1 and content:
from demo_send import send_text
send_text(from_wxid, f"[自动回复] 你说了:{content}")
# 必须返回 200,否则平台会重试
return jsonify({"ret": 200, "msg": "ok"}), 200
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080)
7.3 回调服务的常见坑
坑一:回调地址必须公网可达。 本地 localhost 平台推不进来,开发阶段用 ngrok 或 frp 做内网穿透。ngrok 免费版每次重启 URL 会变,注意重启后要重新调 setCallback 更新地址。
坑二:接口必须在 1~3 秒内返回 200。 如果你的业务逻辑比较重(比如要查数据库、调 LLM 接口),不能在回调里同步处理,否则超时导致平台认为推送失败并重试,同一条消息会被推送多次。正确做法是先把消息扔进队列(哪怕只是一个简单的 Python queue.Queue),立即返回 200,由另一个线程或协程异步消费处理。
坑三:主动发出的消息不会触发回调。 只有你的账号真实收到的消息才会被推送,自己发出的那条不会。很多人第一次测试时觉得"怎么只收到了一半消息",其实是把发出去的消息也算进去了。
坑四:回调重试的幂等处理。 即使你返回了 200,极端情况下(网络抖动导致平台没收到你的响应)平台仍可能重推。回调处理逻辑最好对 msgId 做去重判断,避免同一条消息被处理两遍,比如用 Redis SET 或者本地内存字典记录最近 N 条 msgId。
八、频率控制:怎么做才不容易触发风控
这是很多人忽略但非常关键的一环。微信有行为风控机制,接口调用过快或行为异常容易导致账号受限。
pythonimport time
import random
def safe_send(to_wxids: list, content: str, min_interval: float = 3.0, max_interval: float = 8.0):
"""
批量发消息时加随机间隔,模拟人工操作节奏。
min/max_interval 单位为秒。
"""
for wxid in to_wxids:
send_text(wxid, content)
sleep_time = random.uniform(min_interval, max_interval)
print(f"等待 {sleep_time:.1f}s 后发下一条...")
time.sleep(sleep_time)
几条实用原则:
| 操作 | 建议频率 | 备注 |
|---|---|---|
| 加好友 | 每 2h ≤5 个,每天 5-15 个 | 新号需在线 3 天后再操作 |
| 批量发消息 | 每条间隔 3-8 秒(随机) | 固定间隔更容易被识别 |
| 获取联系人 / 群成员 | 按需调用,非高频轮询 | 避免在短时间内大量拉取 |
| 朋友圈点赞评论 | 随机间隔 5-20 秒 | 新号需在线 1 天后再操作 |
风控是一个概率问题,不是绝对阈值。同样的操作频率,老号比新号抗风险能力强,历史行为正常的账号比活跃度异常的账号更不容易被限。因此,新号前三天一定不要做任何自动化操作,先让账号"暖"起来,有正常的聊天记录、朋友圈互动,之后再逐步开启自动化功能,从低频到高频慢慢加量,不要上来就拉满。
另外,消息内容的同质化也是风控信号之一。如果你向一百个人发完全相同的内容,被检测到群发特征的概率远高于每条消息有细微差异的情况。可以在消息里插入动态变量,比如加上收件人的昵称、发送时间等个性化内容,降低相似度。
九、常见错误排查
| 现象 | 可能原因 | 排查方向 |
|---|---|---|
ret 非 200,msg 提示鉴权失败 | Token 错误或已过期 | 检查 config.py,重新从平台控制台获取 |
| 发消息返回成功但对方收不到 | 账号未在线 / toWxid 填错 | 调 checkOnline 接口确认在线状态;确认填的是 wxid 而非微信号 |
| 回调一直收不到消息 | 回调地址不可达 | 用 curl 从外网测试你的地址是否返回 200 |
| 接口调用频率过高报错 | 触发速率限制 | 加随机 sleep,降低调用密度 |
| 操作敏感接口返回失败 | 账号在线天数不足 | 等账号在线满 3 天后再调用 |
| 回调消息重复收到 | 超时后平台重推 | 对 msgId 做去重,接口必须在 3 秒内返回 200 |
| 二维码扫了没反应 | 二维码已过期 | 重新调 getLoginQrCode 刷新二维码 |
| 发图片比发文本慢很多 | 平台每次重新下载图片 | 第一次发完拿 fileId,后续改用 forwardImage 转发 |
十、小结
走到这里,你已经打通了整个微信消息收发的基础链路:从技术路线选型、环境搭建、HTTP 客户端封装,到登录上线、发文本/图片/链接卡片,再到 Webhook 回调接收消息,以及频率控制和常见错误的排查思路。
这套基础框架的价值在于它的可扩展性。文本消息打通之后,群管理、朋友圈操作、联系人同步这些功能只需要按照同样的接口风格继续叠加即可,不需要重写底层。callback.py 里的回调逻辑可以扩展成完整的消息路由,根据消息类型和来源分发给不同的处理器;safe_send 的频率控制可以升级成带令牌桶的限速队列;client.py 可以加入重试机制和日志记录。
每增加一项功能,你都会对微信的行为模型理解更深,踩坑也会越来越少。
