webhook

webhook

WebhookChannel — generic outbound webhook adapter (zero extra deps).

Classes

WebhookChannel

WebhookChannel(
    url: str = "",
    *,
    secret: str = "",
    method: str = "POST",
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Generic outbound webhook channel (send-only).

PARAMETER	DESCRIPTION
url	Target webhook URL. TYPE: str DEFAULT: ''
secret	Optional shared secret sent in the X-Webhook-Secret header. TYPE: str DEFAULT: ''
method	HTTP method (default POST). TYPE: str DEFAULT: 'POST'
bus	Optional event bus for publishing channel events. TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/webhook.py

def __init__(
    self,
    url: str = "",
    *,
    secret: str = "",
    method: str = "POST",
    bus: Optional[EventBus] = None,
) -> None:
    self._url = url
    self._secret = secret
    self._method = method.upper()
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED

Methods:

connect

connect() -> None

Mark as connected (send-only — no persistent connection).

Source code in src/diapason/channels/webhook.py

def connect(self) -> None:
    """Mark as connected (send-only — no persistent connection)."""
    if not self._url:
        logger.warning("No webhook URL configured")
        self._status = ChannelStatus.ERROR
        return
    self._status = ChannelStatus.CONNECTED

disconnect

disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/webhook.py

def disconnect(self) -> None:
    """Mark as disconnected."""
    self._status = ChannelStatus.DISCONNECTED

send

send(
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool

POST a JSON payload to the configured webhook URL.

Source code in src/diapason/channels/webhook.py

def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """POST a JSON payload to the configured webhook URL."""
    if not self._url:
        logger.warning("Cannot send: no webhook URL configured")
        return False

    try:
        import httpx

        payload: Dict[str, Any] = {
            "channel": channel,
            "content": content,
        }
        if conversation_id:
            payload["conversation_id"] = conversation_id
        if metadata:
            payload["metadata"] = metadata

        headers: Dict[str, str] = {}
        if self._secret:
            headers["X-Webhook-Secret"] = self._secret

        resp = httpx.request(
            self._method,
            self._url,
            json=payload,
            headers=headers,
            timeout=10.0,
        )
        if resp.status_code < 300:
            self._publish_sent(channel, content, conversation_id)
            return True
        logger.warning(
            "Webhook returned status %d",
            resp.status_code,
        )
        return False
    except Exception:
        logger.debug("Webhook send failed", exc_info=True)
        return False

status

status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/webhook.py

def status(self) -> ChannelStatus:
    """Return the current connection status."""
    return self._status

list_channels

list_channels() -> List[str]

Return available channel identifiers.

Source code in src/diapason/channels/webhook.py

def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["webhook"]

on_message

on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages (no-op for webhook).

Source code in src/diapason/channels/webhook.py

def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages (no-op for webhook)."""
    self._handlers.append(handler)