Eyun 开发者文档

Appearance

Webhook 本地联调 ​

这条路径适合还没有接收程序或公网地址的开发者。准备 Python 3 和一个内网穿透工具，按本页完成配置与测试，不用先编写业务代码。

开始前，在控制台完成扫码和发送测试，保持收发测试页面打开。本页用于控制台 Webhook；已经使用 /setHttpCallbackUrl 的项目仍按兼容接口说明接入。

下面只配置测试账号的接收地址；已有正式配置时，更新地址会改变消息去向，请使用独立测试应用。

1. 保存接收程序 ​

新建文件夹，把下方代码保存为 webhook_receiver.py，先不用运行。只需 Python 3，无额外依赖。程序会按当前配置校验来源、去掉重复事件，将消息保存到本地数据库后再返回成功。

完整代码：保存为 webhook_receiver.py
python
import getpass
import hmac
import json
import os
import sqlite3
from http.server import BaseHTTPRequestHandler, HTTPServer

secret = os.environ.get('EYUN_WEBHOOK_SECRET')
if secret is None:
    secret = getpass.getpass('Webhook secret (press Enter only if signing is disabled): ')

db = sqlite3.connect('personal-webhook-receipts.sqlite3', timeout=1)
db.execute("""
    CREATE TABLE IF NOT EXISTS receipts (
        event_id TEXT PRIMARY KEY,
        delivery TEXT NOT NULL,
        instance_id TEXT,
        event TEXT NOT NULL,
        body BLOB NOT NULL
    )
""")
db.commit()


class Receiver(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path != '/webhook':
            return self.reply(404)
        try:
            length = int(self.headers.get('Content-Length', '0'))
        except ValueError:
            return self.reply(400)
        if not 0 < length <= 1024 * 1024:
            return self.reply(413)
        self.connection.settimeout(5)
        try:
            raw = self.rfile.read(length)
        except TimeoutError:
            return self.reply(408)
        signature = self.headers.get('X-Eyun-Signature', '')
        if secret and not hmac.compare_digest(secret.encode(), signature.encode()):
            return self.reply(401)
        event_id = self.headers.get('X-Eyun-Event-Id')
        delivery = self.headers.get('X-Eyun-Delivery-Attempt', '')
        event = self.headers.get('X-Eyun-Event')
        if not event_id or not event or len(raw) != length:
            return self.reply(400)
        try:
            body = json.loads(raw)
            if not isinstance(body, dict):
                return self.reply(400)
            instance_id = body.get('instanceId')
        except (ValueError, UnicodeDecodeError):
            return self.reply(400)
        try:
            with db:
                saved = db.execute(
                    'INSERT OR IGNORE INTO receipts VALUES (?, ?, ?, ?, ?)',
                    (event_id, delivery, instance_id, event, raw),
                ).rowcount
        except sqlite3.Error:
            return self.reply(500)
        self.reply(200)
        print(json.dumps({
            'event': event, 'eventId': event_id, 'delivery': delivery, 'saved': bool(saved)
        }), flush=True)

    def reply(self, status):
        self.send_response(status)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', '2')
        self.end_headers()
        self.wfile.write(b'{}')

    def log_message(self, format, *args):
        pass


port = int(os.environ.get('EYUN_WEBHOOK_PORT', '3000'))
print(f'Listening on localhost:{port}/webhook', flush=True)
HTTPServer(('localhost', port), Receiver).serve_forever()

2. 取得地址，在控制台保存 ​

Eyun 不能直接访问你电脑上的 localhost。需要用内网穿透工具把本机 3000 端口提供给公网访问；取得公网地址后，在末尾加上接收路径。

还没有内网穿透工具？以 ngrok 为例

按 ngrok 官方入门说明完成安装和账号连接。在另一个终端运行：

bash
ngrok http 3000

复制工具显示的 HTTPS 公网地址，例如 https://your-domain.ngrok-free.app。ngrok 的账号凭证与 Eyun API 凭证不同。已有其他内网穿透工具时，也可以用它提供同样的公网转发。

回调地址要允许服务器直接发送 POST，不能要求先通过浏览器登录。这里使用接收程序校验来源；不需要配置该入门教程中的 Google 登录示例。

本例接收路径为 /webhook。例如工具给出的地址是 https://your-domain.ngrok-free.app，应填入的完整地址就是 https://your-domain.ngrok-free.app/webhook。

回到控制台「测试消息收发」，在 Webhook URL 中填入完整地址，点击「保存并启用 Webhook」。页面显示新密钥时，先单独保存；已有配置继续使用原密钥。Webhook 密钥与调用 API 的凭证不同。

保持内网穿透运行；地址变化后，在控制台更新并重新测试。

3. 启动程序，点击测试 ​

在保存代码的文件夹打开终端，运行：

bash
python3 webhook_receiver.py

程序询问 Webhook secret 时：已开启来源校验的回调，粘贴自己保存的 Webhook 密钥并回车；只有确认当前回调没有开启来源校验时，才直接回车。 输入时屏幕不会显示密钥，这是正常的。密钥遗失不能用跳过校验代替恢复。

个人微信当前的 X-Eyun-Signature 直接等于配置的密钥，程序已按此核对；它不使用企微的 HMAC 算法。完整规则见个人微信回调说明。

看到 Listening on localhost:3000/webhook 后，保持终端运行，回到控制台：

  1. 页面显示过新密钥时，勾选「我已保存 Webhook 密钥」。
  2. 如果提示未开启接收记录，阅读说明后点击「同意开启接收记录」。开启后会保存消息正文，供控制台核对，需要你明确同意。
  3. 点击「发送测试投递」。控制台显示测试成功，且终端显示 webhook.test，才继续下一步。 程序目录会生成 personal-webhook-receipts.sqlite3，保存收到的内容。

4. 查看收到的消息 ​

用另一个微信账号，在手机上给当前登录账号发送控制台显示的验证文字。终端显示 message.received 后，在同一个文件夹的另一终端运行：

bash
python3 - <<'PYTHON'
import json
import sqlite3

with sqlite3.connect('file:personal-webhook-receipts.sqlite3?mode=ro', uri=True) as db:
    row = db.execute(
        "SELECT body FROM receipts WHERE event = ? ORDER BY rowid DESC LIMIT 1",
        ('message.received',),
    ).fetchone()

if row:
    print(json.dumps(json.loads(row[0]), ensure_ascii=False, indent=2))
else:
    print('还没有收到消息，请从另一个微信账号的客户端发送一条文本。')
PYTHON

展开 data，找到 content，核对是否与刚发的文字相同。普通私聊文本的 messageType 为字符串 "60001"。用 API 发消息不能代替这一步。

确认读到正文后，回到控制台点击「接收端已读到这条真实正文」。页面显示「接入完成」后，继续用 curl 调用接口。

本地示例只用于开发

单次接收上限为 1 MB，这是示例自身的限制。数据库保存原始消息，请妥善保管，不要提交到代码仓库。停止程序或内网穿透后，地址会停止接收。正式上线前还需做好消息保存、按账号内消息 ID 去重和后台处理，使用稳定的 HTTPS 地址并重新测试。

检查与上线 ​

遇到的问题先检查什么
测试无法送达接收程序和内网穿透是否都在运行，地址末尾是否包含 /webhook
返回 401当前回调的 Webhook 密钥是否正确；不要填写 API 凭证或套用企微签名算法
测试通过，没有正文账号是否在线；是否从另一个微信账号的客户端发送了控制台显示的文字
同一事件重试程序按 X-Eyun-Event-Id 去重；普通私聊正文读取 data.content
只想查看请求内容，或沿用旧接口

在线请求接收器可以用于测试账号查看内容，但不能代替自己的接收服务验证。要完成控制台首次接入，请按本页四步执行。

旧项目使用的 /setHttpCallbackUrl 与控制台 Webhook 是不同配置入口，参数、测试行为和推送规则见兼容接口说明。

正式上线前，按上线检查清单逐项核对。