三种通信方式

开发微信机器人，搞清楚"怎么主动发"和"怎么被动收"是第一步。目前主流API框架都用这三种方式组合：

方式	方向	典型用途
REST API	你主动 → 微信	发送消息、查好友、发朋友圈等
Webhook	微信 → 你的服务	接收微信推送的消息事件
WebSocket	双向长连接	部分框架提供实时订阅能力

REST API：所有"主动操作"都走它

REST API就是普通的HTTP请求，你想让微信做什么，就POST或GET对应接口。

发消息

import requests, json

BASE = "https://wx.chuapi.com"
TOKEN = "your_token"
APP_ID = "your_app_id"

def send_text(to_wxid, content):
    resp = requests.post(
        f"{BASE}/finder/v2/api/message/postText",
        headers={"X-finder-TOKEN": TOKEN, "Content-Type": "application/json"},
        data=json.dumps({
            "appId": APP_ID,
            "toWxid": to_wxid,
            "content": content
        })
    )
    return resp.json()

发图片

def send_image(to_wxid, image_url):
    resp = requests.post(
        f"{BASE}/finder/v2/api/message/postImage",
        headers={"X-finder-TOKEN": TOKEN, "Content-Type": "application/json"},
        data=json.dumps({
            "appId": APP_ID,
            "toWxid": to_wxid,
            "imgUrl": image_url  # 图片URL，服务端自动下载上传
        })
    )
    return resp.json()

获取好友列表

def get_contacts():
    resp = requests.get(
        f"{BASE}/finder/v2/api/contact/getContactList",
        headers={"X-finder-TOKEN": TOKEN},
        params={"appId": APP_ID}
    )
    return resp.json()

获取群列表

def get_groups():
    resp = requests.get(
        f"{BASE}/finder/v2/api/chatroom/getChatRoomList",
        headers={"X-finder-TOKEN": TOKEN},
        params={"appId": APP_ID}
    )
    return resp.json()

所有接口统一要求：Header带 X-finder-TOKEN，Body是JSON格式。返回值格式统一：

{
    "ret": 200,           // 200=成功，其他=失败
    "msg": "操作成功",
    "data": { ... }       // 业务数据
}

Webhook：被动接收微信消息

你不可能一直轮询"有没有新消息"——微信有新消息时，API平台直接POST推送到你配置的回调URL。

配置回调

在API平台控制台找到Webhook配置，填入你的回调地址，比如 https://your-domain.com/callback。

写一个回调接口

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/callback", methods=["POST"])
def on_message():
    """
    接收微信消息回调
    
    微信用户发消息 → 微信服务器 → API平台解密 → 你的回调URL
    """
    data = request.json
    
    # 关键字段
    msg_type = data.get("msgType")       # text / image / emoji / link / appmsg...
    from_user = data.get("fromUser")     # 发送者wxid
    nick_name = data.get("nickName")     # 发送者昵称
    content = data.get("content", "")    # 消息内容
    chat_room = data.get("chatRoomId")   # 群聊ID（群消息才有）
    create_time = data.get("createTime") # 时间戳
    
    # 按类型分发处理
    if msg_type == "text":
        handle_text(from_user, content, chat_room)
    elif msg_type == "image":
        handle_image(from_user, data.get("imageUrl"))
    elif msg_type == "revoke":
        handle_revoke(data)
    
    # 必须返回ret:200，否则平台会重推
    return jsonify({"ret": 200})

def handle_text(from_user, content, chat_room):
    print(f"[{from_user}] → {content}")
    # 业务逻辑：关键词匹配、调用AI、入库...
    pass

回调数据长什么样？

实际推送过来的JSON结构（WTAPI）：

{
    "appId": "wx_e2PiMSX8ySDV6tQGroCDc",
    "msgType": "text",
    "fromUser": "wxid_tyyu4v9ykz3712",
    "nickName": "张三",
    "content": "你好",
    "createTime": 1725672000,
    "chatRoomId": "",
    "chatRoomNickName": ""
}

群聊消息会多两个字段：chatRoomId（群的wxid）和 chatRoomNickName（发送者在群里的昵称）。

WebSocket：实时订阅（部分平台提供）

有些框架提供WebSocket长连接，你可以订阅特定事件，比如：

import websocket

def on_message(ws, message):
    data = json.loads(message)
    print(f"收到事件: {data}")

def on_open(ws):
    # 订阅所有消息事件
    ws.send(json.dumps({
        "action": "subscribe",
        "events": ["message", "friend", "chatroom"]
    }))

ws = websocket.WebSocketApp(
    "wss://ws.chuapi.com/ws?token=your_token",
    on_message=on_message,
    on_open=on_open
)
ws.run_forever()

完整通信流程图

                    主动操作（REST API）
        ┌─────────────────────────────────────────┐
        │                                         │
你的代码 │  POST /message/postText  →  API平台   │
        │  POST /sns/sendSns        →  API平台   │
        │  GET  /contact/getList     →  API平台   │
        └────────────────────┬────────────────────┘
                             │ 协议加密解密
                             ▼
                        微信服务器
                             ▲
                             │ 消息推送（Webhook）
        ┌────────────────────┴────────────────────┐
        │                                         │
你的服务 │  Flask接收POST → 业务处理 → 调用REST回复 │
        │  ngrok/内网穿透 → 公网可访问              │
        └─────────────────────────────────────────┘

实战：两条通道串起来

# ====== 主动发送 ======
def send_text(to_wxid, content):
    requests.post(
        f"{BASE}/finder/v2/api/message/postText",
        headers={"X-finder-TOKEN": TOKEN, "Content-Type": "application/json"},
        data=json.dumps({"appId": APP_ID, "toWxid": to_wxid, "content": content})
    )

# ====== 被动接收 ======
@app.route("/callback", methods=["POST"])
def on_message():
    data = request.json
    if data.get("msgType") == "text":
        # 收到消息 → 处理 → 回复
        send_text(data["fromUser"], f"收到：{data['content']}")
    return jsonify({"ret": 200})

就这两段，一个发一个收，机器人的骨架就搭好了。

注意事项

1. REST API的Header必须带 X-finder-TOKEN，别漏
2. 回调接口必须返回 {"ret": 200}，超时或不返回平台会重推
3. 回调处理别做耗时操作（比如调AI、写数据库），放异步队列
4. 同一条消息可能收到多次回调，做一下去重（用createTime + fromUser做key）
5. 本地开发要用内网穿透工具，API平台必须能访问到你的回调地址