Skip to content

本地联调 ​

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

开始前,在控制台完成扫码和发送测试,保持收发测试页面打开。下面只配置测试账号的接收地址;已有正式配置时,更新地址会改变消息去向,请使用独立测试应用。

1. 保存接收程序 ​

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

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

secret = os.environ.get('EYUN_WEBHOOK_SECRET') or getpass.getpass('Webhook secret: ')
if not secret:
    raise SystemExit('Webhook secret is required')

db = sqlite3.connect('webhook-event-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 != '/hook':
            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)
        expected = 'sha256=' + hmac.new(
            secret.encode(), raw, hashlib.sha256
        ).hexdigest()
        signature = self.headers.get('X-Eyun-Signature', '')
        if not hmac.compare_digest(expected.encode(), signature.encode()):
            return self.reply(401)
        event_id = self.headers.get('X-Eyun-Event-Id')
        delivery = self.headers.get('X-Eyun-Delivery')
        instance_id = self.headers.get('X-Eyun-Instance-Id')
        event = self.headers.get('X-Eyun-Event')
        if not event_id or not delivery or not event or len(raw) != length:
            return self.reply(400)
        try:
            json.loads(raw)
        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}/hook', 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 登录示例。

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

回到控制台「测试消息收发」:

  1. 在 Webhook URL 中填入完整地址,点击「保存并启用 Webhook」。
  2. 首次创建后,保存页面显示的 Webhook 密钥。它用于下一步启动接收程序,与调用 API 的凭证不同。
  3. 已有配置使用以前保存的 Webhook 密钥;更新地址不会更换密钥。遗失时先按密钥恢复说明处理。

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

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

在保存代码的文件夹打开终端,运行:

bash
python3 webhook_receiver.py

出现 Webhook secret: 时粘贴上一步保存的 Webhook 密钥并回车。输入时屏幕不会显示密钥,这是正常的。看到 Listening on localhost:3000/hook,说明程序已启动,保持这个终端运行。

回到控制台,新配置勾选「我已保存 Webhook 密钥」,再点击「发送测试投递」。控制台显示测试成功,且终端显示 webhook.test,才继续下一步。 程序目录会生成 webhook-event-receipts.sqlite3,保存收到的内容。

4. 查看收到的消息 ​

用另一个企微账号,在手机上给当前登录账号发送控制台显示的验证文字。终端显示 message.received 后,在同一个文件夹的另一终端运行下面的查看命令:

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

with sqlite3.connect('file:webhook-event-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

展开 content 数组,找到 text,核对是否包含刚发送的相同文字。文本的 contentType 为 0 或 2;2001 是已读回执,不能当作正文。用 API 发消息不能代替这一步。

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

本地示例只用于开发

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

检查与排错 ​

遇到的问题先检查什么
测试无法送达接收程序和内网穿透是否都在运行,地址末尾是否包含 /hook
返回 401Webhook 密钥是否正确;它不是 API 凭证,更新地址也不会更新密钥
测试通过,没有正文账号是否在线;是否从另一个企微账号的客户端发送了控制台显示的文字
同一事件重试程序按 X-Eyun-Event-Id 去重;X-Eyun-Delivery 只用于排错

其他问题见回调排查。

已有 API 接入,或只想查看请求内容

用 API 配置与测试的示例见回调说明。测试接口需同时满足 data.ok=true 和 data.httpStatus 为 200~299。

在线请求接收器可用于测试账号查看请求,但不能代替自己服务的签名校验、消息保存和真实接收验证。要完成控制台首次接入,请按本页四步执行。

下一步 ​