API Reference

Protocols

WebhookDeliveryServiceProtocol

High-level protocol for dispatching webhook events.

async def dispatch(event: WebhookEvent) -> None

Deliver an event to all matching active subscriptions.

Parameters

Parameter	Type	Description
`event`	WebhookEvent	Event to deliver.

async def redeliver(attempt_id: str) -> Any

Redeliver a failed or dead-lettered attempt.

Returns Result[None, WebhookError]. Typed as Any because Result lives in lexigram (not lexigram-contracts).

Parameters

Parameter	Type	Description
`attempt_id`	str	ID of the attempt to redeliver.

Returns

Type	Description
Any	Result[None, WebhookError]

---

WebhookDeliveryStoreProtocol

Storage for webhook delivery attempts and dead-letter queue.

async def record_attempt(attempt: DeliveryAttempt) -> None

Persist a delivery attempt record.

Parameters

Parameter	Type	Description
`attempt`	DeliveryAttempt	Attempt to record.

async def get_attempts(
    *,
    subscription_id: str | None = None,
    event_id: str | None = None,
    status: DeliveryStatus | None = None,
    limit: int = 100,
    offset: int = 0
) -> list[DeliveryAttempt]

Query delivery attempts matching filters, newest-first.

Parameters

Parameter	Type	Description
`subscription_id`	str | None	Filter by subscription.
`event_id`	str | None	Filter by event.
`status`	DeliveryStatus | None	Filter by delivery status.
`limit`	int	Maximum results.
`offset`	int	Results to skip.

Returns

Type	Description
list[DeliveryAttempt]	Matching attempts in newest-first order.

async def get_dead_letters(
    *,
    subscription_id: str | None = None,
    limit: int = 100,
    offset: int = 0
) -> list[DeliveryAttempt]

Return attempts in dead-letter status.

Parameters

Parameter	Type	Description
`subscription_id`	str | None	Optional filter by subscription.
`limit`	int	Maximum results.
`offset`	int	Results to skip.

Returns

Type	Description
list[DeliveryAttempt]	Dead-lettered attempts.

async def count_recent_failures(
    subscription_id: str,
    since: datetime
) -> int

Count failed attempts for a subscription since a given time.

Parameters

Parameter	Type	Description
`subscription_id`	str	Subscription to check.
`since`	datetime	Count attempts after this timestamp.

Returns

Type	Description
int	Count of failed attempts.

---

WebhookSubscriptionStoreProtocol

Storage-agnostic CRUD for webhook subscriptions.

async def create(subscription: WebhookSubscription) -> None

Persist a new subscription.

Parameters

Parameter	Type	Description
`subscription`	WebhookSubscription	Subscription to store.

async def get(subscription_id: str) -> WebhookSubscription | None

Return subscription by ID, or None.

Parameters

Parameter	Type	Description
`subscription_id`	str	UUID to look up.

Returns

Type	Description
WebhookSubscription | None	The subscription if found, else None.

async def list(
    *,
    active_only: bool = True,
    event_type: str | None = None,
    tenant_id: str | None = None,
    limit: int = 100,
    offset: int = 0
) -> list[WebhookSubscription]

List subscriptions matching filters.

Parameters

Parameter	Type	Description
`active_only`	bool	Only return active subscriptions.
`event_type`	str | None	Filter to subscriptions that handle this event type.
`tenant_id`	str | None	Filter by tenant.
`limit`	int	Maximum results.
`offset`	int	Results to skip.

Returns

Type	Description
list[WebhookSubscription]	Filtered and paginated subscriptions.

async def update(subscription: WebhookSubscription) -> None

Update an existing subscription.

Parameters

Parameter	Type	Description
`subscription`	WebhookSubscription	Updated subscription data.

async def delete(subscription_id: str) -> None

Remove a subscription.

Parameters

Parameter	Type	Description
`subscription_id`	str	UUID to remove.

---

Classes

DeliveryAttempt

Record of a single webhook delivery attempt.

Attributes: attempt_id: Unique identifier for this attempt. subscription_id: Target subscription. event_id: The event being delivered. event_type: Event type name. status: Outcome of this attempt. status_code: HTTP response status code (None if connection failed). attempt_number: 1-based attempt counter. attempted_at: When this attempt was made. next_retry_at: Scheduled time for the next retry (None if terminal). error_message: Error details on failure. duration_ms: Round-trip time in milliseconds.

---

DeliveryStatus

Status of a webhook delivery attempt.

---

WebhookConfig

Configuration for the webhook management subsystem.

Attributes: store_backend: Persistence backend. “sql” requires lexigram-webhook[sql]. retry_max_attempts: Maximum delivery attempts before dead-letter. retry_base_delay: Initial retry delay in seconds. retry_max_delay: Maximum retry delay ceiling in seconds. retry_backoff_factor: Exponential backoff multiplier. secret_length: Secret length in bytes (hex-encoded output is 2x). secret_rotation_grace_hours: Hours both old and new secrets are accepted. delivery_timeout_seconds: HTTP request timeout per delivery attempt. disable_after_consecutive_failures: Auto-disable threshold. failure_window_hours: Window for counting consecutive failures. signature_algorithm: HMAC algorithm name. enable_admin: Whether to register the admin contributor. delivery_log_retention_days: Days to retain delivery log (0 = indefinite). signature_header: HTTP header for the HMAC signature. event_type_header: HTTP header for the event type. event_id_header: HTTP header for the event ID. timestamp_header: HTTP header for the delivery timestamp.

---

WebhookEvent

An event prepared for webhook delivery.

Attributes: event_id: Unique identifier for deduplication (UUID). event_type: Dot-notation event type (e.g. “user.created”). payload: JSON-serializable event data. occurred_at: When the original event occurred. source: Originating service or module.

---

WebhookModule

Lexigram webhook management module.

Provides subscription CRUD, delivery pipeline, HMAC signature verification, and dead-letter queue management.

Usage

app.use(WebhookModule.configure())
app.use(WebhookModule.configure(WebhookConfig(store_backend="sql")))

def configure(
    cls,
    config: WebhookConfig | None = None
) -> DynamicModule

Configure the webhook module with optional settings.

Parameters

Parameter	Type	Description
`config`	WebhookConfig | None	Custom webhook configuration. Defaults to ``WebhookConfig()``.

Returns

Type	Description
DynamicModule	DynamicModule ready for registration with the app container.

---

WebhookSubscription

A registered webhook endpoint subscription.

Attributes: subscription_id: Unique identifier (UUID). url: HTTP(S) endpoint URL to deliver events to. secret: Shared secret for HMAC signature computation. event_types: Set of event type names to deliver. None = all events. active: Whether the subscription is currently active. description: Human-readable label for the subscription. tenant_id: Optional multi-tenant scoping. created_at: When the subscription was created. metadata: Arbitrary key-value context.

---

Exceptions

WebhookError

Base exception for webhook domain operations.

---