67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Webhook delivery helpers for MACP events."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
|
|
def _warn(message: str) -> None:
|
|
print(f"[macp-webhook] {message}", file=sys.stderr)
|
|
|
|
|
|
def _webhook_config(config: dict[str, Any]) -> dict[str, Any]:
|
|
macp = config.get("macp")
|
|
if isinstance(macp, dict) and isinstance(macp.get("webhook"), dict):
|
|
return dict(macp["webhook"])
|
|
return dict(config)
|
|
|
|
|
|
def send_webhook(event: dict[str, Any], config: dict[str, Any]) -> bool:
|
|
"""POST event to webhook URL. Returns True on success."""
|
|
|
|
webhook = _webhook_config(config)
|
|
url = str(webhook.get("url") or "").strip()
|
|
if not url:
|
|
_warn("missing webhook url")
|
|
return False
|
|
|
|
timeout_seconds = max(1.0, float(webhook.get("timeout_seconds") or 10))
|
|
retry_count = max(0, int(webhook.get("retry_count") or 0))
|
|
auth_token = str(webhook.get("auth_token") or "").strip()
|
|
payload = json.dumps(event, ensure_ascii=True).encode("utf-8")
|
|
headers = {"Content-Type": "application/json"}
|
|
if auth_token:
|
|
headers["Authorization"] = f"Bearer {auth_token}"
|
|
|
|
attempts = retry_count + 1
|
|
for attempt in range(1, attempts + 1):
|
|
request = urllib.request.Request(url, data=payload, headers=headers, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
|
|
status = getattr(response, "status", response.getcode())
|
|
if 200 <= int(status) < 300:
|
|
return True
|
|
_warn(f"webhook returned HTTP {status} on attempt {attempt}/{attempts}")
|
|
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, ValueError) as exc:
|
|
_warn(f"webhook attempt {attempt}/{attempts} failed: {exc}")
|
|
if attempt < attempts:
|
|
time.sleep(min(timeout_seconds, 2 ** (attempt - 1)))
|
|
return False
|
|
|
|
|
|
def create_webhook_callback(config: dict[str, Any]) -> Callable[[dict[str, Any]], None]:
|
|
"""Factory that creates a watcher callback from config."""
|
|
|
|
def callback(event: dict[str, Any]) -> None:
|
|
if not send_webhook(event, config):
|
|
_warn(f"delivery failed for event {event.get('event_type', '<unknown>')}")
|
|
|
|
return callback
|