Appearance
本地联调
这条路径适合还没有接收程序或公网地址的开发者。准备 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。
回到控制台「测试消息收发」:
- 在 Webhook URL 中填入完整地址,点击「保存并启用 Webhook」。
- 首次创建后,保存页面显示的 Webhook 密钥。它用于下一步启动接收程序,与调用 API 的凭证不同。
- 已有配置使用以前保存的 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 |
| 返回 401 | Webhook 密钥是否正确;它不是 API 凭证,更新地址也不会更新密钥 |
| 测试通过,没有正文 | 账号是否在线;是否从另一个企微账号的客户端发送了控制台显示的文字 |
| 同一事件重试 | 程序按 X-Eyun-Event-Id 去重;X-Eyun-Delivery 只用于排错 |
其他问题见回调排查。
已有 API 接入,或只想查看请求内容
用 API 配置与测试的示例见回调说明。测试接口需同时满足 data.ok=true 和 data.httpStatus 为 200~299。
在线请求接收器可用于测试账号查看请求,但不能代替自己服务的签名校验、消息保存和真实接收验证。要完成控制台首次接入,请按本页四步执行。