Send HTTP POST notifications to external endpoints when pipeline events occur using WebhookConfig and WebhookDispatcher
The webhook system builds on top of the EventBus to send HTTP POST requests with JSON payloads to external endpoints when pipeline events occur. Webhook deliveries happen in background threads and never block the approval pipeline. Zero external dependencies — uses urllib.request from stdlib.
from attesta import Attestafrom attesta.events import EventBus, EventTypefrom attesta.webhooks import WebhookConfig, WebhookDispatcher# Set up event busbus = EventBus()# Configure webhooksslack_hook = WebhookConfig( url="https://hooks.slack.com/services/T00/B00/xxx", events=[EventType.DENIED], # Only notify on denials)audit_hook = WebhookConfig( url="https://audit.internal.company.com/attesta", events=[], # All events secret="shared-secret-123", retry_count=3,)# Create dispatcher (auto-subscribes to event bus)dispatcher = WebhookDispatcher(bus, [slack_hook, audit_hook])# Pass the event bus to Attestaattesta = Attesta(event_bus=bus)
Webhook deliveries run in daemon threads. If the main process exits before a delivery completes, the webhook may not be sent. For critical notifications, consider increasing the timeout and retry_count values.