运行自动化
Webhook
任务结束或手机需要处理时,向你的服务器发送回调。
添加Webhook
在控制台中:设置 → 连接 → 事件 → Webhook。或通过API:
订阅
POST /v1/webhooks
{"url": "https://example.com/clousd", "events": ["run.completed", "run.failed"], "description": "orders bot"}
# → 201 {"webhook": {"id": "wh_3c1a", …}, "secret": "whsec_…"} the secret is shown onceevents留空表示订阅全部事件。一个密钥最多可有十个Webhook,每个都有自己的签名密钥。POST /v1/webhooks/{id}/rotate生成新的签名密钥,/toggle暂停投递,/test立即发送一个测试事件,GET …/deliveries列出最近的投递记录及其响应码和耗时。
事件
| 事件 | 触发时机 |
|---|---|
| run.completed | 任务已完成;请求体中包含该运行及其答案。 |
| run.failed | 任务未能完成;请求体中说明了原因。 |
| run.cancelled | 任务已被停止。 |
| run.waiting | 任务暂停,等待你回答问题或授权。 |
| device.problem | 手机健康检查未通过。 |
| device.resolved | 问题已解除。 |
验证签名
每次投递都是一个带JSON请求体的POST,请求头为X-Clousd-Signature: t=<unix time>,v1=<hex>,其中hex是以你的Webhook签名密钥对"<t>.<raw body>"计算的HMAC-SHA256。拒绝过旧的时间戳,以防重放攻击。
Python
import hashlib, hmac, time
def valid(body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
parts = dict(p.split("=", 1) for p in header.split(","))
t, sig = parts.get("t", ""), parts.get("v1", "")
if not t.isdigit() or abs(time.time() - int(t)) > tolerance:
return False
mac = hmac.new(secret.encode(), t.encode() + b"." + body, hashlib.sha256).hexdigest()
return hmac.compare_digest(mac, sig)重试
每个事件最多投递四次:立即一次,然后在30秒、5分钟和30分钟后各一次。返回任意2xx即表示确认。请求体为{"id", "type", "created", "data"},同一个id也会出现在X-Clousd-Event-Id中:如果在一次较慢的成功之后又收到重试,跳过已处理过的id即可。连续20次投递失败后,Webhook会自动暂停;可在控制台中或用/toggle重新开启。