Advanced: Webhook Event Notifications

pull (polling):   you → platform   "done yet?"   ×N times
push (Webhook):   platform → you   "done"        ×1 time

curl -s -X POST "$AGENTSTUDIO_URL/webhook_endpoints" \
  -H "Authorization: Bearer $DASHSCOPE_API_KEY" -H "Content-Type: application/json" \
  -d '{
    "description": "拾贝电商 对账任务通知",
    "url": "http://118.178.120.71:8080/hook",
    "events": ["session.status_run_started", "session.status_idled", "deployment_run.started", "deployment_run.succeeded", "deployment_run.failed"]
  }'

{
  "id": "wep_01M1G63CXHX6C7C1801RD2J19W",
  "url": "http://118.178.120.71:8080/hook",
  "events": ["session.status_run_started", "session.status_idled", "..."],
  "status": "ACTIVE",
  "consecutive_fail": 0,
  "last_success_at": null,
  "signing_secret": "whsec_XgYOlI...",
  "created_at": "..."
}

#!/usr/bin/env python3
# receiver.py —— Webhook receiver
import json, os, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

LOG = "/root/webhook_hits.jsonl"

class H(BaseHTTPRequestHandler):
    def _send(self, code, body):
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        n = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(n)
        if self.path != "/hook":
            return self._send(404, b'{"error":"not found"}')
        rec = {"ts": time.time(), "path": self.path,
               "headers": dict(self.headers.items()),
               "body": raw.decode("utf-8", "replace")}
        with open(LOG, "a") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        self._send(200, b'{"ok":true}')

    def do_GET(self):
        if self.path == "/health":
            return self._send(200, b'{"status":"up"}')
        if self.path == "/dump":
            data = open(LOG).read() if os.path.exists(LOG) else ""
            return self._send(200, data.encode())
        if self.path == "/clear":
            open(LOG, "w").close()
            return self._send(200, b'{"cleared":true}')
        self._send(404, b'{"error":"not found"}')

    def log_message(self, *a):
        pass

ThreadingHTTPServer(("0.0.0.0", 8080), H).serve_forever()

# /etc/systemd/system/webhook-receiver.service
[Unit]
Description=Bailian webhook receiver
After=network.target

[Service]
ExecStart=/usr/bin/python3 /root/receiver.py
Restart=always
RestartSec=2

[Install]
WantedBy=multi-user.target

systemctl enable --now webhook-receiver

curl -s -X POST "$AGENTSTUDIO_URL/webhook_endpoints/wep_xxx/test" -H "Authorization: Bearer $DASHSCOPE_API_KEY"

{"type": "event", "id": "whe_01M1G6C24QPR2V9J2QTKFXJCTH",
 "created_at": "2026-09-02T04:34:54.743Z",
 "data": {"id": "wep_01M1G63CXHX6C7C1801RD2J19W", "type": "webhook.test", "workspace_id": "ws_xxx"}}

POST /hook HTTP/1.1
User-Agent: Bailian-ManagedAgent-Webhook/1.0
webhook-id: whe_01M1G69TMRHC5ECD4PFE598GYF
webhook-timestamp: 1788323621
webhook-signature: v1,naL38RUIYmTbeQuXfXTJSigDA6J42dNA6Aa4R+1mKK4=
Content-Type: application/json; charset=utf-8

{"type":"event","id":"whe_01M1G69TMRHC5ECD4PFE598GYF","created_at":"2026-09-02T04:33:41.508Z",
 "data":{"id":"sesn_01M1G5RPKRFHPTB72W2HG05KDM","type":"session.status_idled","workspace_id":"llm-czal8nvvwb8d47ks"}}

signed_content = webhook-id + "." + webhook-timestamp + "." + raw_body
key             = Base64Decode( secret with the whsec_ prefix stripped )   # not the whole whsec_ string as the key
signature       = Base64( HMAC-SHA256(key, UTF8(signed_content)) )

import json, base64, hmac, hashlib, sys

secret = sys.argv[1]
secret_bytes = base64.b64decode(secret[len("whsec_"):])   # strip prefix, then decode

for line in open("/root/webhook_hits.jsonl"):
    rec = json.loads(line)
    h = rec["headers"]
    wid, ts, sig = h["webhook-id"], h["webhook-timestamp"], h["webhook-signature"]
    version, signature = sig.split(",", 1)               # "v1,<base64>"
    signed = f"{wid}.{ts}.".encode("utf-8") + rec["body"].encode("utf-8")
    expected = base64.b64encode(
        hmac.new(secret_bytes, signed, hashlib.sha256).digest()).decode()
    print(h["webhook-id"], hmac.compare_digest(signature, expected))

# The health triad: last_success_at / last_failure_at / consecutive_fail
curl -s "$AGENTSTUDIO_URL/webhook_endpoints" -H "Authorization: Bearer $DASHSCOPE_API_KEY" | python3 -m json.tool

{"id": "wep_...", "status": "ACTIVE", "disabled_reason": null,
 "last_success_at": "2026-09-01T19:53:00.350Z", "last_failure_at": null,
 "consecutive_fail": 0}