Webhooks

zenml.webhooks

Webhook provider and trusted event contracts.

Attributes

__all__ = ['BaseWebhookProvider', 'DynamicWebhookTargetEvent', '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

@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

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

@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

@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

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

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

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

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__}."
    )

DynamicWebhookTargetEvent

Bases: WebhookTargetEvent

User-defined webhook event filter over an authenticated JSON body.

Methods:

matches(*, event_type: str, payload: Mapping[str, Any]) -> bool

Match a provider event type and authenticated JSON body.

Parameters:

Name	Type	Description	Default
event_type	str	The provider-specific raw event type.	required
payload	Mapping[str, Any]	The complete authenticated JSON body.	required

Returns:

Type	Description
bool	Whether the event satisfies every configured filter.

Source code in src/zenml/webhooks/providers/dynamic.py

def matches(self, *, event_type: str, payload: Mapping[str, Any]) -> bool:
    """Match a provider event type and authenticated JSON body.

    Args:
        event_type: The provider-specific raw event type.
        payload: The complete authenticated JSON body.

    Returns:
        Whether the event satisfies every configured filter.
    """
    if not matches_string_filter(
        actual=event_type, configured=self.event_type
    ):
        return False
    for path, configured in self.filters.items():
        actual_values = [
            string_value
            for value in _resolve_dynamic_path(
                payload, self._compiled_filters[path]
            )
            if (string_value := _stringify_json_scalar(value)) is not None
        ]
        if not matches_string_collection_filter(
            actual=actual_values, configured=configured
        ):
            return False
    return True

validate_dynamic_filters() -> DynamicWebhookTargetEvent

Validate body filters and compile their paths.

Returns:

Type	Description
DynamicWebhookTargetEvent	The validated dynamic target event.

Source code in src/zenml/webhooks/providers/dynamic.py

@model_validator(mode="after")
def validate_dynamic_filters(self) -> "DynamicWebhookTargetEvent":
    """Validate body filters and compile their paths.

    Returns:
        The validated dynamic target event.
    """
    self._validate_filter(self.event_type, field_name="event_type")

    compiled_filters: dict[str, tuple[_PathToken, ...]] = {}
    for path, configured in self.filters.items():
        self._validate_filter(configured, field_name=path)
        compiled_filters[path] = _parse_dynamic_path(path)
    self._compiled_filters = compiled_filters
    return self

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

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

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

@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

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

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

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

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

@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

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

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

@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

@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

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

@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

@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

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

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

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

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.

DynamicWebhookTargetEvent

Bases: WebhookTargetEvent

User-defined webhook event filter over an authenticated JSON body.

Methods:

matches(*, event_type: str, payload: Mapping[str, Any]) -> bool

Match a provider event type and authenticated JSON body.

Parameters:

Name	Type	Description	Default
event_type	str	The provider-specific raw event type.	required
payload	Mapping[str, Any]	The complete authenticated JSON body.	required

Returns:

Type	Description
bool	Whether the event satisfies every configured filter.

Source code in src/zenml/webhooks/providers/dynamic.py

def matches(self, *, event_type: str, payload: Mapping[str, Any]) -> bool:
    """Match a provider event type and authenticated JSON body.

    Args:
        event_type: The provider-specific raw event type.
        payload: The complete authenticated JSON body.

    Returns:
        Whether the event satisfies every configured filter.
    """
    if not matches_string_filter(
        actual=event_type, configured=self.event_type
    ):
        return False
    for path, configured in self.filters.items():
        actual_values = [
            string_value
            for value in _resolve_dynamic_path(
                payload, self._compiled_filters[path]
            )
            if (string_value := _stringify_json_scalar(value)) is not None
        ]
        if not matches_string_collection_filter(
            actual=actual_values, configured=configured
        ):
            return False
    return True

validate_dynamic_filters() -> DynamicWebhookTargetEvent

Validate body filters and compile their paths.

Returns:

Type	Description
DynamicWebhookTargetEvent	The validated dynamic target event.

Source code in src/zenml/webhooks/providers/dynamic.py

@model_validator(mode="after")
def validate_dynamic_filters(self) -> "DynamicWebhookTargetEvent":
    """Validate body filters and compile their paths.

    Returns:
        The validated dynamic target event.
    """
    self._validate_filter(self.event_type, field_name="event_type")

    compiled_filters: dict[str, tuple[_PathToken, ...]] = {}
    for path, configured in self.filters.items():
        self._validate_filter(configured, field_name=path)
        compiled_filters[path] = _parse_dynamic_path(path)
    self._compiled_filters = compiled_filters
    return self

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

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

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

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

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

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

@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

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

@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

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

@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

@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

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

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

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

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

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

@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

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

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

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

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

@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

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

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

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

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

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

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)
    semantic_matcher = semantic.matches if semantic is not None else None
    matches: list[WebhookTriggerResponse] = []
    for trigger in candidates:
        targets = self._cast_runtime_targets(trigger)
        if any(
            matches_webhook_target(
                target=target,
                event_type=event.event_type,
                payload=event.payload,
                semantic_matcher=semantic_matcher,
            )
            for target in targets
        ):
            matches.append(trigger)
    return WebhookTriggerMatch(
        triggers=matches,
        event=(semantic.model_dump(mode="json") if semantic else None),
    )

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

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

Optional dynamic filtering for a 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

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

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

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 using optional dynamic custom filters.

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

def match_triggers(
    self,
    *,
    event: "WebhookEvent",
    candidates: Sequence["WebhookTriggerResponse"],
) -> "WebhookTriggerMatch[WebhookTriggerResponse]":
    """Match candidates using optional dynamic custom filters.

    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:
            configuration = self.validate_configuration(
                trigger.configuration
            )
        except (TypeError, ValueError):
            logger.exception(
                "Skipping defective webhook trigger configuration %s",
                trigger.id,
            )
            continue
        targets = cast(
            CustomWebhookConfiguration, configuration
        ).target_events
        if not targets or any(
            matches_webhook_target(
                target=target,
                event_type=event.event_type,
                payload=event.payload,
                semantic_matcher=None,
            )
            for target in targets
        ):
            matches.append(trigger)
    return WebhookTriggerMatch(triggers=matches)

Functions:

dynamic

Dynamic webhook target filters over authenticated JSON payloads.

Classes

DynamicWebhookTargetEvent

Bases: WebhookTargetEvent

User-defined webhook event filter over an authenticated JSON body.

Methods:

matches(*, event_type: str, payload: Mapping[str, Any]) -> bool

Match a provider event type and authenticated JSON body.

Parameters:

Name	Type	Description	Default
event_type	str	The provider-specific raw event type.	required
payload	Mapping[str, Any]	The complete authenticated JSON body.	required

Returns:

Type	Description
bool	Whether the event satisfies every configured filter.

Source code in src/zenml/webhooks/providers/dynamic.py

def matches(self, *, event_type: str, payload: Mapping[str, Any]) -> bool:
    """Match a provider event type and authenticated JSON body.

    Args:
        event_type: The provider-specific raw event type.
        payload: The complete authenticated JSON body.

    Returns:
        Whether the event satisfies every configured filter.
    """
    if not matches_string_filter(
        actual=event_type, configured=self.event_type
    ):
        return False
    for path, configured in self.filters.items():
        actual_values = [
            string_value
            for value in _resolve_dynamic_path(
                payload, self._compiled_filters[path]
            )
            if (string_value := _stringify_json_scalar(value)) is not None
        ]
        if not matches_string_collection_filter(
            actual=actual_values, configured=configured
        ):
            return False
    return True

validate_dynamic_filters() -> DynamicWebhookTargetEvent

Validate body filters and compile their paths.

Returns:

Type	Description
DynamicWebhookTargetEvent	The validated dynamic target event.

Source code in src/zenml/webhooks/providers/dynamic.py

@model_validator(mode="after")
def validate_dynamic_filters(self) -> "DynamicWebhookTargetEvent":
    """Validate body filters and compile their paths.

    Returns:
        The validated dynamic target event.
    """
    self._validate_filter(self.event_type, field_name="event_type")

    compiled_filters: dict[str, tuple[_PathToken, ...]] = {}
    for path, configured in self.filters.items():
        self._validate_filter(configured, field_name=path)
        compiled_filters[path] = _parse_dynamic_path(path)
    self._compiled_filters = compiled_filters
    return self

Functions:

matches_webhook_target(*, target: WebhookTargetEvent, event_type: str, payload: Mapping[str, Any], semantic_matcher: Callable[[Any], bool] | None) -> bool

Dispatch matching to a dynamic or provider-semantic target.

Parameters:

Name	Type	Description	Default
target	WebhookTargetEvent	The configured webhook target.	required
event_type	str	The provider-specific raw event type.	required
payload	Mapping[str, Any]	The complete authenticated JSON body.	required
semantic_matcher	Callable[[Any], bool] | None	Provider semantic matching, when parsing succeeded.	required

Returns:

Type	Description
bool	Whether the provider event matches the target.

Source code in src/zenml/webhooks/providers/dynamic.py

def matches_webhook_target(
    *,
    target: WebhookTargetEvent,
    event_type: str,
    payload: Mapping[str, Any],
    semantic_matcher: Callable[[Any], bool] | None,
) -> bool:
    """Dispatch matching to a dynamic or provider-semantic target.

    Args:
        target: The configured webhook target.
        event_type: The provider-specific raw event type.
        payload: The complete authenticated JSON body.
        semantic_matcher: Provider semantic matching, when parsing succeeded.

    Returns:
        Whether the provider event matches the target.
    """
    if isinstance(target, DynamicWebhookTargetEvent):
        return target.matches(event_type=event_type, payload=payload)
    return semantic_matcher is not None and semantic_matcher(target)

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

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

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

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

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

@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

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

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

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

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)
    semantic_matcher = semantic.matches if semantic is not None else None
    matches: list[WebhookTriggerResponse] = []
    for trigger in candidates:
        targets = self._cast_runtime_targets(trigger)
        if any(
            matches_webhook_target(
                target=target,
                event_type=event.event_type,
                payload=event.payload,
                semantic_matcher=semantic_matcher,
            )
            for target in targets
        ):
            matches.append(trigger)
    return WebhookTriggerMatch(
        triggers=matches,
        event=(semantic.model_dump(mode="json") if semantic else None),
    )

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

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 missing event metadata before processing a GitHub delivery.

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

async def pre_validate(
    self, headers: Mapping[str, str]
) -> WebhookPreValidationResult:
    """Reject missing event metadata before processing a GitHub delivery.

    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."
        )
    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

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

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

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

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

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

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

@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: SlackSemanticEventFilter) -> bool

Return whether this mention matches an app-mention target.

Parameters:

Name	Type	Description	Default
target	SlackSemanticEventFilter	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

def matches(self, target: SlackSemanticEventFilter) -> 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: SlackSemanticEventFilter) -> bool

Match the required channel identifier.

Parameters:

Name	Type	Description	Default
target	SlackSemanticEventFilter	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

def matches(self, target: SlackSemanticEventFilter) -> 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: SlackSemanticEventFilter) -> bool

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

Parameters:

Name	Type	Description	Default
target	SlackSemanticEventFilter	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

def matches(self, target: SlackSemanticEventFilter) -> 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: SlackSemanticEventFilter) -> bool

Return whether this message matches a message target.

Parameters:

Name	Type	Description	Default
target	SlackSemanticEventFilter	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

def matches(self, target: SlackSemanticEventFilter) -> 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: SlackSemanticEventFilter) -> bool

Return whether this metadata event matches its typed target.

Parameters:

Name	Type	Description	Default
target	SlackSemanticEventFilter	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

def matches(self, target: SlackSemanticEventFilter) -> 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: SlackSemanticEventFilter) -> bool

Return whether this reaction matches its reaction target.

Parameters:

Name	Type	Description	Default
target	SlackSemanticEventFilter	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

def matches(self, target: SlackSemanticEventFilter) -> 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: SlackSemanticEventFilter) -> bool

Match identifiers shared by all Slack semantic events.

Parameters:

Name	Type	Description	Default
target	SlackSemanticEventFilter	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

def matches(self, target: SlackSemanticEventFilter) -> 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: SlackSemanticEventFilter) -> bool

Match the required user identifier.

Parameters:

Name	Type	Description	Default
target	SlackSemanticEventFilter	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

def matches(self, target: SlackSemanticEventFilter) -> 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

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

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

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)
    semantic_matcher = semantic.matches if semantic is not None else None
    matches: list[WebhookTriggerResponse] = []
    for trigger in candidates:
        targets = self._cast_runtime_targets(trigger)
        if any(
            matches_webhook_target(
                target=target,
                event_type=event.event_type,
                payload=event.payload,
                semantic_matcher=semantic_matcher,
            )
            for target in targets
        ):
            matches.append(trigger)
    return WebhookTriggerMatch(
        triggers=matches,
        event=(semantic.model_dump(mode="json") if semantic else None),
    )

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

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

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.