File size: 4,791 Bytes
0e3d4b8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Webhook support — receive incoming webhooks and send outgoing webhooks.
Incoming: FastAPI endpoint that receives webhooks and routes them to the LLM.
Outgoing: Send webhook notifications when events happen (goal completed, etc.).
Webhooks allow external systems to trigger the LLM or be notified by it.
"""
from __future__ import annotations
import hashlib
import hmac
import json
import logging
import time
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Callable
logger = logging.getLogger(__name__)
@dataclass
class WebhookEndpoint:
"""A registered incoming webhook endpoint."""
path: str
secret: str = ""
handler: Callable[[dict], dict] | None = None
calls: int = 0
last_call: float = 0.0
class WebhookManager:
"""Manages incoming and outgoing webhooks.
Incoming webhooks are registered with a path and optional secret.
When a POST request hits the path, the handler is called.
Outgoing webhooks are sent to external URLs to notify them of events.
"""
def __init__(self) -> None:
self._endpoints: dict[str, WebhookEndpoint] = {}
self._outgoing: list[dict[str, str]] = []
self._stats = {
"incoming_received": 0,
"incoming_processed": 0,
"incoming_rejected": 0,
"outgoing_sent": 0,
"outgoing_failed": 0,
}
def register_endpoint(self, path: str, handler: Callable[[dict], dict],
secret: str = "") -> None:
"""Register an incoming webhook endpoint."""
self._endpoints[path] = WebhookEndpoint(path=path, secret=secret, handler=handler)
logger.info("Registered webhook endpoint: %s", path)
def handle_request(self, path: str, body: dict, signature: str = "") -> dict:
"""Handle an incoming webhook request.
Args:
path: webhook path
body: request body
signature: HMAC signature for verification
Returns:
Response dict
"""
self._stats["incoming_received"] += 1
endpoint = self._endpoints.get(path)
if endpoint is None:
self._stats["incoming_rejected"] += 1
return {"error": "Unknown webhook endpoint", "path": path}
# Verify signature if secret is set
if endpoint.secret:
expected = hmac.new(
endpoint.secret.encode(),
json.dumps(body, sort_keys=True).encode(),
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(expected, signature):
self._stats["incoming_rejected"] += 1
return {"error": "Invalid signature"}
endpoint.calls += 1
endpoint.last_call = time.time()
try:
result = endpoint.handler(body) if endpoint.handler else {"received": True}
self._stats["incoming_processed"] += 1
return result
except Exception as e:
logger.error("Webhook handler error: %s", e)
return {"error": str(e)}
def send(self, url: str, event: str, data: dict, secret: str = "") -> bool:
"""Send an outgoing webhook notification.
Args:
url: target URL
event: event name
data: payload
secret: HMAC secret for signing
Returns:
True if sent successfully
"""
payload = json.dumps({"event": event, "data": data, "timestamp": time.time()}).encode()
headers = {"Content-Type": "application/json"}
if secret:
signature = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
headers["X-Webhook-Signature"] = signature
try:
req = urllib.request.Request(url, data=payload, method="POST", headers=headers)
with urllib.request.urlopen(req, timeout=10) as resp:
if resp.status < 300:
self._stats["outgoing_sent"] += 1
self._outgoing.append({"url": url, "event": event, "time": str(time.time())})
logger.info("Webhook sent to %s: %s", url, event)
return True
except Exception as e:
logger.error("Webhook send failed: %s", e)
self._stats["outgoing_failed"] += 1
return False
def notify_event(self, event: str, data: dict) -> None:
"""Notify all registered outgoing webhooks of an event."""
for hook in list(self._outgoing):
self.send(hook["url"], event, data)
def list_endpoints(self) -> list[str]:
return list(self._endpoints.keys())
def get_stats(self) -> dict[str, Any]:
return {**self._stats, "endpoints": len(self._endpoints)}
|