Skip to content

Webhooks

zenml.webhooks

Webhook provider and trusted event contracts.

Attributes

__all__ = ['BaseWebhookProvider', 'ParsedWebhookDelivery', 'ParsedWebhookEvent', 'WebhookConfiguration', 'WebhookAuthenticationError', 'WebhookEvent', 'WebhookEventHandler', 'WebhookIntakeConfig', 'WebhookIntakeResponse', 'WebhookPayloadError', 'WebhookPreValidationResult', 'WebhookProviderRegistry', 'WebhookTargetEvent', 'WebhookTriggerMatch', 'get_webhook_provider', 'webhook_provider_registry'] module-attribute

webhook_provider_registry = WebhookProviderRegistry() module-attribute

Classes

BaseWebhookProvider

Bases: ABC

Stateless provider behavior used by intake and trigger matching.

Methods:
authenticate(body: bytes, headers: Mapping[str, str], secret: str) -> None abstractmethod

Authenticate the exact raw request body.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required
secret str

The webhook signing secret.

 required

Raises:

Type Description
WebhookAuthenticationError

If authentication fails.
Source code in src/zenml/webhooks/providers/base.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
@abstractmethod
def authenticate(
    self, body: bytes, headers: Mapping[str, str], secret: str
) -> None:
    """Authenticate the exact raw request body.

    Args:
        body: The raw request body.
        headers: The request headers.
        secret: The webhook signing secret.

    Raises:
        WebhookAuthenticationError: If authentication fails.
    """
get_delivery_id(payload: dict[str, Any], headers: Mapping[str, str]) -> str | None

Extract the optional delivery ID.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed JSON payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str | None

The delivery ID, if present.
Source code in src/zenml/webhooks/providers/base.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
def get_delivery_id(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str | None:
    """Extract the optional delivery ID.

    Args:
        payload: The parsed JSON payload.
        headers: The request headers.

    Returns:
        The delivery ID, if present.
    """
    return None
get_event_type(payload: dict[str, Any], headers: Mapping[str, str]) -> str abstractmethod

Extract the provider event type.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed JSON payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str

The provider event type.

Raises:

Type Description
WebhookPayloadError

If the type is missing or invalid.
Source code in src/zenml/webhooks/providers/base.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@abstractmethod
def get_event_type(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str:
    """Extract the provider event type.

    Args:
        payload: The parsed JSON payload.
        headers: The request headers.

    Returns:
        The provider event type.

    Raises:
        WebhookPayloadError: If the type is missing or invalid.
    """
match_triggers(*, event: WebhookEvent, candidates: Sequence[WebhookTriggerResponse]) -> WebhookTriggerMatch[WebhookTriggerResponse] abstractmethod

Match candidates and return the parsed semantic event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted webhook event.

 required
candidates Sequence[WebhookTriggerResponse]

Triggers selected by generic orchestration.

 required

Returns:

Type Description
WebhookTriggerMatch[WebhookTriggerResponse]

The matching triggers and any parsed semantic event.
Source code in src/zenml/webhooks/providers/base.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
@abstractmethod
def match_triggers(
    self,
    *,
    event: "WebhookEvent",
    candidates: Sequence["WebhookTriggerResponse"],
) -> "WebhookTriggerMatch[WebhookTriggerResponse]":
    """Match candidates and return the parsed semantic event.

    Args:
        event: The trusted webhook event.
        candidates: Triggers selected by generic orchestration.

    Returns:
        The matching triggers and any parsed semantic event.
    """
parse(body: bytes, headers: Mapping[str, str]) -> ParsedWebhookEvent

Parse a delivery into provider-neutral event data.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
ParsedWebhookEvent

The parsed delivery.

Raises:

Type Description
WebhookPayloadError

If the body or metadata is invalid.
Source code in src/zenml/webhooks/providers/base.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
def parse(
    self, body: bytes, headers: Mapping[str, str]
) -> ParsedWebhookEvent:
    """Parse a delivery into provider-neutral event data.

    Args:
        body: The raw request body.
        headers: The request headers.

    Returns:
        The parsed delivery.

    Raises:
        WebhookPayloadError: If the body or metadata is invalid.
    """
    try:
        payload = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise WebhookPayloadError(
            "Request body must be valid JSON."
        ) from error
    if not isinstance(payload, dict):
        raise WebhookPayloadError(
            "Request body must contain a top-level JSON object."
        )
    return ParsedWebhookEvent(
        event_type=self.get_event_type(payload=payload, headers=headers),
        delivery_id=self.get_delivery_id(payload=payload, headers=headers),
        payload=payload,
    )
parse_delivery(body: bytes, headers: Mapping[str, str]) -> ParsedWebhookDelivery

Parse a successful delivery and select its intake response.

Existing providers can continue to implement :meth:parse; providers with control deliveries or custom responses can override this method.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
ParsedWebhookDelivery

The parsed delivery and provider-owned response.
Source code in src/zenml/webhooks/providers/base.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def parse_delivery(
    self, body: bytes, headers: Mapping[str, str]
) -> ParsedWebhookDelivery:
    """Parse a successful delivery and select its intake response.

    Existing providers can continue to implement :meth:`parse`; providers
    with control deliveries or custom responses can override this method.

    Args:
        body: The raw request body.
        headers: The request headers.

    Returns:
        The parsed delivery and provider-owned response.
    """
    return ParsedWebhookDelivery(event=self.parse(body, headers))
pre_validate(headers: Mapping[str, str]) -> WebhookPreValidationResult async

Validate headers before webhook lookup and body reading.

Parameters:

Name Type Description Default
headers Mapping[str, str]

The untrusted request headers.

 required

Returns:

Type Description
WebhookPreValidationResult

Whether generic intake should process or ignore the delivery.
Source code in src/zenml/webhooks/providers/base.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
async def pre_validate(
    self, headers: Mapping[str, str]
) -> WebhookPreValidationResult:
    """Validate headers before webhook lookup and body reading.

    Args:
        headers: The untrusted request headers.

    Returns:
        Whether generic intake should process or ignore the delivery.

    """
    return WebhookPreValidationResult.PROCESS
validate_configuration(configuration: WebhookConfiguration | Mapping[str, Any]) -> WebhookConfiguration

Strictly validate a configuration for persistence.

Parameters:

Name Type Description Default
configuration WebhookConfiguration | Mapping[str, Any]

The provider-neutral configuration.

 required

Returns:

Type Description
WebhookConfiguration

A normalized provider-neutral configuration.

Raises:

Type Description
TypeError

If a configuration for another provider is supplied.
Source code in src/zenml/webhooks/providers/base.py 
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
def validate_configuration(
    self,
    configuration: WebhookConfiguration | Mapping[str, Any],
) -> WebhookConfiguration:
    """Strictly validate a configuration for persistence.

    Args:
        configuration: The provider-neutral configuration.

    Returns:
        A normalized provider-neutral configuration.

    Raises:
        TypeError: If a configuration for another provider is supplied.
    """
    if isinstance(configuration, Mapping):
        return self.configuration_class.model_validate(configuration)
    if isinstance(configuration, self.configuration_class):
        return configuration
    raise TypeError(
        "Expected a mapping or an instance of "
        f"{self.configuration_class.__name__}, got "
        f"{type(configuration).__name__}."
    )

ParsedWebhookDelivery

Bases: BaseModel

A successful provider delivery and its intake response.

ParsedWebhookEvent

Bases: BaseModel

Provider delivery parsed into provider-neutral metadata.

WebhookAuthenticationError(message: Optional[str] = None, url: Optional[str] = None)

Bases: CredentialsNotValid

Raised when a webhook request cannot be authenticated.

Source code in src/zenml/exceptions.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(
    self,
    message: Optional[str] = None,
    url: Optional[str] = None,
):
    """The BaseException used to format messages displayed to the user.

    Args:
        message: Message with details of exception. This message
                 will be appended with another message directing user to
                 `url` for more information. If `None`, then default
                 Exception behavior is used.
        url: URL to point to in exception message. If `None`, then no url
             is appended.
    """
    if message and url:
        message += f" For more information, visit {url}."
    super().__init__(message)

WebhookConfiguration

Bases: YAMLSerializationMixin

Base class for provider-owned webhook configuration.

WebhookEvent

Bases: Event

Trusted immutable event handed to registered handlers.

WebhookEventHandler

Bases: EventHandler

Base handler for trusted webhook events.

Methods:
handle_event(event: Event) -> None

Route trusted webhook events to the specialized handler.

Parameters:

Name Type Description Default
event Event

The dispatched event envelope.

 required
Source code in src/zenml/webhooks/handler.py 
25
26
27
28
29
30
31
32
def handle_event(self, event: Event) -> None:
    """Route trusted webhook events to the specialized handler.

    Args:
        event: The dispatched event envelope.
    """
    if isinstance(event, WebhookEvent):
        self.handle_webhook_event(event)
handle_webhook_event(event: WebhookEvent) -> None abstractmethod

Handle a trusted webhook event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted webhook event.

 required
Source code in src/zenml/webhooks/handler.py 
34
35
36
37
38
39
40
@abstractmethod
def handle_webhook_event(self, event: WebhookEvent) -> None:
    """Handle a trusted webhook event.

    Args:
        event: The trusted webhook event.
    """

WebhookIntakeConfig

Bases: BaseModel

Configuration required to authenticate one webhook delivery.

WebhookIntakeResponse

Bases: BaseModel

Provider-owned successful webhook intake response.

WebhookPayloadError

Bases: ValueError

Raised when a webhook payload fails fundamental validation.

WebhookPreValidationResult

Bases: StrEnum

Possible outcomes of provider header pre-validation.

WebhookProviderRegistry()

Registry for webhook provider implementations.

Initialize the webhook provider registry.

Source code in src/zenml/webhooks/providers/registry.py 
27
28
29
30
31
def __init__(self) -> None:
    """Initialize the webhook provider registry."""
    self._provider_classes: dict[str, type[BaseWebhookProvider]] = {}
    self._builtins_registered = False
    self._lock = threading.RLock()
Methods:
get(webhook_type: str) -> BaseWebhookProvider

Instantiate the provider registered for a webhook type.

Parameters:

Name Type Description Default
webhook_type str

The webhook type identifier.

 required

Returns:

Type Description
BaseWebhookProvider

A new provider instance.

Raises:

Type Description
KeyError

If no provider is registered for the webhook type.
Source code in src/zenml/webhooks/providers/registry.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get(self, webhook_type: str) -> BaseWebhookProvider:
    """Instantiate the provider registered for a webhook type.

    Args:
        webhook_type: The webhook type identifier.

    Returns:
        A new provider instance.

    Raises:
        KeyError: If no provider is registered for the webhook type.
    """
    self.register_builtin_providers()
    try:
        provider_class = self._provider_classes[webhook_type]
    except KeyError:
        raise KeyError(
            f"No webhook provider is registered for type {webhook_type}."
        ) from None
    return provider_class()
register(provider_class: type[BaseWebhookProvider], *, overwrite: bool = False) -> None

Register a webhook provider class.

Parameters:

Name Type Description Default
provider_class type[BaseWebhookProvider]

The provider class to register.

 required
overwrite bool

Whether to replace an existing registration.

 False
Source code in src/zenml/webhooks/providers/registry.py 
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
def register(
    self,
    provider_class: type[BaseWebhookProvider],
    *,
    overwrite: bool = False,
) -> None:
    """Register a webhook provider class.

    Args:
        provider_class: The provider class to register.
        overwrite: Whether to replace an existing registration.
    """
    webhook_type = provider_class.webhook_type
    with self._lock:
        if webhook_type in self._provider_classes and not overwrite:
            logger.debug(
                "Webhook provider type %s is already registered. "
                "Skipping registration of %s.",
                webhook_type,
                provider_class.__name__,
            )
            return

        self._provider_classes[webhook_type] = provider_class
        logger.debug(
            "Registered webhook provider %s for type %s.",
            provider_class.__name__,
            webhook_type,
        )
register_builtin_providers() -> None

Register the built-in webhook providers once, on demand.

Source code in src/zenml/webhooks/providers/registry.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def register_builtin_providers(self) -> None:
    """Register the built-in webhook providers once, on demand."""
    with self._lock:
        if self._builtins_registered:
            return

        from zenml.webhooks.providers.clickup import ClickUpWebhookProvider
        from zenml.webhooks.providers.custom import CustomWebhookProvider
        from zenml.webhooks.providers.github import GitHubWebhookProvider
        from zenml.webhooks.providers.slack import SlackWebhookProvider

        self.register(CustomWebhookProvider)
        self.register(GitHubWebhookProvider)
        self.register(ClickUpWebhookProvider)
        self.register(SlackWebhookProvider)
        self._builtins_registered = True

WebhookTargetEvent

Bases: BaseModel

Shared shape of a provider-specific target event.

Methods:
validate_filters() -> WebhookTargetEvent

Validate all configured string filters.

Returns:

Type Description
WebhookTargetEvent

The validated target event.
Source code in src/zenml/webhooks/providers/base.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@model_validator(mode="after")
def validate_filters(self) -> "WebhookTargetEvent":
    """Validate all configured string filters.

    Returns:
        The validated target event.
    """
    for field_name, field in type(self).model_fields.items():
        if cast(Any, field.annotation) != StringFilterOption:
            continue
        self._validate_filter(
            getattr(self, field_name),
            field_name=field_name,
        )
    return self

WebhookTriggerMatch

Bases: BaseModel, Generic[WebhookTriggerT]

Result of matching one trusted event to webhook triggers.

Functions:

get_webhook_provider(webhook_type: str) -> BaseWebhookProvider

Get the provider registered for a webhook type.

Parameters:

Name Type Description Default
webhook_type str

The webhook type identifier.

 required

Returns:

Type Description
BaseWebhookProvider

A new provider instance.
Source code in src/zenml/webhooks/providers/registry.py 
105
106
107
108
109
110
111
112
113
114
def get_webhook_provider(webhook_type: str) -> BaseWebhookProvider:
    """Get the provider registered for a webhook type.

    Args:
        webhook_type: The webhook type identifier.

    Returns:
        A new provider instance.
    """
    return webhook_provider_registry.get(webhook_type)

Modules

events

Provider-neutral trusted webhook event models.

Classes
WebhookEvent

Bases: Event

Trusted immutable event handed to registered handlers.

handler

Event handler contract for trusted webhook events.

Classes
WebhookEventHandler

Bases: EventHandler

Base handler for trusted webhook events.

Methods:
handle_event(event: Event) -> None

Route trusted webhook events to the specialized handler.

Parameters:

Name Type Description Default
event Event

The dispatched event envelope.

 required
Source code in src/zenml/webhooks/handler.py 
25
26
27
28
29
30
31
32
def handle_event(self, event: Event) -> None:
    """Route trusted webhook events to the specialized handler.

    Args:
        event: The dispatched event envelope.
    """
    if isinstance(event, WebhookEvent):
        self.handle_webhook_event(event)
handle_webhook_event(event: WebhookEvent) -> None abstractmethod

Handle a trusted webhook event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted webhook event.

 required
Source code in src/zenml/webhooks/handler.py 
34
35
36
37
38
39
40
@abstractmethod
def handle_webhook_event(self, event: WebhookEvent) -> None:
    """Handle a trusted webhook event.

    Args:
        event: The trusted webhook event.
    """

intake

Internal contracts for webhook intake.

Classes
WebhookIntakeConfig

Bases: BaseModel

Configuration required to authenticate one webhook delivery.

providers

Webhook provider contracts and registry.

Classes
BaseWebhookProvider

Bases: ABC

Stateless provider behavior used by intake and trigger matching.

Methods:
authenticate(body: bytes, headers: Mapping[str, str], secret: str) -> None abstractmethod

Authenticate the exact raw request body.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required
secret str

The webhook signing secret.

 required

Raises:

Type Description
WebhookAuthenticationError

If authentication fails.
Source code in src/zenml/webhooks/providers/base.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
@abstractmethod
def authenticate(
    self, body: bytes, headers: Mapping[str, str], secret: str
) -> None:
    """Authenticate the exact raw request body.

    Args:
        body: The raw request body.
        headers: The request headers.
        secret: The webhook signing secret.

    Raises:
        WebhookAuthenticationError: If authentication fails.
    """
get_delivery_id(payload: dict[str, Any], headers: Mapping[str, str]) -> str | None

Extract the optional delivery ID.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed JSON payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str | None

The delivery ID, if present.
Source code in src/zenml/webhooks/providers/base.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
def get_delivery_id(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str | None:
    """Extract the optional delivery ID.

    Args:
        payload: The parsed JSON payload.
        headers: The request headers.

    Returns:
        The delivery ID, if present.
    """
    return None
get_event_type(payload: dict[str, Any], headers: Mapping[str, str]) -> str abstractmethod

Extract the provider event type.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed JSON payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str

The provider event type.

Raises:

Type Description
WebhookPayloadError

If the type is missing or invalid.
Source code in src/zenml/webhooks/providers/base.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@abstractmethod
def get_event_type(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str:
    """Extract the provider event type.

    Args:
        payload: The parsed JSON payload.
        headers: The request headers.

    Returns:
        The provider event type.

    Raises:
        WebhookPayloadError: If the type is missing or invalid.
    """
match_triggers(*, event: WebhookEvent, candidates: Sequence[WebhookTriggerResponse]) -> WebhookTriggerMatch[WebhookTriggerResponse] abstractmethod

Match candidates and return the parsed semantic event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted webhook event.

 required
candidates Sequence[WebhookTriggerResponse]

Triggers selected by generic orchestration.

 required

Returns:

Type Description
WebhookTriggerMatch[WebhookTriggerResponse]

The matching triggers and any parsed semantic event.
Source code in src/zenml/webhooks/providers/base.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
@abstractmethod
def match_triggers(
    self,
    *,
    event: "WebhookEvent",
    candidates: Sequence["WebhookTriggerResponse"],
) -> "WebhookTriggerMatch[WebhookTriggerResponse]":
    """Match candidates and return the parsed semantic event.

    Args:
        event: The trusted webhook event.
        candidates: Triggers selected by generic orchestration.

    Returns:
        The matching triggers and any parsed semantic event.
    """
parse(body: bytes, headers: Mapping[str, str]) -> ParsedWebhookEvent

Parse a delivery into provider-neutral event data.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
ParsedWebhookEvent

The parsed delivery.

Raises:

Type Description
WebhookPayloadError

If the body or metadata is invalid.
Source code in src/zenml/webhooks/providers/base.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
def parse(
    self, body: bytes, headers: Mapping[str, str]
) -> ParsedWebhookEvent:
    """Parse a delivery into provider-neutral event data.

    Args:
        body: The raw request body.
        headers: The request headers.

    Returns:
        The parsed delivery.

    Raises:
        WebhookPayloadError: If the body or metadata is invalid.
    """
    try:
        payload = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise WebhookPayloadError(
            "Request body must be valid JSON."
        ) from error
    if not isinstance(payload, dict):
        raise WebhookPayloadError(
            "Request body must contain a top-level JSON object."
        )
    return ParsedWebhookEvent(
        event_type=self.get_event_type(payload=payload, headers=headers),
        delivery_id=self.get_delivery_id(payload=payload, headers=headers),
        payload=payload,
    )
parse_delivery(body: bytes, headers: Mapping[str, str]) -> ParsedWebhookDelivery

Parse a successful delivery and select its intake response.

Existing providers can continue to implement :meth:parse; providers with control deliveries or custom responses can override this method.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
ParsedWebhookDelivery

The parsed delivery and provider-owned response.
Source code in src/zenml/webhooks/providers/base.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def parse_delivery(
    self, body: bytes, headers: Mapping[str, str]
) -> ParsedWebhookDelivery:
    """Parse a successful delivery and select its intake response.

    Existing providers can continue to implement :meth:`parse`; providers
    with control deliveries or custom responses can override this method.

    Args:
        body: The raw request body.
        headers: The request headers.

    Returns:
        The parsed delivery and provider-owned response.
    """
    return ParsedWebhookDelivery(event=self.parse(body, headers))
pre_validate(headers: Mapping[str, str]) -> WebhookPreValidationResult async

Validate headers before webhook lookup and body reading.

Parameters:

Name Type Description Default
headers Mapping[str, str]

The untrusted request headers.

 required

Returns:

Type Description
WebhookPreValidationResult

Whether generic intake should process or ignore the delivery.
Source code in src/zenml/webhooks/providers/base.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
async def pre_validate(
    self, headers: Mapping[str, str]
) -> WebhookPreValidationResult:
    """Validate headers before webhook lookup and body reading.

    Args:
        headers: The untrusted request headers.

    Returns:
        Whether generic intake should process or ignore the delivery.

    """
    return WebhookPreValidationResult.PROCESS
validate_configuration(configuration: WebhookConfiguration | Mapping[str, Any]) -> WebhookConfiguration

Strictly validate a configuration for persistence.

Parameters:

Name Type Description Default
configuration WebhookConfiguration | Mapping[str, Any]

The provider-neutral configuration.

 required

Returns:

Type Description
WebhookConfiguration

A normalized provider-neutral configuration.

Raises:

Type Description
TypeError

If a configuration for another provider is supplied.
Source code in src/zenml/webhooks/providers/base.py 
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
def validate_configuration(
    self,
    configuration: WebhookConfiguration | Mapping[str, Any],
) -> WebhookConfiguration:
    """Strictly validate a configuration for persistence.

    Args:
        configuration: The provider-neutral configuration.

    Returns:
        A normalized provider-neutral configuration.

    Raises:
        TypeError: If a configuration for another provider is supplied.
    """
    if isinstance(configuration, Mapping):
        return self.configuration_class.model_validate(configuration)
    if isinstance(configuration, self.configuration_class):
        return configuration
    raise TypeError(
        "Expected a mapping or an instance of "
        f"{self.configuration_class.__name__}, got "
        f"{type(configuration).__name__}."
    )
BuiltinWebhookType

Bases: StrEnum

Webhook provider types bundled with ZenML.

ParsedWebhookDelivery

Bases: BaseModel

A successful provider delivery and its intake response.

ParsedWebhookEvent

Bases: BaseModel

Provider delivery parsed into provider-neutral metadata.

WebhookAuthenticationError(message: Optional[str] = None, url: Optional[str] = None)

Bases: CredentialsNotValid

Raised when a webhook request cannot be authenticated.

Source code in src/zenml/exceptions.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(
    self,
    message: Optional[str] = None,
    url: Optional[str] = None,
):
    """The BaseException used to format messages displayed to the user.

    Args:
        message: Message with details of exception. This message
                 will be appended with another message directing user to
                 `url` for more information. If `None`, then default
                 Exception behavior is used.
        url: URL to point to in exception message. If `None`, then no url
             is appended.
    """
    if message and url:
        message += f" For more information, visit {url}."
    super().__init__(message)
WebhookConfiguration

Bases: YAMLSerializationMixin

Base class for provider-owned webhook configuration.

WebhookIntakeResponse

Bases: BaseModel

Provider-owned successful webhook intake response.

WebhookPayloadError

Bases: ValueError

Raised when a webhook payload fails fundamental validation.

WebhookPreValidationResult

Bases: StrEnum

Possible outcomes of provider header pre-validation.

WebhookProviderRegistry()

Registry for webhook provider implementations.

Initialize the webhook provider registry.

Source code in src/zenml/webhooks/providers/registry.py 
27
28
29
30
31
def __init__(self) -> None:
    """Initialize the webhook provider registry."""
    self._provider_classes: dict[str, type[BaseWebhookProvider]] = {}
    self._builtins_registered = False
    self._lock = threading.RLock()
Methods:
get(webhook_type: str) -> BaseWebhookProvider

Instantiate the provider registered for a webhook type.

Parameters:

Name Type Description Default
webhook_type str

The webhook type identifier.

 required

Returns:

Type Description
BaseWebhookProvider

A new provider instance.

Raises:

Type Description
KeyError

If no provider is registered for the webhook type.
Source code in src/zenml/webhooks/providers/registry.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get(self, webhook_type: str) -> BaseWebhookProvider:
    """Instantiate the provider registered for a webhook type.

    Args:
        webhook_type: The webhook type identifier.

    Returns:
        A new provider instance.

    Raises:
        KeyError: If no provider is registered for the webhook type.
    """
    self.register_builtin_providers()
    try:
        provider_class = self._provider_classes[webhook_type]
    except KeyError:
        raise KeyError(
            f"No webhook provider is registered for type {webhook_type}."
        ) from None
    return provider_class()
register(provider_class: type[BaseWebhookProvider], *, overwrite: bool = False) -> None

Register a webhook provider class.

Parameters:

Name Type Description Default
provider_class type[BaseWebhookProvider]

The provider class to register.

 required
overwrite bool

Whether to replace an existing registration.

 False
Source code in src/zenml/webhooks/providers/registry.py 
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
def register(
    self,
    provider_class: type[BaseWebhookProvider],
    *,
    overwrite: bool = False,
) -> None:
    """Register a webhook provider class.

    Args:
        provider_class: The provider class to register.
        overwrite: Whether to replace an existing registration.
    """
    webhook_type = provider_class.webhook_type
    with self._lock:
        if webhook_type in self._provider_classes and not overwrite:
            logger.debug(
                "Webhook provider type %s is already registered. "
                "Skipping registration of %s.",
                webhook_type,
                provider_class.__name__,
            )
            return

        self._provider_classes[webhook_type] = provider_class
        logger.debug(
            "Registered webhook provider %s for type %s.",
            provider_class.__name__,
            webhook_type,
        )
register_builtin_providers() -> None

Register the built-in webhook providers once, on demand.

Source code in src/zenml/webhooks/providers/registry.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def register_builtin_providers(self) -> None:
    """Register the built-in webhook providers once, on demand."""
    with self._lock:
        if self._builtins_registered:
            return

        from zenml.webhooks.providers.clickup import ClickUpWebhookProvider
        from zenml.webhooks.providers.custom import CustomWebhookProvider
        from zenml.webhooks.providers.github import GitHubWebhookProvider
        from zenml.webhooks.providers.slack import SlackWebhookProvider

        self.register(CustomWebhookProvider)
        self.register(GitHubWebhookProvider)
        self.register(ClickUpWebhookProvider)
        self.register(SlackWebhookProvider)
        self._builtins_registered = True
WebhookTargetEvent

Bases: BaseModel

Shared shape of a provider-specific target event.

Methods:
validate_filters() -> WebhookTargetEvent

Validate all configured string filters.

Returns:

Type Description
WebhookTargetEvent

The validated target event.
Source code in src/zenml/webhooks/providers/base.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@model_validator(mode="after")
def validate_filters(self) -> "WebhookTargetEvent":
    """Validate all configured string filters.

    Returns:
        The validated target event.
    """
    for field_name, field in type(self).model_fields.items():
        if cast(Any, field.annotation) != StringFilterOption:
            continue
        self._validate_filter(
            getattr(self, field_name),
            field_name=field_name,
        )
    return self
WebhookTriggerMatch

Bases: BaseModel, Generic[WebhookTriggerT]

Result of matching one trusted event to webhook triggers.

Functions:
get_webhook_provider(webhook_type: str) -> BaseWebhookProvider

Get the provider registered for a webhook type.

Parameters:

Name Type Description Default
webhook_type str

The webhook type identifier.

 required

Returns:

Type Description
BaseWebhookProvider

A new provider instance.
Source code in src/zenml/webhooks/providers/registry.py 
105
106
107
108
109
110
111
112
113
114
def get_webhook_provider(webhook_type: str) -> BaseWebhookProvider:
    """Get the provider registered for a webhook type.

    Args:
        webhook_type: The webhook type identifier.

    Returns:
        A new provider instance.
    """
    return webhook_provider_registry.get(webhook_type)
Modules
base

Shared contracts for stateless webhook providers.

Classes
BaseWebhookProvider

Bases: ABC

Stateless provider behavior used by intake and trigger matching.

Methods:
authenticate(body: bytes, headers: Mapping[str, str], secret: str) -> None abstractmethod

Authenticate the exact raw request body.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required
secret str

The webhook signing secret.

 required

Raises:

Type Description
WebhookAuthenticationError

If authentication fails.
Source code in src/zenml/webhooks/providers/base.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
@abstractmethod
def authenticate(
    self, body: bytes, headers: Mapping[str, str], secret: str
) -> None:
    """Authenticate the exact raw request body.

    Args:
        body: The raw request body.
        headers: The request headers.
        secret: The webhook signing secret.

    Raises:
        WebhookAuthenticationError: If authentication fails.
    """
get_delivery_id(payload: dict[str, Any], headers: Mapping[str, str]) -> str | None

Extract the optional delivery ID.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed JSON payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str | None

The delivery ID, if present.
Source code in src/zenml/webhooks/providers/base.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
def get_delivery_id(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str | None:
    """Extract the optional delivery ID.

    Args:
        payload: The parsed JSON payload.
        headers: The request headers.

    Returns:
        The delivery ID, if present.
    """
    return None
get_event_type(payload: dict[str, Any], headers: Mapping[str, str]) -> str abstractmethod

Extract the provider event type.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed JSON payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str

The provider event type.

Raises:

Type Description
WebhookPayloadError

If the type is missing or invalid.
Source code in src/zenml/webhooks/providers/base.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@abstractmethod
def get_event_type(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str:
    """Extract the provider event type.

    Args:
        payload: The parsed JSON payload.
        headers: The request headers.

    Returns:
        The provider event type.

    Raises:
        WebhookPayloadError: If the type is missing or invalid.
    """
match_triggers(*, event: WebhookEvent, candidates: Sequence[WebhookTriggerResponse]) -> WebhookTriggerMatch[WebhookTriggerResponse] abstractmethod

Match candidates and return the parsed semantic event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted webhook event.

 required
candidates Sequence[WebhookTriggerResponse]

Triggers selected by generic orchestration.

 required

Returns:

Type Description
WebhookTriggerMatch[WebhookTriggerResponse]

The matching triggers and any parsed semantic event.
Source code in src/zenml/webhooks/providers/base.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
@abstractmethod
def match_triggers(
    self,
    *,
    event: "WebhookEvent",
    candidates: Sequence["WebhookTriggerResponse"],
) -> "WebhookTriggerMatch[WebhookTriggerResponse]":
    """Match candidates and return the parsed semantic event.

    Args:
        event: The trusted webhook event.
        candidates: Triggers selected by generic orchestration.

    Returns:
        The matching triggers and any parsed semantic event.
    """
parse(body: bytes, headers: Mapping[str, str]) -> ParsedWebhookEvent

Parse a delivery into provider-neutral event data.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
ParsedWebhookEvent

The parsed delivery.

Raises:

Type Description
WebhookPayloadError

If the body or metadata is invalid.
Source code in src/zenml/webhooks/providers/base.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
def parse(
    self, body: bytes, headers: Mapping[str, str]
) -> ParsedWebhookEvent:
    """Parse a delivery into provider-neutral event data.

    Args:
        body: The raw request body.
        headers: The request headers.

    Returns:
        The parsed delivery.

    Raises:
        WebhookPayloadError: If the body or metadata is invalid.
    """
    try:
        payload = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise WebhookPayloadError(
            "Request body must be valid JSON."
        ) from error
    if not isinstance(payload, dict):
        raise WebhookPayloadError(
            "Request body must contain a top-level JSON object."
        )
    return ParsedWebhookEvent(
        event_type=self.get_event_type(payload=payload, headers=headers),
        delivery_id=self.get_delivery_id(payload=payload, headers=headers),
        payload=payload,
    )
parse_delivery(body: bytes, headers: Mapping[str, str]) -> ParsedWebhookDelivery

Parse a successful delivery and select its intake response.

Existing providers can continue to implement :meth:parse; providers with control deliveries or custom responses can override this method.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
ParsedWebhookDelivery

The parsed delivery and provider-owned response.
Source code in src/zenml/webhooks/providers/base.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def parse_delivery(
    self, body: bytes, headers: Mapping[str, str]
) -> ParsedWebhookDelivery:
    """Parse a successful delivery and select its intake response.

    Existing providers can continue to implement :meth:`parse`; providers
    with control deliveries or custom responses can override this method.

    Args:
        body: The raw request body.
        headers: The request headers.

    Returns:
        The parsed delivery and provider-owned response.
    """
    return ParsedWebhookDelivery(event=self.parse(body, headers))
pre_validate(headers: Mapping[str, str]) -> WebhookPreValidationResult async

Validate headers before webhook lookup and body reading.

Parameters:

Name Type Description Default
headers Mapping[str, str]

The untrusted request headers.

 required

Returns:

Type Description
WebhookPreValidationResult

Whether generic intake should process or ignore the delivery.
Source code in src/zenml/webhooks/providers/base.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
async def pre_validate(
    self, headers: Mapping[str, str]
) -> WebhookPreValidationResult:
    """Validate headers before webhook lookup and body reading.

    Args:
        headers: The untrusted request headers.

    Returns:
        Whether generic intake should process or ignore the delivery.

    """
    return WebhookPreValidationResult.PROCESS
validate_configuration(configuration: WebhookConfiguration | Mapping[str, Any]) -> WebhookConfiguration

Strictly validate a configuration for persistence.

Parameters:

Name Type Description Default
configuration WebhookConfiguration | Mapping[str, Any]

The provider-neutral configuration.

 required

Returns:

Type Description
WebhookConfiguration

A normalized provider-neutral configuration.

Raises:

Type Description
TypeError

If a configuration for another provider is supplied.
Source code in src/zenml/webhooks/providers/base.py 
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
def validate_configuration(
    self,
    configuration: WebhookConfiguration | Mapping[str, Any],
) -> WebhookConfiguration:
    """Strictly validate a configuration for persistence.

    Args:
        configuration: The provider-neutral configuration.

    Returns:
        A normalized provider-neutral configuration.

    Raises:
        TypeError: If a configuration for another provider is supplied.
    """
    if isinstance(configuration, Mapping):
        return self.configuration_class.model_validate(configuration)
    if isinstance(configuration, self.configuration_class):
        return configuration
    raise TypeError(
        "Expected a mapping or an instance of "
        f"{self.configuration_class.__name__}, got "
        f"{type(configuration).__name__}."
    )
ParsedWebhookDelivery

Bases: BaseModel

A successful provider delivery and its intake response.

ParsedWebhookEvent

Bases: BaseModel

Provider delivery parsed into provider-neutral metadata.

WebhookAuthenticationError(message: Optional[str] = None, url: Optional[str] = None)

Bases: CredentialsNotValid

Raised when a webhook request cannot be authenticated.

Source code in src/zenml/exceptions.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(
    self,
    message: Optional[str] = None,
    url: Optional[str] = None,
):
    """The BaseException used to format messages displayed to the user.

    Args:
        message: Message with details of exception. This message
                 will be appended with another message directing user to
                 `url` for more information. If `None`, then default
                 Exception behavior is used.
        url: URL to point to in exception message. If `None`, then no url
             is appended.
    """
    if message and url:
        message += f" For more information, visit {url}."
    super().__init__(message)
WebhookConfiguration

Bases: YAMLSerializationMixin

Base class for provider-owned webhook configuration.

WebhookIntakeResponse

Bases: BaseModel

Provider-owned successful webhook intake response.

WebhookPayloadError

Bases: ValueError

Raised when a webhook payload fails fundamental validation.

WebhookPreValidationResult

Bases: StrEnum

Possible outcomes of provider header pre-validation.

WebhookTargetEvent

Bases: BaseModel

Shared shape of a provider-specific target event.

Methods:
validate_filters() -> WebhookTargetEvent

Validate all configured string filters.

Returns:

Type Description
WebhookTargetEvent

The validated target event.
Source code in src/zenml/webhooks/providers/base.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@model_validator(mode="after")
def validate_filters(self) -> "WebhookTargetEvent":
    """Validate all configured string filters.

    Returns:
        The validated target event.
    """
    for field_name, field in type(self).model_fields.items():
        if cast(Any, field.annotation) != StringFilterOption:
            continue
        self._validate_filter(
            getattr(self, field_name),
            field_name=field_name,
        )
    return self
WebhookTriggerMatch

Bases: BaseModel, Generic[WebhookTriggerT]

Result of matching one trusted event to webhook triggers.

Functions:
authenticate_hmac_sha256(*, body: bytes, headers: Mapping[str, str], secret: str, header: str, prefixed: bool = True) -> None

Authenticate an HMAC-SHA256 signature.

Parameters:

Name Type Description Default
body bytes

The exact request body.

 required
headers Mapping[str, str]

The request headers.

 required
secret str

The signing secret.

 required
header str

The signature header name.

 required
prefixed bool

If True, require a sha256= prefix (GitHub-style). If False, compare the raw hexadecimal digest (ClickUp-style).

 True

Raises:

Type Description
WebhookAuthenticationError

If the signature is invalid.
Source code in src/zenml/webhooks/providers/base.py 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
def authenticate_hmac_sha256(
    *,
    body: bytes,
    headers: Mapping[str, str],
    secret: str,
    header: str,
    prefixed: bool = True,
) -> None:
    """Authenticate an HMAC-SHA256 signature.

    Args:
        body: The exact request body.
        headers: The request headers.
        secret: The signing secret.
        header: The signature header name.
        prefixed: If `True`, require a `sha256=` prefix (GitHub-style). If
            `False`, compare the raw hexadecimal digest (ClickUp-style).

    Raises:
        WebhookAuthenticationError: If the signature is invalid.
    """
    signature = headers.get(header)
    digest = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    if prefixed:
        if not signature or not signature.startswith("sha256="):
            raise WebhookAuthenticationError(
                f"Missing or malformed {header} header."
            )
        expected = "sha256=" + digest
        if not hmac.compare_digest(signature, expected):
            raise WebhookAuthenticationError("Invalid webhook signature.")
        return
    if not signature:
        raise WebhookAuthenticationError(f"Missing {header} header.")
    if not hmac.compare_digest(signature.lower(), digest.lower()):
        raise WebhookAuthenticationError("Invalid webhook signature.")
matches_string_collection_filter(*, actual: Sequence[str], configured: StringFilterOption) -> bool

Match a collection against a configured string filter.

Parameters:

Name Type Description Default
actual Sequence[str]

Values extracted from the webhook event.

 required
configured StringFilterOption

The configured string filter or OR-list of filters.

 required

Returns:

Type Description
bool

Whether any value satisfies a positive filter or every value satisfies
bool

a negative filter.
Source code in src/zenml/webhooks/providers/base.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def matches_string_collection_filter(
    *, actual: Sequence[str], configured: StringFilterOption
) -> bool:
    """Match a collection against a configured string filter.

    Args:
        actual: Values extracted from the webhook event.
        configured: The configured string filter or OR-list of filters.

    Returns:
        Whether any value satisfies a positive filter or every value satisfies
        a negative filter.
    """
    if configured is None:
        return True
    if not actual:
        return False
    for configured_value in (
        configured if isinstance(configured, list) else [configured]
    ):
        aggregate = (
            all if _is_negative_string_filter(configured_value) else any
        )
        if aggregate(
            _matches_string_filter_value(
                actual=actual_value, configured=configured_value
            )
            for actual_value in actual
        ):
            return True
    return False
matches_string_filter(*, actual: str | None, configured: StringFilterOption) -> bool

Match an extracted value against a supported string filter.

Parameters:

Name Type Description Default
actual str | None

The value extracted from the webhook event.

 required
configured StringFilterOption

The configured string filter or OR-list of filters.

 required

Returns:

Type Description
bool

Whether the extracted value matches the configured filter.
Source code in src/zenml/webhooks/providers/base.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def matches_string_filter(
    *, actual: str | None, configured: StringFilterOption
) -> bool:
    """Match an extracted value against a supported string filter.

    Args:
        actual: The value extracted from the webhook event.
        configured: The configured string filter or OR-list of filters.

    Returns:
        Whether the extracted value matches the configured filter.
    """
    if configured is None:
        return True
    if actual is None:
        return False
    configured_values = (
        configured if isinstance(configured, list) else [configured]
    )
    return any(
        _matches_string_filter_value(actual=actual, configured=value)
        for value in configured_values
    )
clickup

ClickUp webhook provider and semantic target event catalog.

Classes
ClickUpListCreatedEvent

Bases: ClickUpListSemanticEvent

Normalized created-list event.

ClickUpListDeletedEvent

Bases: ClickUpListSemanticEvent

Normalized deleted-list event.

ClickUpListSemanticEvent

Bases: ClickUpSemanticEvent

Normalized ClickUp list event with shared location filters.

Methods:
matches(target: ClickUpWebhookTargetEvent) -> bool

Return whether this event matches a typed list target.

Parameters:

Name Type Description Default
target ClickUpWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether this event matches the target.
Source code in src/zenml/webhooks/providers/clickup.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def matches(self, target: ClickUpWebhookTargetEvent) -> bool:
    """Return whether this event matches a typed list target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether this event matches the target.
    """
    if not isinstance(target, self.event_filter_type):
        return False
    return _matches_location_filters(
        list_id=self.list_id,
        space_id=self.space_id,
        folder_id=self.folder_id,
        target=target,
    )
ClickUpListUpdatedEvent

Bases: ClickUpListSemanticEvent

Normalized updated-list event.

ClickUpSemanticEvent

Bases: BaseModel

Normalized ClickUp event used for trigger matching.

Methods:
matches(target: ClickUpWebhookTargetEvent) -> bool abstractmethod

Return whether the semantic event matches its typed target.

Parameters:

Name Type Description Default
target ClickUpWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether the semantic event matches the target.
Source code in src/zenml/webhooks/providers/clickup.py 
312
313
314
315
316
317
318
319
320
321
@abstractmethod
def matches(self, target: ClickUpWebhookTargetEvent) -> bool:
    """Return whether the semantic event matches its typed target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether the semantic event matches the target.
    """
ClickUpTaskAssigneeUpdatedEvent

Bases: ClickUpTaskSemanticEvent

Normalized task-assignee-updated event.

ClickUpTaskCommentPostedEvent

Bases: ClickUpTaskSemanticEvent

Normalized task-comment-posted event.

ClickUpTaskCreatedEvent

Bases: ClickUpTaskSemanticEvent

Normalized created-task event.

ClickUpTaskDeletedEvent

Bases: ClickUpTaskSemanticEvent

Normalized deleted-task event.

ClickUpTaskMovedEvent

Bases: ClickUpTaskSemanticEvent

Normalized moved-task event.

ClickUpTaskSemanticEvent

Bases: ClickUpSemanticEvent

Normalized ClickUp task event with shared location filters.

Methods:
matches(target: ClickUpWebhookTargetEvent) -> bool

Return whether this event matches a typed task target.

Parameters:

Name Type Description Default
target ClickUpWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether this event matches the target.
Source code in src/zenml/webhooks/providers/clickup.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def matches(self, target: ClickUpWebhookTargetEvent) -> bool:
    """Return whether this event matches a typed task target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether this event matches the target.
    """
    if not isinstance(target, self.event_filter_type):
        return False
    return _matches_task_filters(
        task_id=self.task_id,
        list_id=self.list_id,
        space_id=self.space_id,
        folder_id=self.folder_id,
        target=target,
    )
ClickUpTaskStatusUpdatedEvent

Bases: ClickUpTaskSemanticEvent

Normalized task-status-updated event.

Methods:
matches(target: ClickUpWebhookTargetEvent) -> bool

Return whether this event matches a status-updated target.

Parameters:

Name Type Description Default
target ClickUpWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether this event matches the target.
Source code in src/zenml/webhooks/providers/clickup.py 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
def matches(self, target: ClickUpWebhookTargetEvent) -> bool:
    """Return whether this event matches a status-updated target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether this event matches the target.
    """
    if not isinstance(target, TaskStatusUpdated):
        return False
    return all(
        (
            _matches_task_filters(
                task_id=self.task_id,
                list_id=self.list_id,
                space_id=self.space_id,
                folder_id=self.folder_id,
                target=target,
            ),
            matches_string_filter(
                actual=self.status, configured=target.status
            ),
        )
    )
ClickUpTaskUpdatedEvent

Bases: ClickUpTaskSemanticEvent

Normalized updated-task event.

ClickUpWebhookConfiguration

Bases: WebhookConfiguration

Typed configuration for a ClickUp webhook trigger.

ClickUpWebhookEvent

Bases: StrEnum

ClickUp events supported by webhook triggers.

ClickUpWebhookProvider

Bases: BaseWebhookProvider

Provider for authenticated and semantically matched ClickUp webhooks.

Methods:
authenticate(body: bytes, headers: Mapping[str, str], secret: str) -> None

Authenticate a ClickUp delivery.

Parameters:

Name Type Description Default
body bytes

The exact raw request body.

 required
headers Mapping[str, str]

The request headers.

 required
secret str

The webhook signing secret.

 required
Source code in src/zenml/webhooks/providers/clickup.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def authenticate(
    self, body: bytes, headers: Mapping[str, str], secret: str
) -> None:
    """Authenticate a ClickUp delivery.

    Args:
        body: The exact raw request body.
        headers: The request headers.
        secret: The webhook signing secret.
    """
    authenticate_hmac_sha256(
        body=body,
        headers=headers,
        secret=secret,
        header=CLICKUP_SIGNATURE_HEADER,
        prefixed=False,
    )
get_delivery_id(payload: dict[str, Any], headers: Mapping[str, str]) -> str | None

Build ClickUp's documented idempotency key.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed ClickUp payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str | None

The documented history-based delivery ID, if available. Intake
str | None

generates a unique ID for history-less events.

Raises:

Type Description
WebhookPayloadError

If webhook_id is missing.
Source code in src/zenml/webhooks/providers/clickup.py 
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
def get_delivery_id(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str | None:
    """Build ClickUp's documented idempotency key.

    Args:
        payload: The parsed ClickUp payload.
        headers: The request headers.

    Returns:
        The documented history-based delivery ID, if available. Intake
        generates a unique ID for history-less events.

    Raises:
        WebhookPayloadError: If webhook_id is missing.
    """
    webhook_id = payload.get("webhook_id")
    if not isinstance(webhook_id, str) or not webhook_id:
        raise WebhookPayloadError(
            "Missing or empty ClickUp 'webhook_id' field."
        )
    history_ids = _history_item_ids(payload)
    if history_ids:
        return f"{webhook_id}:{','.join(history_ids)}"
    return None
get_event_type(payload: dict[str, Any], headers: Mapping[str, str]) -> str

Extract the ClickUp event name from the JSON body.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed ClickUp payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str

The ClickUp event name.

Raises:

Type Description
WebhookPayloadError

If the event field is missing or empty.
Source code in src/zenml/webhooks/providers/clickup.py 
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
def get_event_type(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str:
    """Extract the ClickUp event name from the JSON body.

    Args:
        payload: The parsed ClickUp payload.
        headers: The request headers.

    Returns:
        The ClickUp event name.

    Raises:
        WebhookPayloadError: If the event field is missing or empty.
    """
    event_type = payload.get("event")
    if not isinstance(event_type, str) or not event_type:
        raise WebhookPayloadError(
            "Missing or empty ClickUp 'event' field."
        )
    return event_type
match_triggers(*, event: WebhookEvent, candidates: Sequence[WebhookTriggerResponse]) -> WebhookTriggerMatch[WebhookTriggerResponse]

Match ClickUp triggers and return the parsed semantic event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted ClickUp webhook event.

 required
candidates Sequence[WebhookTriggerResponse]

The candidate webhook triggers.

 required

Returns:

Type Description
WebhookTriggerMatch[WebhookTriggerResponse]

Matching triggers and their shared semantic event.
Source code in src/zenml/webhooks/providers/clickup.py 
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
def match_triggers(
    self,
    *,
    event: "WebhookEvent",
    candidates: Sequence["WebhookTriggerResponse"],
) -> "WebhookTriggerMatch[WebhookTriggerResponse]":
    """Match ClickUp triggers and return the parsed semantic event.

    Args:
        event: The trusted ClickUp webhook event.
        candidates: The candidate webhook triggers.

    Returns:
        Matching triggers and their shared semantic event.
    """
    semantic = self.parse_semantic_event(event)
    if semantic is None:
        return WebhookTriggerMatch(triggers=[])
    matches: list[WebhookTriggerResponse] = []
    for trigger in candidates:
        targets = self._cast_runtime_targets(trigger)
        if any(semantic.matches(target) for target in targets):
            matches.append(trigger)
    return WebhookTriggerMatch(
        triggers=matches,
        event=semantic.model_dump(mode="json"),
    )
parse_semantic_event(event: WebhookEvent) -> ClickUpSemanticEvent | None

Parse a trusted delivery into a normalized semantic event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted ClickUp webhook event.

 required

Returns:

Type Description
ClickUpSemanticEvent | None

The normalized semantic event, or None for unsupported events.
Source code in src/zenml/webhooks/providers/clickup.py 
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
def parse_semantic_event(
    self, event: "WebhookEvent"
) -> ClickUpSemanticEvent | None:
    """Parse a trusted delivery into a normalized semantic event.

    Args:
        event: The trusted ClickUp webhook event.

    Returns:
        The normalized semantic event, or `None` for unsupported events.
    """
    try:
        event_type = ClickUpWebhookEvent(event.event_type)
    except ValueError:
        return None
    payload = event.payload
    list_id = _id_string(payload, "list_id")
    space_id = _id_string(payload, "space_id")
    folder_id = _id_string(payload, "folder_id")
    semantic_cls = next(
        (
            candidate
            for candidate in _SEMANTIC_EVENTS
            if candidate.model_fields["type"].default == event_type
        ),
        None,
    )
    if semantic_cls is None:
        return None
    if semantic_cls is ClickUpTaskStatusUpdatedEvent:
        return ClickUpTaskStatusUpdatedEvent(
            type=event_type,
            task_id=_id_string(payload, "task_id"),
            list_id=list_id,
            space_id=space_id,
            folder_id=folder_id,
            status=_status_after(payload),
        )
    if issubclass(semantic_cls, ClickUpTaskSemanticEvent):
        return semantic_cls(
            type=event_type,
            task_id=_id_string(payload, "task_id"),
            list_id=list_id,
            space_id=space_id,
            folder_id=folder_id,
        )
    return semantic_cls(
        type=event_type,
        list_id=list_id,
        space_id=space_id,
        folder_id=folder_id,
    )
ListCreated

Bases: _ClickUpListTarget

Filters for a created ClickUp list.

ListDeleted

Bases: _ClickUpListTarget

Filters for a deleted ClickUp list.

ListUpdated

Bases: _ClickUpListTarget

Filters for an updated ClickUp list.

TaskAssigneeUpdated

Bases: _ClickUpTaskTarget

Filters for a ClickUp task assignee change.

TaskCommentPosted

Bases: _ClickUpTaskTarget

Filters for a comment posted on a ClickUp task.

TaskCreated

Bases: _ClickUpTaskTarget

Filters for a created ClickUp task.

TaskDeleted

Bases: _ClickUpTaskTarget

Filters for a deleted ClickUp task.

TaskMoved

Bases: _ClickUpTaskTarget

Filters for a ClickUp task moved to another list.

TaskStatusUpdated

Bases: _ClickUpTaskStatusTarget

Filters for a ClickUp task status change.

TaskUpdated

Bases: _ClickUpTaskTarget

Filters for an updated ClickUp task.

Functions:
custom

Custom webhook provider.

Classes
CustomWebhookConfiguration

Bases: WebhookConfiguration

Configuration for an unfiltered custom webhook trigger.

CustomWebhookProvider

Bases: BaseWebhookProvider

Provider for signed custom JSON webhook deliveries.

Methods:
authenticate(body: bytes, headers: Mapping[str, str], secret: str) -> None

Authenticate a custom delivery.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required
secret str

The signing secret.

 required
Source code in src/zenml/webhooks/providers/custom.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def authenticate(
    self, body: bytes, headers: Mapping[str, str], secret: str
) -> None:
    """Authenticate a custom delivery.

    Args:
        body: The raw request body.
        headers: The request headers.
        secret: The signing secret.
    """
    authenticate_hmac_sha256(
        body=body,
        headers=headers,
        secret=secret,
        header=CUSTOM_SIGNATURE_HEADER,
    )
get_delivery_id(payload: dict[str, Any], headers: Mapping[str, str]) -> str | None

Extract the optional custom delivery ID.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str | None

The delivery ID, if present.
Source code in src/zenml/webhooks/providers/custom.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
def get_delivery_id(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str | None:
    """Extract the optional custom delivery ID.

    Args:
        payload: The parsed payload.
        headers: The request headers.

    Returns:
        The delivery ID, if present.
    """
    return headers.get(CUSTOM_DELIVERY_HEADER)
get_event_type(payload: dict[str, Any], headers: Mapping[str, str]) -> str

Extract the custom event type.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str

The event type.

Raises:

Type Description
WebhookPayloadError

If the event header is missing.
Source code in src/zenml/webhooks/providers/custom.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def get_event_type(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str:
    """Extract the custom event type.

    Args:
        payload: The parsed payload.
        headers: The request headers.

    Returns:
        The event type.

    Raises:
        WebhookPayloadError: If the event header is missing.
    """
    event_type = headers.get(CUSTOM_EVENT_HEADER)
    if not event_type:
        raise WebhookPayloadError(
            f"Missing required {CUSTOM_EVENT_HEADER} header."
        )
    return event_type
match_triggers(*, event: WebhookEvent, candidates: Sequence[WebhookTriggerResponse]) -> WebhookTriggerMatch[WebhookTriggerResponse]

Match candidates with valid unfiltered custom configuration.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted custom event.

 required
candidates Sequence[WebhookTriggerResponse]

The associated candidate triggers.

 required

Returns:

Type Description
WebhookTriggerMatch[WebhookTriggerResponse]

Matching candidates without semantic event metadata.
Source code in src/zenml/webhooks/providers/custom.py 
 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
def match_triggers(
    self,
    *,
    event: "WebhookEvent",
    candidates: Sequence["WebhookTriggerResponse"],
) -> "WebhookTriggerMatch[WebhookTriggerResponse]":
    """Match candidates with valid unfiltered custom configuration.

    Args:
        event: The trusted custom event.
        candidates: The associated candidate triggers.

    Returns:
        Matching candidates without semantic event metadata.
    """
    matches: list[WebhookTriggerResponse] = []
    for trigger in candidates:
        try:
            self.validate_configuration(trigger.configuration)
        except (TypeError, ValueError):
            logger.exception(
                "Skipping defective webhook trigger configuration %s",
                trigger.id,
            )
            continue
        matches.append(trigger)
    return WebhookTriggerMatch(triggers=matches)
Functions:
github

GitHub webhook provider and semantic target event catalog.

Classes
GitHubCommit

Bases: BaseModel

Commit metadata associated with a GitHub semantic event.

GitHubIssueOpenedEvent

Bases: GitHubSemanticEvent

Normalized newly opened issue event.

Methods:
matches(target: GitHubWebhookTargetEvent) -> bool

Return whether this event matches an opened-issue target.

Parameters:

Name Type Description Default
target GitHubWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether this event matches the target.
Source code in src/zenml/webhooks/providers/github.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
def matches(self, target: GitHubWebhookTargetEvent) -> bool:
    """Return whether this event matches an opened-issue target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether this event matches the target.
    """
    if not isinstance(target, IssueOpened):
        return False
    return all(
        (
            matches_string_filter(
                actual=self.repo, configured=target.repo
            ),
            matches_string_filter(
                actual=self.author, configured=target.author
            ),
            matches_string_filter(
                actual=self.author_association,
                configured=target.author_association,
            ),
            matches_string_collection_filter(
                actual=self.labels, configured=target.labels
            ),
            matches_string_collection_filter(
                actual=self.assignees, configured=target.assignees
            ),
            matches_string_filter(
                actual=self.milestone, configured=target.milestone
            ),
        )
    )
GitHubMergedPullRequestEvent

Bases: GitHubSemanticEvent

Normalized merged pull request event.

Methods:
matches(target: GitHubWebhookTargetEvent) -> bool

Return whether this event matches a merged-pull-request target.

Parameters:

Name Type Description Default
target GitHubWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether this event matches the target.
Source code in src/zenml/webhooks/providers/github.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def matches(self, target: GitHubWebhookTargetEvent) -> bool:
    """Return whether this event matches a merged-pull-request target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether this event matches the target.
    """
    if not isinstance(target, MergedPullRequest):
        return False
    return all(
        (
            matches_string_filter(
                actual=self.repo, configured=target.repo
            ),
            matches_string_filter(
                actual=self.target_branch, configured=target.target_branch
            ),
            matches_string_filter(
                actual=self.source_branch, configured=target.source_branch
            ),
            matches_string_filter(
                actual=self.author, configured=target.author
            ),
        )
    )
GitHubPushEvent

Bases: GitHubSemanticEvent

Normalized branch push event.

Methods:
matches(target: GitHubWebhookTargetEvent) -> bool

Return whether this event matches a push target.

Parameters:

Name Type Description Default
target GitHubWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether this event matches the target.
Source code in src/zenml/webhooks/providers/github.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def matches(self, target: GitHubWebhookTargetEvent) -> bool:
    """Return whether this event matches a push target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether this event matches the target.
    """
    if not isinstance(target, PushEvent):
        return False
    return all(
        (
            matches_string_filter(
                actual=self.repo, configured=target.repo
            ),
            matches_string_filter(
                actual=self.branch, configured=target.branch
            ),
            matches_string_filter(
                actual=self.actor, configured=target.actor
            ),
        )
    )
GitHubReleasePublishedEvent

Bases: GitHubSemanticEvent

Normalized published release event.

Methods:
matches(target: GitHubWebhookTargetEvent) -> bool

Return whether this event matches a published-release target.

Parameters:

Name Type Description Default
target GitHubWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether this event matches the target.
Source code in src/zenml/webhooks/providers/github.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def matches(self, target: GitHubWebhookTargetEvent) -> bool:
    """Return whether this event matches a published-release target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether this event matches the target.
    """
    if not isinstance(target, ReleasePublished):
        return False
    return all(
        (
            matches_string_filter(
                actual=self.repo, configured=target.repo
            ),
            matches_string_filter(actual=self.tag, configured=target.tag),
            matches_string_filter(
                actual=self.target_branch, configured=target.target_branch
            ),
            matches_string_filter(
                actual=self.actor, configured=target.actor
            ),
        )
    )
GitHubSemanticEvent

Bases: BaseModel

Provider event normalized for semantic trigger matching.

Methods:
matches(target: GitHubWebhookTargetEvent) -> bool abstractmethod

Return whether the semantic event matches its typed target.

Parameters:

Name Type Description Default
target GitHubWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether the semantic event matches the target.
Source code in src/zenml/webhooks/providers/github.py 
182
183
184
185
186
187
188
189
190
191
@abstractmethod
def matches(self, target: GitHubWebhookTargetEvent) -> bool:
    """Return whether the semantic event matches its typed target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether the semantic event matches the target.
    """
GitHubWebhookConfiguration

Bases: WebhookConfiguration

Typed configuration for a GitHub webhook trigger.

GitHubWebhookEvent

Bases: StrEnum

Semantic GitHub events supported by webhook triggers.

GitHubWebhookEventType

Bases: StrEnum

Raw GitHub event families supported by semantic target events.

GitHubWebhookProvider

Bases: BaseWebhookProvider

Provider for authenticated and semantically matched GitHub webhooks.

Methods:
authenticate(body: bytes, headers: Mapping[str, str], secret: str) -> None

Authenticate a GitHub delivery.

Parameters:

Name Type Description Default
body bytes

The exact raw request body.

 required
headers Mapping[str, str]

The request headers.

 required
secret str

The webhook signing secret.

 required
Source code in src/zenml/webhooks/providers/github.py 
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
def authenticate(
    self, body: bytes, headers: Mapping[str, str], secret: str
) -> None:
    """Authenticate a GitHub delivery.

    Args:
        body: The exact raw request body.
        headers: The request headers.
        secret: The webhook signing secret.
    """
    authenticate_hmac_sha256(
        body=body,
        headers=headers,
        secret=secret,
        header=GITHUB_SIGNATURE_HEADER,
    )
get_delivery_id(payload: dict[str, Any], headers: Mapping[str, str]) -> str | None

Extract the optional GitHub delivery ID.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed GitHub payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str | None

The delivery ID, if present.
Source code in src/zenml/webhooks/providers/github.py 
478
479
480
481
482
483
484
485
486
487
488
489
490
def get_delivery_id(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str | None:
    """Extract the optional GitHub delivery ID.

    Args:
        payload: The parsed GitHub payload.
        headers: The request headers.

    Returns:
        The delivery ID, if present.
    """
    return headers.get(GITHUB_DELIVERY_HEADER)
get_event_type(payload: dict[str, Any], headers: Mapping[str, str]) -> str

Extract the raw GitHub event family.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed GitHub payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str

The raw GitHub event family.

Raises:

Type Description
WebhookPayloadError

If the GitHub event header is missing.
Source code in src/zenml/webhooks/providers/github.py 
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
def get_event_type(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str:
    """Extract the raw GitHub event family.

    Args:
        payload: The parsed GitHub payload.
        headers: The request headers.

    Returns:
        The raw GitHub event family.

    Raises:
        WebhookPayloadError: If the GitHub event header is missing.
    """
    event_type = headers.get(GITHUB_EVENT_HEADER)
    if not event_type:
        raise WebhookPayloadError(
            f"Missing required {GITHUB_EVENT_HEADER} header."
        )
    return event_type
match_triggers(*, event: WebhookEvent, candidates: Sequence[WebhookTriggerResponse]) -> WebhookTriggerMatch[WebhookTriggerResponse]

Match GitHub triggers and return the parsed semantic event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted GitHub webhook event.

 required
candidates Sequence[WebhookTriggerResponse]

The candidate webhook triggers.

 required

Returns:

Type Description
WebhookTriggerMatch[WebhookTriggerResponse]

Matching triggers and their shared semantic event.
Source code in src/zenml/webhooks/providers/github.py 
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
def match_triggers(
    self,
    *,
    event: "WebhookEvent",
    candidates: Sequence["WebhookTriggerResponse"],
) -> "WebhookTriggerMatch[WebhookTriggerResponse]":
    """Match GitHub triggers and return the parsed semantic event.

    Args:
        event: The trusted GitHub webhook event.
        candidates: The candidate webhook triggers.

    Returns:
        Matching triggers and their shared semantic event.
    """
    semantic = self.parse_semantic_event(event)
    if semantic is None:
        return WebhookTriggerMatch(triggers=[])
    matches: list[WebhookTriggerResponse] = []
    for trigger in candidates:
        targets = self._cast_runtime_targets(trigger)
        if any(semantic.matches(target) for target in targets):
            matches.append(trigger)
    return WebhookTriggerMatch(
        triggers=matches,
        event=semantic.model_dump(mode="json"),
    )
parse_semantic_event(event: WebhookEvent) -> GitHubSemanticEvent | None

Parse a trusted delivery into a normalized semantic event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted GitHub webhook event.

 required

Returns:

Type Description
GitHubSemanticEvent | None

The normalized semantic event, or None for irrelevant payloads.
Source code in src/zenml/webhooks/providers/github.py 
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
def parse_semantic_event(
    self, event: "WebhookEvent"
) -> GitHubSemanticEvent | None:
    """Parse a trusted delivery into a normalized semantic event.

    Args:
        event: The trusted GitHub webhook event.

    Returns:
        The normalized semantic event, or `None` for irrelevant payloads.
    """
    payload = event.payload
    if event.event_type == "pull_request":
        pull_request = payload.get("pull_request")
        if (
            payload.get("action") != "closed"
            or not isinstance(pull_request, Mapping)
            or pull_request.get("merged") is not True
        ):
            return None
        repo = _string_at(payload, "repository", "full_name")
        target = _string_at(payload, "pull_request", "base", "ref")
        if repo is None or target is None:
            return None
        merge_commit_sha = _string_at(
            payload, "pull_request", "merge_commit_sha"
        )
        return GitHubMergedPullRequestEvent(
            repo=repo,
            target_branch=target,
            source_branch=_string_at(
                payload, "pull_request", "head", "ref"
            ),
            author=_string_at(payload, "pull_request", "user", "login"),
            commit=(
                GitHubCommit(
                    name=_string_at(payload, "pull_request", "title"),
                    sha=merge_commit_sha,
                )
                if merge_commit_sha
                else None
            ),
        )
    if event.event_type == "workflow_run":
        if payload.get("action") != "completed":
            return None
        workflow = _string_at(payload, "workflow_run", "name")
        if workflow is None:
            return None
        return GitHubWorkflowRunCompletedEvent(
            workflow=workflow,
            conclusion=_string_at(payload, "workflow_run", "conclusion"),
            actor=_string_at(payload, "workflow_run", "actor", "login"),
        )
    if event.event_type == "push":
        ref = _string_at(payload, "ref")
        repo = _string_at(payload, "repository", "full_name")
        prefix = "refs/heads/"
        if ref is None or repo is None or not ref.startswith(prefix):
            return None
        head_commit_sha = _string_at(payload, "head_commit", "id")
        return GitHubPushEvent(
            repo=repo,
            branch=ref.removeprefix(prefix),
            actor=_string_at(payload, "sender", "login"),
            commit=(
                GitHubCommit(
                    name=_string_at(payload, "head_commit", "message"),
                    sha=head_commit_sha,
                )
                if head_commit_sha
                else None
            ),
        )
    if event.event_type == "release":
        if payload.get("action") != "published":
            return None
        repo = _string_at(payload, "repository", "full_name")
        tag = _string_at(payload, "release", "tag_name")
        if repo is None or tag is None:
            return None
        return GitHubReleasePublishedEvent(
            repo=repo,
            tag=tag,
            target_branch=_string_at(
                payload, "release", "target_commitish"
            ),
            actor=_string_at(payload, "release", "author", "login"),
        )
    if event.event_type == "issues":
        issue = payload.get("issue")
        if payload.get("action") != "opened" or not isinstance(
            issue, Mapping
        ):
            return None
        repo = _string_at(payload, "repository", "full_name")
        title = _string_at(payload, "issue", "title")
        number = issue.get("number")
        if (
            repo is None
            or title is None
            or not isinstance(number, int)
            or isinstance(number, bool)
        ):
            return None
        return GitHubIssueOpenedEvent(
            repo=repo,
            number=number,
            title=title,
            author=_string_at(payload, "issue", "user", "login"),
            author_association=_string_at(
                payload, "issue", "author_association"
            ),
            labels=_object_strings_at(
                payload, "issue", "labels", item_field="name"
            ),
            assignees=_object_strings_at(
                payload, "issue", "assignees", item_field="login"
            ),
            milestone=_string_at(payload, "issue", "milestone", "title"),
            issue_type=_string_at(payload, "issue", "type", "name"),
        )
    return None
pre_validate(headers: Mapping[str, str]) -> WebhookPreValidationResult async

Reject malformed and ignore unsupported GitHub event families.

Parameters:

Name Type Description Default
headers Mapping[str, str]

The untrusted request headers.

 required

Returns:

Type Description
WebhookPreValidationResult

Whether generic intake should process or ignore the delivery.

Raises:

Type Description
WebhookPayloadError

If the GitHub event header is missing.
Source code in src/zenml/webhooks/providers/github.py 
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
async def pre_validate(
    self, headers: Mapping[str, str]
) -> WebhookPreValidationResult:
    """Reject malformed and ignore unsupported GitHub event families.

    Args:
        headers: The untrusted request headers.

    Returns:
        Whether generic intake should process or ignore the delivery.

    Raises:
        WebhookPayloadError: If the GitHub event header is missing.
    """
    event_type = headers.get(GITHUB_EVENT_HEADER)
    if not event_type:
        raise WebhookPayloadError(
            f"Missing or empty {GITHUB_EVENT_HEADER} header."
        )
    try:
        GitHubWebhookEventType(event_type)
    except ValueError:
        return WebhookPreValidationResult.IGNORE
    return WebhookPreValidationResult.PROCESS
GitHubWorkflowRunCompletedEvent

Bases: GitHubSemanticEvent

Normalized completed workflow run event.

Methods:
matches(target: GitHubWebhookTargetEvent) -> bool

Return whether this event matches a workflow-run target.

Parameters:

Name Type Description Default
target GitHubWebhookTargetEvent

The typed target event configuration.

 required

Returns:

Type Description
bool

Whether this event matches the target.
Source code in src/zenml/webhooks/providers/github.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def matches(self, target: GitHubWebhookTargetEvent) -> bool:
    """Return whether this event matches a workflow-run target.

    Args:
        target: The typed target event configuration.

    Returns:
        Whether this event matches the target.
    """
    if not isinstance(target, WorkflowRunCompleted):
        return False
    return all(
        (
            matches_string_filter(
                actual=self.workflow, configured=target.workflow
            ),
            matches_string_filter(
                actual=self.conclusion, configured=target.conclusion
            ),
            matches_string_filter(
                actual=self.actor, configured=target.actor
            ),
        )
    )
IssueOpened

Bases: WebhookTargetEvent

Filters for a newly opened GitHub issue.

MergedPullRequest

Bases: WebhookTargetEvent

Filters for a merged GitHub pull request.

PushEvent

Bases: WebhookTargetEvent

Filters for a GitHub branch push.

ReleasePublished

Bases: WebhookTargetEvent

Filters for a published GitHub release.

WorkflowRunCompleted

Bases: WebhookTargetEvent

Filters for a completed GitHub workflow run.

Functions:
registry

Registry for webhook provider implementations.

Classes
WebhookProviderRegistry()

Registry for webhook provider implementations.

Initialize the webhook provider registry.

Source code in src/zenml/webhooks/providers/registry.py 
27
28
29
30
31
def __init__(self) -> None:
    """Initialize the webhook provider registry."""
    self._provider_classes: dict[str, type[BaseWebhookProvider]] = {}
    self._builtins_registered = False
    self._lock = threading.RLock()
Methods:
get(webhook_type: str) -> BaseWebhookProvider

Instantiate the provider registered for a webhook type.

Parameters:

Name Type Description Default
webhook_type str

The webhook type identifier.

 required

Returns:

Type Description
BaseWebhookProvider

A new provider instance.

Raises:

Type Description
KeyError

If no provider is registered for the webhook type.
Source code in src/zenml/webhooks/providers/registry.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get(self, webhook_type: str) -> BaseWebhookProvider:
    """Instantiate the provider registered for a webhook type.

    Args:
        webhook_type: The webhook type identifier.

    Returns:
        A new provider instance.

    Raises:
        KeyError: If no provider is registered for the webhook type.
    """
    self.register_builtin_providers()
    try:
        provider_class = self._provider_classes[webhook_type]
    except KeyError:
        raise KeyError(
            f"No webhook provider is registered for type {webhook_type}."
        ) from None
    return provider_class()
register(provider_class: type[BaseWebhookProvider], *, overwrite: bool = False) -> None

Register a webhook provider class.

Parameters:

Name Type Description Default
provider_class type[BaseWebhookProvider]

The provider class to register.

 required
overwrite bool

Whether to replace an existing registration.

 False
Source code in src/zenml/webhooks/providers/registry.py 
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
def register(
    self,
    provider_class: type[BaseWebhookProvider],
    *,
    overwrite: bool = False,
) -> None:
    """Register a webhook provider class.

    Args:
        provider_class: The provider class to register.
        overwrite: Whether to replace an existing registration.
    """
    webhook_type = provider_class.webhook_type
    with self._lock:
        if webhook_type in self._provider_classes and not overwrite:
            logger.debug(
                "Webhook provider type %s is already registered. "
                "Skipping registration of %s.",
                webhook_type,
                provider_class.__name__,
            )
            return

        self._provider_classes[webhook_type] = provider_class
        logger.debug(
            "Registered webhook provider %s for type %s.",
            provider_class.__name__,
            webhook_type,
        )
register_builtin_providers() -> None

Register the built-in webhook providers once, on demand.

Source code in src/zenml/webhooks/providers/registry.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def register_builtin_providers(self) -> None:
    """Register the built-in webhook providers once, on demand."""
    with self._lock:
        if self._builtins_registered:
            return

        from zenml.webhooks.providers.clickup import ClickUpWebhookProvider
        from zenml.webhooks.providers.custom import CustomWebhookProvider
        from zenml.webhooks.providers.github import GitHubWebhookProvider
        from zenml.webhooks.providers.slack import SlackWebhookProvider

        self.register(CustomWebhookProvider)
        self.register(GitHubWebhookProvider)
        self.register(ClickUpWebhookProvider)
        self.register(SlackWebhookProvider)
        self._builtins_registered = True
Functions:
get_webhook_provider(webhook_type: str) -> BaseWebhookProvider

Get the provider registered for a webhook type.

Parameters:

Name Type Description Default
webhook_type str

The webhook type identifier.

 required

Returns:

Type Description
BaseWebhookProvider

A new provider instance.
Source code in src/zenml/webhooks/providers/registry.py 
105
106
107
108
109
110
111
112
113
114
def get_webhook_provider(webhook_type: str) -> BaseWebhookProvider:
    """Get the provider registered for a webhook type.

    Args:
        webhook_type: The webhook type identifier.

    Returns:
        A new provider instance.
    """
    return webhook_provider_registry.get(webhook_type)
slack

Slack Events API webhook provider.

Classes
AppMentionEventFilter

Bases: SlackEventFilter

Filters for a Slack app mention.

FileSharedEventFilter

Bases: SlackEventFilter

Filters for a Slack file being shared.

MessageEventFilter

Bases: SlackEventFilter

Filters for a Slack message event.

Methods:
validate_subtype_filters() -> MessageEventFilter

Reject competing broad and targeted subtype selection.

Returns:

Type Description
MessageEventFilter

The validated message event filter.

Raises:

Type Description
ValueError

If all subtypes and named subtypes are both selected.
Source code in src/zenml/webhooks/providers/slack.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@model_validator(mode="after")
def validate_subtype_filters(self) -> "MessageEventFilter":
    """Reject competing broad and targeted subtype selection.

    Returns:
        The validated message event filter.

    Raises:
        ValueError: If all subtypes and named subtypes are both selected.
    """
    if self.include_subtypes and self.subtype is not None:
        raise ValueError(
            "Slack message filters cannot combine 'include_subtypes' "
            "with 'subtype'."
        )
    return self
MessageMetadataPostedEventFilter

Bases: _MessageMetadataEventFilter

Filters for Slack message metadata being posted.

MessageMetadataUpdatedEventFilter

Bases: _MessageMetadataEventFilter

Filters for Slack message metadata being updated.

ReactionAddedEventFilter

Bases: _ReactionEventFilter

Filters for a Slack reaction being added.

ReactionRemovedEventFilter

Bases: _ReactionEventFilter

Filters for a Slack reaction being removed.

SlackAppMentionEvent

Bases: SlackChannelEvent

Normalized Slack app mention.

Methods:
matches(target: SlackWebhookEventFilter) -> bool

Return whether this mention matches an app-mention target.

Parameters:

Name Type Description Default
target SlackWebhookEventFilter

The typed Slack target event.

 required

Returns:

Type Description
bool

Whether this mention matches the target.
Source code in src/zenml/webhooks/providers/slack.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def matches(self, target: SlackWebhookEventFilter) -> bool:
    """Return whether this mention matches an app-mention target.

    Args:
        target: The typed Slack target event.

    Returns:
        Whether this mention matches the target.
    """
    if not isinstance(target, AppMentionEventFilter):
        return False
    return all(
        (
            super().matches(target),
            matches_string_filter(
                actual=self.text, configured=target.text
            ),
            _matches_bool(
                actual=self.thread_ts is not None,
                configured=target.threaded,
            ),
        )
    )
SlackChannelEvent

Bases: SlackUserEvent

Base event with a required Slack channel identifier.

Methods:
matches(target: SlackWebhookEventFilter) -> bool

Match the required channel identifier.

Parameters:

Name Type Description Default
target SlackWebhookEventFilter

The typed Slack target event.

 required

Returns:

Type Description
bool

Whether the event's team, user, and channel match the target.
Source code in src/zenml/webhooks/providers/slack.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def matches(self, target: SlackWebhookEventFilter) -> bool:
    """Match the required channel identifier.

    Args:
        target: The typed Slack target event.

    Returns:
        Whether the event's team, user, and channel match the target.
    """
    return all(
        (
            super().matches(target),
            matches_string_filter(
                actual=self.channel_id, configured=target.channel_id
            ),
        )
    )
SlackEventFilter

Bases: WebhookTargetEvent

Base filters shared by Slack target events.

SlackFileSharedEvent

Bases: SlackChannelEvent

Normalized Slack file-shared event.

Methods:
matches(target: SlackWebhookEventFilter) -> bool

Return whether this file share matches a file-share target.

Parameters:

Name Type Description Default
target SlackWebhookEventFilter

The typed Slack target event.

 required

Returns:

Type Description
bool

Whether this file share matches the target.
Source code in src/zenml/webhooks/providers/slack.py 
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
def matches(self, target: SlackWebhookEventFilter) -> bool:
    """Return whether this file share matches a file-share target.

    Args:
        target: The typed Slack target event.

    Returns:
        Whether this file share matches the target.
    """
    if not isinstance(target, FileSharedEventFilter):
        return False
    return all(
        (
            super().matches(target),
            matches_string_filter(
                actual=self.file_id, configured=target.file_id
            ),
        )
    )
SlackMessageEvent

Bases: SlackSemanticEvent

Normalized Slack message event.

Methods:
matches(target: SlackWebhookEventFilter) -> bool

Return whether this message matches a message target.

Parameters:

Name Type Description Default
target SlackWebhookEventFilter

The typed Slack target event.

 required

Returns:

Type Description
bool

Whether this message matches the target.
Source code in src/zenml/webhooks/providers/slack.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def matches(self, target: SlackWebhookEventFilter) -> bool:
    """Return whether this message matches a message target.

    Args:
        target: The typed Slack target event.

    Returns:
        Whether this message matches the target.
    """
    if not isinstance(target, MessageEventFilter):
        return False
    if target.subtype is not None:
        subtype_matches = matches_string_filter(
            actual=self.subtype, configured=target.subtype
        )
    elif target.include_subtypes:
        subtype_matches = True
    else:
        subtype_matches = self.subtype is None and not self.bot_authored
    return all(
        (
            super().matches(target),
            matches_string_filter(
                actual=self.user_id, configured=target.user_id
            ),
            matches_string_filter(
                actual=self.channel_id, configured=target.channel_id
            ),
            matches_string_filter(
                actual=self.channel_type, configured=target.channel_type
            ),
            matches_string_filter(
                actual=self.text, configured=target.text
            ),
            subtype_matches,
            _matches_bool(
                actual=self.thread_ts is not None,
                configured=target.threaded,
            ),
        )
    )
SlackMessageMetadata

Bases: BaseModel

Structured metadata attached to a Slack message.

SlackMessageMetadataEvent

Bases: SlackChannelEvent

Shared normalized fields for a Slack message-metadata event.

Methods:
matches(target: SlackWebhookEventFilter) -> bool

Return whether this metadata event matches its typed target.

Parameters:

Name Type Description Default
target SlackWebhookEventFilter

The typed Slack target event.

 required

Returns:

Type Description
bool

Whether this metadata event matches the target.
Source code in src/zenml/webhooks/providers/slack.py 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
def matches(self, target: SlackWebhookEventFilter) -> bool:
    """Return whether this metadata event matches its typed target.

    Args:
        target: The typed Slack target event.

    Returns:
        Whether this metadata event matches the target.
    """
    if not isinstance(target, self.event_filter_type):
        return False
    metadata_target = cast(_MessageMetadataEventFilter, target)
    return all(
        (
            super().matches(target),
            matches_string_filter(
                actual=self.app_id, configured=metadata_target.app_id
            ),
            matches_string_filter(
                actual=self.bot_id, configured=metadata_target.bot_id
            ),
            matches_string_filter(
                actual=self.metadata.event_type,
                configured=metadata_target.metadata_event_type,
            ),
        )
    )
SlackMessageMetadataPostedEvent

Bases: SlackMessageMetadataEvent

Normalized Slack message-metadata-posted event.

SlackMessageMetadataUpdatedEvent

Bases: SlackMessageMetadataEvent

Normalized Slack message-metadata-updated event.

SlackReactionAddedEvent

Bases: SlackReactionEvent

Normalized Slack reaction-added event.

SlackReactionEvent

Bases: SlackUserEvent

Shared normalized fields for a Slack reaction event.

Methods:
matches(target: SlackWebhookEventFilter) -> bool

Return whether this reaction matches its reaction target.

Parameters:

Name Type Description Default
target SlackWebhookEventFilter

The typed Slack target event.

 required

Returns:

Type Description
bool

Whether this reaction matches the target.
Source code in src/zenml/webhooks/providers/slack.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
def matches(self, target: SlackWebhookEventFilter) -> bool:
    """Return whether this reaction matches its reaction target.

    Args:
        target: The typed Slack target event.

    Returns:
        Whether this reaction matches the target.
    """
    if not isinstance(target, self.event_filter_type):
        return False
    reaction_target = cast(_ReactionEventFilter, target)
    return all(
        (
            super().matches(target),
            matches_string_filter(
                actual=self.channel_id, configured=target.channel_id
            ),
            matches_string_filter(
                actual=self.reaction, configured=reaction_target.reaction
            ),
            matches_string_filter(
                actual=self.item_user_id,
                configured=reaction_target.item_user_id,
            ),
            matches_string_filter(
                actual=self.item.type,
                configured=reaction_target.item_type,
            ),
            matches_string_filter(
                actual=self.item.id, configured=reaction_target.item_id
            ),
        )
    )
SlackReactionItem

Bases: BaseModel

Normalized item referenced by a Slack reaction event.

SlackReactionRemovedEvent

Bases: SlackReactionEvent

Normalized Slack reaction-removed event.

SlackSemanticEvent

Bases: BaseModel

Provider event normalized for semantic trigger matching.

Methods:
matches(target: SlackWebhookEventFilter) -> bool

Match identifiers shared by all Slack semantic events.

Parameters:

Name Type Description Default
target SlackWebhookEventFilter

The typed Slack target event.

 required

Returns:

Type Description
bool

Whether the event's team matches the target.
Source code in src/zenml/webhooks/providers/slack.py 
223
224
225
226
227
228
229
230
231
232
233
234
def matches(self, target: SlackWebhookEventFilter) -> bool:
    """Match identifiers shared by all Slack semantic events.

    Args:
        target: The typed Slack target event.

    Returns:
        Whether the event's team matches the target.
    """
    return matches_string_filter(
        actual=self.team_id, configured=target.team_id
    )
SlackUserEvent

Bases: SlackSemanticEvent

Base event with a required Slack user identifier.

Methods:
matches(target: SlackWebhookEventFilter) -> bool

Match the required user identifier.

Parameters:

Name Type Description Default
target SlackWebhookEventFilter

The typed Slack target event.

 required

Returns:

Type Description
bool

Whether the event's team and user match the target.
Source code in src/zenml/webhooks/providers/slack.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def matches(self, target: SlackWebhookEventFilter) -> bool:
    """Match the required user identifier.

    Args:
        target: The typed Slack target event.

    Returns:
        Whether the event's team and user match the target.
    """
    return all(
        (
            super().matches(target),
            matches_string_filter(
                actual=self.user_id, configured=target.user_id
            ),
        )
    )
SlackWebhookConfiguration

Bases: WebhookConfiguration

Typed configuration for a Slack webhook trigger.

SlackWebhookEventType

Bases: StrEnum

Slack event types used during intake and semantic matching.

SlackWebhookProvider

Bases: BaseWebhookProvider

Provider for signed Slack Events API deliveries.

Methods:
authenticate(body: bytes, headers: Mapping[str, str], secret: str) -> None

Authenticate a Slack delivery using its exact raw body.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required
secret str

The Slack app signing secret.

 required

Raises:

Type Description
WebhookAuthenticationError

If the request cannot be authenticated.
Source code in src/zenml/webhooks/providers/slack.py 
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
def authenticate(
    self, body: bytes, headers: Mapping[str, str], secret: str
) -> None:
    """Authenticate a Slack delivery using its exact raw body.

    Args:
        body: The raw request body.
        headers: The request headers.
        secret: The Slack app signing secret.

    Raises:
        WebhookAuthenticationError: If the request cannot be authenticated.
    """
    signature = headers.get(SLACK_SIGNATURE_HEADER)
    if not signature or not signature.startswith(
        f"{SLACK_SIGNATURE_VERSION}="
    ):
        raise WebhookAuthenticationError(
            f"Missing or malformed {SLACK_SIGNATURE_HEADER} header."
        )

    timestamp = headers.get(SLACK_REQUEST_TIMESTAMP_HEADER)
    if (
        not timestamp
        or not timestamp.isascii()
        or not timestamp.isdigit()
        or len(timestamp) > 20
    ):
        raise WebhookAuthenticationError(
            "Missing or malformed "
            f"{SLACK_REQUEST_TIMESTAMP_HEADER} header."
        )
    if (
        abs(time.time() - int(timestamp))
        > SLACK_TIMESTAMP_TOLERANCE_SECONDS
    ):
        raise WebhookAuthenticationError(
            "Slack request timestamp is outside the allowed tolerance."
        )

    signature_base = (
        f"{SLACK_SIGNATURE_VERSION}:{timestamp}:".encode() + body
    )
    expected = (
        f"{SLACK_SIGNATURE_VERSION}="
        + hmac.new(
            secret.encode(), signature_base, hashlib.sha256
        ).hexdigest()
    )
    if not hmac.compare_digest(signature, expected):
        raise WebhookAuthenticationError("Invalid webhook signature.")
get_event_type(payload: dict[str, Any], headers: Mapping[str, str]) -> str

Reject direct event parsing in favor of delivery parsing.

Parameters:

Name Type Description Default
payload dict[str, Any]

The parsed Slack payload.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
str

This method never returns successfully.

Raises:

Type Description
NotImplementedError

Always, because Slack has control deliveries.
Source code in src/zenml/webhooks/providers/slack.py 
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
def get_event_type(
    self, payload: dict[str, Any], headers: Mapping[str, str]
) -> str:
    """Reject direct event parsing in favor of delivery parsing.

    Args:
        payload: The parsed Slack payload.
        headers: The request headers.

    Returns:
        This method never returns successfully.

    Raises:
        NotImplementedError: Always, because Slack has control deliveries.
    """
    raise NotImplementedError(
        "Slack deliveries must be parsed with parse_delivery()."
    )
match_triggers(*, event: WebhookEvent, candidates: Sequence[WebhookTriggerResponse]) -> WebhookTriggerMatch[WebhookTriggerResponse]

Match candidates and return the parsed semantic event.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted webhook event.

 required
candidates Sequence[WebhookTriggerResponse]

Triggers selected by generic orchestration.

 required

Returns:

Type Description
WebhookTriggerMatch[WebhookTriggerResponse]

The matching triggers and any parsed semantic event.
Source code in src/zenml/webhooks/providers/slack.py 
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
def match_triggers(
    self,
    *,
    event: "WebhookEvent",
    candidates: Sequence["WebhookTriggerResponse"],
) -> "WebhookTriggerMatch[WebhookTriggerResponse]":
    """Match candidates and return the parsed semantic event.

    Args:
        event: The trusted webhook event.
        candidates: Triggers selected by generic orchestration.

    Returns:
        The matching triggers and any parsed semantic event.
    """
    semantic = self.parse_semantic_event(event)
    if semantic is None:
        return WebhookTriggerMatch(triggers=[])
    matches: list[WebhookTriggerResponse] = []
    for trigger in candidates:
        targets = self._cast_runtime_targets(trigger)
        if any(semantic.matches(target) for target in targets):
            matches.append(trigger)
    return WebhookTriggerMatch(
        triggers=matches,
        event=semantic.model_dump(mode="json"),
    )
parse_delivery(body: bytes, headers: Mapping[str, str]) -> ParsedWebhookDelivery

Parse a Slack event or control delivery.

Parameters:

Name Type Description Default
body bytes

The raw request body.

 required
headers Mapping[str, str]

The request headers.

 required

Returns:

Type Description
ParsedWebhookDelivery

The optional event and Slack-compatible response.

Raises:

Type Description
WebhookPayloadError

If the Slack envelope is malformed.
Source code in src/zenml/webhooks/providers/slack.py 
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
def parse_delivery(
    self, body: bytes, headers: Mapping[str, str]
) -> ParsedWebhookDelivery:
    """Parse a Slack event or control delivery.

    Args:
        body: The raw request body.
        headers: The request headers.

    Returns:
        The optional event and Slack-compatible response.

    Raises:
        WebhookPayloadError: If the Slack envelope is malformed.
    """
    try:
        payload = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise WebhookPayloadError(
            "Request body must be valid JSON."
        ) from error
    if not isinstance(payload, dict):
        raise WebhookPayloadError(
            "Request body must contain a top-level JSON object."
        )

    delivery_type = payload.get("type")
    if delivery_type == SLACK_EVENT_CALLBACK:
        return self._parse_event_callback(payload)
    if delivery_type == SLACK_URL_VERIFICATION:
        challenge = payload.get("challenge")
        if not isinstance(challenge, str) or not challenge:
            raise WebhookPayloadError(
                "Slack URL verification requires a non-empty challenge."
            )
        return ParsedWebhookDelivery(
            event=None,
            response=WebhookIntakeResponse(
                status_code=200,
                body=challenge,
                media_type="text/plain",
            ),
        )
    if delivery_type == SLACK_APP_RATE_LIMITED:
        self._validate_rate_limited_delivery(payload)
        return ParsedWebhookDelivery(
            event=None,
            response=WebhookIntakeResponse(status_code=200),
        )
    raise WebhookPayloadError(
        "Unsupported Slack delivery type."
        if isinstance(delivery_type, str) and delivery_type
        else "Slack delivery requires a non-empty type."
    )
parse_semantic_event(event: WebhookEvent) -> SlackSemanticEvent | None

Normalize a trusted Slack event for trigger matching.

Parameters:

Name Type Description Default
event WebhookEvent

The trusted Slack webhook event.

 required

Returns:

Type Description
SlackSemanticEvent | None

The normalized supported event, or None if it is not matchable.
Source code in src/zenml/webhooks/providers/slack.py 
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
def parse_semantic_event(
    self, event: "WebhookEvent"
) -> SlackSemanticEvent | None:
    """Normalize a trusted Slack event for trigger matching.

    Args:
        event: The trusted Slack webhook event.

    Returns:
        The normalized supported event, or `None` if it is not matchable.
    """
    payload = event.payload.get("event")
    if not isinstance(payload, Mapping):
        return None
    common_fields = self._common_event_fields(event, payload)
    if common_fields is None:
        return None
    if event.event_type == SlackWebhookEventType.APP_MENTION:
        return self._parse_app_mention(payload, common_fields)
    if event.event_type == SlackWebhookEventType.MESSAGE:
        return self._parse_message(payload, common_fields)
    if event.event_type == SlackWebhookEventType.REACTION_ADDED:
        return self._parse_reaction(
            payload, common_fields, SlackReactionAddedEvent
        )
    if event.event_type == SlackWebhookEventType.REACTION_REMOVED:
        return self._parse_reaction(
            payload, common_fields, SlackReactionRemovedEvent
        )
    if event.event_type == SlackWebhookEventType.MESSAGE_METADATA_POSTED:
        fields = self._parse_message_metadata_fields(
            payload, common_fields
        )
        return (
            SlackMessageMetadataPostedEvent(**fields)
            if fields is not None
            else None
        )
    if event.event_type == SlackWebhookEventType.MESSAGE_METADATA_UPDATED:
        return self._parse_message_metadata_updated(payload, common_fields)
    if event.event_type == SlackWebhookEventType.FILE_SHARED:
        return self._parse_file_shared(payload, common_fields)
    return None
Functions:
types

Identifiers for webhook providers bundled with ZenML.

Classes
BuiltinWebhookType

Bases: StrEnum

Webhook provider types bundled with ZenML.