Module adcp.webhooks

Webhook creation, signing, and reception for AdCP agents.

Single front door for both senders and receivers. Underlying modules in adcp.signing.webhook_* and adcp.webhook_receiver are implementation details kept for internal organization — prefer the re-exports here for stability.

Which sender helper to use

Functions

async def challenge_webhook_destination(*,
url: str | AnyUrl,
account_id: str,
subscriber_id: str,
sender: WebhookSender | None = None,
authentication: AdCPBaseModel | Mapping[str, Any] | None = None,
challenge: str | None = None,
timeout_seconds: float | None = None,
policy: WebhookDestinationPolicy | None = None,
field: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookChallengeResult
Expand source code 
async def challenge_webhook_destination(
    *,
    url: str | AnyUrl,
    account_id: str,
    subscriber_id: str,
    sender: WebhookSender | None = None,
    authentication: AdCPBaseModel | Mapping[str, Any] | None = None,
    challenge: str | None = None,
    timeout_seconds: float | None = None,
    policy: WebhookDestinationPolicy | None = None,
    field: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookChallengeResult:
    """Validate and prove control of a durable webhook destination.

    Use before activating a new or changed active
    ``sync_accounts.accounts[].notification_configs[]`` entry. Inactive
    configs can be persisted without calling this helper.

    ``authentication`` follows the durable config's legacy auth selector:
    when present, the challenge is sent with Bearer or HMAC-SHA256. When
    omitted, pass an RFC 9421 :class:`WebhookSender`; the helper uses that
    sender's webhook-signing key and the SDK-managed pinned transport.
    """

    error_url = str(url) if isinstance(url, (str, AnyUrl)) else None
    if sender is not None and authentication is not None:
        raise WebhookChallengeError(
            "pass either sender= for RFC 9421 or authentication= for legacy auth, not both",
            reason="ambiguous_auth_mode",
            field=field,
            url=error_url,
        )
    sender_owns_client = bool(getattr(cast(Any, sender), "_owns_client", False))
    sender_transport_hooks = tuple(getattr(cast(Any, sender), "_transport_hooks", ()))
    if sender is not None and not sender_owns_client:
        raise WebhookChallengeError(
            "proof-of-control requires a WebhookSender constructed without client=",
            reason="unsafe_sender_client",
            field=field,
            url=error_url,
        )
    if sender is not None and sender_transport_hooks:
        raise WebhookChallengeError(
            "proof-of-control does not support sender transport_hooks",
            reason="unsupported_sender_hooks",
            field=field,
            url=error_url,
        )
    if sender is not None and not sender.signs_with_rfc9421:
        raise WebhookChallengeError(
            "proof-of-control requires an RFC 9421 WebhookSender when authentication is omitted",
            reason="sender_auth_mode_mismatch",
            field=field,
            url=error_url,
            suggestion=(
                "Use WebhookSender.from_jwk(...) for default durable configs, "
                "or pass config.authentication for legacy Bearer/HMAC configs."
            ),
        )
    if sender is None and authentication is None:
        raise WebhookChallengeError(
            "webhook challenge requires sender= when authentication is omitted",
            reason="sender_required",
            field=field,
            url=error_url,
            suggestion=(
                "Pass the seller's WebhookSender, or pass config.authentication " "for legacy auth."
            ),
        )
    try:
        destination = validate_webhook_destination_url(url, policy=policy, field=field)
        payload = create_webhook_challenge_payload(
            account_id=account_id,
            subscriber_id=subscriber_id,
            challenge=challenge,
        )
    except WebhookDestinationValidationError as exc:
        raise WebhookChallengeError(
            str(exc),
            reason=exc.reason,
            field=exc.field,
            url=exc.url,
            suggestion=exc.suggestion,
        ) from exc
    except ValueError as exc:
        raise WebhookChallengeError(
            f"webhook challenge configuration is invalid: {exc}",
            reason="invalid_configuration",
            field=field,
            url=error_url,
        ) from exc
    challenge_value = payload["challenge"]

    try:
        if sender is not None:
            effective_timeout = (
                timeout_seconds
                if timeout_seconds is not None
                else float(getattr(cast(Any, sender), "_timeout", _DEFAULT_TIMEOUT_SECONDS))
            )
            response = await _send_sender_webhook_challenge(
                url=destination.effective_url,
                sender=sender,
                payload=payload,
                timeout_seconds=effective_timeout,
                policy=destination.policy,
                extra_headers=extra_headers,
            )
            status_code = response.status_code
            response_headers = dict(response.headers)
            response_body = response.content
        else:
            auth_config = _authentication_to_config(cast(Any, authentication))
            response = await _send_legacy_webhook_challenge(
                url=destination.effective_url,
                authentication=auth_config,
                payload=payload,
                extra_headers=extra_headers,
                timeout_seconds=timeout_seconds,
                policy=destination.policy,
            )
            status_code = response.status_code
            response_headers = dict(response.headers)
            response_body = response.content
    except httpx.TimeoutException as exc:
        raise WebhookChallengeError(
            "webhook challenge timed out",
            reason="timeout",
            field=field,
            url=destination.original_url,
        ) from exc
    except httpx.HTTPError as exc:
        raise WebhookChallengeError(
            f"webhook challenge request failed: {exc}",
            reason="request_failed",
            field=field,
            url=destination.original_url,
        ) from exc
    except ValueError as exc:
        raise WebhookChallengeError(
            f"webhook challenge configuration is invalid: {exc}",
            reason="invalid_configuration",
            field=field,
            url=destination.original_url,
        ) from exc

    if not 200 <= status_code < 300:
        raise WebhookChallengeError(
            f"webhook challenge failed with HTTP {status_code}",
            reason="http_status",
            field=field,
            url=destination.original_url,
            status_code=status_code,
        )

    echoed_field = validate_webhook_challenge_response(
        response_body,
        challenge=challenge_value,
        field=field,
        url=destination.original_url,
    )
    return WebhookChallengeResult(
        challenge=challenge_value,
        echoed_field=echoed_field,
        destination=destination,
        status_code=status_code,
        response_headers=response_headers,
        response_body=response_body,
    )

Validate and prove control of a durable webhook destination.

Use before activating a new or changed active sync_accounts.accounts[].notification_configs[] entry. Inactive configs can be persisted without calling this helper.

authentication follows the durable config's legacy auth selector: when present, the challenge is sent with Bearer or HMAC-SHA256. When omitted, pass an RFC 9421 :class:WebhookSender; the helper uses that sender's webhook-signing key and the SDK-managed pinned transport.

def create_a2a_webhook_payload(task_id: str,
status: GeneratedTaskStatus,
context_id: str,
result: PydanticBaseModel | dict[str, Any],
timestamp: datetime | None = None) ‑> a2a_pb2.Task | a2a_pb2.TaskStatusUpdateEvent
Expand source code 
def create_a2a_webhook_payload(
    task_id: str,
    status: GeneratedTaskStatus,
    context_id: str,
    result: PydanticBaseModel | dict[str, Any],
    timestamp: datetime | None = None,
) -> Task | TaskStatusUpdateEvent:
    """
    Create A2A webhook payload (Task or TaskStatusUpdateEvent).

    Per A2A specification:
    - Terminated statuses (completed, failed, canceled, rejected): Returns Task
      with artifacts[].parts[]
    - Intermediate statuses (working, input-required, submitted, auth-required):
      Returns TaskStatusUpdateEvent with status.message.parts[]

    This function helps agent implementations construct properly formatted A2A webhook
    payloads for sending to clients.

    Args:
        task_id: Unique identifier for the task
        status: Current task status
        context_id: Session/conversation identifier (required by A2A protocol)
        timestamp: When the webhook was generated (defaults to current UTC time)
        result: Task-specific payload — any Pydantic model or plain dict

    Returns:
        Task object for terminated statuses, TaskStatusUpdateEvent for intermediate statuses

    Examples:
        Create a completed Task webhook:
        >>> from adcp.webhooks import create_a2a_webhook_payload
        >>> from adcp.types import GeneratedTaskStatus
        >>>
        >>> task = create_a2a_webhook_payload(
        ...     task_id="task_123",
        ...     context_id="ctx_123",
        ...     status=GeneratedTaskStatus.completed,
        ...     result={"products": [...]},
        ... )
        >>> # task is a Task object with artifacts containing the result

        Create a working status update:
        >>> event = create_a2a_webhook_payload(
        ...     task_id="task_456",
        ...     context_id="ctx_456",
        ...     status=GeneratedTaskStatus.working,
        ...     result={"current_step": "processing", "percentage": 30},
        ... )
        >>> # event is a TaskStatusUpdateEvent with status.message

        Send A2A webhook via HTTP POST:
        >>> import httpx
        >>> from a2a.types import Task
        >>>
        >>> payload = create_a2a_webhook_payload(...)
        >>> # Serialize to dict for JSON
        >>> if isinstance(payload, Task):
        ...     payload_dict = payload.model_dump(mode='json')
        ... else:
        ...     payload_dict = payload.model_dump(mode='json')
        >>>
        >>> response = await httpx.post(webhook_url, json=payload_dict)
    """
    if timestamp is None:
        timestamp = datetime.now(timezone.utc)

    # Convert datetime to ISO string for A2A protocol
    timestamp_str = timestamp.isoformat() if isinstance(timestamp, datetime) else timestamp
    timestamp_proto = _isoformat_to_proto_timestamp(timestamp_str) if timestamp_str else None

    # Map GeneratedTaskStatus to A2A TaskState enum value.
    # GeneratedTaskStatus is always an Enum so .value is guaranteed.
    status_value = status.value
    adcp_to_task_state: dict[str, int] = {
        "completed": pb.TaskState.TASK_STATE_COMPLETED,
        "failed": pb.TaskState.TASK_STATE_FAILED,
        "canceled": pb.TaskState.TASK_STATE_CANCELED,
        "rejected": pb.TaskState.TASK_STATE_REJECTED,
        "working": pb.TaskState.TASK_STATE_WORKING,
        "submitted": pb.TaskState.TASK_STATE_SUBMITTED,
        # GeneratedTaskStatus enum values are hyphenated ("input-required",
        # "auth-required"). The underscore forms are accepted as a convenience
        # for callers passing raw strings rather than enum members.
        "input_required": pb.TaskState.TASK_STATE_INPUT_REQUIRED,
        "input-required": pb.TaskState.TASK_STATE_INPUT_REQUIRED,
        "auth_required": pb.TaskState.TASK_STATE_AUTH_REQUIRED,
        "auth-required": pb.TaskState.TASK_STATE_AUTH_REQUIRED,
    }
    task_state_enum = adcp_to_task_state.get(status_value)
    if task_state_enum is None:
        # Falling back to TASK_STATE_UNSPECIFIED (proto3 zero) would be
        # silently omitted by MessageToDict, producing an invalid wire
        # shape ``{"status": {}}`` that A2A v0.3 receivers reject as
        # missing the required ``state`` field. Fail loud at the builder
        # boundary so callers can't ship a broken envelope.
        known = [
            "submitted",
            "working",
            "input-required",
            "completed",
            "canceled",
            "failed",
            "rejected",
            "auth-required",
        ]
        raise ValueError(
            f"create_a2a_webhook_payload: unknown status {status_value!r}. "
            f"Known AdCP→A2A states: {known}. "
            "Note: 'unknown' has no a2a-sdk 1.0 protobuf constant; build a "
            "Task manually and pass it through to_wire_dict if you need to "
            "emit that state."
        )

    # Build parts for the message/artifact.
    parts: list[pb.Part] = []

    # Convert Pydantic model to dict if needed
    if hasattr(result, "model_dump"):
        result_dict: dict[str, Any] = result.model_dump(mode="json")
    else:
        result_dict = result

    value = Value()
    ParseDict(result_dict, value)
    parts.append(pb.Part(data=value))

    # Determine if this is a terminated status (Task) or intermediate (TaskStatusUpdateEvent).
    # canceled and rejected are terminal: the task will not continue.
    is_terminated = status in (
        GeneratedTaskStatus.completed,
        GeneratedTaskStatus.failed,
        GeneratedTaskStatus.canceled,
        GeneratedTaskStatus.rejected,
    )

    if is_terminated:
        status_kwargs: dict[str, Any] = {"state": task_state_enum}
        if timestamp_proto is not None:
            status_kwargs["timestamp"] = timestamp_proto
        task_status = pb.TaskStatus(**status_kwargs)

        artifacts = (
            [
                pb.Artifact(
                    artifact_id=f"{task_id}_result",
                    parts=parts,
                )
            ]
            if parts
            else []
        )

        return pb.Task(
            id=task_id,
            status=task_status,
            artifacts=artifacts,
            context_id=context_id,
        )

    # Intermediate status: build a Message carrying the parts and nest it
    # inside TaskStatus.message so the event mirrors the spec shape.
    message_obj = None
    if parts:
        message_obj = pb.Message(
            message_id=f"{task_id}_msg",
            role=pb.Role.ROLE_AGENT,
            parts=parts,
        )

    status_kwargs = {"state": task_state_enum}
    if timestamp_proto is not None:
        status_kwargs["timestamp"] = timestamp_proto
    if message_obj is not None:
        status_kwargs["message"] = message_obj
    task_status = pb.TaskStatus(**status_kwargs)

    return pb.TaskStatusUpdateEvent(
        task_id=task_id,
        status=task_status,
        context_id=context_id,
    )

Create A2A webhook payload (Task or TaskStatusUpdateEvent).

Per A2A specification: - Terminated statuses (completed, failed, canceled, rejected): Returns Task with artifacts[].parts[] - Intermediate statuses (working, input-required, submitted, auth-required): Returns TaskStatusUpdateEvent with status.message.parts[]

This function helps agent implementations construct properly formatted A2A webhook payloads for sending to clients.

Args
-----=
task_id
Unique identifier for the task
status
Current task status
context_id
Session/conversation identifier (required by A2A protocol)
timestamp
When the webhook was generated (defaults to current UTC time)
result
Task-specific payload — any Pydantic model or plain dict

Returns -----= Task object for terminated statuses, TaskStatusUpdateEvent for intermediate statuses

Examples -----= Create a completed Task webhook:

>>> from adcp.webhooks import create_a2a_webhook_payload
>>> from adcp.types import GeneratedTaskStatus
>>>
>>> task = create_a2a_webhook_payload(
...     task_id="task_123",
...     context_id="ctx_123",
...     status=GeneratedTaskStatus.completed,
...     result={"products": [...]},
... )
>>> # task is a Task object with artifacts containing the result

Create a working status update:

>>> event = create_a2a_webhook_payload(
...     task_id="task_456",
...     context_id="ctx_456",
...     status=GeneratedTaskStatus.working,
...     result={"current_step": "processing", "percentage": 30},
... )
>>> # event is a TaskStatusUpdateEvent with status.message

Send A2A webhook via HTTP POST:

>>> import httpx
>>> from a2a.types import Task
>>>
>>> payload = create_a2a_webhook_payload(...)
>>> # Serialize to dict for JSON
>>> if isinstance(payload, Task):
...     payload_dict = payload.model_dump(mode='json')
... else:
...     payload_dict = payload.model_dump(mode='json')
>>>
>>> response = await httpx.post(webhook_url, json=payload_dict)
def create_mcp_webhook_payload(task_id: str,
status: GeneratedTaskStatus | str,
task_type: TaskType | str,
*,
result: PydanticBaseModel | dict[str, Any] | None = None,
timestamp: datetime | None = None,
operation_id: str,
notification_id: str | None = None,
message: str | None = None,
context_id: str | None = None,
protocol: AdcpProtocol | str | None = None,
idempotency_key: str | None = None,
token: str | None = None) ‑> adcp.types.generated_poc.core.mcp_webhook_payload.McpWebhookPayload
Expand source code 
def create_mcp_webhook_payload(
    task_id: str,
    status: GeneratedTaskStatus | str,
    task_type: TaskType | str,
    *,
    result: PydanticBaseModel | dict[str, Any] | None = None,
    timestamp: datetime | None = None,
    operation_id: str,
    notification_id: str | None = None,
    message: str | None = None,
    context_id: str | None = None,
    protocol: AdcpProtocol | str | None = None,
    idempotency_key: str | None = None,
    token: str | None = None,
) -> McpWebhookPayload:
    """
    Build an :class:`McpWebhookPayload` for a tracked async task.

    Pair with :func:`to_wire_dict` for HTTP transport — Pydantic-typed at
    construction so the publisher catches schema drift before it leaves
    the process.

    ``task_type`` is restricted to the closed :class:`TaskType` enum (the
    spec's complete set of async/tracked operations). Passing a value not
    present in the enum produces a validation error before an invalid webhook
    payload can leave the process.

    Args:
        task_id: Unique identifier for the task.
        status: Current task status.
        task_type: Type of AdCP async operation (see :class:`TaskType`).
        result: Task-specific payload — any Pydantic model or plain dict.
            Plain dicts are validated against
            :class:`AdcpAsyncResponseData`'s discriminated union.
        timestamp: When the webhook was generated. Defaults to current UTC.
        operation_id: Client-generated identifier supplied through
            ``push_notification_config.operation_id``. Required for every
            task webhook; publishers echo it verbatim and MUST NOT derive it
            from the receiver URL.
        notification_id: Stable identity for one logical notification.
            Terminal task webhooks default to ``"{task_id}.terminal"`` so
            re-emissions under different delivery keys still converge.
        message: Human-readable summary of task state.
        context_id: Session/conversation identifier.
        protocol: AdCP protocol this task belongs to (see :class:`AdcpProtocol`).
            Auto-derived from ``task_type`` when omitted, matching the JS
            SDK's ``protocolForTool`` so cross-SDK bodies classify
            operations identically. Pass an explicit value to override.
        idempotency_key: Sender-generated key stable across retries of the
            same event. Defaults to a freshly-generated UUID v4 — callers
            retrying delivery of the same event MUST pass the key from
            their first attempt; passing None twice mints two keys and
            defeats dedup.
        token: Buyer-supplied token from ``push_notification_config.token``,
            echoed back per spec for authenticity validation.

    Returns:
        :class:`McpWebhookPayload` instance. Use :func:`to_wire_dict` (or
        ``payload.model_dump(mode="json", exclude_none=True)``) to get the
        JSON-ready dict for HTTP transport.

    Examples:
        Create a completed webhook with results:
        >>> from adcp.webhooks import create_mcp_webhook_payload, to_wire_dict
        >>> from adcp.types import GeneratedTaskStatus
        >>>
        >>> payload = create_mcp_webhook_payload(
        ...     task_id="task_123",
        ...     operation_id="op_123",
        ...     status=GeneratedTaskStatus.completed,
        ...     task_type="create_media_buy",
        ...     result={"media_buy_id": "mb_1", "buyer_ref": "ref_1"},
        ...     message="Created campaign"
        ... )
        >>> wire = to_wire_dict(payload)

        Create a failed webhook with error:
        >>> payload = create_mcp_webhook_payload(
        ...     task_id="task_456",
        ...     operation_id="op_456",
        ...     status=GeneratedTaskStatus.failed,
        ...     task_type="create_media_buy",
        ...     result={"errors": [{"code": "INVALID_INPUT", "message": "..."}]},
        ...     message="Validation failed"
        ... )

        Create a working status update:
        >>> payload = create_mcp_webhook_payload(
        ...     task_id="task_789",
        ...     operation_id="op_789",
        ...     status=GeneratedTaskStatus.working,
        ...     task_type="sync_creatives",
        ...     message="Processing 3 of 10 creatives"
        ... )
    """
    if timestamp is None:
        timestamp = datetime.now(timezone.utc)
    if idempotency_key is None:
        idempotency_key = generate_webhook_idempotency_key()
    if not operation_id:
        raise ValueError(
            "operation_id is required for AdCP task webhooks; copy "
            "push_notification_config.operation_id verbatim"
        )

    status_value = status.value if hasattr(status, "value") else str(status)
    if notification_id is None and status_value in {
        "completed",
        "failed",
        "canceled",
        "rejected",
    }:
        notification_id = f"{task_id}.terminal"

    # Auto-derive `protocol` from `task_type` when caller doesn't override.
    # Matches `protocolForTool` in the JS reference SDK so cross-SDK bodies
    # classify operations identically.
    if protocol is None:
        try:
            task_type_enum = task_type if isinstance(task_type, TaskType) else TaskType(task_type)
        except ValueError:
            # Unknown string — let `model_validate` raise the canonical
            # task_type error below rather than swallow it here.
            task_type_enum = None
        if task_type_enum is not None:
            protocol = _TASK_TYPE_TO_PROTOCOL.get(task_type_enum)

    # Foreign BaseModel subclasses (anything outside AdcpAsyncResponseData)
    # don't match the discriminated-union variants by identity — dump to a
    # dict so the union picks by shape, matching the dict path.
    result_value: PydanticBaseModel | dict[str, Any] | None
    if isinstance(result, PydanticBaseModel):
        result_value = result.model_dump(mode="json")
    else:
        result_value = result

    payload = McpWebhookPayload.model_validate(
        {
            "idempotency_key": idempotency_key,
            "notification_id": notification_id,
            "task_id": task_id,
            "task_type": task_type,
            "protocol": protocol,
            "status": status_value,
            "timestamp": timestamp,
            "operation_id": operation_id,
            "message": message,
            "context_id": context_id,
            "token": token,
        }
    )
    # Preserve task result payloads byte-for-byte. Validating through the
    # generated AdcpAsyncResponseData union can coerce arbitrary dicts into
    # typed response models and inject response defaults, changing webhook
    # bodies before signing.
    payload.result = result_value  # type: ignore[assignment]
    return payload

Build an :class:McpWebhookPayload for a tracked async task.

Pair with :func:to_wire_dict() for HTTP transport — Pydantic-typed at construction so the publisher catches schema drift before it leaves the process.

task_type is restricted to the closed :class:TaskType enum (the spec's complete set of async/tracked operations). Passing a value not present in the enum produces a validation error before an invalid webhook payload can leave the process.

Args
-----=
task_id
Unique identifier for the task.
status
Current task status.
task_type
Type of AdCP async operation (see :class:TaskType).
result
Task-specific payload — any Pydantic model or plain dict. Plain dicts are validated against :class:AdcpAsyncResponseData's discriminated union.
timestamp
When the webhook was generated. Defaults to current UTC.
operation_id
Client-generated identifier supplied through push_notification_config.operation_id. Required for every task webhook; publishers echo it verbatim and MUST NOT derive it from the receiver URL.
notification_id
Stable identity for one logical notification. Terminal task webhooks default to "{task_id}.terminal" so re-emissions under different delivery keys still converge.
message
Human-readable summary of task state.
context_id
Session/conversation identifier.
protocol
AdCP protocol this task belongs to (see :class:AdcpProtocol). Auto-derived from task_type when omitted, matching the JS SDK's protocolForTool so cross-SDK bodies classify operations identically. Pass an explicit value to override.
idempotency_key
Sender-generated key stable across retries of the same event. Defaults to a freshly-generated UUID v4 — callers retrying delivery of the same event MUST pass the key from their first attempt; passing None twice mints two keys and defeats dedup.
token
Buyer-supplied token from push_notification_config.token, echoed back per spec for authenticity validation.

Returns -----= :class:McpWebhookPayload instance. Use :func:to_wire_dict() (or payload.model_dump(mode="json", exclude_none=True)) to get the JSON-ready dict for HTTP transport.

Examples -----= Create a completed webhook with results:

>>> from adcp.webhooks import create_mcp_webhook_payload, to_wire_dict
>>> from adcp.types import GeneratedTaskStatus
>>>
>>> payload = create_mcp_webhook_payload(
...     task_id="task_123",
...     operation_id="op_123",
...     status=GeneratedTaskStatus.completed,
...     task_type="create_media_buy",
...     result={"media_buy_id": "mb_1", "buyer_ref": "ref_1"},
...     message="Created campaign"
... )
>>> wire = to_wire_dict(payload)

Create a failed webhook with error:

>>> payload = create_mcp_webhook_payload(
...     task_id="task_456",
...     operation_id="op_456",
...     status=GeneratedTaskStatus.failed,
...     task_type="create_media_buy",
...     result={"errors": [{"code": "INVALID_INPUT", "message": "..."}]},
...     message="Validation failed"
... )

Create a working status update:

>>> payload = create_mcp_webhook_payload(
...     task_id="task_789",
...     operation_id="op_789",
...     status=GeneratedTaskStatus.working,
...     task_type="sync_creatives",
...     message="Processing 3 of 10 creatives"
... )
def create_webhook_challenge_payload(*, account_id: str, subscriber_id: str, challenge: str | None = None) ‑> dict[str, str]
Expand source code 
def create_webhook_challenge_payload(
    *,
    account_id: str,
    subscriber_id: str,
    challenge: str | None = None,
) -> dict[str, str]:
    """Build the durable ``notification_configs[]`` challenge payload."""

    if not isinstance(account_id, str) or not account_id:
        raise ValueError("account_id must be a non-empty string")
    if not isinstance(subscriber_id, str) or not subscriber_id:
        raise ValueError("subscriber_id must be a non-empty string")
    challenge_value = generate_webhook_challenge_value() if challenge is None else challenge
    if not isinstance(challenge_value, str) or not challenge_value:
        raise ValueError("challenge must be a non-empty string")
    return {
        "type": "webhook.challenge",
        "challenge": challenge_value,
        "account_id": account_id,
        "subscriber_id": subscriber_id,
    }

Build the durable notification_configs[] challenge payload.

async def deliver(config: AdCPBaseModel | Mapping[str, Any],
payload: AdCPBaseModel | Task | TaskStatusUpdateEvent | Mapping[str, Any],
*,
client: httpx.AsyncClient | None = None,
extra_headers: Mapping[str, str] | None = None,
timeout_seconds: float | None = None,
token_field: str | None = None,
allow_private: bool = False,
allowed_ports: frozenset[int] | None = None) ‑> httpx.Response
Expand source code 
async def deliver(
    config: AdCPBaseModel | Mapping[str, Any],
    payload: AdCPBaseModel | Task | TaskStatusUpdateEvent | Mapping[str, Any],
    *,
    client: httpx.AsyncClient | None = None,
    extra_headers: Mapping[str, str] | None = None,
    timeout_seconds: float | None = None,
    token_field: str | None = None,
    allow_private: bool = False,
    allowed_ports: frozenset[int] | None = None,
) -> httpx.Response:
    """Dispatch one legacy-auth webhook in a single call.

    Collapses the sender's six-step boilerplate (build envelope, serialize,
    sign, merge headers, POST, echo token) into one call so the signer and
    the wire see the *same bytes*. The serialization-format drift that
    plagued the hand-rolled path — ``json=`` in httpx re-serializes the dict
    and breaks ``Content-Digest`` — is structurally impossible here: the
    helper JSON-serializes once, signs those bytes, and POSTs those bytes
    via ``content=``.

    This helper is for the **legacy** AdCP 3.x authentication schemes
    (``Bearer`` / ``HMAC-SHA256``) and emits a :class:`DeprecationWarning`
    on first use. For 4.0+ integrations use :class:`WebhookSender` (RFC 9421).

    Args:
        config: A :class:`PushNotificationConfig`, :class:`ReportingWebhook`,
            or equivalent dict. Must carry ``url`` (``https://`` only) and
            ``authentication.{schemes, credentials}``.
        payload: The webhook body. Accepts a Pydantic model (e.g. built via
            :func:`create_mcp_webhook_payload` / :func:`create_a2a_webhook_payload`),
            an a2a ``Task`` / ``TaskStatusUpdateEvent``, or a plain dict.
            Models are dumped with ``mode="json", exclude_none=True``.
        client: Optional shared ``httpx.AsyncClient``. When supplied, the
            caller owns SSRF guarantees — the helper trusts the operator's
            transport completely (typically a vetted egress proxy with
            mTLS, or an ASGI transport for testing). When omitted, the
            helper builds a per-request :class:`adcp.signing.IpPinnedTransport`
            so the URL is resolved, SSRF-validated, and pinned to the
            resolved IP — same defense applied to :class:`WebhookSender`.
        allow_private: Forwarded to the per-request pinned transport
            (owned-client path only). ``False`` (default) rejects URLs
            whose resolved IP is in a private / loopback / link-local
            range. Set ``True`` for dev/CI fixtures that post to internal
            endpoints; production should leave it ``False``.
        allowed_ports: Forwarded to the per-request pinned transport
            (owned-client path only). ``None`` (default) imposes no port
            filter — AdCP doesn't constrain webhook ports. Hardened
            deployments pass :data:`adcp.signing.DEFAULT_ALLOWED_PORTS`
            (`{443, 8443}`) or a custom set.
        extra_headers: Merged last. May not override any of
            ``Content-Type``, ``Content-Digest``, ``Content-Length``,
            ``Host``, ``Authorization``, ``Signature``, ``Signature-Input``,
            ``X-AdCP-Signature``, or ``X-AdCP-Timestamp``. Auth and
            signature-binding headers are sender-owned so the signer and
            the wire cannot disagree.
        timeout_seconds: Per-request timeout applied only when the helper
            creates its own client. Raises ``ValueError`` if set alongside
            ``client=`` — configure the timeout on the shared client instead.
        token_field: Opt-in field name for echoing ``config.token`` into
            the payload body (top-level for MCP dicts, under ``metadata``
            for ``Task`` / ``TaskStatusUpdateEvent``). Default ``None``
            disables echo; there is no spec-defined field name, so the
            caller must pick one the receiver agrees to read.

    Returns:
        The raw ``httpx.Response``. Caller is responsible for
        ``response.status_code`` inspection and retry scheduling. For retry,
        pass the *same, unmutated* payload again — serialization is
        deterministic so retries produce byte-identical bodies (spec-correct
        receiver dedup via ``idempotency_key``). Mutating the payload dict
        between attempts breaks byte-identity; callers who need byte-identical
        HTTP envelopes across retries (including headers) should use
        :class:`WebhookSender` and :meth:`WebhookSender.resend`. There is
        intentionally no ``resend()`` here — the retry contract is "call
        ``deliver`` again with the same inputs".

    Raises:
        ValueError: missing ``url``, non-HTTPS URL, control characters in
            header values, missing / unknown ``authentication`` (use
            :class:`WebhookSender` for RFC 9421), overriding a reserved
            header, or setting ``timeout_seconds`` alongside ``client``.
        DeprecationWarning (fires once): ``authentication`` is a 3.x fallback.

    Security notes:
        * ``config.url`` is buyer-controlled. The helper enforces HTTPS,
          rejects control characters, AND (on the owned-client path)
          builds a per-request IP-pinned transport that runs the full
          SSRF range check (loopback / RFC 1918 / link-local / CGNAT /
          IPv6 ULA / multicast / cloud metadata) and pins the connection
          to the validated IP. Operator-supplied clients skip the SSRF
          guard — they own egress policy on their transport.
        * ``config.token`` sits in the request body, so any receiver that
          logs bodies retains it indefinitely. Treat the token as a
          medium-sensitivity correlator, not a long-lived secret.
        * At ``httpx`` DEBUG log level, ``Authorization`` and
          ``X-AdCP-Signature`` appear in logs — gate DEBUG in production.
    """
    if client is not None and timeout_seconds is not None:
        raise ValueError(
            "timeout_seconds cannot be set when client= is provided; "
            "configure the timeout on your shared httpx.AsyncClient instead."
        )

    url, token, auth_scheme, credentials = _extract_config_fields(config)

    if auth_scheme is None:
        raise ValueError(
            "config.authentication is required for deliver(). "
            "For RFC 9421 signing (the AdCP 4.0 default), use "
            "adcp.webhooks.WebhookSender — no helper for unsigned webhooks "
            "is provided because the spec requires signing."
        )
    if auth_scheme not in ("Bearer", "HMAC-SHA256"):
        raise ValueError(
            f"unknown authentication scheme {auth_scheme!r}; "
            "supported legacy schemes are 'Bearer' and 'HMAC-SHA256'. "
            "For RFC 9421 use adcp.webhooks.WebhookSender."
        )

    _warn_auth_deprecation_once()

    # Build the pinned transport up-front (owned-client path). SSRF
    # validation runs synchronously inside ``build_async_ip_pinned_transport``
    # — a hostile URL raises ``SSRFValidationError`` before we serialize
    # the body or compute the HMAC, so a buyer-supplied 127.0.0.1 URL
    # does not produce an HMAC-over-buyer-body sitting in process memory
    # for fault-handlers / custom logging to capture on exception.
    # Mirrors the WebhookSender._send_bytes ordering.
    transport: Any = None
    if client is None:
        from adcp.signing.ip_pinned_transport import build_async_ip_pinned_transport

        transport = build_async_ip_pinned_transport(
            url,
            allow_private=allow_private,
            allowed_ports=allowed_ports,
        )

    body_dict = to_wire_dict(payload)
    if token is not None and token_field is not None:
        _validate_header_value("config.token", token)
        _inject_push_token(body_dict, token, payload, token_field)

    # Compact separators so the signer and the wire see byte-identical
    # payloads, matching the canonical on-wire form pinned by
    # adcontextprotocol/adcp#2478. ``_compute_legacy_signature`` returns the
    # same compact body bytes below — we serialize here for the size check
    # and Bearer path, which both operate on the final transmitted bytes.
    body_bytes = json.dumps(body_dict, separators=(",", ":")).encode("utf-8")
    if len(body_bytes) > _MAX_BODY_BYTES:
        raise ValueError(
            f"serialized webhook body is {len(body_bytes):,} bytes, over the "
            f"{_MAX_BODY_BYTES:,}-byte cap. Split into smaller webhooks or use "
            "the batch-reporting endpoints — most receivers reject bodies over "
            "10MB at the reverse proxy anyway."
        )

    headers: dict[str, str] = {"Content-Type": "application/json"}

    if auth_scheme == "Bearer":
        if not credentials:
            raise ValueError(
                "config.authentication.schemes=['Bearer'] requires "
                "authentication.credentials (min 32 characters — token "
                "exchanged out-of-band with the receiver)."
            )
        _validate_header_value("authentication.credentials", credentials)
        headers["Authorization"] = f"Bearer {credentials}"
    else:  # HMAC-SHA256
        if not credentials:
            raise ValueError(
                "config.authentication.schemes=['HMAC-SHA256'] requires "
                "authentication.credentials (min 32 characters — shared "
                "secret exchanged out-of-band with the receiver)."
            )
        _validate_header_value("authentication.credentials", credentials)
        get_adcp_signed_headers_for_webhook(
            headers,
            secret=credentials,
            timestamp=str(int(time.time())),
            payload=body_dict,
        )

    if extra_headers:
        if len(extra_headers) > _MAX_EXTRA_HEADERS:
            raise ValueError(
                f"extra_headers has {len(extra_headers)} entries; "
                f"helper caps at {_MAX_EXTRA_HEADERS}. Pass only the custom "
                "headers you actually need (trace IDs, correlation IDs)."
            )
        for key in extra_headers:
            normalized = str(key).lower()
            if normalized in _RESERVED_HEADERS or normalized.startswith(":"):
                raise ValueError(_reserved_header_message(normalized, key))
        for key, value in extra_headers.items():
            _validate_header_value(f"extra_headers[{key!r}]", value)
            headers[key] = value

    effective_timeout = timeout_seconds if timeout_seconds is not None else _DEFAULT_TIMEOUT_SECONDS
    if client is None:
        # Owned-client path. ``transport`` was built up-front so SSRF
        # rejected before signing; here we just construct the per-request
        # client. ``follow_redirects=False`` closes rebinding-via-redirect;
        # ``trust_env=False`` blocks ``HTTPS_PROXY`` env-var bypass.
        # Same shape as ``WebhookSender._send_bytes``.
        async with httpx.AsyncClient(
            transport=transport,
            timeout=effective_timeout,
            follow_redirects=False,
            trust_env=False,
        ) as http_client:
            return await http_client.post(url, content=body_bytes, headers=headers)
    # Operator-supplied client: trust them completely; they own SSRF
    # guarantees on their transport (vetted egress proxy, ASGI test
    # transport, etc.).
    return await client.post(url, content=body_bytes, headers=headers)

Dispatch one legacy-auth webhook in a single call.

Collapses the sender's six-step boilerplate (build envelope, serialize, sign, merge headers, POST, echo token) into one call so the signer and the wire see the same bytes. The serialization-format drift that plagued the hand-rolled path — json= in httpx re-serializes the dict and breaks Content-Digest — is structurally impossible here: the helper JSON-serializes once, signs those bytes, and POSTs those bytes via content=.

This helper is for the legacy AdCP 3.x authentication schemes (Bearer / HMAC-SHA256) and emits a :class:DeprecationWarning on first use. For 4.0+ integrations use :class:WebhookSender (RFC 9421).

Args
-----=
config
A :class:PushNotificationConfig, :class:ReportingWebhook, or equivalent dict. Must carry url (https:// only) and authentication.{schemes, credentials}.
payload
The webhook body. Accepts a Pydantic model (e.g. built via :func:create_mcp_webhook_payload() / :func:create_a2a_webhook_payload()), an a2a Task / TaskStatusUpdateEvent, or a plain dict. Models are dumped with mode="json", exclude_none=True.
client
Optional shared httpx.AsyncClient. When supplied, the caller owns SSRF guarantees — the helper trusts the operator's transport completely (typically a vetted egress proxy with mTLS, or an ASGI transport for testing). When omitted, the helper builds a per-request :class:IpPinnedTransport so the URL is resolved, SSRF-validated, and pinned to the resolved IP — same defense applied to :class:WebhookSender.
allow_private
Forwarded to the per-request pinned transport (owned-client path only). False (default) rejects URLs whose resolved IP is in a private / loopback / link-local range. Set True for dev/CI fixtures that post to internal endpoints; production should leave it False.
allowed_ports
Forwarded to the per-request pinned transport (owned-client path only). None (default) imposes no port filter — AdCP doesn't constrain webhook ports. Hardened deployments pass :data:adcp.signing.DEFAULT_ALLOWED_PORTS ({443, 8443}) or a custom set.
extra_headers
Merged last. May not override any of Content-Type, Content-Digest, Content-Length, Host, Authorization, Signature, Signature-Input, X-AdCP-Signature, or X-AdCP-Timestamp. Auth and signature-binding headers are sender-owned so the signer and the wire cannot disagree.
timeout_seconds
Per-request timeout applied only when the helper creates its own client. Raises ValueError if set alongside client= — configure the timeout on the shared client instead.
token_field
Opt-in field name for echoing config.token into the payload body (top-level for MCP dicts, under metadata for Task / TaskStatusUpdateEvent). Default None disables echo; there is no spec-defined field name, so the caller must pick one the receiver agrees to read.

Returns -----= The raw httpx.Response. Caller is responsible for response.status_code inspection and retry scheduling. For retry, pass the same, unmutated payload again — serialization is deterministic so retries produce byte-identical bodies (spec-correct receiver dedup via idempotency_key). Mutating the payload dict between attempts breaks byte-identity; callers who need byte-identical HTTP envelopes across retries (including headers) should use :class:WebhookSender and :meth:WebhookSender.resend(). There is intentionally no resend() here — the retry contract is "call deliver() again with the same inputs".

Raises
-----=
ValueError
missing url, non-HTTPS URL, control characters in header values, missing / unknown authentication (use :class:WebhookSender for RFC 9421), overriding a reserved header, or setting timeout_seconds alongside client.

DeprecationWarning (fires once): authentication is a 3.x fallback. Security notes: * config.url is buyer-controlled. The helper enforces HTTPS, rejects control characters, AND (on the owned-client path) builds a per-request IP-pinned transport that runs the full SSRF range check (loopback / RFC 1918 / link-local / CGNAT / IPv6 ULA / multicast / cloud metadata) and pins the connection to the validated IP. Operator-supplied clients skip the SSRF guard — they own egress policy on their transport. * config.token sits in the request body, so any receiver that logs bodies retains it indefinitely. Treat the token as a medium-sensitivity correlator, not a long-lived secret. * At httpx DEBUG log level, Authorization and X-AdCP-Signature appear in logs — gate DEBUG in production.

def extract_webhook_result_data(webhook_payload: dict[str, Any]) ‑> dict[str, typing.Any] | None
Expand source code 
def extract_webhook_result_data(webhook_payload: dict[str, Any]) -> dict[str, Any] | None:
    """
    Extract result data from webhook payload (MCP or A2A format).

    This utility function handles webhook payloads from both MCP and A2A protocols,
    extracting the result data regardless of the webhook format. Useful for quick
    inspection, logging, or custom webhook routing logic without requiring full
    client initialization.

    Protocol Detection:
    - A2A Task: Has "artifacts" field (terminated statuses: completed, failed, canceled, rejected)
    - A2A TaskStatusUpdateEvent: Has nested "status.message" structure (intermediate statuses)
    - MCP: Has "result" field directly

    Args:
        webhook_payload: Raw webhook dictionary from HTTP request (JSON-deserialized)

    Returns:
        dict[str, Any] containing the extracted AdCP response data, or None if no
        result is present. For A2A webhooks, unwraps data from artifacts/message parts
        structure. For MCP webhooks, returns the result field directly.

    Examples:
        Extract from MCP webhook:
        >>> mcp_payload = {
        ...     "task_id": "task_123",
        ...     "task_type": "create_media_buy",
        ...     "status": "completed",
        ...     "timestamp": "2025-01-15T10:00:00Z",
        ...     "result": {"media_buy_id": "mb_123", "buyer_ref": "ref_123", "packages": []}
        ... }
        >>> result = extract_webhook_result_data(mcp_payload)
        >>> print(result["media_buy_id"])
        mb_123

        Extract from A2A Task webhook:
        >>> a2a_task_payload = {
        ...     "id": "task_456",
        ...     "context_id": "ctx_456",
        ...     "status": {"state": "completed", "timestamp": "2025-01-15T10:00:00Z"},
        ...     "artifacts": [
        ...         {
        ...             "artifact_id": "artifact_456",
        ...             "parts": [
        ...                 {
        ...                     "data": {
        ...                         "media_buy_id": "mb_456",
        ...                         "buyer_ref": "ref_456",
        ...                         "packages": []
        ...                     }
        ...                 }
        ...             ]
        ...         }
        ...     ]
        ... }
        >>> result = extract_webhook_result_data(a2a_task_payload)
        >>> print(result["media_buy_id"])
        mb_456

        Extract from A2A TaskStatusUpdateEvent webhook:
        >>> a2a_event_payload = {
        ...     "task_id": "task_789",
        ...     "context_id": "ctx_789",
        ...     "status": {
        ...         "state": "working",
        ...         "timestamp": "2025-01-15T10:00:00Z",
        ...         "message": {
        ...             "message_id": "msg_789",
        ...             "role": "agent",
        ...             "parts": [
        ...                 {"data": {"current_step": "processing", "percentage": 50}}
        ...             ]
        ...         }
        ...     },
        ...     "final": False
        ... }
        >>> result = extract_webhook_result_data(a2a_event_payload)
        >>> print(result["percentage"])
        50

        Handle webhook with no result:
        >>> empty_payload = {"task_id": "task_000", "status": "working", "timestamp": "..."}
        >>> result = extract_webhook_result_data(empty_payload)
        >>> print(result)
        None
    """
    # Detect A2A Task format (has "artifacts" field)
    if "artifacts" in webhook_payload:
        # Extract from task.artifacts[].parts[]
        artifacts = webhook_payload.get("artifacts", [])
        if not artifacts:
            return None

        # Use last artifact (most recent)
        target_artifact = artifacts[-1]
        parts = target_artifact.get("parts", [])
        if not parts:
            return None

        # Find DataPart (skip TextPart)
        for part in parts:
            # Check if this part has "data" field (DataPart)
            if "data" in part:
                data = part["data"]
                # Unwrap {"response": {...}} wrapper if present (A2A convention)
                if isinstance(data, dict) and "response" in data and len(data) == 1:
                    return cast(dict[str, Any], data["response"])
                return cast(dict[str, Any], data)

        return None

    # Detect A2A TaskStatusUpdateEvent format (has nested "status.message")
    status = webhook_payload.get("status")
    if isinstance(status, dict):
        message = status.get("message")
        if isinstance(message, dict):
            # Extract from status.message.parts[]
            parts = message.get("parts", [])
            if not parts:
                return None

            # Find DataPart
            for part in parts:
                if "data" in part:
                    data = part["data"]
                    # Unwrap {"response": {...}} wrapper if present
                    if isinstance(data, dict) and "response" in data and len(data) == 1:
                        return cast(dict[str, Any], data["response"])
                    return cast(dict[str, Any], data)

            return None

    # MCP format: result field directly
    return cast(dict[str, Any] | None, webhook_payload.get("result"))

Extract result data from webhook payload (MCP or A2A format).

This utility function handles webhook payloads from both MCP and A2A protocols, extracting the result data regardless of the webhook format. Useful for quick inspection, logging, or custom webhook routing logic without requiring full client initialization.

Protocol Detection: - A2A Task: Has "artifacts" field (terminated statuses: completed, failed, canceled, rejected) - A2A TaskStatusUpdateEvent: Has nested "status.message" structure (intermediate statuses) - MCP: Has "result" field directly

Args
-----=
webhook_payload
Raw webhook dictionary from HTTP request (JSON-deserialized)

Returns -----= dict[str, Any] containing the extracted AdCP response data, or None if no result is present. For A2A webhooks, unwraps data from artifacts/message parts structure. For MCP webhooks, returns the result field directly.

Examples -----= Extract from MCP webhook:

>>> mcp_payload = {
...     "task_id": "task_123",
...     "task_type": "create_media_buy",
...     "status": "completed",
...     "timestamp": "2025-01-15T10:00:00Z",
...     "result": {"media_buy_id": "mb_123", "buyer_ref": "ref_123", "packages": []}
... }
>>> result = extract_webhook_result_data(mcp_payload)
>>> print(result["media_buy_id"])
mb_123

Extract from A2A Task webhook:

>>> a2a_task_payload = {
...     "id": "task_456",
...     "context_id": "ctx_456",
...     "status": {"state": "completed", "timestamp": "2025-01-15T10:00:00Z"},
...     "artifacts": [
...         {
...             "artifact_id": "artifact_456",
...             "parts": [
...                 {
...                     "data": {
...                         "media_buy_id": "mb_456",
...                         "buyer_ref": "ref_456",
...                         "packages": []
...                     }
...                 }
...             ]
...         }
...     ]
... }
>>> result = extract_webhook_result_data(a2a_task_payload)
>>> print(result["media_buy_id"])
mb_456

Extract from A2A TaskStatusUpdateEvent webhook:

>>> a2a_event_payload = {
...     "task_id": "task_789",
...     "context_id": "ctx_789",
...     "status": {
...         "state": "working",
...         "timestamp": "2025-01-15T10:00:00Z",
...         "message": {
...             "message_id": "msg_789",
...             "role": "agent",
...             "parts": [
...                 {"data": {"current_step": "processing", "percentage": 50}}
...             ]
...         }
...     },
...     "final": False
... }
>>> result = extract_webhook_result_data(a2a_event_payload)
>>> print(result["percentage"])
50

Handle webhook with no result:

>>> empty_payload = {"task_id": "task_000", "status": "working", "timestamp": "..."}
>>> result = extract_webhook_result_data(empty_payload)
>>> print(result)
None
def generate_webhook_challenge_value() ‑> str
Expand source code 
def generate_webhook_challenge_value() -> str:
    """Generate an opaque random value for a proof-of-control challenge."""

    return f"wch_{secrets.token_urlsafe(32)}"

Generate an opaque random value for a proof-of-control challenge.

def generate_webhook_idempotency_key() ‑> str
Expand source code 
def generate_webhook_idempotency_key() -> str:
    """Generate a cryptographically random idempotency_key for a webhook event.

    Returns a UUID v4 prefixed with ``whk_`` — matches the example format in
    ``webhooks.mdx`` and stays within the spec's length + charset bounds
    (``^[A-Za-z0-9_.:-]{16,255}$``).

    Publishers generate this once per delivery and reuse it for exact retries.
    A re-emission of the same logical event uses a new delivery key while
    retaining its stable ``notification_id``.
    """
    return f"whk_{uuid.uuid4()}"

Generate a cryptographically random idempotency_key for a webhook event.

Returns a UUID v4 prefixed with whk_ — matches the example format in webhooks.mdx and stays within the spec's length + charset bounds (^[A-Za-z0-9_.:-]{16,255}$).

Publishers generate this once per delivery and reuse it for exact retries. A re-emission of the same logical event uses a new delivery key while retaining its stable notification_id.

def get_adcp_signed_headers_for_webhook(headers: dict[str, Any],
secret: str,
timestamp: str | int | None,
payload: dict[str, Any] | AdCPBaseModel) ‑> dict[str, typing.Any]
Expand source code 
def get_adcp_signed_headers_for_webhook(
    headers: dict[str, Any],
    secret: str,
    timestamp: str | int | None,
    payload: dict[str, Any] | AdCPBaseModel,
) -> dict[str, Any]:
    """
    Generate AdCP-compliant signed headers for webhook delivery.

    This function creates a cryptographic signature that proves the webhook
    came from an authorized agent and protects against replay attacks by
    including a timestamp in the signed message.

    The function adds two headers to the provided headers dict:
    - X-AdCP-Signature: HMAC-SHA256 signature in format "sha256=<hex_digest>"
    - X-AdCP-Timestamp: Unix timestamp in seconds

    The signing algorithm:
    1. Constructs message as "{timestamp}.{json_payload}"
    2. JSON-serializes payload with default separators (matches wire format from json= kwarg)
    3. UTF-8 encodes the message
    4. HMAC-SHA256 signs with the shared secret
    5. Hex-encodes and prefixes with "sha256="

    Args:
        headers: Existing headers dictionary to add signature headers to
        secret: Shared secret key for HMAC signing
        timestamp: Unix timestamp in seconds (str or int). If None, uses current time.
        payload: Webhook payload (dict or Pydantic model - will be JSON-serialized)

    Returns:
        The modified headers dictionary with signature headers added

    Examples:
        Sign and send an MCP webhook:
        >>> import time
        >>> from adcp.webhooks import create_mcp_webhook_payload
        >>> from adcp.webhooks import get_adcp_signed_headers_for_webhook
        >>>
        >>> payload = create_mcp_webhook_payload(
        ...     task_id="task_123",
        ...     operation_id="op_123",
        ...     status="completed",
        ...     task_type="create_media_buy",
        ...     result={"media_buy_id": "mb_1"},
        ... )
        >>> headers = {"Content-Type": "application/json"}
        >>> signed_headers = get_adcp_signed_headers_for_webhook(
        ...     headers, secret="my-webhook-secret", timestamp=str(int(time.time())),
        ...     payload=payload,
        ... )
        >>>
        >>> # Send webhook with signed headers
        >>> import httpx
        >>> response = await httpx.post(
        ...     webhook_url,
        ...     json=payload,
        ...     headers=signed_headers
        ... )

        Headers will contain:
        >>> print(signed_headers)
        {
            "Content-Type": "application/json",
            "X-AdCP-Signature": "sha256=a1b2c3...",
            "X-AdCP-Timestamp": "1773185740"
        }
    """
    signature_headers, _body_bytes = _compute_legacy_signature(
        secret=secret, timestamp=timestamp, payload=payload
    )
    headers.update(signature_headers)
    return headers

Generate AdCP-compliant signed headers for webhook delivery.

This function creates a cryptographic signature that proves the webhook came from an authorized agent and protects against replay attacks by including a timestamp in the signed message.

The function adds two headers to the provided headers dict: - X-AdCP-Signature: HMAC-SHA256 signature in format "sha256=" - X-AdCP-Timestamp: Unix timestamp in seconds

The signing algorithm: 1. Constructs message as "{timestamp}.{json_payload}" 2. JSON-serializes payload with default separators (matches wire format from json= kwarg) 3. UTF-8 encodes the message 4. HMAC-SHA256 signs with the shared secret 5. Hex-encodes and prefixes with "sha256="

Args
-----=
headers
Existing headers dictionary to add signature headers to
secret
Shared secret key for HMAC signing
timestamp
Unix timestamp in seconds (str or int). If None, uses current time.
payload
Webhook payload (dict or Pydantic model - will be JSON-serialized)

Returns -----= The modified headers dictionary with signature headers added

Examples -----= Sign and send an MCP webhook:

>>> import time
>>> from adcp.webhooks import create_mcp_webhook_payload
>>> from adcp.webhooks import get_adcp_signed_headers_for_webhook
>>>
>>> payload = create_mcp_webhook_payload(
...     task_id="task_123",
...     operation_id="op_123",
...     status="completed",
...     task_type="create_media_buy",
...     result={"media_buy_id": "mb_1"},
... )
>>> headers = {"Content-Type": "application/json"}
>>> signed_headers = get_adcp_signed_headers_for_webhook(
...     headers, secret="my-webhook-secret", timestamp=str(int(time.time())),
...     payload=payload,
... )
>>>
>>> # Send webhook with signed headers
>>> import httpx
>>> response = await httpx.post(
...     webhook_url,
...     json=payload,
...     headers=signed_headers
... )

Headers will contain:

>>> print(signed_headers)
{
    "Content-Type": "application/json",
    "X-AdCP-Signature": "sha256=a1b2c3...",
    "X-AdCP-Timestamp": "1773185740"
}
def sign_legacy_webhook(secret: str,
payload: dict[str, Any] | AdCPBaseModel,
*,
timestamp: str | int | None = None,
headers: dict[str, Any] | None = None) ‑> tuple[dict[str, str], bytes]
Expand source code 
def sign_legacy_webhook(
    secret: str,
    payload: dict[str, Any] | AdCPBaseModel,
    *,
    timestamp: str | int | None = None,
    headers: dict[str, Any] | None = None,
) -> tuple[dict[str, str], bytes]:
    """Return ``(signed_headers, body_bytes)`` for a legacy HMAC webhook.

    Byte-equality between signature input and HTTP body is guaranteed —
    callers POST ``content=body_bytes`` instead of ``json=payload``, so the
    separator-drift trap that caused silent 401s in every spaced-vs-compact
    interop is structurally impossible here.

    This is a lower-level companion to :func:`deliver` for callers who need
    to own the HTTP transport themselves (custom auth, pre-configured
    ``httpx.AsyncClient``, non-httpx clients). For the one-shot "send a
    webhook" path, prefer :func:`deliver`.

    The returned ``body_bytes`` use compact separators (``","``/``":"``)
    matching the canonical on-wire form pinned by adcontextprotocol/adcp#2478.

    Example:
        >>> signed, body = sign_legacy_webhook("shared-secret", payload)
        >>> headers = {**signed, "Content-Type": "application/json"}
        >>> await client.post(url, content=body, headers=headers)
    """
    signature_headers, body_bytes = _compute_legacy_signature(
        secret=secret, timestamp=timestamp, payload=payload
    )
    if headers is not None:
        merged = {str(k): str(v) for k, v in headers.items()}
        merged.update(signature_headers)
        return merged, body_bytes
    return signature_headers, body_bytes

Return (signed_headers, body_bytes) for a legacy HMAC webhook.

Byte-equality between signature input and HTTP body is guaranteed — callers POST content=body_bytes instead of json=payload, so the separator-drift trap that caused silent 401s in every spaced-vs-compact interop is structurally impossible here.

This is a lower-level companion to :func:deliver() for callers who need to own the HTTP transport themselves (custom auth, pre-configured httpx.AsyncClient, non-httpx clients). For the one-shot "send a webhook" path, prefer :func:deliver().

The returned body_bytes use compact separators (","/":") matching the canonical on-wire form pinned by adcontextprotocol/adcp#2478.

Example -----=

>>> signed, body = sign_legacy_webhook("shared-secret", payload)
>>> headers = {**signed, "Content-Type": "application/json"}
>>> await client.post(url, content=body, headers=headers)
def sign_webhook(*,
method: str,
url: str,
headers: Mapping[str, str],
body: bytes,
private_key: PrivateKey,
key_id: str,
alg: str,
created: int | None = None,
expires_in_seconds: int = 300,
nonce: str | None = None,
label: str = 'sig1') ‑> SignedHeaders
Expand source code 
def sign_webhook(
    *,
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes,
    private_key: PrivateKey,
    key_id: str,
    alg: str,
    created: int | None = None,
    expires_in_seconds: int = DEFAULT_EXPIRES_IN_SECONDS,
    nonce: str | None = None,
    label: str = SIG_LABEL_DEFAULT,
) -> SignedHeaders:
    """Sign an outgoing webhook POST per adcp/webhook-signing/v1.

    ``cover_content_digest=True`` and ``tag=WEBHOOK_TAG`` are pinned. The
    caller attaches ``SignedHeaders.as_dict()`` to the outgoing HTTP request.

    The ``method`` is normally ``"POST"`` for webhook delivery; passed through
    unchanged so callers signing a retried ``PUT`` or variant delivery verb
    are not forced into an extra translation.

    See also:
        :class:`adcp.webhooks.WebhookSender` — higher-level one-call helper
        that builds the payload, signs, and POSTs in a single call. Prefer it
        unless you need to own the HTTP transport yourself.
    """
    return sign_request(
        method=method,
        url=url,
        headers=headers,
        body=body,
        private_key=private_key,
        key_id=key_id,
        alg=alg,
        cover_content_digest=True,
        created=created,
        expires_in_seconds=expires_in_seconds,
        nonce=nonce,
        tag=WEBHOOK_TAG,
        label=label,
        signing_profile_version="3.2",
    )

Sign an outgoing webhook POST per adcp/webhook-signing/v1.

cover_content_digest=True and tag=WEBHOOK_TAG are pinned. The caller attaches SignedHeaders.as_dict() to the outgoing HTTP request.

The method is normally "POST" for webhook delivery; passed through unchanged so callers signing a retried PUT or variant delivery verb are not forced into an extra translation.

See also: :class:WebhookSender — higher-level one-call helper that builds the payload, signs, and POSTs in a single call. Prefer it unless you need to own the HTTP transport yourself.

def to_wire_dict(payload: AdCPBaseModel | Task | TaskStatusUpdateEvent | Mapping[str, Any]) ‑> dict[str, typing.Any]
Expand source code 
def to_wire_dict(
    payload: AdCPBaseModel | Task | TaskStatusUpdateEvent | Mapping[str, Any],
) -> dict[str, Any]:
    """Serialize any AdCP webhook payload to a JSON-ready dict.

    Single seam for adopters that accept "any AdCP webhook payload" — a
    sender wrapping :func:`create_a2a_webhook_payload` and
    :func:`create_mcp_webhook_payload` would otherwise have to write
    per-shape dispatch (``isinstance`` checks, ``MessageToDict`` for
    protobuf, ``model_dump`` for Pydantic, passthrough for dict). Brittle:
    a future a2a-sdk that swaps protobuf for a Pydantic façade silently
    changes which branch runs, and adopters duplicate the dispatch in
    every send path. Use this helper instead — the dispatch lives here.

    Behaviour by input shape:

    * a2a ``Task`` / ``TaskStatusUpdateEvent`` (protobuf, a2a-sdk 1.0+) →
      ``MessageToDict(..., preserving_proto_field_name=False)`` so JSON
      keys match the A2A wire spec (camelCase: ``id``, ``contextId``,
      ``artifactId``). Enum values are normalized from the 1.0 protobuf
      form (``TASK_STATE_COMPLETED``, ``ROLE_AGENT``) to the 0.3-spec
      lowercase form (``completed``, ``agent``) so 0.3 buyer receivers
      keep parsing.
    * Any Pydantic model (``McpWebhookPayload``, future Pydantic façades,
      :class:`AdCPBaseModel` subclasses) → ``model_dump(mode="json",
      exclude_none=True)``.
    * ``Mapping`` → coerced to ``dict``. Legacy adopter passthrough for
      callers that build the wire dict by hand.

    Raises:
        TypeError: payload is none of the above.
    """
    if isinstance(payload, (Task, TaskStatusUpdateEvent)):
        data = MessageToDict(payload, preserving_proto_field_name=False)
        _normalize_a2a_task_state_to_v03(data)
        return data
    if hasattr(payload, "model_dump"):
        model = cast(AdCPBaseModel, payload)
        return model.model_dump(mode="json", exclude_none=True)
    if isinstance(payload, Mapping):
        return dict(payload)
    raise TypeError(
        f"Unsupported webhook payload type {type(payload).__name__}: expected "
        "a2a Task / TaskStatusUpdateEvent (protobuf), an AdCP Pydantic model "
        "(e.g. McpWebhookPayload), or a Mapping[str, Any]."
    )

Serialize any AdCP webhook payload to a JSON-ready dict.

Single seam for adopters that accept "any AdCP webhook payload" — a sender wrapping :func:create_a2a_webhook_payload() and :func:create_mcp_webhook_payload() would otherwise have to write per-shape dispatch (isinstance checks, MessageToDict for protobuf, model_dump for Pydantic, passthrough for dict). Brittle: a future a2a-sdk that swaps protobuf for a Pydantic façade silently changes which branch runs, and adopters duplicate the dispatch in every send path. Use this helper instead — the dispatch lives here.

Behaviour by input shape:

  • a2a Task / TaskStatusUpdateEvent (protobuf, a2a-sdk 1.0+) → MessageToDict(..., preserving_proto_field_name=False) so JSON keys match the A2A wire spec (camelCase: id, contextId, artifactId). Enum values are normalized from the 1.0 protobuf form (TASK_STATE_COMPLETED, ROLE_AGENT) to the 0.3-spec lowercase form (completed, agent) so 0.3 buyer receivers keep parsing.
  • Any Pydantic model (McpWebhookPayload, future Pydantic façades, :class:AdCPBaseModel subclasses) → model_dump(mode="json", exclude_none=True).
  • Mapping → coerced to dict. Legacy adopter passthrough for callers that build the wire dict by hand.
Raises
-----=
TypeError
payload is none of the above.
def validate_webhook_challenge_response(response: bytes | Mapping[str, Any],
*,
challenge: str,
field: str | None = None,
url: str | None = None) ‑> str
Expand source code 
def validate_webhook_challenge_response(
    response: bytes | Mapping[str, Any],
    *,
    challenge: str,
    field: str | None = None,
    url: str | None = None,
) -> str:
    """Validate that a receiver echoed the challenge value.

    Receivers may respond with either ``{"challenge": "<value>"}`` or
    ``{"token": "<value>"}``. The return value is the field that matched.
    """

    try:
        if isinstance(response, bytes):
            decoded = json.loads(response.decode("utf-8"))
        else:
            decoded = dict(response)
    except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as exc:
        raise WebhookChallengeError(
            "webhook challenge response must be a JSON object",
            reason="invalid_json",
            field=field,
            url=url,
        ) from exc

    if not isinstance(decoded, Mapping):
        raise WebhookChallengeError(
            "webhook challenge response must be a JSON object",
            reason="invalid_json",
            field=field,
            url=url,
        )

    for key in ("challenge", "token"):
        value = decoded.get(key)
        if value == challenge:
            return key

    if "challenge" in decoded or "token" in decoded:
        reason = "challenge_mismatch"
        message = "webhook challenge response did not echo the expected value"
    else:
        reason = "missing_echo"
        message = "webhook challenge response must include 'challenge' or 'token'"
    raise WebhookChallengeError(message, reason=reason, field=field, url=url)

Validate that a receiver echoed the challenge value.

Receivers may respond with either {"challenge": "<value>"} or {"token": "<value>"}. The return value is the field that matched.

def validate_webhook_destination_url(url: str | AnyUrl,
*,
policy: WebhookDestinationPolicy | None = None,
field: str | None = None) ‑> WebhookDestinationValidation
Expand source code 
def validate_webhook_destination_url(
    url: str | AnyUrl,
    *,
    policy: WebhookDestinationPolicy | None = None,
    field: str | None = None,
) -> WebhookDestinationValidation:
    """Validate a buyer webhook URL before storing it.

    The helper is the registration-time counterpart to ``WebhookSender``'s
    delivery-time SSRF guard. It applies optional transport hooks, enforces
    production HTTPS policy, resolves the destination once through the shared
    SSRF classifier, and returns the effective URL plus the validated IP.
    Sellers should normally persist ``original_url``. ``effective_url`` is for
    the immediate validation/delivery decision after transport hooks such as
    Docker localhost rewrites; do not persist a test-only rewrite as the
    buyer's registered URL.

    Raises :class:`WebhookDestinationValidationError` with structured fields
    sellers can map to ``INVALID_REQUEST`` protocol errors.
    """

    active_policy = policy or WebhookDestinationPolicy.production()
    _validate_policy_hooks(active_policy)

    if isinstance(url, str):
        url_text = url
    else:
        url_text = str(url)

    if not url_text:
        _raise_webhook_destination_error(
            "webhook destination URL must be a non-empty string",
            reason="missing_url",
            field=field,
            url=url_text,
            effective_url=None,
            policy=active_policy,
        )
    if any(c in url_text for c in _HEADER_FORBIDDEN_CHARS):
        _raise_webhook_destination_error(
            "webhook destination URL contains control characters",
            reason="control_characters",
            field=field,
            url=url_text,
            effective_url=None,
            policy=active_policy,
        )

    try:
        effective_url = apply_hooks(url_text, active_policy.transport_hooks)
    except ValueError as exc:
        _raise_webhook_destination_error(
            f"webhook destination URL failed transport hook policy: {exc}",
            reason="transport_hook_rejected",
            field=field,
            url=url_text,
            effective_url=None,
            policy=active_policy,
        )
    if any(c in effective_url for c in _HEADER_FORBIDDEN_CHARS):
        _raise_webhook_destination_error(
            "webhook destination URL contains control characters after transport hooks",
            reason="control_characters",
            field=field,
            url=url_text,
            effective_url=effective_url,
            policy=active_policy,
        )

    parsed = urlsplit(effective_url)
    if parsed.username is not None or parsed.password is not None:
        _raise_webhook_destination_error(
            "webhook destination URL must not embed userinfo (user:pass@host)",
            reason="userinfo_not_allowed",
            field=field,
            url=url_text,
            effective_url=effective_url,
            policy=active_policy,
            suggestion="Pass credentials in webhook authentication settings instead of the URL.",
        )
    if parsed.fragment:
        _raise_webhook_destination_error(
            "webhook destination URL must not include a fragment",
            reason="fragment_not_allowed",
            field=field,
            url=url_text,
            effective_url=effective_url,
            policy=active_policy,
            suggestion=(
                "Move routing state into the webhook path or query string; "
                "URL fragments are never sent in HTTP requests."
            ),
        )
    if parsed.scheme not in ("http", "https"):
        _raise_webhook_destination_error(
            f"webhook destination URL must use http:// or https:// (got {parsed.scheme!r})",
            reason="invalid_scheme",
            field=field,
            url=url_text,
            effective_url=effective_url,
            policy=active_policy,
        )
    if active_policy.require_https and parsed.scheme != "https":
        _raise_webhook_destination_error(
            f"webhook destination URL must use https:// under {active_policy.name} policy",
            reason="https_required",
            field=field,
            url=url_text,
            effective_url=effective_url,
            policy=active_policy,
            suggestion=(
                "Use an HTTPS webhook URL, or pass "
                "WebhookDestinationPolicy.local_development() for local tests."
            ),
        )

    try:
        hostname, resolved_ip, port = resolve_and_validate_host(
            effective_url,
            allow_private=active_policy.allow_private_destinations,
            allowed_ports=active_policy.allowed_destination_ports,
        )
    except SSRFValidationError as exc:
        _raise_webhook_destination_error(
            f"webhook destination URL failed SSRF validation: {exc}",
            reason="ssrf_rejected",
            field=field,
            url=url_text,
            effective_url=effective_url,
            policy=active_policy,
        )

    return WebhookDestinationValidation(
        original_url=url_text,
        effective_url=effective_url,
        hostname=hostname,
        resolved_ip=resolved_ip,
        port=port,
        policy=active_policy,
    )

Validate a buyer webhook URL before storing it.

The helper is the registration-time counterpart to WebhookSender's delivery-time SSRF guard. It applies optional transport hooks, enforces production HTTPS policy, resolves the destination once through the shared SSRF classifier, and returns the effective URL plus the validated IP. Sellers should normally persist original_url. effective_url is for the immediate validation/delivery decision after transport hooks such as Docker localhost rewrites; do not persist a test-only rewrite as the buyer's registered URL.

Raises :class:WebhookDestinationValidationError with structured fields sellers can map to INVALID_REQUEST protocol errors.

def verify_webhook_hmac(*,
headers: Mapping[str, str],
body: bytes,
options: LegacyWebhookHmacOptions) ‑> VerifiedLegacyWebhookSender
Expand source code 
def verify_webhook_hmac(
    *,
    headers: Mapping[str, str],
    body: bytes,
    options: LegacyWebhookHmacOptions,
) -> VerifiedLegacyWebhookSender:
    """Verify an HMAC-SHA256-signed webhook body per the legacy scheme.

    Raises :class:`LegacyWebhookHmacError` on any failure. Fires a one-time
    :class:`DeprecationWarning` — operators SHOULD migrate to 9421 before AdCP
    4.0 removes the ``authentication`` field.

    ``headers`` can be any ``Mapping[str, str]`` — ``dict``,
    ``werkzeug.datastructures.EnvironHeaders``, Starlette's ``Headers``, etc.
    Keys are case-folded internally.
    """
    _warn_once()

    header_map = {str(k).lower(): str(v) for k, v in headers.items()}
    sig_value = header_map.get(_SIGNATURE_HEADER)
    ts_value = header_map.get(_TIMESTAMP_HEADER)
    if sig_value is None or ts_value is None:
        raise LegacyWebhookHmacError("missing X-AdCP-Signature or X-AdCP-Timestamp header")
    if not sig_value.startswith(_HEX_PREFIX):
        raise LegacyWebhookHmacError(
            f"signature must start with {_HEX_PREFIX!r}, got {sig_value[:16]!r}"
        )
    hex_sig = sig_value[len(_HEX_PREFIX) :]

    try:
        ts_int = int(ts_value)
    except ValueError as exc:
        raise LegacyWebhookHmacError(f"invalid timestamp {ts_value!r}") from exc

    # Bound on the skew window. Matches the 9421 max window (300s) exactly —
    # the 9421 pipeline applies DEFAULT_SKEW_SECONDS inside its window check,
    # so both schemes have the same "skew budget" on the wire. Do NOT add
    # DEFAULT_SKEW_SECONDS on top; that would double-count and yield a 360s
    # budget for HMAC vs 300s for 9421.
    skew = abs(options.now - ts_int)
    if skew > options.window_seconds:
        raise LegacyWebhookHmacError(
            f"timestamp skew {skew:.0f}s exceeds window {options.window_seconds}s"
        )

    # The sender constructs the message as f"{timestamp}.{json_payload}"
    # where json_payload is the body bytes as UTF-8. Re-decoding a dict would
    # re-serialize with potentially different key order and break the
    # signature — verify against the raw bytes as received.
    message = f"{ts_value}.".encode() + body
    expected = hmac.new(options.secret, message, hashlib.sha256).hexdigest()
    # Constant-time compare — hmac.compare_digest handles str/str.
    if not hmac.compare_digest(expected, hex_sig):
        raise LegacyWebhookHmacError("signature did not match")

    return VerifiedLegacyWebhookSender(sender_identity=options.sender_identity)

Verify an HMAC-SHA256-signed webhook body per the legacy scheme.

Raises :class:LegacyWebhookHmacError on any failure. Fires a one-time :class:DeprecationWarning — operators SHOULD migrate to 9421 before AdCP 4.0 removes the authentication field.

headers can be any Mapping[str, str] — dict, werkzeug.datastructures.EnvironHeaders, Starlette's Headers, etc. Keys are case-folded internally.

def verify_webhook_signature(*,
method: str,
url: str,
headers: Mapping[str, str],
body: bytes,
options: WebhookVerifyOptions) ‑> VerifiedWebhookSender
Expand source code 
def verify_webhook_signature(
    *,
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes,
    options: WebhookVerifyOptions,
) -> VerifiedWebhookSender:
    """Verify an incoming signed webhook per the adcp/webhook-signing/v1 profile.

    Raises :class:`SignatureVerificationError` with a ``webhook_signature_*``
    code on failure. Success returns a :class:`VerifiedWebhookSender` carrying
    the identity to scope dedup state by.
    """
    _precheck_webhook_has_required_components(headers)

    request_options = VerifyOptions(
        now=options.clock(),
        capability=VerifierCapability(
            supported=True,
            covers_content_digest="required",
            required_for=frozenset({"webhook"}),
        ),
        operation="webhook",
        jwks_resolver=options.jwks_resolver,
        replay_store=options.replay_store,
        revocation_checker=options.revocation_checker,
        revocation_list=options.revocation_list,
        max_skew_seconds=options.max_skew_seconds,
        max_window_seconds=options.max_window_seconds,
        label=options.label,
        expected_tag=WEBHOOK_TAG,
        accepted_adcp_uses=frozenset({ADCP_USE_REQUEST, ADCP_USE_WEBHOOK}),
        allowed_algs=options.allowed_algs,
        agent_url=options.sender_url,
        expected_key_origins=options.expected_key_origins,
        signing_purpose="webhook_signing",
        posture=options.posture,
    )

    try:
        signer: VerifiedSigner = verify_request_signature(
            method=method, url=url, headers=headers, body=body, options=request_options
        )
    except SignatureVerificationError as exc:
        raise _retag_to_webhook(exc) from exc

    return VerifiedWebhookSender(
        key_id=signer.key_id,
        alg=signer.alg,
        label=signer.label,
        verified_at=signer.verified_at,
        sender_url=signer.agent_url,
    )

Verify an incoming signed webhook per the adcp/webhook-signing/v1 profile.

Raises :class:SignatureVerificationError with a webhook_signature_* code on failure. Success returns a :class:VerifiedWebhookSender carrying the identity to scope dedup state by.

Classes

class DockerLocalhostRewrite (rewrite_to: str = 'host.docker.internal')
Expand source code 
@dataclass(frozen=True)
class DockerLocalhostRewrite:
    """Rewrite ``localhost`` / ``127.0.0.1`` / ``::1`` to a Docker-host alias.

    Activated by adopters running e2e tests against host-side services
    from inside a Docker container. The default ``host.docker.internal``
    works on Docker Desktop (Mac/Windows). On Linux, pass
    ``rewrite_to="172.17.0.1"`` (default bridge gateway) or
    ``rewrite_to="host.docker.internal"`` after adding
    ``--add-host=host.docker.internal:host-gateway`` to the container
    run.

    Construction-time validation: this hook is only meaningful when the
    sender has ``allow_private_destinations=True``. The construct
    method on the sender side checks the flag — a hook attached to a
    sender without it raises :class:`ValueError` so the misconfiguration
    surfaces at wiring time rather than at the first delivery.

    The check happens via :meth:`validate_for_sender`, called by
    :meth:`WebhookSender._from_strategy` (and ``__init__``) when
    ``transport_hooks`` is set.

    ``rewrite_to`` is validated and canonicalized at construction: it
    must be a hostname or IP literal, is lower-cased, and bare IPv6
    literals are bracketed automatically (``"::1"`` is stored as
    ``"[::1]"``) so the assembled authority is unambiguous with a port.
    IPv4-mapped IPv6 is re-formatted to its canonical compressed form
    (``"::ffff:127.0.0.1"`` becomes ``"[::ffff:7f00:1]"`` — the same
    address, spelled canonically). Non-ASCII hostnames are IDNA-encoded
    to A-labels. IPv6 zone IDs (``"fe80::1%eth0"``) are rejected: RFC
    6874 requires the ``%`` be percent-encoded inside a URI, and
    silently emitting an invalid authority is worse than failing at
    wiring time.

    The validation is deliberately **structural**, not a hostname-syntax
    check. ASCII names are accepted as-is once they cannot alter the
    URL's shape, because Docker Compose service names legally contain
    underscores (``my_service``, ``host_gateway``) which RFC 952/1123
    and IDNA both reject — and Docker's embedded DNS resolves them.
    Enforcing hostname syntax here would refuse the exact configuration
    this class exists to serve. Whether the name resolves is the
    resolver's business; whether it rewrites the signed URL is ours.
    """

    rewrite_to: str = "host.docker.internal"

    def __post_init__(self) -> None:
        # Deferred import: ``adcp.signing``'s package __init__ is an
        # order of magnitude heavier than this leaf module, and both
        # real consumers (webhook_sender, webhooks) already import it.
        # This runs once per hook construction, never per delivery.
        from adcp.signing._idna_canonicalize import canonicalize_host

        value = self.rewrite_to
        if not value:
            raise ValueError(
                "DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
                "literal; got an empty string"
            )

        # Accept an already-bracketed IPv6 literal by unwrapping it first;
        # the brackets are re-applied below from the canonical form.
        inner = value[1:-1] if value.startswith("[") and value.endswith("]") else value
        if not inner:
            # `"[]"` survives the non-empty check above and empties here. An
            # empty host is the one outcome this guard exists to prevent: it
            # assembles to `https://:9000/hook`, an authority with a port and
            # no host -- exactly the shape @target-uri canonicalization rejects.
            raise ValueError(
                f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
                f"literal; got {value!r}, which has no host"
            )

        # Structural rejection comes FIRST and is the actual point of this
        # guard: `rewrite_to` is interpolated straight into the netloc, so any
        # character that can move the boundary between authority, path, query,
        # fragment or userinfo rewrites the signed URL into a different URL.
        # `%` is here for RFC 6874 IPv6 zone IDs, which are not representable
        # in a URI authority without percent-encoding.
        bad = {ch for ch in inner if ch in _STRUCTURAL_CHARS or ord(ch) < 0x21 or ord(ch) == 0x7F}
        if bad:
            raise ValueError(
                f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
                f"literal; got {value!r}, which contains {sorted(bad)!r} and would "
                f"change the structure of the signed URL"
            )

        try:
            ip = ipaddress.ip_address(inner)
        except ValueError:
            pass
        else:
            # Bracket v6 so the assembled authority is unambiguous with a port
            # -- the defect this guard exists to close. `str(ip)` also folds the
            # literal to its canonical compressed form.
            canonical = f"[{ip}]" if ip.version == 6 else str(ip)
            object.__setattr__(self, "rewrite_to", canonical)
            return

        if ":" in inner:
            # Not an IP literal, so a colon is a port -- which `rewrite_url`
            # re-appends itself, and which `apply_hooks`' port guard would then
            # see as a port change.
            raise ValueError(
                f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
                f"literal without a port; got {value!r}"
            )

        if inner.isascii():
            # Deliberately NOT routed through `canonicalize_host`: this is a
            # Docker helper, and Docker Compose service names legally contain
            # underscores (`my_service`, `host_gateway`), which IDNA rejects
            # under RFC 952/1123. Docker's embedded DNS resolves them, so
            # refusing them here would break the case this class exists for.
            # Structural safety is already established above; anything further
            # is the resolver's business, not ours.
            # One trailing root dot, matching `canonicalize_host` -- `rstrip`
            # would eat every dot, so `"."` and `".."` normalized to the empty
            # host rather than being rejected.
            ascii_host = inner.lower()
            if ascii_host.endswith("."):
                ascii_host = ascii_host[:-1]
            if not ascii_host or any(label == "" for label in ascii_host.split(".")):
                # Catches `"."`, `".."` and `"a..b"`. An empty label is not a
                # host, and `".."` in particular survives a single-dot strip as
                # `"."` -- non-empty, but still no host.
                raise ValueError(
                    f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
                    f"literal; got {value!r}, which normalizes to an empty label"
                )
            object.__setattr__(self, "rewrite_to", ascii_host)
            return

        # Non-ASCII: convert to A-labels so the netloc is wire-legal. Failure
        # here is a genuinely unusable host, not a naming-convention quibble.
        try:
            canonical = canonicalize_host(inner)
        except (UnicodeError, ValueError) as exc:
            # ``idna.IDNAError`` subclasses ``UnicodeError``.
            raise ValueError(
                f"DockerLocalhostRewrite(rewrite_to=...) must be a hostname or IP "
                f"literal; got {value!r} ({exc})"
            ) from exc
        object.__setattr__(self, "rewrite_to", canonical)

    def rewrite_url(self, url: str) -> str | None:
        parsed = urlsplit(url)
        # ``hostname`` lower-cases and strips brackets from IPv6 — match
        # against both bare and bracketed forms above.
        host = (parsed.hostname or "").lower()
        if host not in _LOCALHOST_HOSTS:
            return None
        # Reassemble with the rewritten host, preserving port, path,
        # query, fragment. Userinfo (``user:pass@``) is intentionally
        # dropped — webhook URLs in AdCP do not carry credentials in the
        # URL, and ``_extract_config_fields`` rejects userinfo upstream.
        # If a future caller needs it, propagate ``parsed.username`` /
        # ``parsed.password`` here.
        netloc = self.rewrite_to
        if parsed.port is not None:
            netloc = f"{self.rewrite_to}:{parsed.port}"
        return urlunsplit((parsed.scheme, netloc, parsed.path, parsed.query, parsed.fragment))

    def validate_for_sender(self, *, allow_private_destinations: bool) -> None:
        """Reject misconfiguration at sender-construction time.

        Without ``allow_private_destinations=True``, SSRF would reject
        the post-rewrite URL — silently making this hook a no-op at
        best, confusing failure at worst. Raise.
        """
        if not allow_private_destinations:
            raise ValueError(
                "DockerLocalhostRewrite requires the sender to be constructed "
                "with allow_private_destinations=True. The hook rewrites "
                "localhost to a private-IP destination; SSRF would reject the "
                "rewritten URL otherwise. Pass allow_private_destinations=True "
                "to opt in explicitly, or remove the hook for production senders."
            )

Rewrite localhost / 127.0.0.1 / ::1 to a Docker-host alias.

Activated by adopters running e2e tests against host-side services from inside a Docker container. The default host.docker.internal works on Docker Desktop (Mac/Windows). On Linux, pass rewrite_to="172.17.0.1" (default bridge gateway) or rewrite_to="host.docker.internal" after adding --add-host=host.docker.internal:host-gateway to the container run.

Construction-time validation: this hook is only meaningful when the sender has allow_private_destinations=True. The construct method on the sender side checks the flag — a hook attached to a sender without it raises :class:ValueError so the misconfiguration surfaces at wiring time rather than at the first delivery.

The check happens via :meth:validate_for_sender, called by :meth:WebhookSender._from_strategy (and __init__) when transport_hooks is set.

rewrite_to is validated and canonicalized at construction: it must be a hostname or IP literal, is lower-cased, and bare IPv6 literals are bracketed automatically ("::1" is stored as "[::1]") so the assembled authority is unambiguous with a port. IPv4-mapped IPv6 is re-formatted to its canonical compressed form ("::ffff:127.0.0.1" becomes "[::ffff:7f00:1]" — the same address, spelled canonically). Non-ASCII hostnames are IDNA-encoded to A-labels. IPv6 zone IDs ("fe80::1%eth0") are rejected: RFC 6874 requires the % be percent-encoded inside a URI, and silently emitting an invalid authority is worse than failing at wiring time.

The validation is deliberately structural, not a hostname-syntax check. ASCII names are accepted as-is once they cannot alter the URL's shape, because Docker Compose service names legally contain underscores (my_service, host_gateway) which RFC 952/1123 and IDNA both reject — and Docker's embedded DNS resolves them. Enforcing hostname syntax here would refuse the exact configuration this class exists to serve. Whether the name resolves is the resolver's business; whether it rewrites the signed URL is ours.

Instance variables

var rewrite_to : str

Methods

def rewrite_url(self, url: str) ‑> str | None
Expand source code 
def rewrite_url(self, url: str) -> str | None:
    parsed = urlsplit(url)
    # ``hostname`` lower-cases and strips brackets from IPv6 — match
    # against both bare and bracketed forms above.
    host = (parsed.hostname or "").lower()
    if host not in _LOCALHOST_HOSTS:
        return None
    # Reassemble with the rewritten host, preserving port, path,
    # query, fragment. Userinfo (``user:pass@``) is intentionally
    # dropped — webhook URLs in AdCP do not carry credentials in the
    # URL, and ``_extract_config_fields`` rejects userinfo upstream.
    # If a future caller needs it, propagate ``parsed.username`` /
    # ``parsed.password`` here.
    netloc = self.rewrite_to
    if parsed.port is not None:
        netloc = f"{self.rewrite_to}:{parsed.port}"
    return urlunsplit((parsed.scheme, netloc, parsed.path, parsed.query, parsed.fragment))
def validate_for_sender(self, *, allow_private_destinations: bool) ‑> None
Expand source code 
def validate_for_sender(self, *, allow_private_destinations: bool) -> None:
    """Reject misconfiguration at sender-construction time.

    Without ``allow_private_destinations=True``, SSRF would reject
    the post-rewrite URL — silently making this hook a no-op at
    best, confusing failure at worst. Raise.
    """
    if not allow_private_destinations:
        raise ValueError(
            "DockerLocalhostRewrite requires the sender to be constructed "
            "with allow_private_destinations=True. The hook rewrites "
            "localhost to a private-IP destination; SSRF would reject the "
            "rewritten URL otherwise. Pass allow_private_destinations=True "
            "to opt in explicitly, or remove the hook for production senders."
        )

Reject misconfiguration at sender-construction time.

Without allow_private_destinations=True, SSRF would reject the post-rewrite URL — silently making this hook a no-op at best, confusing failure at worst. Raise.

class LegacyHmacFallback (options_for: Callable[[Mapping[str, str]], LegacyWebhookHmacOptions | None],
only_when_9421_absent: bool = True)
Expand source code 
@dataclass(frozen=True)
class LegacyHmacFallback:
    """Opt-in policy for accepting HMAC-SHA256 senders during 3.x migration.

    The default behavior of the receiver is to reject any request that fails
    9421 verification. Pass an instance of this class to ``WebhookReceiverConfig``
    to accept HMAC-signed webhooks as a fallback.

    :param options_for: callback that returns a populated
        :class:`LegacyWebhookHmacOptions` given the incoming request headers.
        Your implementation resolves the sender (from Bearer, hostname, or
        legacy shared-secret tag) and returns the secret + sender_identity
        tuple the verifier needs. Return ``None`` to decline the fallback
        for this request (rejection follows the 9421-only failure path).
    :param only_when_9421_absent: when ``True`` (default), HMAC fallback only
        fires when no 9421 headers are present at all. When a request carries
        9421 headers that FAIL verification, it still rejects — preventing a
        downgrade attack where a MITM strips the 9421 signature and replaces
        it with a forged HMAC one it knows the secret for. When ``False``,
        HMAC is tried on any 9421 failure; only set this for testing or known
        homogenous sender cohorts.
    """

    options_for: Callable[[Mapping[str, str]], LegacyWebhookHmacOptions | None]
    only_when_9421_absent: bool = True

    @classmethod
    def from_shared_secret(
        cls,
        *,
        secret: bytes,
        sender_identity: str,
        only_when_9421_absent: bool = True,
        window_seconds: int = 300,
    ) -> LegacyHmacFallback:
        """Convenience constructor for the "one secret, one sender" case.

        Covers the common 3.x migration setup where the receiver has exactly
        one publisher on the legacy scheme and binds them to a known
        ``sender_identity`` (typically a buyer-defined string). For multi-
        sender or header-derived-identity setups, construct with an
        ``options_for`` callback directly.
        """
        import time as _time

        def _options_for(_headers: Mapping[str, str]) -> LegacyWebhookHmacOptions:
            return LegacyWebhookHmacOptions(
                secret=secret,
                sender_identity=sender_identity,
                now=_time.time(),
                window_seconds=window_seconds,
            )

        return cls(
            options_for=_options_for,
            only_when_9421_absent=only_when_9421_absent,
        )

Opt-in policy for accepting HMAC-SHA256 senders during 3.x migration.

The default behavior of the receiver is to reject any request that fails 9421 verification. Pass an instance of this class to WebhookReceiverConfig to accept HMAC-signed webhooks as a fallback.

:param options_for: callback that returns a populated :class:LegacyWebhookHmacOptions given the incoming request headers. Your implementation resolves the sender (from Bearer, hostname, or legacy shared-secret tag) and returns the secret + sender_identity tuple the verifier needs. Return None to decline the fallback for this request (rejection follows the 9421-only failure path). :param only_when_9421_absent: when True (default), HMAC fallback only fires when no 9421 headers are present at all. When a request carries 9421 headers that FAIL verification, it still rejects — preventing a downgrade attack where a MITM strips the 9421 signature and replaces it with a forged HMAC one it knows the secret for. When False, HMAC is tried on any 9421 failure; only set this for testing or known homogenous sender cohorts.

Static methods

def from_shared_secret(*,
secret: bytes,
sender_identity: str,
only_when_9421_absent: bool = True,
window_seconds: int = 300) ‑> LegacyHmacFallback

Convenience constructor for the "one secret, one sender" case.

Covers the common 3.x migration setup where the receiver has exactly one publisher on the legacy scheme and binds them to a known sender_identity (typically a buyer-defined string). For multi- sender or header-derived-identity setups, construct with an options_for callback directly.

Instance variables

var only_when_9421_absent : bool
var options_for : Callable[[Mapping[str, str]], LegacyWebhookHmacOptions | None]
class LegacyWebhookHmacError (reason: str)
Expand source code 
class LegacyWebhookHmacError(Exception):
    """Raised when HMAC-SHA256 legacy verification fails.

    Distinct from :class:`adcp.signing.errors.SignatureVerificationError` so
    callers can distinguish legacy-path failures from 9421-path failures —
    operators want to know which scheme fired when diagnosing a 401.
    """

    def __init__(self, reason: str) -> None:
        super().__init__(reason)
        self.reason = reason

Raised when HMAC-SHA256 legacy verification fails.

Distinct from :class:SignatureVerificationError so callers can distinguish legacy-path failures from 9421-path failures — operators want to know which scheme fired when diagnosing a 401.

Ancestors

  • builtins.Exception
  • builtins.BaseException
class LegacyWebhookHmacOptions (secret: bytes, sender_identity: str, now: float, window_seconds: int = 300)
Expand source code 
@dataclass(frozen=True)
class LegacyWebhookHmacOptions:
    """Options for the HMAC verifier.

    :param secret_resolver: callable ``(header_map) -> bytes | None`` that
        returns the shared secret for this incoming request. The receiver is
        responsible for determining sender identity from headers (Bearer
        token, IP allowlist, hostname) and looking up the secret bound to
        that sender. The resolver returns ``None`` when no sender can be
        authenticated — the verifier then rejects without attempting compare.
    :param sender_identity: trusted authentication/audit label for this legacy
        secret. In HMAC-legacy, there is no cryptographic sender identity (the
        secret IS the identity), so the caller provides one. Receivers must
        still resolve it to a stable publisher scope that survives secret
        rotation; never derive that scope from the webhook body.
    :param now: current time, epoch seconds. Defaults fetched at call time.
    :param window_seconds: accepted skew. Sender timestamp outside ``[now -
        window, now + window]`` rejects.
    """

    secret: bytes
    sender_identity: str
    now: float
    window_seconds: int = _DEFAULT_WINDOW_SECONDS

Options for the HMAC verifier.

:param secret_resolver: callable (header_map) -> bytes | None that returns the shared secret for this incoming request. The receiver is responsible for determining sender identity from headers (Bearer token, IP allowlist, hostname) and looking up the secret bound to that sender. The resolver returns None when no sender can be authenticated — the verifier then rejects without attempting compare. :param sender_identity: trusted authentication/audit label for this legacy secret. In HMAC-legacy, there is no cryptographic sender identity (the secret IS the identity), so the caller provides one. Receivers must still resolve it to a stable publisher scope that survives secret rotation; never derive that scope from the webhook body. :param now: current time, epoch seconds. Defaults fetched at call time. :param window_seconds: accepted skew. Sender timestamp outside [now - window, now + window] rejects.

Instance variables

var now : float
var secret : bytes
var sender_identity : str
var window_seconds : int
class MemoryBackend (*, clock: Callable[[], float] = <built-in function time>)
Expand source code 
class MemoryBackend(IdempotencyBackend):
    """In-process dict-backed store.

    Suitable for tests, single-process reference implementations, and local
    development. **Not suitable for multi-process deployments** — each worker
    has its own cache, so a retry that lands on a different worker is treated
    as a fresh request.

    Thread safety: the backend uses an :class:`asyncio.Lock` to serialize
    mutations of the shared dict. Reads go through the lock too; for a pure
    in-process backend this is cheap and prevents torn reads across concurrent
    ``get``/``put`` interleaving.

    :param clock: Callable returning the current epoch seconds. Override for
        tests that need to advance time deterministically without monkeypatching
        :mod:`time`. Defaults to :func:`time.time`.
    """

    def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
        self._store: dict[tuple[str, str], CachedResponse] = {}
        self._lock = asyncio.Lock()
        self._key_locks: weakref.WeakValueDictionary[tuple[str, str], asyncio.Lock] = (
            weakref.WeakValueDictionary()
        )
        self._clock = clock

    async def get(self, scope_key: str, key: str) -> CachedResponse | None:
        async with self._lock:
            entry = self._store.get((scope_key, key))
            if entry is None:
                return None
            if entry.expires_at_epoch <= self._clock():
                # Lazy expiry — drop the stale entry so the next request
                # treats the slot as fresh and races to repopulate.
                del self._store[(scope_key, key)]
                return None
            return entry

    async def put(
        self,
        scope_key: str,
        key: str,
        entry: CachedResponse,
    ) -> None:
        async with self._lock:
            self._store[(scope_key, key)] = entry

    async def current_time(self) -> float:
        return self._clock()

    @asynccontextmanager
    async def hold(self, scope_key: str, key: str) -> AsyncIterator[None]:
        """Serialize one idempotent handler execution in this process."""
        slot = (scope_key, key)
        async with self._lock:
            key_lock = self._key_locks.get(slot)
            if key_lock is None:
                key_lock = asyncio.Lock()
                self._key_locks[slot] = key_lock
        async with key_lock:
            yield

    async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool:
        """Atomically claim a missing or expired slot."""
        slot = (scope_key, key)
        async with self._lock:
            existing = self._store.get(slot)
            if existing is not None and existing.expires_at_epoch > self._clock():
                return False
            self._store[slot] = entry
            return True

    async def delete_expired(self, now_epoch: float | None = None) -> int:
        cutoff = now_epoch if now_epoch is not None else self._clock()
        async with self._lock:
            stale = [k for k, v in self._store.items() if v.expires_at_epoch <= cutoff]
            for k in stale:
                del self._store[k]
            return len(stale)

    async def clear(self) -> None:
        """Remove all cached entries.

        Test-suite hook — handy for resetting state between fixtures when a
        single :class:`MemoryBackend` is shared across multiple tests.
        """
        async with self._lock:
            self._store.clear()

    async def _size(self) -> int:
        """Test-only: return the current entry count."""
        async with self._lock:
            return len(self._store)

In-process dict-backed store.

Suitable for tests, single-process reference implementations, and local development. Not suitable for multi-process deployments — each worker has its own cache, so a retry that lands on a different worker is treated as a fresh request.

Thread safety: the backend uses an :class:asyncio.Lock to serialize mutations of the shared dict. Reads go through the lock too; for a pure in-process backend this is cheap and prevents torn reads across concurrent get/put interleaving.

:param clock: Callable returning the current epoch seconds. Override for tests that need to advance time deterministically without monkeypatching :mod:time. Defaults to :func:time.time.

Ancestors

Methods

async def clear(self) ‑> None
Expand source code 
async def clear(self) -> None:
    """Remove all cached entries.

    Test-suite hook — handy for resetting state between fixtures when a
    single :class:`MemoryBackend` is shared across multiple tests.
    """
    async with self._lock:
        self._store.clear()

Remove all cached entries.

Test-suite hook — handy for resetting state between fixtures when a single :class:MemoryBackend is shared across multiple tests.

async def hold(self, scope_key: str, key: str) ‑> AsyncIterator[None]
Expand source code 
@asynccontextmanager
async def hold(self, scope_key: str, key: str) -> AsyncIterator[None]:
    """Serialize one idempotent handler execution in this process."""
    slot = (scope_key, key)
    async with self._lock:
        key_lock = self._key_locks.get(slot)
        if key_lock is None:
            key_lock = asyncio.Lock()
            self._key_locks[slot] = key_lock
    async with key_lock:
        yield

Serialize one idempotent handler execution in this process.

async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) ‑> bool
Expand source code 
async def put_if_absent(self, scope_key: str, key: str, entry: CachedResponse) -> bool:
    """Atomically claim a missing or expired slot."""
    slot = (scope_key, key)
    async with self._lock:
        existing = self._store.get(slot)
        if existing is not None and existing.expires_at_epoch > self._clock():
            return False
        self._store[slot] = entry
        return True

Atomically claim a missing or expired slot.

Inherited members

class PgWebhookDeliverySupervisor (pool: Any,
sender: WebhookSender | None,
*,
retry: RetryPolicy | None = None,
circuit: CircuitBreakerPolicy | None = None,
log_sink: DeliveryLogSink | None = None,
circuit_table: str = 'adcp_webhook_circuit_state',
queue_table: str = 'adcp_webhook_delivery_queue',
log_table: str = 'adcp_webhook_delivery_log')
Expand source code 
class PgWebhookDeliverySupervisor:
    """Postgres-backed :class:`~adcp.webhook_supervisor.WebhookDeliverySupervisor`.

    Parameters
    ----------
    pool:
        A ``psycopg_pool.AsyncConnectionPool`` owned by the caller. Each
        operation acquires a short-lived connection (or holds one for the
        duration of a delivery). We don't open, own, or close the pool.
    sender:
        The underlying :class:`~adcp.webhook_sender.WebhookSender` used for
        the actual HTTP-Signatures POST. Must be non-None.
    retry:
        Retry policy (default: 3 attempts, exponential backoff with jitter).
    circuit:
        Circuit-breaker tuning (default: 5 failures open, 60s recovery,
        2 successes close).
    log_sink:
        Optional :class:`~adcp.webhook_supervisor.DeliveryLogSink` called
        after each attempt. Failures are swallowed. Use for custom
        observability pipelines in addition to the built-in log table.
    circuit_table / queue_table / log_table:
        Override table names (ASCII only, 1–63 chars). Useful for multi-
        tenant schemas. Defaults are ``adcp_webhook_circuit_state``,
        ``adcp_webhook_delivery_queue``, and ``adcp_webhook_delivery_log``.

    Concurrency
    -----------
    Safe to share across asyncio tasks in a single process. Multiple
    processes/pods each running :meth:`run_worker` are explicitly supported
    via ``FOR UPDATE SKIP LOCKED``.
    """

    # The queue survives process loss, but successful/finally-failed rows are
    # removed and therefore do not yet retain an immutable delivery binding
    # for the complete AdCP 3.2 advertised horizon.
    delivery_state_is_durable = False
    delivery_retry_horizon_seconds: int | None = None

    def __init__(
        self,
        pool: Any,  # psycopg_pool.AsyncConnectionPool; Any avoids import at runtime
        sender: WebhookSender | None,
        *,
        retry: RetryPolicy | None = None,
        circuit: CircuitBreakerPolicy | None = None,
        log_sink: DeliveryLogSink | None = None,
        circuit_table: str = DEFAULT_CIRCUIT_TABLE,
        queue_table: str = DEFAULT_QUEUE_TABLE,
        log_table: str = DEFAULT_LOG_TABLE,
    ) -> None:
        if not PG_AVAILABLE:
            raise ImportError(_INSTALL_HINT)
        if sender is None:
            raise ValueError(
                "PgWebhookDeliverySupervisor requires a non-None WebhookSender. "
                "Construct one via WebhookSender.from_jwk(...) or "
                "WebhookSender.from_pem(...) and pass it as the first positional argument."
            )
        self._pool = pool
        self._sender = sender
        self._retry = retry or RetryPolicy()
        self._circuit_policy = circuit or CircuitBreakerPolicy()
        self._log_sink = log_sink

        ct = _safe_identifier(circuit_table)
        qt = _safe_identifier(queue_table)
        lt = _safe_identifier(log_table)
        self._circuit_t = ct
        self._queue_t = qt
        self._log_t = lt

        # Pre-format SQL at construction time to avoid per-call f-string overhead
        # and to bake in the validated table names once.
        self._sql_circuit_get = (
            f"SELECT state, opened_at FROM {ct} WHERE breaker_key = %s"  # noqa: S608
        )
        self._sql_circuit_set_half_open = (
            f"UPDATE {ct} SET state = 'half_open', success_count = 0, "  # noqa: S608
            f"updated_at = now() WHERE breaker_key = %s AND state = 'open'"
        )
        # Atomic upsert: increment failure_count; open circuit when threshold crossed.
        # {ct}.column_name in DO UPDATE SET refers to the *existing* row's value
        # (before the update), which is what Postgres requires for self-referential
        # ON CONFLICT expressions.
        # Coerce thresholds via ``int()`` before f-string interpolation —
        # an adopter passing an ``IntEnum`` / ``bool`` / custom ``__index__``
        # subclass for ``failure_threshold`` would format verbatim into
        # SQL otherwise. Coercion produces a plain int whose ``str()``
        # is byte-safe for static SQL emission.
        failure_t = int(self._circuit_policy.failure_threshold)
        self._sql_circuit_failure = (
            f"INSERT INTO {ct} "  # noqa: S608
            f"(breaker_key, state, failure_count, success_count, updated_at) "
            f"VALUES (%s, 'closed', 1, 0, now()) "
            f"ON CONFLICT (breaker_key) DO UPDATE SET "
            f"  failure_count = {ct}.failure_count + 1, "
            f"  success_count = 0, "
            f"  state = CASE "
            f"    WHEN {ct}.failure_count + 1 >= {failure_t} OR {ct}.state = 'half_open' "
            f"    THEN 'open' ELSE {ct}.state END, "
            f"  opened_at = CASE "
            f"    WHEN ({ct}.failure_count + 1 >= {failure_t} OR {ct}.state = 'half_open') "
            f"      AND {ct}.state != 'open' THEN now() "
            f"    ELSE {ct}.opened_at END, "
            f"  updated_at = now()"
        )
        # Atomic upsert: increment success_count; close circuit when threshold crossed.
        # RETURNING lets us read the post-update state without a second query, which
        # eliminates the half-open race: two concurrent workers see different final
        # counts and only the worker whose UPDATE produces count >= threshold
        # transitions to 'closed'.
        success_t = int(self._circuit_policy.success_threshold)
        self._sql_circuit_success = (
            f"INSERT INTO {ct} "  # noqa: S608
            f"(breaker_key, state, failure_count, success_count, updated_at) "
            f"VALUES (%s, 'closed', 0, 0, now()) "
            f"ON CONFLICT (breaker_key) DO UPDATE SET "
            f"  failure_count = 0, "
            f"  success_count = CASE "
            f"    WHEN {ct}.state = 'half_open' THEN {ct}.success_count + 1 ELSE 0 END, "
            f"  state = CASE "
            f"    WHEN {ct}.state = 'half_open' AND {ct}.success_count + 1 >= {success_t} "
            f"    THEN 'closed' ELSE {ct}.state END, "
            f"  updated_at = now() "
            f"RETURNING state, success_count"
        )
        self._sql_enqueue = (
            f"INSERT INTO {qt} "  # noqa: S608
            f"(breaker_key, url, task_id, task_type, status_str, result_json, "
            f"token, sequence_key, max_attempts, notification_type, operation_id) "
            f"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id"
        )
        self._sql_poll = (
            f"SELECT id, breaker_key, url, task_id, task_type, status_str, "  # noqa: S608
            f"result_json, token, sequence_key, attempt_count, max_attempts, "
            f"idempotency_key, sent_body, notification_type, operation_id "
            f"FROM {qt} "
            f"WHERE status_str IN ('pending', 'retry') AND scheduled_at <= now() "
            f"ORDER BY scheduled_at LIMIT 1 FOR UPDATE SKIP LOCKED"
        )
        self._sql_delete_job = f"DELETE FROM {qt} WHERE id = %s"  # noqa: S608
        self._sql_reschedule = (
            f"UPDATE {qt} SET "  # noqa: S608
            f"  status_str = 'retry', "
            f"  attempt_count = %s, "
            f"  scheduled_at = %s, "
            f"  sent_body = %s, "
            f"  idempotency_key = %s "
            f"WHERE id = %s"
        )
        self._sql_log_insert = (
            f"INSERT INTO {lt} "  # noqa: S608
            f"(queue_id, breaker_key, url, task_id, sequence_key, sequence_number, "
            f"attempt_number, max_attempts, outcome, http_status_code, error_message, "
            f"response_time_ms, occurred_at, will_retry, next_retry_at, task_type, "
            f"payload_size_bytes, notification_type) "
            f"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
        )

        self._worker_started = False
        self._worker_warned = False

    # ------------------------------------------------------------------ schema

    async def create_schema(self) -> None:
        """Bootstrap all three tables and their indexes. Idempotent.

        Safe to call on every app boot. Each DDL statement is executed
        separately — psycopg does not split on ``;``.

        ::

            pool = AsyncConnectionPool("postgresql://...")
            supervisor = PgWebhookDeliverySupervisor(pool=pool, sender=sender)
            await supervisor.create_schema()
        """
        ct, qt, lt = self._circuit_t, self._queue_t, self._log_t
        statements = [
            f"""CREATE TABLE IF NOT EXISTS {ct} (
                breaker_key   TEXT        COLLATE "C" NOT NULL PRIMARY KEY,
                state         TEXT        NOT NULL DEFAULT 'closed',
                failure_count INT         NOT NULL DEFAULT 0,
                success_count INT         NOT NULL DEFAULT 0,
                opened_at     TIMESTAMPTZ,
                updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
            )""",
            f"""CREATE TABLE IF NOT EXISTS {qt} (
                id              BIGSERIAL   PRIMARY KEY,
                breaker_key     TEXT        NOT NULL,
                url             TEXT        NOT NULL,
                task_id         TEXT        NOT NULL,
                task_type       TEXT,
                status_str      TEXT        NOT NULL DEFAULT 'pending',
                result_json     TEXT,
                token           TEXT,
                sequence_key    TEXT,
                attempt_count   INT         NOT NULL DEFAULT 0,
                max_attempts    INT         NOT NULL DEFAULT 3,
                scheduled_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
                created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
                idempotency_key TEXT,
                sent_body       BYTEA,
                notification_type TEXT,
                operation_id    TEXT
            )""",
            # Backfill the operation_id column on tables created before the
            # async-completion-webhook operation_id echo landed. ADD COLUMN
            # IF NOT EXISTS is a no-op on fresh tables (the column is already
            # in the CREATE above) and an in-place add on pre-existing ones —
            # so a CREATE-only deployment and an upgrading one converge.
            f"ALTER TABLE {qt} ADD COLUMN IF NOT EXISTS operation_id TEXT",
            # Partial index on work-eligible rows; avoids scanning completed/in-flight rows.
            f"""CREATE INDEX IF NOT EXISTS {qt}_work_idx
                ON {qt} (status_str, scheduled_at)
                WHERE status_str IN ('pending', 'retry')""",
            f"""CREATE TABLE IF NOT EXISTS {lt} (
                id                BIGSERIAL   PRIMARY KEY,
                queue_id          BIGINT,
                breaker_key       TEXT        NOT NULL,
                url               TEXT        NOT NULL,
                task_id           TEXT        NOT NULL,
                sequence_key      TEXT,
                sequence_number   BIGINT,
                attempt_number    INT         NOT NULL,
                max_attempts      INT         NOT NULL,
                outcome           TEXT        NOT NULL,
                http_status_code  INT,
                error_message     TEXT,
                response_time_ms  INT         NOT NULL DEFAULT 0,
                occurred_at       TIMESTAMPTZ NOT NULL,
                will_retry        BOOLEAN     NOT NULL DEFAULT false,
                next_retry_at     TIMESTAMPTZ,
                task_type         TEXT,
                payload_size_bytes INT,
                notification_type TEXT
            )""",
            f"CREATE INDEX IF NOT EXISTS {lt}_queue_id_idx ON {lt} (queue_id)",
            f"CREATE INDEX IF NOT EXISTS {lt}_task_id_idx ON {lt} (task_id)",
        ]
        async with self._pool.connection() as conn:
            for stmt in statements:
                await conn.execute(stmt)

    # ----------------------------------------------------------------- send

    async def send_mcp(
        self,
        *,
        url: str,
        task_id: str,
        status: GeneratedTaskStatus | str,
        task_type: TaskType | str,
        result: Any = None,
        operation_id: str,
        token: str | None = None,
        sequence_key: str | None = None,
        breaker_key: str | None = None,
        notification_type: str | None = None,
    ) -> WebhookDeliveryResult | None:
        """Enqueue one MCP-style webhook delivery; **always returns None**.

        The actual HTTP send happens in :meth:`run_worker`. This method
        writes the job to ``adcp_webhook_delivery_queue`` and returns
        immediately. ``None`` is returned whether the circuit is open
        (delivery skipped) or the job was successfully enqueued.

        For delivery outcomes use a
        :class:`~adcp.webhook_supervisor.DeliveryLogSink` or query the
        ``adcp_webhook_delivery_log`` table directly.

        :param breaker_key: Override the circuit-breaker lookup key (default:
            ``url``). Multi-tenant sellers whose buyers share a SaaS receiver
            URL MUST pass a tenant-scoped key (e.g. ``f"{tenant_id}:{url}"``).
        :param operation_id: Buyer-supplied correlation id echoed verbatim
            into the webhook payload per spec. Persisted on the queue row and
            replayed to :meth:`WebhookSender.send_mcp` when the worker
            delivers the job.
        :param notification_type: Passed through to the delivery log for
            delivery-report webhooks (``scheduled`` / ``final`` / etc.).
        """
        if not self._worker_started and not self._worker_warned:
            self._worker_warned = True
            logger.warning(
                "[adcp.webhook_supervisor_pg] send_mcp() called before run_worker() "
                "has been started. Deliveries will be queued but not sent until a worker "
                "is running. Call asyncio.create_task(supervisor.run_worker()) at startup."
            )

        # The queue column is TEXT; psycopg would otherwise bind a TaskType
        # enum via repr (`"TaskType.create_media_buy"`), corrupting the wire
        # value. Normalize once here so every persistence + log site sees
        # the on-wire string.
        task_type_str: str = task_type.value if isinstance(task_type, TaskType) else task_type

        bkey = breaker_key or url

        # Check circuit state; reject immediately if OPEN within the timeout window.
        async with self._pool.connection() as conn:
            cur = await conn.execute(self._sql_circuit_get, (bkey,))
            row = await cur.fetchone()

        if row is not None:
            state: str = row[0]
            opened_at: datetime | None = row[1]
            if state == "open" and opened_at is not None:
                if opened_at.tzinfo is None:
                    opened_at = opened_at.replace(tzinfo=UTC)
                elapsed = (datetime.now(UTC) - opened_at).total_seconds()
                if elapsed < self._circuit_policy.open_timeout_seconds:
                    occurred_at = datetime.now(UTC)
                    attempt = DeliveryAttempt(
                        url=url,
                        sequence_key=sequence_key,
                        sequence_number=None,
                        attempt_number=0,
                        max_attempts=self._retry.max_attempts,
                        outcome="circuit_open",
                        http_status_code=None,
                        error_message=f"circuit breaker OPEN for {bkey} — skipped delivery",
                        response_time_ms=0,
                        occurred_at=occurred_at,
                        will_retry=False,
                        next_retry_at=None,
                        task_type=task_type_str,
                        task_id=task_id,
                        payload_size_bytes=None,
                        notification_type=notification_type,
                    )
                    await self._log_circuit_open(bkey, attempt)
                    await self._call_sink(attempt)
                    logger.warning(
                        "[adcp.webhook_supervisor_pg] circuit OPEN for %s — skipped %s",
                        bkey,
                        task_type_str,
                    )
                    return None
                # Open timeout elapsed; transition to half_open so next worker probes.
                async with self._pool.connection() as conn:
                    await conn.execute(self._sql_circuit_set_half_open, (bkey,))

        status_str = status if isinstance(status, str) else str(status)
        result_json = json.dumps(result) if result is not None else None

        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_enqueue,
                (
                    bkey,
                    url,
                    task_id,
                    task_type_str,
                    status_str,
                    result_json,
                    token,
                    sequence_key,
                    self._retry.max_attempts,
                    notification_type,
                    operation_id,
                ),
            )
            enqueue_row = await cur.fetchone()

        queue_id = enqueue_row[0] if enqueue_row else None
        logger.debug(
            "[adcp.webhook_supervisor_pg] enqueued %s → %s (queue_id=%s)",
            task_id,
            url,
            queue_id,
        )
        return None

    # ----------------------------------------------------------------- worker

    async def run_worker(self, *, poll_interval: float = 0.5) -> None:
        """Poll the delivery queue with ``FOR UPDATE SKIP LOCKED``; runs forever.

        **REQUIRED** — start at app startup::

            asyncio.create_task(supervisor.run_worker())

        Multiple processes or coroutines can run concurrently; ``SKIP LOCKED``
        ensures each job is processed by exactly one worker. The DB connection
        is held for the duration of the HTTP send so a crashed worker
        automatically releases the job back to the queue via transaction
        rollback (no separate recovery sweep needed).

        :param poll_interval: Seconds to sleep when the queue is empty.
        """
        self._worker_started = True
        logger.info(
            "[adcp.webhook_supervisor_pg] worker started (poll_interval=%.2fs)",
            poll_interval,
        )
        while True:
            try:
                delivered = await self._poll_and_process()
                if not delivered:
                    await asyncio.sleep(poll_interval)
            except asyncio.CancelledError:
                logger.info("[adcp.webhook_supervisor_pg] worker cancelled — shut down")
                raise
            except Exception:
                logger.exception(
                    "[adcp.webhook_supervisor_pg] worker error; will retry after %.2fs",
                    poll_interval,
                )
                await asyncio.sleep(poll_interval)

    # ---------------------------------------------------------- internal helpers

    async def _poll_and_process(self) -> bool:
        """Lease one job from the queue, send it, update state. Returns True if processed."""
        # The connection stays open for the duration of the HTTP send so the
        # FOR UPDATE lock is held throughout. On crash (or CancelledError),
        # the transaction rolls back and the job returns to 'pending'.
        async with self._pool.connection() as conn:
            cur = await conn.execute(self._sql_poll)
            row = await cur.fetchone()
            if row is None:
                return False

            (
                queue_id,
                bkey,
                url,
                task_id,
                task_type,
                status_str,
                result_json,
                token,
                sequence_key,
                attempt_count,
                max_attempts,
                idempotency_key,
                sent_body,
                notification_type,
                operation_id,
            ) = row

            attempt_number = attempt_count + 1
            will_retry = attempt_number < max_attempts

            # Deferred import avoids circular module dependency at package init.
            from adcp.webhook_sender import WebhookDeliveryResult  # noqa: PLC0415

            t0 = time.monotonic()
            occurred_at = datetime.now(UTC)
            delivery_result: WebhookDeliveryResult | None = None
            exc_caught: BaseException | None = None

            try:
                if sent_body:
                    # Spec-compliant retry: replay the exact bytes so the receiver
                    # can dedup via the same idempotency_key (per mcp-webhook-payload.json:
                    # "Publishers MUST … reuse the same key on every retry").
                    prev = WebhookDeliveryResult(
                        status_code=0,
                        idempotency_key=idempotency_key or "",
                        url=url,
                        response_headers={},
                        response_body=b"",
                        sent_body=sent_body,
                    )
                    delivery_result = await self._sender.resend(prev)
                else:
                    result_obj = json.loads(result_json) if result_json else None
                    delivery_result = await self._sender.send_mcp(
                        url=url,
                        task_id=task_id,
                        status=status_str,
                        task_type=task_type,
                        result=result_obj,
                        operation_id=operation_id,
                        token=token,
                    )
            except Exception as exc:
                exc_caught = exc

            response_time_ms = int((time.monotonic() - t0) * 1000)
            success = delivery_result is not None and delivery_result.ok

            if success:
                assert delivery_result is not None  # narrowing for mypy
                await conn.execute(self._sql_delete_job, (queue_id,))
                await conn.execute(self._sql_circuit_success, (bkey,))
                attempt = DeliveryAttempt(
                    url=url,
                    sequence_key=sequence_key,
                    sequence_number=queue_id,
                    attempt_number=attempt_number,
                    max_attempts=max_attempts,
                    outcome="success",
                    http_status_code=delivery_result.status_code,
                    error_message=None,
                    response_time_ms=response_time_ms,
                    occurred_at=occurred_at,
                    will_retry=False,
                    next_retry_at=None,
                    task_type=task_type,
                    task_id=task_id,
                    payload_size_bytes=(
                        len(delivery_result.sent_body) if delivery_result.sent_body else None
                    ),
                    notification_type=notification_type,
                )
            else:
                await conn.execute(self._sql_circuit_failure, (bkey,))

                next_delay = (
                    self._retry.delay_for_attempt(attempt_number + 1) if will_retry else None
                )
                next_retry_at = (
                    occurred_at + timedelta(seconds=next_delay) if next_delay is not None else None
                )

                if will_retry:
                    stored_body = delivery_result.sent_body if delivery_result is not None else None
                    stored_ikey = (
                        delivery_result.idempotency_key if delivery_result is not None else None
                    )
                    await conn.execute(
                        self._sql_reschedule,
                        (
                            attempt_number,
                            next_retry_at or occurred_at,
                            stored_body,
                            stored_ikey,
                            queue_id,
                        ),
                    )
                else:
                    await conn.execute(self._sql_delete_job, (queue_id,))

                if delivery_result is not None:
                    err_msg: str | None = (
                        f"HTTP {delivery_result.status_code}: "
                        f"{delivery_result.response_body[:200]!r}"
                    )
                    http_status: int | None = delivery_result.status_code
                    psize: int | None = (
                        len(delivery_result.sent_body) if delivery_result.sent_body else None
                    )
                elif exc_caught is not None:
                    err_msg = f"{type(exc_caught).__name__}: {exc_caught}"
                    http_status = None
                    psize = None
                else:
                    err_msg = None
                    http_status = None
                    psize = None

                attempt = DeliveryAttempt(
                    url=url,
                    sequence_key=sequence_key,
                    sequence_number=queue_id,
                    attempt_number=attempt_number,
                    max_attempts=max_attempts,
                    outcome="failure",
                    http_status_code=http_status,
                    error_message=err_msg,
                    response_time_ms=response_time_ms,
                    occurred_at=occurred_at,
                    will_retry=will_retry,
                    next_retry_at=next_retry_at,
                    task_type=task_type,
                    task_id=task_id,
                    payload_size_bytes=psize,
                    notification_type=notification_type,
                )

            await self._log_attempt_via_conn(conn, queue_id, bkey, attempt)
            # connection __aexit__ commits here; lock released

        # Sink is called outside the transaction so a slow/broken sink
        # cannot interfere with the queue update.
        await self._call_sink(attempt)

        if exc_caught is not None and not will_retry:
            raise exc_caught  # propagate to run_worker's exception logger

        return True

    async def _log_attempt_via_conn(
        self,
        conn: Any,
        queue_id: int | None,
        bkey: str,
        attempt: DeliveryAttempt,
    ) -> None:
        """Write a delivery attempt row using an already-open connection."""
        try:
            await conn.execute(
                self._sql_log_insert,
                (
                    queue_id,
                    bkey,
                    attempt.url,
                    attempt.task_id,
                    attempt.sequence_key,
                    attempt.sequence_number,
                    attempt.attempt_number,
                    attempt.max_attempts,
                    attempt.outcome,
                    attempt.http_status_code,
                    attempt.error_message,
                    attempt.response_time_ms,
                    attempt.occurred_at,
                    attempt.will_retry,
                    attempt.next_retry_at,
                    attempt.task_type,
                    attempt.payload_size_bytes,
                    attempt.notification_type,
                ),
            )
        except Exception:
            logger.warning(
                "[adcp.webhook_supervisor_pg] failed to log attempt for %s — delivery unaffected",
                attempt.url,
                exc_info=True,
            )

    async def _log_circuit_open(self, bkey: str, attempt: DeliveryAttempt) -> None:
        """Write a circuit_open attempt row using a fresh connection."""
        try:
            async with self._pool.connection() as conn:
                await self._log_attempt_via_conn(conn, None, bkey, attempt)
        except Exception:
            logger.warning(
                "[adcp.webhook_supervisor_pg] failed to log circuit_open for %s",
                attempt.url,
                exc_info=True,
            )

    async def _call_sink(self, attempt: DeliveryAttempt) -> None:
        if self._log_sink is None:
            return
        try:
            await asyncio.wait_for(
                self._log_sink.record(attempt),
                timeout=self._retry.sink_timeout_seconds,
            )
        except asyncio.TimeoutError:
            logger.warning(
                "[adcp.webhook_supervisor_pg] DeliveryLogSink timed out for %s",
                attempt.url,
            )
        except Exception:
            logger.warning(
                "[adcp.webhook_supervisor_pg] DeliveryLogSink raised for %s",
                attempt.url,
                exc_info=True,
            )

Postgres-backed :class:~adcp.webhook_supervisor.WebhookDeliverySupervisor.

Parameters

pool: A psycopg_pool.AsyncConnectionPool owned by the caller. Each operation acquires a short-lived connection (or holds one for the duration of a delivery). We don't open, own, or close the pool. sender: The underlying :class:~adcp.webhook_sender.WebhookSender used for the actual HTTP-Signatures POST. Must be non-None. retry: Retry policy (default: 3 attempts, exponential backoff with jitter). circuit: Circuit-breaker tuning (default: 5 failures open, 60s recovery, 2 successes close). log_sink: Optional :class:~adcp.webhook_supervisor.DeliveryLogSink called after each attempt. Failures are swallowed. Use for custom observability pipelines in addition to the built-in log table. circuit_table / queue_table / log_table: Override table names (ASCII only, 1–63 chars). Useful for multi- tenant schemas. Defaults are adcp_webhook_circuit_state, adcp_webhook_delivery_queue, and adcp_webhook_delivery_log.

Concurrency

Safe to share across asyncio tasks in a single process. Multiple processes/pods each running :meth:run_worker are explicitly supported via FOR UPDATE SKIP LOCKED.

Class variables

var delivery_retry_horizon_seconds : int | None
var delivery_state_is_durable

Methods

async def create_schema(self) ‑> None
Expand source code 
async def create_schema(self) -> None:
    """Bootstrap all three tables and their indexes. Idempotent.

    Safe to call on every app boot. Each DDL statement is executed
    separately — psycopg does not split on ``;``.

    ::

        pool = AsyncConnectionPool("postgresql://...")
        supervisor = PgWebhookDeliverySupervisor(pool=pool, sender=sender)
        await supervisor.create_schema()
    """
    ct, qt, lt = self._circuit_t, self._queue_t, self._log_t
    statements = [
        f"""CREATE TABLE IF NOT EXISTS {ct} (
            breaker_key   TEXT        COLLATE "C" NOT NULL PRIMARY KEY,
            state         TEXT        NOT NULL DEFAULT 'closed',
            failure_count INT         NOT NULL DEFAULT 0,
            success_count INT         NOT NULL DEFAULT 0,
            opened_at     TIMESTAMPTZ,
            updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
        )""",
        f"""CREATE TABLE IF NOT EXISTS {qt} (
            id              BIGSERIAL   PRIMARY KEY,
            breaker_key     TEXT        NOT NULL,
            url             TEXT        NOT NULL,
            task_id         TEXT        NOT NULL,
            task_type       TEXT,
            status_str      TEXT        NOT NULL DEFAULT 'pending',
            result_json     TEXT,
            token           TEXT,
            sequence_key    TEXT,
            attempt_count   INT         NOT NULL DEFAULT 0,
            max_attempts    INT         NOT NULL DEFAULT 3,
            scheduled_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
            created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
            idempotency_key TEXT,
            sent_body       BYTEA,
            notification_type TEXT,
            operation_id    TEXT
        )""",
        # Backfill the operation_id column on tables created before the
        # async-completion-webhook operation_id echo landed. ADD COLUMN
        # IF NOT EXISTS is a no-op on fresh tables (the column is already
        # in the CREATE above) and an in-place add on pre-existing ones —
        # so a CREATE-only deployment and an upgrading one converge.
        f"ALTER TABLE {qt} ADD COLUMN IF NOT EXISTS operation_id TEXT",
        # Partial index on work-eligible rows; avoids scanning completed/in-flight rows.
        f"""CREATE INDEX IF NOT EXISTS {qt}_work_idx
            ON {qt} (status_str, scheduled_at)
            WHERE status_str IN ('pending', 'retry')""",
        f"""CREATE TABLE IF NOT EXISTS {lt} (
            id                BIGSERIAL   PRIMARY KEY,
            queue_id          BIGINT,
            breaker_key       TEXT        NOT NULL,
            url               TEXT        NOT NULL,
            task_id           TEXT        NOT NULL,
            sequence_key      TEXT,
            sequence_number   BIGINT,
            attempt_number    INT         NOT NULL,
            max_attempts      INT         NOT NULL,
            outcome           TEXT        NOT NULL,
            http_status_code  INT,
            error_message     TEXT,
            response_time_ms  INT         NOT NULL DEFAULT 0,
            occurred_at       TIMESTAMPTZ NOT NULL,
            will_retry        BOOLEAN     NOT NULL DEFAULT false,
            next_retry_at     TIMESTAMPTZ,
            task_type         TEXT,
            payload_size_bytes INT,
            notification_type TEXT
        )""",
        f"CREATE INDEX IF NOT EXISTS {lt}_queue_id_idx ON {lt} (queue_id)",
        f"CREATE INDEX IF NOT EXISTS {lt}_task_id_idx ON {lt} (task_id)",
    ]
    async with self._pool.connection() as conn:
        for stmt in statements:
            await conn.execute(stmt)

Bootstrap all three tables and their indexes. Idempotent.

Safe to call on every app boot. Each DDL statement is executed separately — psycopg does not split on ;.

::

pool = AsyncConnectionPool("postgresql://...")
supervisor = PgWebhookDeliverySupervisor(pool=pool, sender=sender)
await supervisor.create_schema()
async def run_worker(self, *, poll_interval: float = 0.5) ‑> None
Expand source code 
async def run_worker(self, *, poll_interval: float = 0.5) -> None:
    """Poll the delivery queue with ``FOR UPDATE SKIP LOCKED``; runs forever.

    **REQUIRED** — start at app startup::

        asyncio.create_task(supervisor.run_worker())

    Multiple processes or coroutines can run concurrently; ``SKIP LOCKED``
    ensures each job is processed by exactly one worker. The DB connection
    is held for the duration of the HTTP send so a crashed worker
    automatically releases the job back to the queue via transaction
    rollback (no separate recovery sweep needed).

    :param poll_interval: Seconds to sleep when the queue is empty.
    """
    self._worker_started = True
    logger.info(
        "[adcp.webhook_supervisor_pg] worker started (poll_interval=%.2fs)",
        poll_interval,
    )
    while True:
        try:
            delivered = await self._poll_and_process()
            if not delivered:
                await asyncio.sleep(poll_interval)
        except asyncio.CancelledError:
            logger.info("[adcp.webhook_supervisor_pg] worker cancelled — shut down")
            raise
        except Exception:
            logger.exception(
                "[adcp.webhook_supervisor_pg] worker error; will retry after %.2fs",
                poll_interval,
            )
            await asyncio.sleep(poll_interval)

Poll the delivery queue with FOR UPDATE SKIP LOCKED; runs forever.

REQUIRED — start at app startup::

asyncio.create_task(supervisor.run_worker())

Multiple processes or coroutines can run concurrently; SKIP LOCKED ensures each job is processed by exactly one worker. The DB connection is held for the duration of the HTTP send so a crashed worker automatically releases the job back to the queue via transaction rollback (no separate recovery sweep needed).

:param poll_interval: Seconds to sleep when the queue is empty.

async def send_mcp(self,
*,
url: str,
task_id: str,
status: GeneratedTaskStatus | str,
task_type: TaskType | str,
result: Any = None,
operation_id: str,
token: str | None = None,
sequence_key: str | None = None,
breaker_key: str | None = None,
notification_type: str | None = None) ‑> WebhookDeliveryResult | None
Expand source code 
async def send_mcp(
    self,
    *,
    url: str,
    task_id: str,
    status: GeneratedTaskStatus | str,
    task_type: TaskType | str,
    result: Any = None,
    operation_id: str,
    token: str | None = None,
    sequence_key: str | None = None,
    breaker_key: str | None = None,
    notification_type: str | None = None,
) -> WebhookDeliveryResult | None:
    """Enqueue one MCP-style webhook delivery; **always returns None**.

    The actual HTTP send happens in :meth:`run_worker`. This method
    writes the job to ``adcp_webhook_delivery_queue`` and returns
    immediately. ``None`` is returned whether the circuit is open
    (delivery skipped) or the job was successfully enqueued.

    For delivery outcomes use a
    :class:`~adcp.webhook_supervisor.DeliveryLogSink` or query the
    ``adcp_webhook_delivery_log`` table directly.

    :param breaker_key: Override the circuit-breaker lookup key (default:
        ``url``). Multi-tenant sellers whose buyers share a SaaS receiver
        URL MUST pass a tenant-scoped key (e.g. ``f"{tenant_id}:{url}"``).
    :param operation_id: Buyer-supplied correlation id echoed verbatim
        into the webhook payload per spec. Persisted on the queue row and
        replayed to :meth:`WebhookSender.send_mcp` when the worker
        delivers the job.
    :param notification_type: Passed through to the delivery log for
        delivery-report webhooks (``scheduled`` / ``final`` / etc.).
    """
    if not self._worker_started and not self._worker_warned:
        self._worker_warned = True
        logger.warning(
            "[adcp.webhook_supervisor_pg] send_mcp() called before run_worker() "
            "has been started. Deliveries will be queued but not sent until a worker "
            "is running. Call asyncio.create_task(supervisor.run_worker()) at startup."
        )

    # The queue column is TEXT; psycopg would otherwise bind a TaskType
    # enum via repr (`"TaskType.create_media_buy"`), corrupting the wire
    # value. Normalize once here so every persistence + log site sees
    # the on-wire string.
    task_type_str: str = task_type.value if isinstance(task_type, TaskType) else task_type

    bkey = breaker_key or url

    # Check circuit state; reject immediately if OPEN within the timeout window.
    async with self._pool.connection() as conn:
        cur = await conn.execute(self._sql_circuit_get, (bkey,))
        row = await cur.fetchone()

    if row is not None:
        state: str = row[0]
        opened_at: datetime | None = row[1]
        if state == "open" and opened_at is not None:
            if opened_at.tzinfo is None:
                opened_at = opened_at.replace(tzinfo=UTC)
            elapsed = (datetime.now(UTC) - opened_at).total_seconds()
            if elapsed < self._circuit_policy.open_timeout_seconds:
                occurred_at = datetime.now(UTC)
                attempt = DeliveryAttempt(
                    url=url,
                    sequence_key=sequence_key,
                    sequence_number=None,
                    attempt_number=0,
                    max_attempts=self._retry.max_attempts,
                    outcome="circuit_open",
                    http_status_code=None,
                    error_message=f"circuit breaker OPEN for {bkey} — skipped delivery",
                    response_time_ms=0,
                    occurred_at=occurred_at,
                    will_retry=False,
                    next_retry_at=None,
                    task_type=task_type_str,
                    task_id=task_id,
                    payload_size_bytes=None,
                    notification_type=notification_type,
                )
                await self._log_circuit_open(bkey, attempt)
                await self._call_sink(attempt)
                logger.warning(
                    "[adcp.webhook_supervisor_pg] circuit OPEN for %s — skipped %s",
                    bkey,
                    task_type_str,
                )
                return None
            # Open timeout elapsed; transition to half_open so next worker probes.
            async with self._pool.connection() as conn:
                await conn.execute(self._sql_circuit_set_half_open, (bkey,))

    status_str = status if isinstance(status, str) else str(status)
    result_json = json.dumps(result) if result is not None else None

    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_enqueue,
            (
                bkey,
                url,
                task_id,
                task_type_str,
                status_str,
                result_json,
                token,
                sequence_key,
                self._retry.max_attempts,
                notification_type,
                operation_id,
            ),
        )
        enqueue_row = await cur.fetchone()

    queue_id = enqueue_row[0] if enqueue_row else None
    logger.debug(
        "[adcp.webhook_supervisor_pg] enqueued %s → %s (queue_id=%s)",
        task_id,
        url,
        queue_id,
    )
    return None

Enqueue one MCP-style webhook delivery; always returns None.

The actual HTTP send happens in :meth:run_worker. This method writes the job to adcp_webhook_delivery_queue and returns immediately. None is returned whether the circuit is open (delivery skipped) or the job was successfully enqueued.

For delivery outcomes use a :class:~adcp.webhook_supervisor.DeliveryLogSink or query the adcp_webhook_delivery_log table directly.

:param breaker_key: Override the circuit-breaker lookup key (default: url). Multi-tenant sellers whose buyers share a SaaS receiver URL MUST pass a tenant-scoped key (e.g. f"{tenant_id}:{url}"). :param operation_id: Buyer-supplied correlation id echoed verbatim into the webhook payload per spec. Persisted on the queue row and replayed to :meth:WebhookSender.send_mcp() when the worker delivers the job. :param notification_type: Passed through to the delivery log for delivery-report webhooks (scheduled / final / etc.).

class PreparedWebhook (url: str,
idempotency_key: str,
body: bytes,
extra_headers: Mapping[str, str] = <factory>)
Expand source code 
@dataclass(frozen=True)
class PreparedWebhook:
    """Immutable webhook request prepared for durable outbox storage.

    The serialized body and idempotency key are bound once, before the
    outbox transaction commits. Workers may then replay this object under a
    fresh RFC 9421 signature without regenerating timestamps or JSON bytes.
    """

    url: str
    idempotency_key: str
    body: bytes
    extra_headers: Mapping[str, str] = field(default_factory=dict)

Immutable webhook request prepared for durable outbox storage.

The serialized body and idempotency key are bound once, before the outbox transaction commits. Workers may then replay this object under a fresh RFC 9421 signature without regenerating timestamps or JSON bytes.

Instance variables

var body : bytes
var extra_headers : Mapping[str, str]
var idempotency_key : str
var url : str
class TransportHook (*args, **kwargs)
Expand source code 
class TransportHook(Protocol):
    """Rewrite the destination URL before SSRF runs.

    Implementations return either ``None`` (no rewrite — pass through)
    or a new URL string. The framework validates that the new URL has
    the same scheme and port as the input, and reassembles
    path/query/fragment from the original; only the hostname is
    permitted to change.

    Hooks may be called many times per sender (once per delivery), so
    they should be cheap and side-effect-free.
    """

    def rewrite_url(self, url: str) -> str | None: ...

Rewrite the destination URL before SSRF runs.

Implementations return either None (no rewrite — pass through) or a new URL string. The framework validates that the new URL has the same scheme and port as the input, and reassembles path/query/fragment from the original; only the hostname is permitted to change.

Hooks may be called many times per sender (once per delivery), so they should be cheap and side-effect-free.

Ancestors

  • typing.Protocol
  • typing.Generic

Methods

def rewrite_url(self, url: str) ‑> str | None
Expand source code 
def rewrite_url(self, url: str) -> str | None: ...
class VerifiedLegacyWebhookSender (sender_identity: str)
Expand source code 
@dataclass(frozen=True)
class VerifiedLegacyWebhookSender:
    """Identity returned by the HMAC verifier on success.

    Shape-compatible with the 9421 ``VerifiedWebhookSender.as_sender_identity``
    so downstream dedup code treats both the same.
    """

    sender_identity: str

    def as_sender_identity(self) -> str:
        return self.sender_identity

Identity returned by the HMAC verifier on success.

Shape-compatible with the 9421 VerifiedWebhookSender.as_sender_identity() so downstream dedup code treats both the same.

Instance variables

var sender_identity : str

Methods

def as_sender_identity(self) ‑> str
Expand source code 
def as_sender_identity(self) -> str:
    return self.sender_identity
class VerifiedSignerLike (*args, **kwargs)
Expand source code 
@runtime_checkable
class VerifiedSignerLike(Protocol):
    """Anything with ``as_sender_identity() -> str``.

    Both :class:`VerifiedWebhookSender` (9421) and :class:`VerifiedLegacyWebhookSender`
    (HMAC) implement this shape, so the receiver treats both verification
    paths identically downstream.
    """

    def as_sender_identity(self) -> str: ...

Anything with as_sender_identity() -> str.

Both :class:VerifiedWebhookSender (9421) and :class:VerifiedLegacyWebhookSender (HMAC) implement this shape, so the receiver treats both verification paths identically downstream.

Ancestors

  • typing.Protocol
  • typing.Generic

Methods

def as_sender_identity(self) ‑> str
Expand source code 
def as_sender_identity(self) -> str: ...
class VerifiedWebhookSender (key_id: str,
alg: str,
label: str,
verified_at: float,
sender_url: str | None = None)
Expand source code 
@dataclass(frozen=True)
class VerifiedWebhookSender:
    """Returned on successful webhook verification.

    Distinct type from :class:`VerifiedSigner` so a caller that mistakenly
    passes a request-verified signer into a webhook-scoped dedup store (or the
    reverse) will fail to type-check. Both carry the same bytes; the type
    separation is a guardrail, not a data difference.
    """

    key_id: str
    alg: str
    label: str
    verified_at: float
    sender_url: str | None = None

    def as_sender_identity(self) -> str:
        """Verified signing-key identity for authentication and audit.

        This value changes when a publisher rotates keys, so it MUST NOT be
        used as the stable publisher scope for webhook deduplication. Configure
        :attr:`WebhookReceiverConfig.publisher_scope_for` to resolve verified
        key evidence to a trusted seller identity that survives rotation.
        """
        if self.sender_url is not None:
            return f"{self.sender_url}|{self.key_id}"
        return self.key_id

Returned on successful webhook verification.

Distinct type from :class:VerifiedSigner so a caller that mistakenly passes a request-verified signer into a webhook-scoped dedup store (or the reverse) will fail to type-check. Both carry the same bytes; the type separation is a guardrail, not a data difference.

Instance variables

var alg : str
var key_id : str
var label : str
var sender_url : str | None
var verified_at : float

Methods

def as_sender_identity(self) ‑> str
Expand source code 
def as_sender_identity(self) -> str:
    """Verified signing-key identity for authentication and audit.

    This value changes when a publisher rotates keys, so it MUST NOT be
    used as the stable publisher scope for webhook deduplication. Configure
    :attr:`WebhookReceiverConfig.publisher_scope_for` to resolve verified
    key evidence to a trusted seller identity that survives rotation.
    """
    if self.sender_url is not None:
        return f"{self.sender_url}|{self.key_id}"
    return self.key_id

Verified signing-key identity for authentication and audit.

This value changes when a publisher rotates keys, so it MUST NOT be used as the stable publisher scope for webhook deduplication. Configure :attr:WebhookReceiverConfig.publisher_scope_for to resolve verified key evidence to a trusted seller identity that survives rotation.

class WebhookChallengeError (message: str,
*,
reason: str,
field: str | None = None,
url: str | None = None,
status_code: int | None = None,
suggestion: str | None = None)
Expand source code 
class WebhookChallengeError(ValueError):
    """Typed proof-of-control failure suitable for ``sync_accounts`` errors."""

    code = "INVALID_REQUEST"

    def __init__(
        self,
        message: str,
        *,
        reason: str,
        field: str | None = None,
        url: str | None = None,
        status_code: int | None = None,
        suggestion: str | None = None,
    ) -> None:
        super().__init__(message)
        self.reason = reason
        self.field = field
        self.url = url
        self.status_code = status_code
        self.suggestion = suggestion

    def to_error(self) -> dict[str, str]:
        """Return a small ``errors[]``-compatible dict for seller handlers."""

        error = {"code": self.code, "message": str(self)}
        if self.field is not None:
            error["field"] = self.field
        if self.suggestion is not None:
            error["suggestion"] = self.suggestion
        return error

Typed proof-of-control failure suitable for sync_accounts errors.

Ancestors

  • builtins.ValueError
  • builtins.Exception
  • builtins.BaseException

Class variables

var code

Methods

def to_error(self) ‑> dict[str, str]
Expand source code 
def to_error(self) -> dict[str, str]:
    """Return a small ``errors[]``-compatible dict for seller handlers."""

    error = {"code": self.code, "message": str(self)}
    if self.field is not None:
        error["field"] = self.field
    if self.suggestion is not None:
        error["suggestion"] = self.suggestion
    return error

Return a small errors[]-compatible dict for seller handlers.

class WebhookChallengeResult (challenge: str,
echoed_field: str,
destination: WebhookDestinationValidation,
status_code: int,
response_headers: Mapping[str, str],
response_body: bytes)
Expand source code 
@dataclass(frozen=True)
class WebhookChallengeResult:
    """Successful durable webhook proof-of-control challenge."""

    challenge: str
    echoed_field: str
    destination: WebhookDestinationValidation
    status_code: int
    response_headers: Mapping[str, str]
    response_body: bytes

    @property
    def ok(self) -> bool:
        return 200 <= self.status_code < 300

Successful durable webhook proof-of-control challenge.

Instance variables

var challenge : str
var destination : WebhookDestinationValidation
var echoed_field : str
prop ok : bool
Expand source code 
@property
def ok(self) -> bool:
    return 200 <= self.status_code < 300
var response_body : bytes
var response_headers : Mapping[str, str]
var status_code : int
class WebhookDedupStore (backend: IdempotencyBackend,
ttl_seconds: int = 86400,
*,
processing_ttl_seconds: int | None = None,
namespace: str = 'webhook',
clock: Callable[[], float] = <built-in function time>)
Expand source code 
class WebhookDedupStore:
    """Dedup ``(sender_id, idempotency_key)`` pairs to suppress retried webhooks.

    :param backend: any :class:`IdempotencyBackend`. Same MemoryBackend or
        PgBackend type used by :class:`IdempotencyStore` is fine — the
        ``namespace`` parameter prefixes all sender IDs so request-side and
        webhook-side scopes can't alias even when sharing one backend instance.
    :param ttl_seconds: payload-binding and replay window. Must be within ``[86400, 604800]`` per
        the spec minimum. Defaults to 86400 (24h).
    :param processing_ttl_seconds: processing-owner lease. An exact
        retry receives ``in_progress`` while the lease is live and may claim
        after it expires. Defaults to 300 seconds so a crashed handler cannot
        fence retries for the complete advertised delivery horizon. Configure
        it above the receiver's normal publication timeout; shorter leases
        increase the chance of overlapping owners and therefore require
        idempotent application publication. Must be positive and no longer
        than ``ttl_seconds``.
    :param namespace: prefix applied to every ``sender_id`` before it hits
        the backend. Defaults to ``"webhook"``, which is safe when the same
        backend is shared with :class:`IdempotencyStore` (request-side keys
        are scoped by a principal_id that isn't wrapped in this namespace,
        so collisions are impossible). Override only if you run multiple
        webhook scopes against one backend (e.g., separate dedup spaces for
        task webhooks vs list-change webhooks).
    :param clock: Deprecated compatibility argument. Expiry and lease decisions
        use :meth:`IdempotencyBackend.current_time` so durable backends and
        replicas share one authoritative clock.
    """

    def __init__(
        self,
        backend: IdempotencyBackend,
        ttl_seconds: int = _MIN_TTL_SECONDS,
        *,
        processing_ttl_seconds: int | None = None,
        namespace: str = "webhook",
        clock: Callable[[], float] = time.time,
    ) -> None:
        if not _MIN_TTL_SECONDS <= ttl_seconds <= _MAX_TTL_SECONDS:
            raise ValueError(
                f"ttl_seconds must be in [{_MIN_TTL_SECONDS}, {_MAX_TTL_SECONDS}] "
                f"per webhook spec minimum, got {ttl_seconds}"
            )
        if not namespace:
            raise ValueError("namespace must be a non-empty string")
        effective_processing_ttl = (
            _DEFAULT_PROCESSING_TTL_SECONDS
            if processing_ttl_seconds is None
            else processing_ttl_seconds
        )
        if not 1 <= effective_processing_ttl <= ttl_seconds:
            raise ValueError(
                "processing_ttl_seconds must be positive and no greater than "
                f"ttl_seconds, got {effective_processing_ttl}"
            )
        self.backend = backend
        self.ttl_seconds = ttl_seconds
        self.processing_ttl_seconds = effective_processing_ttl
        self.namespace = namespace
        _ = clock
        self._warned_legacy_backend = False

    @asynccontextmanager
    async def _hold(self, scope_key: str, key: str) -> AsyncIterator[None]:
        """Use the backend lock or a warned process-local compatibility lock."""
        if await self.backend.supports_atomic_hold():
            async with self.backend.hold(scope_key, key):
                yield
            return
        if not self._warned_legacy_backend:
            warnings.warn(
                f"{type(self.backend).__name__} does not implement hold(); using "
                "process-local webhook claim locking. This does not satisfy the "
                "multi-process AdCP 3.2 receiver contract; implement a durable "
                "atomic hold operation.",
                DeprecationWarning,
                stacklevel=3,
            )
            self._warned_legacy_backend = True

        slot = (scope_key, key)
        state = _legacy_backend_lock_state(self.backend)
        async with state.guard:
            lock = state.locks.get(slot)
            if lock is None:
                lock = asyncio.Lock()
                state.locks[slot] = lock
        async with lock:
            yield

    async def claim(
        self,
        sender_id: str,
        idempotency_key: str,
        payload_hash: str,
    ) -> WebhookDedupClaim:
        """Atomically claim a delivery or classify its retained state."""
        if not sender_id:
            raise ValueError("sender_id must be a non-empty string")
        if not idempotency_key:
            raise ValueError("idempotency_key must be a non-empty string")
        if not payload_hash:
            raise ValueError("payload_hash must be a non-empty string")

        scoped_sender = f"{self.namespace}:{sender_id}"
        async with self._hold(scoped_sender, idempotency_key):
            now = await self.backend.current_time()
            existing = await self.backend.get(scoped_sender, idempotency_key)
            if existing is not None:
                if existing.payload_hash != payload_hash:
                    return WebhookDedupClaim(status="conflict")
                state = existing.response.get("webhook_state")
                if state == "handled":
                    return WebhookDedupClaim(status="handled")
                lease_expires_at = existing.response.get("lease_expires_at")
                if (
                    state == "processing"
                    and isinstance(lease_expires_at, (int, float))
                    and lease_expires_at > now
                ):
                    return WebhookDedupClaim(status="in_progress")
                retain_until = existing.expires_at_epoch
            else:
                retain_until = now + self.ttl_seconds

            owner = secrets.token_urlsafe(32)
            claim_entry = CachedResponse(
                payload_hash=payload_hash,
                response={
                    "webhook_state": "processing",
                    "owner": owner,
                    "lease_expires_at": now + self.processing_ttl_seconds,
                },
                expires_at_epoch=retain_until,
            )
            if existing is None:
                await self.backend.put(scoped_sender, idempotency_key, claim_entry)
            else:
                await self.backend.replace(scoped_sender, idempotency_key, claim_entry)
            return WebhookDedupClaim(status="claimed", claim_token=owner)

    async def complete(
        self,
        sender_id: str,
        idempotency_key: str,
        payload_hash: str,
        claim_token: str,
    ) -> bool:
        """Durably mark an owned delivery handled.

        Returns ``True`` for the owner transition and ``False`` when the exact
        payload was already handled. Any missing, changed, or superseded claim
        fails closed with :class:`WebhookDedupOwnershipError`.
        """
        return await self._settle(
            sender_id,
            idempotency_key,
            payload_hash,
            claim_token,
            handled=True,
        )

    async def release(
        self,
        sender_id: str,
        idempotency_key: str,
        payload_hash: str,
        claim_token: str,
    ) -> bool:
        """Release an owned failed claim while retaining payload binding."""
        return await self._settle(
            sender_id,
            idempotency_key,
            payload_hash,
            claim_token,
            handled=False,
        )

    async def _settle(
        self,
        sender_id: str,
        idempotency_key: str,
        payload_hash: str,
        claim_token: str,
        *,
        handled: bool,
    ) -> bool:
        if not all((sender_id, idempotency_key, payload_hash, claim_token)):
            raise ValueError(
                "sender_id, idempotency_key, payload_hash, and claim_token are required"
            )
        scoped_sender = f"{self.namespace}:{sender_id}"
        async with self._hold(scoped_sender, idempotency_key):
            existing = await self.backend.get(scoped_sender, idempotency_key)
            if existing is None or existing.payload_hash != payload_hash:
                raise WebhookDedupOwnershipError("webhook delivery claim is missing or changed")
            state = existing.response.get("webhook_state")
            if handled and state == "handled":
                return False
            if state != "processing" or existing.response.get("owner") != claim_token:
                raise WebhookDedupOwnershipError("webhook delivery claim is not owned")
            response: dict[str, object]
            if handled:
                response = {"webhook_state": "handled"}
            else:
                response = {"webhook_state": "retryable"}
            await self.backend.replace(
                scoped_sender,
                idempotency_key,
                CachedResponse(
                    payload_hash=payload_hash,
                    response=response,
                    expires_at_epoch=existing.expires_at_epoch,
                ),
            )
            return True

Dedup (sender_id, idempotency_key) pairs to suppress retried webhooks.

:param backend: any :class:IdempotencyBackend. Same MemoryBackend or PgBackend type used by :class:IdempotencyStore is fine — the namespace parameter prefixes all sender IDs so request-side and webhook-side scopes can't alias even when sharing one backend instance. :param ttl_seconds: payload-binding and replay window. Must be within [86400, 604800] per the spec minimum. Defaults to 86400 (24h). :param processing_ttl_seconds: processing-owner lease. An exact retry receives in_progress while the lease is live and may claim after it expires. Defaults to 300 seconds so a crashed handler cannot fence retries for the complete advertised delivery horizon. Configure it above the receiver's normal publication timeout; shorter leases increase the chance of overlapping owners and therefore require idempotent application publication. Must be positive and no longer than ttl_seconds. :param namespace: prefix applied to every sender_id before it hits the backend. Defaults to "webhook", which is safe when the same backend is shared with :class:IdempotencyStore (request-side keys are scoped by a principal_id that isn't wrapped in this namespace, so collisions are impossible). Override only if you run multiple webhook scopes against one backend (e.g., separate dedup spaces for task webhooks vs list-change webhooks). :param clock: Deprecated compatibility argument. Expiry and lease decisions use :meth:IdempotencyBackend.current_time so durable backends and replicas share one authoritative clock.

Methods

async def claim(self, sender_id: str, idempotency_key: str, payload_hash: str) ‑> WebhookDedupClaim
Expand source code 
async def claim(
    self,
    sender_id: str,
    idempotency_key: str,
    payload_hash: str,
) -> WebhookDedupClaim:
    """Atomically claim a delivery or classify its retained state."""
    if not sender_id:
        raise ValueError("sender_id must be a non-empty string")
    if not idempotency_key:
        raise ValueError("idempotency_key must be a non-empty string")
    if not payload_hash:
        raise ValueError("payload_hash must be a non-empty string")

    scoped_sender = f"{self.namespace}:{sender_id}"
    async with self._hold(scoped_sender, idempotency_key):
        now = await self.backend.current_time()
        existing = await self.backend.get(scoped_sender, idempotency_key)
        if existing is not None:
            if existing.payload_hash != payload_hash:
                return WebhookDedupClaim(status="conflict")
            state = existing.response.get("webhook_state")
            if state == "handled":
                return WebhookDedupClaim(status="handled")
            lease_expires_at = existing.response.get("lease_expires_at")
            if (
                state == "processing"
                and isinstance(lease_expires_at, (int, float))
                and lease_expires_at > now
            ):
                return WebhookDedupClaim(status="in_progress")
            retain_until = existing.expires_at_epoch
        else:
            retain_until = now + self.ttl_seconds

        owner = secrets.token_urlsafe(32)
        claim_entry = CachedResponse(
            payload_hash=payload_hash,
            response={
                "webhook_state": "processing",
                "owner": owner,
                "lease_expires_at": now + self.processing_ttl_seconds,
            },
            expires_at_epoch=retain_until,
        )
        if existing is None:
            await self.backend.put(scoped_sender, idempotency_key, claim_entry)
        else:
            await self.backend.replace(scoped_sender, idempotency_key, claim_entry)
        return WebhookDedupClaim(status="claimed", claim_token=owner)

Atomically claim a delivery or classify its retained state.

async def complete(self, sender_id: str, idempotency_key: str, payload_hash: str, claim_token: str) ‑> bool
Expand source code 
async def complete(
    self,
    sender_id: str,
    idempotency_key: str,
    payload_hash: str,
    claim_token: str,
) -> bool:
    """Durably mark an owned delivery handled.

    Returns ``True`` for the owner transition and ``False`` when the exact
    payload was already handled. Any missing, changed, or superseded claim
    fails closed with :class:`WebhookDedupOwnershipError`.
    """
    return await self._settle(
        sender_id,
        idempotency_key,
        payload_hash,
        claim_token,
        handled=True,
    )

Durably mark an owned delivery handled.

Returns True for the owner transition and False when the exact payload was already handled. Any missing, changed, or superseded claim fails closed with :class:WebhookDedupOwnershipError.

async def release(self, sender_id: str, idempotency_key: str, payload_hash: str, claim_token: str) ‑> bool
Expand source code 
async def release(
    self,
    sender_id: str,
    idempotency_key: str,
    payload_hash: str,
    claim_token: str,
) -> bool:
    """Release an owned failed claim while retaining payload binding."""
    return await self._settle(
        sender_id,
        idempotency_key,
        payload_hash,
        claim_token,
        handled=False,
    )

Release an owned failed claim while retaining payload binding.

class WebhookDeliveryResult (status_code: int,
idempotency_key: str,
url: str,
response_headers: Mapping[str, str],
response_body: bytes,
sent_body: bytes = b'',
sent_extra_headers: Mapping[str, str] = <factory>)
Expand source code 
@dataclass(frozen=True)
class WebhookDeliveryResult:
    """Outcome of one ``send_*`` call.

    Senders care about: did it land (``ok``), what key was used (for logs
    and retry), what did the receiver say (``status_code``, ``response_body``).

    The ``sent_body`` and ``sent_extra_headers`` fields capture exactly what
    was signed and POSTed — the sender's :meth:`WebhookSender.resend` replays
    them under a fresh signature (preserving ``idempotency_key`` for dedup)
    rather than re-serializing from a user-supplied dict, which would drift
    if any field (``timestamp``, nested ``result``) differs between calls.
    """

    status_code: int
    idempotency_key: str
    url: str
    response_headers: Mapping[str, str]
    response_body: bytes
    sent_body: bytes = b""
    sent_extra_headers: Mapping[str, str] = field(default_factory=dict)

    @property
    def ok(self) -> bool:
        """True on 2xx. Note: receivers MUST return 2xx on duplicates too, so
        a 200 with ``duplicate=true`` in the body is still ``ok``."""
        return 200 <= self.status_code < 300

Outcome of one send_* call.

Senders care about: did it land (ok), what key was used (for logs and retry), what did the receiver say (status_code, response_body).

The sent_body and sent_extra_headers fields capture exactly what was signed and POSTed — the sender's :meth:WebhookSender.resend() replays them under a fresh signature (preserving idempotency_key for dedup) rather than re-serializing from a user-supplied dict, which would drift if any field (timestamp, nested result) differs between calls.

Instance variables

var idempotency_key : str
prop ok : bool
Expand source code 
@property
def ok(self) -> bool:
    """True on 2xx. Note: receivers MUST return 2xx on duplicates too, so
    a 200 with ``duplicate=true`` in the body is still ``ok``."""
    return 200 <= self.status_code < 300

True on 2xx. Note: receivers MUST return 2xx on duplicates too, so a 200 with duplicate=true in the body is still ok.

var response_body : bytes
var response_headers : Mapping[str, str]
var sent_body : bytes
var sent_extra_headers : Mapping[str, str]
var status_code : int
var url : str
class WebhookDestinationPolicy (require_https: bool = True,
allow_private_destinations: bool = False,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = (),
name: str = 'production')
Expand source code 
@dataclass(frozen=True)
class WebhookDestinationPolicy:
    """Registration-time policy for durable buyer webhook URLs.

    Use :meth:`production` before persisting buyer-provided
    ``push_notification_config.url`` or
    ``accounts[].notification_configs[].url``. Use
    :meth:`local_development` only for tests and local fixtures that need
    HTTP localhost or private-network endpoints.
    """

    require_https: bool = True
    allow_private_destinations: bool = False
    allowed_destination_ports: frozenset[int] | None = None
    transport_hooks: tuple[TransportHook, ...] = ()
    name: str = "production"

    @classmethod
    def production(
        cls,
        *,
        allowed_destination_ports: frozenset[int] | None = None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> WebhookDestinationPolicy:
        """Production webhook policy: HTTPS and public routable IPs only."""

        return cls(
            require_https=True,
            allow_private_destinations=False,
            allowed_destination_ports=allowed_destination_ports,
            transport_hooks=transport_hooks,
            name="production",
        )

    @classmethod
    def local_development(
        cls,
        *,
        allowed_destination_ports: frozenset[int] | None = None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> WebhookDestinationPolicy:
        """Explicit dev/test policy: allows HTTP and private destinations.

        Cloud metadata endpoints remain blocked by the shared SSRF
        validator even when private destinations are allowed.
        """

        return cls(
            require_https=False,
            allow_private_destinations=True,
            allowed_destination_ports=allowed_destination_ports,
            transport_hooks=transport_hooks,
            name="local_development",
        )

Registration-time policy for durable buyer webhook URLs.

Use :meth:production before persisting buyer-provided push_notification_config.url or accounts[].notification_configs[].url. Use :meth:local_development only for tests and local fixtures that need HTTP localhost or private-network endpoints.

Static methods

def local_development(*,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = ()) ‑> WebhookDestinationPolicy

Explicit dev/test policy: allows HTTP and private destinations.

Cloud metadata endpoints remain blocked by the shared SSRF validator even when private destinations are allowed.

def production(*,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = ()) ‑> WebhookDestinationPolicy

Production webhook policy: HTTPS and public routable IPs only.

Instance variables

var allow_private_destinations : bool
var allowed_destination_ports : frozenset[int] | None
var name : str
var require_https : bool
var transport_hooks : tuple[TransportHook, ...]
class WebhookDestinationValidation (original_url: str,
effective_url: str,
hostname: str,
resolved_ip: str,
port: int,
policy: WebhookDestinationPolicy)
Expand source code 
@dataclass(frozen=True)
class WebhookDestinationValidation:
    """Resolved result of a registration-time webhook URL validation."""

    original_url: str
    effective_url: str
    hostname: str
    resolved_ip: str
    port: int
    policy: WebhookDestinationPolicy

Resolved result of a registration-time webhook URL validation.

Instance variables

var effective_url : str
var hostname : str
var original_url : str
var policy : WebhookDestinationPolicy
var port : int
var resolved_ip : str
class WebhookDestinationValidationError (message: str,
*,
reason: str,
field: str | None = None,
url: str | None = None,
effective_url: str | None = None,
policy: WebhookDestinationPolicy | None = None,
suggestion: str | None = None)
Expand source code 
class WebhookDestinationValidationError(ValueError):
    """Typed URL-policy failure suitable for protocol error mapping.

    ``code`` is intentionally the protocol-level bucket sellers commonly
    return in ``errors[]``; ``reason`` carries the SDK-specific detail.
    ``field`` should be set by callers to values such as
    ``push_notification_config.url`` or
    ``accounts[0].notification_configs[0].url``.
    """

    code = "INVALID_REQUEST"

    def __init__(
        self,
        message: str,
        *,
        reason: str,
        field: str | None = None,
        url: str | None = None,
        effective_url: str | None = None,
        policy: WebhookDestinationPolicy | None = None,
        suggestion: str | None = None,
    ) -> None:
        super().__init__(message)
        self.reason = reason
        self.field = field
        self.url = url
        self.effective_url = effective_url
        self.policy = policy
        self.suggestion = suggestion

    def to_error(self) -> dict[str, str]:
        """Return a small ``errors[]``-compatible dict for seller handlers."""

        error = {"code": self.code, "message": str(self)}
        if self.field is not None:
            error["field"] = self.field
        if self.suggestion is not None:
            error["suggestion"] = self.suggestion
        return error

Typed URL-policy failure suitable for protocol error mapping.

code is intentionally the protocol-level bucket sellers commonly return in errors[]; reason carries the SDK-specific detail. field should be set by callers to values such as push_notification_config.url or accounts[0].notification_configs[0].url.

Ancestors

  • builtins.ValueError
  • builtins.Exception
  • builtins.BaseException

Class variables

var code

Methods

def to_error(self) ‑> dict[str, str]
Expand source code 
def to_error(self) -> dict[str, str]:
    """Return a small ``errors[]``-compatible dict for seller handlers."""

    error = {"code": self.code, "message": str(self)}
    if self.field is not None:
        error["field"] = self.field
    if self.suggestion is not None:
        error["suggestion"] = self.suggestion
    return error

Return a small errors[]-compatible dict for seller handlers.

class WebhookOutcome (rejected: bool = False,
rejection_reason: RejectionReason | None = None,
response_headers: Mapping[str, str] = <factory>,
sender_identity: str | None = None,
payload: WebhookPayload | None = None,
duplicate: bool = False,
in_progress: bool = False,
handled: bool = False,
idempotency_key: str | None = None)
Expand source code 
@dataclass(frozen=True)
class WebhookOutcome:
    """Result of a single ``receive`` call.

    ``duplicate`` means the exact delivery was durably handled and should be
    acknowledged with 2xx. ``in_progress`` means another owner is publishing
    the identical delivery and should receive 503. A fresh claim has neither
    flag and must be passed to :meth:`WebhookReceiver.acknowledge` only after
    application publication succeeds (or to ``release`` on failure).
    """

    rejected: bool = False
    rejection_reason: RejectionReason | None = None
    response_headers: Mapping[str, str] = field(default_factory=dict)
    # Populated on successful verify (even when rejected downstream of crypto)
    sender_identity: str | None = None
    # Populated on successful verify + parse
    payload: WebhookPayload | None = None
    duplicate: bool = False
    in_progress: bool = False
    handled: bool = False
    idempotency_key: str | None = None
    _payload_hash: str | None = field(default=None, repr=False, compare=False)
    _claim_token: str | None = field(default=None, repr=False, compare=False)
    _dedup_scope: str | None = field(default=None, repr=False, compare=False)
    _observation_scope: str | None = field(default=None, repr=False, compare=False)
    _observation_key: str | None = field(default=None, repr=False, compare=False)
    _observation_hash: str | None = field(default=None, repr=False, compare=False)
    _observation_token: str | None = field(default=None, repr=False, compare=False)

    @property
    def http_status(self) -> int | None:
        """Recommended immediate HTTP status, or ``None`` for a fresh claim."""
        if self.rejected:
            if self.rejection_reason == "body_too_large":
                return 413
            if self.rejection_reason == "idempotency_conflict":
                return 409
            if self.rejection_reason in {
                "signature_missing",
                "signature_invalid",
                "signature_legacy_failed",
            }:
                return 401
            return 400
        if self.in_progress:
            return 503
        if self.duplicate or self.handled:
            return 200
        return None

Result of a single receive call.

duplicate means the exact delivery was durably handled and should be acknowledged with 2xx. in_progress means another owner is publishing the identical delivery and should receive 503. A fresh claim has neither flag and must be passed to :meth:WebhookReceiver.acknowledge() only after application publication succeeds (or to release on failure).

Instance variables

var duplicate : bool
var handled : bool
prop http_status : int | None
Expand source code 
@property
def http_status(self) -> int | None:
    """Recommended immediate HTTP status, or ``None`` for a fresh claim."""
    if self.rejected:
        if self.rejection_reason == "body_too_large":
            return 413
        if self.rejection_reason == "idempotency_conflict":
            return 409
        if self.rejection_reason in {
            "signature_missing",
            "signature_invalid",
            "signature_legacy_failed",
        }:
            return 401
        return 400
    if self.in_progress:
        return 503
    if self.duplicate or self.handled:
        return 200
    return None

Recommended immediate HTTP status, or None for a fresh claim.

var idempotency_key : str | None
var in_progress : bool
var payload : adcp.types.generated_poc.core.mcp_webhook_payload.McpWebhookPayload | adcp.types.generated_poc.brand.revocation_notification.RevocationNotification | adcp.types.generated_poc.collection.collection_list_changed_webhook.CollectionListChangedWebhook | adcp.types.generated_poc.property.property_list_changed_webhook.PropertyListChangedWebhook | adcp.types.generated_poc.content_standards.artifact_webhook_payload.ArtifactWebhookPayload | None
var rejected : bool
var rejection_reason : Literal['signature_missing', 'signature_invalid', 'signature_legacy_failed', 'content_type_invalid', 'body_invalid_json', 'payload_invalid', 'idempotency_key_missing', 'idempotency_key_invalid', 'idempotency_conflict', 'payload_not_i_json', 'body_too_large'] | None
var response_headers : Mapping[str, str]
var sender_identity : str | None
class WebhookReceiver (config: WebhookReceiverConfig)
Expand source code 
class WebhookReceiver:
    """Stateless webhook entry point, one instance per receiver configuration.

    Instance state (``config``) is read-only after construction. Per-request
    state lives in the :class:`WebhookOutcome` returned from :meth:`receive`.
    """

    def __init__(self, config: WebhookReceiverConfig) -> None:
        if not config.receiver_scope:
            raise ValueError("WebhookReceiverConfig.receiver_scope must be non-empty")
        self._config = config

    async def receive(
        self,
        *,
        method: str,
        url: str,
        headers: Mapping[str, str],
        body: bytes,
    ) -> WebhookOutcome:
        """Verify, dedupe, parse. Returns a :class:`WebhookOutcome`.

        Never raises for sender-caused cryptographic or protocol failures —
        returns an outcome with ``rejected=True`` and populated
        ``response_headers`` so the caller can convert to an HTTP response
        without try/except around every call. Operational failures inside
        the dedup backend or verify-options factory MAY still raise; wrap
        the call if you need to 5xx cleanly on internal errors.
        """
        if not _content_type_is_json(headers):
            return _reject("content_type_invalid", sender_identity=None)
        if len(body) > _MAX_WEBHOOK_BODY_BYTES:
            return _reject("body_too_large", sender_identity=None)

        signer, rejection = await self._verify(method=method, url=url, headers=headers, body=body)
        if rejection is not None:
            return rejection
        assert signer is not None  # verification succeeded

        sender_id = signer.as_sender_identity()
        publisher_scope = self._config.publisher_scope_for(signer)
        if not isinstance(publisher_scope, str) or not publisher_scope:
            raise ValueError("publisher_scope_for must return a non-empty string")
        dedup_scope = json.dumps(
            [self._config.receiver_scope, publisher_scope, "delivery"],
            ensure_ascii=True,
            separators=(",", ":"),
        )
        observation_scope = json.dumps(
            [self._config.receiver_scope, publisher_scope, "terminal-observation"],
            ensure_ascii=True,
            separators=(",", ":"),
        )

        try:
            payload_dict = json.loads(body, object_pairs_hook=_unique_json_object)
        except (json.JSONDecodeError, _DuplicateJsonKeyError, RecursionError):
            return _reject("body_invalid_json", sender_identity=sender_id)
        if not isinstance(payload_dict, dict):
            return _reject("body_invalid_json", sender_identity=sender_id)
        try:
            _validate_json_depth(payload_dict, max_depth=_MAX_JSON_DEPTH)
        except _JsonDepthError:
            return _reject("payload_not_i_json", sender_identity=sender_id)

        idempotency_key = payload_dict.get("idempotency_key")
        if not isinstance(idempotency_key, str) or not idempotency_key:
            # Spec 3.0-rc: idempotency_key is REQUIRED on every webhook payload.
            return _reject("idempotency_key_missing", sender_identity=sender_id)
        if not _IDEMPOTENCY_KEY_RE.match(idempotency_key):
            # Non-conformant format — charset or length out of bounds.
            return _reject("idempotency_key_invalid", sender_identity=sender_id)

        try:
            canonical_payload = rfc8785.dumps(payload_dict)
        except (TypeError, ValueError, RecursionError):
            return _reject("payload_not_i_json", sender_identity=sender_id)
        payload_hash = hashlib.sha256(canonical_payload).hexdigest()
        claim = await self._config.dedup.claim(
            sender_id=dedup_scope,
            idempotency_key=idempotency_key,
            payload_hash=payload_hash,
        )

        if claim.status == "conflict":
            return WebhookOutcome(
                rejected=True,
                rejection_reason="idempotency_conflict",
                sender_identity=sender_id,
                idempotency_key=idempotency_key,
                _payload_hash=payload_hash,
                _dedup_scope=dedup_scope,
            )

        parsed = self._parse(payload_dict)
        if parsed is None:
            if claim.status == "claimed":
                assert claim.claim_token is not None
                await self._config.dedup.complete(
                    dedup_scope,
                    idempotency_key,
                    payload_hash,
                    claim.claim_token,
                )
            return _reject("payload_invalid", sender_identity=sender_id)

        observation = _terminal_observation(payload_dict, self._config.kind)
        observation_claim = None
        if claim.status == "claimed" and observation is not None:
            observation_key, observation_hash = observation
            observation_claim = await self._config.dedup.claim(
                sender_id=observation_scope,
                idempotency_key=observation_key,
                payload_hash=observation_hash,
            )
            assert claim.claim_token is not None
            if observation_claim.status == "conflict":
                await self._config.dedup.release(
                    dedup_scope,
                    idempotency_key,
                    payload_hash,
                    claim.claim_token,
                )
                return WebhookOutcome(
                    rejected=True,
                    rejection_reason="idempotency_conflict",
                    sender_identity=sender_id,
                    payload=parsed,
                    idempotency_key=idempotency_key,
                    _payload_hash=payload_hash,
                    _dedup_scope=dedup_scope,
                )
            if observation_claim.status == "handled":
                await self._config.dedup.complete(
                    dedup_scope,
                    idempotency_key,
                    payload_hash,
                    claim.claim_token,
                )
                return WebhookOutcome(
                    sender_identity=sender_id,
                    payload=parsed,
                    duplicate=True,
                    idempotency_key=idempotency_key,
                    _payload_hash=payload_hash,
                    _dedup_scope=dedup_scope,
                )
            if observation_claim.status == "in_progress":
                await self._config.dedup.release(
                    dedup_scope,
                    idempotency_key,
                    payload_hash,
                    claim.claim_token,
                )
                return WebhookOutcome(
                    sender_identity=sender_id,
                    payload=parsed,
                    in_progress=True,
                    idempotency_key=idempotency_key,
                    _payload_hash=payload_hash,
                    _dedup_scope=dedup_scope,
                )

        return WebhookOutcome(
            sender_identity=sender_id,
            payload=parsed,
            duplicate=claim.status == "handled",
            in_progress=claim.status == "in_progress",
            idempotency_key=idempotency_key,
            _payload_hash=payload_hash,
            _claim_token=claim.claim_token,
            _dedup_scope=dedup_scope,
            _observation_scope=observation_scope if observation is not None else None,
            _observation_key=observation[0] if observation is not None else None,
            _observation_hash=observation[1] if observation is not None else None,
            _observation_token=(
                observation_claim.claim_token if observation_claim is not None else None
            ),
        )

    async def acknowledge(self, outcome: WebhookOutcome) -> WebhookOutcome:
        """Durably acknowledge a fresh claim after publication succeeds."""
        sender_id, key, payload_hash, token = self._owned_claim(outcome)
        if outcome._observation_token is not None:
            assert outcome._observation_scope is not None
            assert outcome._observation_key is not None
            assert outcome._observation_hash is not None
            await self._config.dedup.complete(
                outcome._observation_scope,
                outcome._observation_key,
                outcome._observation_hash,
                outcome._observation_token,
            )
        await self._config.dedup.complete(sender_id, key, payload_hash, token)
        return replace(
            outcome,
            handled=True,
            _claim_token=None,
            _observation_token=None,
        )

    async def release(self, outcome: WebhookOutcome) -> WebhookOutcome:
        """Release a failed publication while retaining immutable payload binding."""
        sender_id, key, payload_hash, token = self._owned_claim(outcome)
        if outcome._observation_token is not None:
            assert outcome._observation_scope is not None
            assert outcome._observation_key is not None
            assert outcome._observation_hash is not None
            await self._config.dedup.release(
                outcome._observation_scope,
                outcome._observation_key,
                outcome._observation_hash,
                outcome._observation_token,
            )
        await self._config.dedup.release(sender_id, key, payload_hash, token)
        return replace(outcome, _claim_token=None, _observation_token=None)

    async def receive_and_process(
        self,
        *,
        method: str,
        url: str,
        headers: Mapping[str, str],
        body: bytes,
        handler: Callable[[WebhookPayload], Awaitable[None] | None],
    ) -> WebhookOutcome:
        """Verify, claim, publish, and durably acknowledge one delivery.

        Existing duplicates, conflicts, and in-progress deliveries return
        without calling ``handler``. Handler failures release the owner lease
        for an exact retry and are re-raised so the HTTP framework can return
        a retryable 5xx.
        """
        outcome = await self.receive(method=method, url=url, headers=headers, body=body)
        if outcome.http_status is not None:
            return outcome
        assert outcome.payload is not None
        try:
            result = handler(outcome.payload)
            if inspect.isawaitable(result):
                await result
        except BaseException:
            await self.release(outcome)
            raise
        return await self.acknowledge(outcome)

    @staticmethod
    def _owned_claim(outcome: WebhookOutcome) -> tuple[str, str, str, str]:
        values = (
            outcome._dedup_scope,
            outcome.idempotency_key,
            outcome._payload_hash,
            outcome._claim_token,
        )
        if not all(isinstance(value, str) and value for value in values):
            raise ValueError("WebhookOutcome does not carry an owned delivery claim")
        return cast(tuple[str, str, str, str], values)

    def receive_sync(
        self,
        *,
        method: str,
        url: str,
        headers: Mapping[str, str],
        body: bytes,
    ) -> WebhookOutcome:
        """Synchronous wrapper around :meth:`receive` for WSGI-style frameworks.

        This is the low-level verification/claim surface. A fresh outcome
        still requires asynchronous ``acknowledge`` or ``release``. Sync-only
        applications should normally use :meth:`receive_and_process_sync`,
        which owns the complete claim lifecycle in one event loop.

            @app.post("/webhooks/adcp")
            def hook():
                outcome = receiver.receive_sync(
                    method=request.method,
                    url=request.url,
                    headers=dict(request.headers),
                    body=request.get_data(),
                )
                ...

        Raises :class:`RuntimeError` if invoked from a thread that already has
        a running event loop — the underlying verify / dedup path is async and
        cannot be driven from inside an active loop without blocking it. From
        async code, call :meth:`receive` directly.
        """
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            # No running loop in this thread — safe to spin one up.
            return asyncio.run(self.receive(method=method, url=url, headers=headers, body=body))
        raise RuntimeError(
            "WebhookReceiver.receive_sync() cannot be called from a running "
            "event loop. Use `await receiver.receive(...)` instead."
        )

    def receive_and_process_sync(
        self,
        *,
        method: str,
        url: str,
        headers: Mapping[str, str],
        body: bytes,
        handler: Callable[[WebhookPayload], None],
    ) -> WebhookOutcome:
        """Sync entry point that publishes and settles a delivery atomically."""
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(
                self.receive_and_process(
                    method=method,
                    url=url,
                    headers=headers,
                    body=body,
                    handler=handler,
                )
            )
        raise RuntimeError(
            "WebhookReceiver.receive_and_process_sync() cannot be called from a "
            "running event loop. Use `await receiver.receive_and_process(...)` instead."
        )

    async def _verify(
        self,
        *,
        method: str,
        url: str,
        headers: Mapping[str, str],
        body: bytes,
    ) -> tuple[VerifiedSignerLike | None, WebhookOutcome | None]:
        """Returns (signer, None) on success or (None, rejection_outcome)."""
        has_9421 = _has_9421_headers(headers)

        if has_9421:
            try:
                signer = verify_webhook_signature(
                    method=method,
                    url=url,
                    headers=headers,
                    body=body,
                    options=self._config.verify_options,
                )
                return signer, None
            except SignatureVerificationError as exc:
                # Downgrade defense: when 9421 IS present but fails, do NOT
                # consult HMAC fallback by default. A MITM that stripped a
                # valid 9421 signature and replaced it with a forged HMAC one
                # is exactly what the downgrade guard exists for.
                fallback = self._config.legacy_hmac
                allow_hmac = fallback is not None and not fallback.only_when_9421_absent
                if not allow_hmac:
                    return None, WebhookOutcome(
                        rejected=True,
                        rejection_reason="signature_invalid",
                        response_headers=_www_authenticate_header(exc.code),
                    )
                logger.warning(
                    "9421 webhook verify failed (%s); trying HMAC legacy because "
                    "legacy_hmac.only_when_9421_absent=False is set",
                    exc.code,
                )

        fallback = self._config.legacy_hmac
        if fallback is None:
            # No 9421 headers AND no HMAC fallback configured → spec says 9421
            # is baseline-required in 3.0, so this is non-conformant.
            return None, WebhookOutcome(
                rejected=True,
                rejection_reason="signature_missing",
                response_headers=_www_authenticate_header("webhook_signature_required"),
            )

        hmac_options = fallback.options_for(headers)
        if hmac_options is None:
            return None, WebhookOutcome(
                rejected=True,
                rejection_reason="signature_missing",
                response_headers=_www_authenticate_header("webhook_signature_required"),
            )
        try:
            legacy_signer = verify_webhook_hmac(headers=headers, body=body, options=hmac_options)
            return legacy_signer, None
        except LegacyWebhookHmacError:
            return None, WebhookOutcome(
                rejected=True,
                rejection_reason="signature_legacy_failed",
                response_headers=_www_authenticate_header("webhook_signature_invalid"),
            )

    def _parse(self, payload_dict: dict[str, Any]) -> WebhookPayload | None:
        model = _MODEL_BY_KIND[self._config.kind]
        try:
            return cast(WebhookPayload, model.model_validate(payload_dict))
        except ValidationError as exc:
            # Operators need the field-level reason to diagnose sender bugs.
            # The receiver still returns payload_invalid downstream; this is
            # just observability.
            logger.warning(
                "webhook payload failed %s validation: %s",
                self._config.kind,
                exc.errors(include_url=False),
            )
            return None

Stateless webhook entry point, one instance per receiver configuration.

Instance state (config) is read-only after construction. Per-request state lives in the :class:WebhookOutcome returned from :meth:receive.

Methods

async def acknowledge(self,
outcome: WebhookOutcome) ‑> WebhookOutcome
Expand source code 
async def acknowledge(self, outcome: WebhookOutcome) -> WebhookOutcome:
    """Durably acknowledge a fresh claim after publication succeeds."""
    sender_id, key, payload_hash, token = self._owned_claim(outcome)
    if outcome._observation_token is not None:
        assert outcome._observation_scope is not None
        assert outcome._observation_key is not None
        assert outcome._observation_hash is not None
        await self._config.dedup.complete(
            outcome._observation_scope,
            outcome._observation_key,
            outcome._observation_hash,
            outcome._observation_token,
        )
    await self._config.dedup.complete(sender_id, key, payload_hash, token)
    return replace(
        outcome,
        handled=True,
        _claim_token=None,
        _observation_token=None,
    )

Durably acknowledge a fresh claim after publication succeeds.

async def receive(self, *, method: str, url: str, headers: Mapping[str, str], body: bytes) ‑> WebhookOutcome
Expand source code 
async def receive(
    self,
    *,
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes,
) -> WebhookOutcome:
    """Verify, dedupe, parse. Returns a :class:`WebhookOutcome`.

    Never raises for sender-caused cryptographic or protocol failures —
    returns an outcome with ``rejected=True`` and populated
    ``response_headers`` so the caller can convert to an HTTP response
    without try/except around every call. Operational failures inside
    the dedup backend or verify-options factory MAY still raise; wrap
    the call if you need to 5xx cleanly on internal errors.
    """
    if not _content_type_is_json(headers):
        return _reject("content_type_invalid", sender_identity=None)
    if len(body) > _MAX_WEBHOOK_BODY_BYTES:
        return _reject("body_too_large", sender_identity=None)

    signer, rejection = await self._verify(method=method, url=url, headers=headers, body=body)
    if rejection is not None:
        return rejection
    assert signer is not None  # verification succeeded

    sender_id = signer.as_sender_identity()
    publisher_scope = self._config.publisher_scope_for(signer)
    if not isinstance(publisher_scope, str) or not publisher_scope:
        raise ValueError("publisher_scope_for must return a non-empty string")
    dedup_scope = json.dumps(
        [self._config.receiver_scope, publisher_scope, "delivery"],
        ensure_ascii=True,
        separators=(",", ":"),
    )
    observation_scope = json.dumps(
        [self._config.receiver_scope, publisher_scope, "terminal-observation"],
        ensure_ascii=True,
        separators=(",", ":"),
    )

    try:
        payload_dict = json.loads(body, object_pairs_hook=_unique_json_object)
    except (json.JSONDecodeError, _DuplicateJsonKeyError, RecursionError):
        return _reject("body_invalid_json", sender_identity=sender_id)
    if not isinstance(payload_dict, dict):
        return _reject("body_invalid_json", sender_identity=sender_id)
    try:
        _validate_json_depth(payload_dict, max_depth=_MAX_JSON_DEPTH)
    except _JsonDepthError:
        return _reject("payload_not_i_json", sender_identity=sender_id)

    idempotency_key = payload_dict.get("idempotency_key")
    if not isinstance(idempotency_key, str) or not idempotency_key:
        # Spec 3.0-rc: idempotency_key is REQUIRED on every webhook payload.
        return _reject("idempotency_key_missing", sender_identity=sender_id)
    if not _IDEMPOTENCY_KEY_RE.match(idempotency_key):
        # Non-conformant format — charset or length out of bounds.
        return _reject("idempotency_key_invalid", sender_identity=sender_id)

    try:
        canonical_payload = rfc8785.dumps(payload_dict)
    except (TypeError, ValueError, RecursionError):
        return _reject("payload_not_i_json", sender_identity=sender_id)
    payload_hash = hashlib.sha256(canonical_payload).hexdigest()
    claim = await self._config.dedup.claim(
        sender_id=dedup_scope,
        idempotency_key=idempotency_key,
        payload_hash=payload_hash,
    )

    if claim.status == "conflict":
        return WebhookOutcome(
            rejected=True,
            rejection_reason="idempotency_conflict",
            sender_identity=sender_id,
            idempotency_key=idempotency_key,
            _payload_hash=payload_hash,
            _dedup_scope=dedup_scope,
        )

    parsed = self._parse(payload_dict)
    if parsed is None:
        if claim.status == "claimed":
            assert claim.claim_token is not None
            await self._config.dedup.complete(
                dedup_scope,
                idempotency_key,
                payload_hash,
                claim.claim_token,
            )
        return _reject("payload_invalid", sender_identity=sender_id)

    observation = _terminal_observation(payload_dict, self._config.kind)
    observation_claim = None
    if claim.status == "claimed" and observation is not None:
        observation_key, observation_hash = observation
        observation_claim = await self._config.dedup.claim(
            sender_id=observation_scope,
            idempotency_key=observation_key,
            payload_hash=observation_hash,
        )
        assert claim.claim_token is not None
        if observation_claim.status == "conflict":
            await self._config.dedup.release(
                dedup_scope,
                idempotency_key,
                payload_hash,
                claim.claim_token,
            )
            return WebhookOutcome(
                rejected=True,
                rejection_reason="idempotency_conflict",
                sender_identity=sender_id,
                payload=parsed,
                idempotency_key=idempotency_key,
                _payload_hash=payload_hash,
                _dedup_scope=dedup_scope,
            )
        if observation_claim.status == "handled":
            await self._config.dedup.complete(
                dedup_scope,
                idempotency_key,
                payload_hash,
                claim.claim_token,
            )
            return WebhookOutcome(
                sender_identity=sender_id,
                payload=parsed,
                duplicate=True,
                idempotency_key=idempotency_key,
                _payload_hash=payload_hash,
                _dedup_scope=dedup_scope,
            )
        if observation_claim.status == "in_progress":
            await self._config.dedup.release(
                dedup_scope,
                idempotency_key,
                payload_hash,
                claim.claim_token,
            )
            return WebhookOutcome(
                sender_identity=sender_id,
                payload=parsed,
                in_progress=True,
                idempotency_key=idempotency_key,
                _payload_hash=payload_hash,
                _dedup_scope=dedup_scope,
            )

    return WebhookOutcome(
        sender_identity=sender_id,
        payload=parsed,
        duplicate=claim.status == "handled",
        in_progress=claim.status == "in_progress",
        idempotency_key=idempotency_key,
        _payload_hash=payload_hash,
        _claim_token=claim.claim_token,
        _dedup_scope=dedup_scope,
        _observation_scope=observation_scope if observation is not None else None,
        _observation_key=observation[0] if observation is not None else None,
        _observation_hash=observation[1] if observation is not None else None,
        _observation_token=(
            observation_claim.claim_token if observation_claim is not None else None
        ),
    )

Verify, dedupe, parse. Returns a :class:WebhookOutcome.

Never raises for sender-caused cryptographic or protocol failures — returns an outcome with rejected=True and populated response_headers so the caller can convert to an HTTP response without try/except around every call. Operational failures inside the dedup backend or verify-options factory MAY still raise; wrap the call if you need to 5xx cleanly on internal errors.

async def receive_and_process(self,
*,
method: str,
url: str,
headers: Mapping[str, str],
body: bytes,
handler: Callable[[WebhookPayload], Awaitable[None] | None]) ‑> WebhookOutcome
Expand source code 
async def receive_and_process(
    self,
    *,
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes,
    handler: Callable[[WebhookPayload], Awaitable[None] | None],
) -> WebhookOutcome:
    """Verify, claim, publish, and durably acknowledge one delivery.

    Existing duplicates, conflicts, and in-progress deliveries return
    without calling ``handler``. Handler failures release the owner lease
    for an exact retry and are re-raised so the HTTP framework can return
    a retryable 5xx.
    """
    outcome = await self.receive(method=method, url=url, headers=headers, body=body)
    if outcome.http_status is not None:
        return outcome
    assert outcome.payload is not None
    try:
        result = handler(outcome.payload)
        if inspect.isawaitable(result):
            await result
    except BaseException:
        await self.release(outcome)
        raise
    return await self.acknowledge(outcome)

Verify, claim, publish, and durably acknowledge one delivery.

Existing duplicates, conflicts, and in-progress deliveries return without calling handler. Handler failures release the owner lease for an exact retry and are re-raised so the HTTP framework can return a retryable 5xx.

def receive_and_process_sync(self,
*,
method: str,
url: str,
headers: Mapping[str, str],
body: bytes,
handler: Callable[[WebhookPayload], None]) ‑> WebhookOutcome
Expand source code 
def receive_and_process_sync(
    self,
    *,
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes,
    handler: Callable[[WebhookPayload], None],
) -> WebhookOutcome:
    """Sync entry point that publishes and settles a delivery atomically."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(
            self.receive_and_process(
                method=method,
                url=url,
                headers=headers,
                body=body,
                handler=handler,
            )
        )
    raise RuntimeError(
        "WebhookReceiver.receive_and_process_sync() cannot be called from a "
        "running event loop. Use `await receiver.receive_and_process(...)` instead."
    )

Sync entry point that publishes and settles a delivery atomically.

def receive_sync(self, *, method: str, url: str, headers: Mapping[str, str], body: bytes) ‑> WebhookOutcome
Expand source code 
def receive_sync(
    self,
    *,
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes,
) -> WebhookOutcome:
    """Synchronous wrapper around :meth:`receive` for WSGI-style frameworks.

    This is the low-level verification/claim surface. A fresh outcome
    still requires asynchronous ``acknowledge`` or ``release``. Sync-only
    applications should normally use :meth:`receive_and_process_sync`,
    which owns the complete claim lifecycle in one event loop.

        @app.post("/webhooks/adcp")
        def hook():
            outcome = receiver.receive_sync(
                method=request.method,
                url=request.url,
                headers=dict(request.headers),
                body=request.get_data(),
            )
            ...

    Raises :class:`RuntimeError` if invoked from a thread that already has
    a running event loop — the underlying verify / dedup path is async and
    cannot be driven from inside an active loop without blocking it. From
    async code, call :meth:`receive` directly.
    """
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        # No running loop in this thread — safe to spin one up.
        return asyncio.run(self.receive(method=method, url=url, headers=headers, body=body))
    raise RuntimeError(
        "WebhookReceiver.receive_sync() cannot be called from a running "
        "event loop. Use `await receiver.receive(...)` instead."
    )

Synchronous wrapper around :meth:receive for WSGI-style frameworks.

This is the low-level verification/claim surface. A fresh outcome still requires asynchronous acknowledge or release. Sync-only applications should normally use :meth:receive_and_process_sync, which owns the complete claim lifecycle in one event loop.

@app.post("/webhooks/adcp")
def hook():
    outcome = receiver.receive_sync(
        method=request.method,
        url=request.url,
        headers=dict(request.headers),
        body=request.get_data(),
    )
    ...

Raises :class:RuntimeError if invoked from a thread that already has a running event loop — the underlying verify / dedup path is async and cannot be driven from inside an active loop without blocking it. From async code, call :meth:receive directly.

async def release(self,
outcome: WebhookOutcome) ‑> WebhookOutcome
Expand source code 
async def release(self, outcome: WebhookOutcome) -> WebhookOutcome:
    """Release a failed publication while retaining immutable payload binding."""
    sender_id, key, payload_hash, token = self._owned_claim(outcome)
    if outcome._observation_token is not None:
        assert outcome._observation_scope is not None
        assert outcome._observation_key is not None
        assert outcome._observation_hash is not None
        await self._config.dedup.release(
            outcome._observation_scope,
            outcome._observation_key,
            outcome._observation_hash,
            outcome._observation_token,
        )
    await self._config.dedup.release(sender_id, key, payload_hash, token)
    return replace(outcome, _claim_token=None, _observation_token=None)

Release a failed publication while retaining immutable payload binding.

class WebhookReceiverConfig (verify_options: WebhookVerifyOptions,
dedup: WebhookDedupStore,
receiver_scope: str,
publisher_scope_for: Callable[[VerifiedSignerLike], str],
legacy_hmac: LegacyHmacFallback | None = None,
kind: WebhookKind = 'mcp')
Expand source code 
@dataclass(frozen=True)
class WebhookReceiverConfig:
    """Configuration bundle.

    :param verify_options: verifier configuration (JWKS, replay store, etc.).
        A single instance is reused for every request — the verifier stamps
        ``now`` itself via ``verify_options.clock()``, so there's no need to
        refresh a time field per request.
    :param dedup: webhook-dedup store.
    :param receiver_scope: trusted tenant/subscription/endpoint scope for this
        receiver instance. Never derive it from the webhook payload.
    :param publisher_scope_for: maps the verified signing identity to a stable
        seller/publisher identity. The returned value must survive signing-key
        rotation; key IDs are authentication evidence, not publication scope.
    :param legacy_hmac: optional HMAC-SHA256 fallback for 3.x migration.
    :param kind: which webhook payload type to parse into. Default ``"mcp"``
        (the task-status webhook that dominates most integrations); pass
        explicitly for list-change / artifact / revocation receivers.
    """

    verify_options: WebhookVerifyOptions
    dedup: WebhookDedupStore
    receiver_scope: str
    publisher_scope_for: Callable[[VerifiedSignerLike], str]
    legacy_hmac: LegacyHmacFallback | None = None
    kind: WebhookKind = "mcp"

Configuration bundle.

:param verify_options: verifier configuration (JWKS, replay store, etc.). A single instance is reused for every request — the verifier stamps now itself via verify_options.clock(), so there's no need to refresh a time field per request. :param dedup: webhook-dedup store. :param receiver_scope: trusted tenant/subscription/endpoint scope for this receiver instance. Never derive it from the webhook payload. :param publisher_scope_for: maps the verified signing identity to a stable seller/publisher identity. The returned value must survive signing-key rotation; key IDs are authentication evidence, not publication scope. :param legacy_hmac: optional HMAC-SHA256 fallback for 3.x migration. :param kind: which webhook payload type to parse into. Default "mcp" (the task-status webhook that dominates most integrations); pass explicitly for list-change / artifact / revocation receivers.

Instance variables

var dedup : WebhookDedupStore
var kind : Literal['mcp', 'revocation_notification', 'collection_list_changed', 'property_list_changed', 'artifact']
var legacy_hmac : LegacyHmacFallback | None
var publisher_scope_for : Callable[[VerifiedSignerLike], str]
var receiver_scope : str
var verify_options : WebhookVerifyOptions
class WebhookSender (*,
private_key: PrivateKey,
key_id: str,
alg: str,
client: httpx.AsyncClient | None = None,
timeout_seconds: float = 10.0,
allow_private_destinations: bool = False,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = ())
Expand source code 
class WebhookSender:
    """Outbound signed-webhook delivery client.

    Owns one webhook-signing private key. Reuses a single :class:`httpx.AsyncClient`
    across requests for connection pooling — pass your own via ``client=`` if
    you want to share it with other SDK surfaces.

    Thread/task safety: safe to call concurrent ``send_*`` from many asyncio
    tasks. The underlying ``httpx.AsyncClient`` manages its own pool.
    """

    def __init__(
        self,
        *,
        private_key: PrivateKey,
        key_id: str,
        alg: str,
        client: httpx.AsyncClient | None = None,
        timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS,
        allow_private_destinations: bool = False,
        allowed_destination_ports: frozenset[int] | None = None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> None:
        """Construct a sender wired to RFC 9421 JWK signing.

        The HMAC and bearer modes are reached via :meth:`from_bearer_token`,
        :meth:`from_adcp_legacy_hmac`, and :meth:`from_standard_webhooks_secret`
        — those classmethods bypass this initializer through
        :meth:`_from_strategy` because their key material has different
        types (``bytes`` / ``str`` rather than ``PrivateKey``).

        ``transport_hooks`` runs URL rewrites before SSRF validation —
        see :class:`adcp.webhook_transport_hooks.DockerLocalhostRewrite`
        for the canonical use case. SSRF remains authoritative on the
        rewritten URL; hooks cannot punch through the range check.
        """
        self._auth: WebhookAuthStrategy = JwkSignerStrategy(
            private_key=private_key, key_id=key_id, alg=alg
        )
        self._key_id = key_id
        self._timeout = timeout_seconds
        self._client = client
        self._owns_client = client is None
        self._allow_private_destinations = allow_private_destinations
        self._allowed_destination_ports = allowed_destination_ports
        self._transport_hooks = tuple(transport_hooks)
        _validate_hooks(self._transport_hooks, allow_private_destinations)

    @classmethod
    def _from_strategy(
        cls,
        auth: WebhookAuthStrategy,
        *,
        key_id: str,
        client: httpx.AsyncClient | None,
        timeout_seconds: float,
        allow_private_destinations: bool,
        allowed_destination_ports: frozenset[int] | None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> WebhookSender:
        """Build a sender around a pre-constructed auth strategy.

        Internal constructor for the HMAC/bearer paths. The public
        ``__init__`` is locked to the JWK signature for back-compat;
        new modes don't fit that signature, so they bypass it here.
        """
        sender = cls.__new__(cls)
        sender._auth = auth
        sender._key_id = key_id
        sender._timeout = timeout_seconds
        sender._client = client
        sender._owns_client = client is None
        sender._allow_private_destinations = allow_private_destinations
        sender._allowed_destination_ports = allowed_destination_ports
        sender._transport_hooks = tuple(transport_hooks)
        _validate_hooks(sender._transport_hooks, allow_private_destinations)
        return sender

    @classmethod
    def from_jwk(
        cls,
        jwk: Mapping[str, Any],
        *,
        d_field: str = "d",
        client: httpx.AsyncClient | None = None,
        timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS,
        allow_private_destinations: bool = False,
        allowed_destination_ports: frozenset[int] | None = None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> WebhookSender:
        """Construct from a JWK that includes the private scalar.

        The JWK MUST have ``adcp_use == "webhook-signing"`` — the sender
        doesn't validate this (you're signing with your own key; validation
        happens at the receiver), but a key whose adcp_use is wrong will be
        rejected by every conformant verifier.

        ``allow_private_destinations`` and ``allowed_destination_ports``
        forward to :meth:`__init__` — see that signature for semantics.
        """
        # Snapshot the mapping once — a live Mapping could otherwise return
        # different values across the adcp_use / kid / d / alg reads.
        jwk_snapshot = dict(jwk)
        if jwk_snapshot.get("adcp_use") != "webhook-signing":
            raise ValueError(
                f"WebhookSender requires a JWK with adcp_use='webhook-signing' "
                f"(got {jwk_snapshot.get('adcp_use')!r}). Webhook-signing and "
                f"request-signing keys MUST be distinct so a signature from one "
                f"surface cannot be replayed as the other. Generate a separate "
                f"key with adcp_use='webhook-signing' and publish it in your "
                f"adagents.json alongside your request-signing key. See "
                f"https://adcontextprotocol.org/docs/building/implementation/security"
            )
        alg = jwk_snapshot.get("alg")
        if alg == "EdDSA":
            alg = "ed25519"
        elif alg == "ES256":
            alg = "ecdsa-p256-sha256"
        if alg not in ("ed25519", "ecdsa-p256-sha256"):
            raise ValueError(f"unsupported JWK alg {jwk_snapshot.get('alg')!r}")
        private_key = private_key_from_jwk(jwk_snapshot, d_field=d_field)
        return cls(
            private_key=private_key,
            key_id=str(jwk_snapshot["kid"]),
            alg=alg,
            client=client,
            timeout_seconds=timeout_seconds,
            allow_private_destinations=allow_private_destinations,
            allowed_destination_ports=allowed_destination_ports,
            transport_hooks=transport_hooks,
        )

    @classmethod
    def from_pem(
        cls,
        pem_path: str | Path | bytes,
        *,
        key_id: str,
        alg: str = "ed25519",
        passphrase: bytes | None = None,
        client: httpx.AsyncClient | None = None,
        timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS,
        allow_private_destinations: bool = False,
        allowed_destination_ports: frozenset[int] | None = None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> WebhookSender:
        """Load a private key from a PEM file and bind it as a webhook sender.

        Companion to ``adcp-keygen --purpose webhook-signing``, which writes
        the PEM and prints the public JWK. The JWK is published at your
        ``jwks_uri``; the PEM holds the private key material. ``from_pem``
        reads the PEM, constructs the right ``PrivateKey`` type for ``alg``,
        and returns a sender ready to send.

        Args:
            pem_path: Path to the PKCS#8 PEM, or the PEM bytes directly.
            key_id: JWK ``kid`` claim — must match the published JWK.
            alg: Signature algorithm. ``ed25519`` (default) or ``es256``.
                Also accepts the RFC 9421 form ``ecdsa-p256-sha256``.
            passphrase: Required if the PEM is encrypted
                (``adcp-keygen --encrypt``).
            client: Optional pre-built :class:`httpx.AsyncClient` to share
                across the SDK; the sender owns its own client when omitted.
            timeout_seconds: Per-request timeout for the owned client.
            allow_private_destinations: Forwarded to :meth:`__init__`.
            allowed_destination_ports: Forwarded to :meth:`__init__`.

        Raises:
            ValueError: ``alg`` is not ed25519 / es256, or the PEM contains
                a key whose type doesn't match ``alg``.
        """
        if alg in ("es256", "ES256"):
            alg = ALG_ES256
        elif alg == "EdDSA":
            alg = ALG_ED25519
        if alg not in (ALG_ED25519, ALG_ES256):
            raise ValueError(
                f"unsupported alg {alg!r} — use 'ed25519' or 'es256' "
                f"(the two AdCP webhook-signing algorithms)"
            )

        if isinstance(pem_path, bytes):
            pem_bytes = pem_path
        else:
            pem_bytes = Path(pem_path).read_bytes()

        private_key = load_private_key_pem(pem_bytes, password=passphrase)

        # The PEM's key type must match the requested alg — mixing them
        # would produce signatures no verifier can validate, and the
        # resulting error at delivery time would point at the receiver.
        # Fail here so the misconfiguration surfaces at construction.
        if alg == ALG_ED25519 and not isinstance(private_key, ed25519.Ed25519PrivateKey):
            raise ValueError(
                f"PEM holds a {type(private_key).__name__} but alg='ed25519' "
                f"was requested. Re-run adcp-keygen with --alg ed25519, or "
                f"pass alg='es256' to match the existing PEM."
            )
        if alg == ALG_ES256 and not isinstance(private_key, ec.EllipticCurvePrivateKey):
            raise ValueError(
                f"PEM holds a {type(private_key).__name__} but alg='es256' "
                f"was requested. Re-run adcp-keygen with --alg es256, or "
                f"pass alg='ed25519' to match the existing PEM."
            )

        return cls(
            private_key=private_key,
            key_id=key_id,
            alg=alg,
            client=client,
            timeout_seconds=timeout_seconds,
            allow_private_destinations=allow_private_destinations,
            allowed_destination_ports=allowed_destination_ports,
            transport_hooks=transport_hooks,
        )

    @classmethod
    def from_bearer_token(
        cls,
        token: str,
        *,
        client: httpx.AsyncClient | None = None,
        timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS,
        allow_private_destinations: bool = False,
        allowed_destination_ports: frozenset[int] | None = None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> WebhookSender:
        """Build a sender that POSTs with ``Authorization: Bearer <token>``.

        For buyers who authenticate the sender at the gateway and don't
        verify body signatures. The sender's marshaling guarantees still
        apply (byte-exact JSON, idempotency_key in body); body signing
        is skipped.

        A buyer treating bearer tokens as the sole authenticity signal
        SHOULD also enforce TLS/mTLS at the transport layer — a stolen
        token is a complete forgery. Prefer JWK signing (:meth:`from_jwk`)
        for AdCP-conformant deliveries.
        """
        if not isinstance(token, str) or not token:
            raise ValueError("bearer token must be a non-empty string")
        return cls._from_strategy(
            BearerTokenStrategy(token=token),
            key_id="bearer",
            client=client,
            timeout_seconds=timeout_seconds,
            allow_private_destinations=allow_private_destinations,
            allowed_destination_ports=allowed_destination_ports,
            transport_hooks=transport_hooks,
        )

    @classmethod
    def from_adcp_legacy_hmac(
        cls,
        secret: bytes,
        *,
        key_id: str,
        client: httpx.AsyncClient | None = None,
        timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS,
        allow_private_destinations: bool = False,
        allowed_destination_ports: frozenset[int] | None = None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> WebhookSender:
        """Build a sender wired to AdCP-legacy HMAC-SHA256.

        Wire format matches :func:`adcp.signing.webhook_hmac.verify_webhook_hmac`:
        ``X-AdCP-Signature: sha256=<hex>`` over ``f"{timestamp}.{body}"``,
        with ``X-AdCP-Timestamp`` set fresh per delivery (resends produce
        a new signature over the same body).

        ``secret`` is the raw HMAC key — the AdCP-legacy scheme has no
        canonical encoding, so callers pass bytes directly. ``key_id``
        is echoed in ``X-AdCP-Key-Id`` for receiver-side multi-key
        rotation; it is not used in the signature itself.

        AdCP-legacy HMAC will be removed in AdCP 4.0 — operators SHOULD
        migrate to JWK signing (:meth:`from_jwk`) ahead of that boundary.
        """
        if not isinstance(secret, bytes) or not secret:
            raise ValueError("hmac secret must be non-empty bytes")
        if not isinstance(key_id, str) or not key_id:
            raise ValueError("key_id must be a non-empty string")
        # Mirror the receiver-side _warn_once() in webhook_hmac so a
        # sender-only operator (no receiver in this process) still sees
        # the AdCP 4.0 deprecation signal at runtime, not just in the
        # docstring.
        _warn_legacy_hmac_once()
        return cls._from_strategy(
            AdcpLegacyHmacStrategy(secret=secret, key_id=key_id),
            key_id=key_id,
            client=client,
            timeout_seconds=timeout_seconds,
            allow_private_destinations=allow_private_destinations,
            allowed_destination_ports=allowed_destination_ports,
            transport_hooks=transport_hooks,
        )

    @classmethod
    def from_standard_webhooks_secret(
        cls,
        secret: str,
        *,
        key_id: str,
        client: httpx.AsyncClient | None = None,
        timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS,
        allow_private_destinations: bool = False,
        allowed_destination_ports: frozenset[int] | None = None,
        transport_hooks: tuple[TransportHook, ...] = (),
    ) -> WebhookSender:
        """Build a sender wired to standardwebhooks.com v1 (Svix/Resend interop).

        ``secret`` is the canonical ``whsec_<base64>`` form distributed
        by buyers running Svix, Resend, or any other Standard Webhooks
        verifier. The constructor base64-decodes the prefix-stripped
        payload internally — passing the literal ``whsec_...`` to
        :meth:`from_adcp_legacy_hmac` would silently produce signatures
        Svix rejects, which is exactly the footgun this typed split
        prevents.

        Wire format per spec: ``webhook-id`` / ``webhook-timestamp`` /
        ``webhook-signature: v1,<base64>`` over
        ``f"{webhook_id}.{webhook_timestamp}.{body}"``. Each delivery
        gets a fresh ``webhook-id`` so a receiver using webhook-id for
        its own replay cache doesn't false-positive on a legitimate
        retry — :meth:`resend` re-signs and gets a new id.
        """
        if not isinstance(secret, str) or not secret:
            raise ValueError("secret must be a non-empty string (whsec_<base64>)")
        if not isinstance(key_id, str) or not key_id:
            raise ValueError("key_id must be a non-empty string")
        decoded = _decode_sw_secret(secret)
        return cls._from_strategy(
            StandardWebhooksHmacStrategy(secret=decoded, key_id=key_id),
            key_id=key_id,
            client=client,
            timeout_seconds=timeout_seconds,
            allow_private_destinations=allow_private_destinations,
            allowed_destination_ports=allowed_destination_ports,
            transport_hooks=transport_hooks,
        )

    def __repr__(self) -> str:
        # Explicit repr so no future debug helper or error traceback auto-
        # renders self.__dict__ and pulls the private key (or HMAC secret /
        # bearer token) into logs.
        return f"WebhookSender(auth={type(self._auth).__name__}, " f"key_id={self._key_id!r})"

    @property
    def signs_with_rfc9421(self) -> bool:
        """``True`` iff this sender uses the RFC 9421 webhook-signing profile.

        Boot-time validators read this to enforce the
        ``webhook_signing.supported=true`` capability invariant:
        capabilities advertise RFC 9421 → wired sender must produce
        ``Signature`` / ``Signature-Input`` headers. ``from_bearer_token``,
        ``from_adcp_legacy_hmac``, and ``from_standard_webhooks_secret``
        senders return ``False``.
        """
        return isinstance(self._auth, JwkSignerStrategy)

    async def aclose(self) -> None:
        """Close the internal httpx client if we own it."""
        if self._owns_client and self._client is not None:
            await self._client.aclose()
            self._client = None

    async def __aenter__(self) -> WebhookSender:
        if not self._owns_client:
            await self._get_client()
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.aclose()

    async def _get_client(self) -> httpx.AsyncClient:
        if self._client is None:
            self._client = httpx.AsyncClient(timeout=self._timeout)
        return self._client

    async def send_mcp(
        self,
        *,
        url: str,
        task_id: str,
        status: GeneratedTaskStatus | str,
        task_type: TaskType | str,
        result: AdcpAsyncResponseData | dict[str, Any] | None = None,
        timestamp: datetime | None = None,
        operation_id: str,
        notification_id: str | None = None,
        message: str | None = None,
        context_id: str | None = None,
        protocol: AdcpProtocol | str | None = None,
        idempotency_key: str | None = None,
        token: str | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """POST a signed MCP-style task-status webhook.

        On retry, prefer :meth:`resend` over calling this again — ``resend``
        replays the exact same bytes, whereas re-invoking ``send_mcp`` with
        the "same" args would produce a fresh ``timestamp`` and potentially
        a different serialized body, which conflicts with the delivery key's
        immutable payload binding.

        :param token: Buyer-supplied token from
            ``push_notification_config.token`` echoed back on the
            payload's ``token`` field per spec
            (``schemas/cache/core/push_notification_config.json``: "Echoed
            back in webhook payload to validate request authenticity").
            Cross-language wire-parity with the JS implementation.
        """
        return await self.send_prepared(
            self.prepare_mcp(
                url=url,
                task_id=task_id,
                status=status,
                task_type=task_type,
                result=result,
                timestamp=timestamp,
                operation_id=operation_id,
                notification_id=notification_id,
                message=message,
                context_id=context_id,
                protocol=protocol,
                idempotency_key=idempotency_key,
                token=token,
                extra_headers=extra_headers,
            )
        )

    def prepare_mcp(
        self,
        *,
        url: str,
        task_id: str,
        status: GeneratedTaskStatus | str,
        task_type: TaskType | str,
        result: AdcpAsyncResponseData | dict[str, Any] | None = None,
        timestamp: datetime | None = None,
        operation_id: str,
        notification_id: str | None = None,
        message: str | None = None,
        context_id: str | None = None,
        protocol: AdcpProtocol | str | None = None,
        idempotency_key: str | None = None,
        token: str | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> PreparedWebhook:
        """Prepare an MCP webhook without performing network I/O.

        Durable publishers call this inside (or immediately before) their
        outbox transaction and persist all returned fields verbatim. The
        returned body is exactly what :meth:`send_prepared` signs and posts.
        """
        payload = create_mcp_webhook_payload(
            task_id=task_id,
            status=status,
            task_type=task_type,
            result=result,
            timestamp=timestamp,
            operation_id=operation_id,
            notification_id=notification_id,
            message=message,
            context_id=context_id,
            protocol=protocol,
            idempotency_key=idempotency_key,
            token=token,
        )
        body_dict = {
            **to_wire_dict(payload),
            "idempotency_key": payload.idempotency_key,
        }
        body = json.dumps(body_dict).encode("utf-8")
        if len(body) > _MAX_BODY_BYTES:
            raise ValueError(
                f"serialized webhook body is {len(body):,} bytes, over the "
                f"{_MAX_BODY_BYTES:,}-byte cap. Split into smaller webhooks "
                "or use batch-reporting endpoints."
            )
        return PreparedWebhook(
            url=url,
            idempotency_key=payload.idempotency_key,
            body=body,
            extra_headers=dict(extra_headers) if extra_headers else {},
        )

    async def send_prepared(self, prepared: PreparedWebhook) -> WebhookDeliveryResult:
        """Sign and post a previously prepared immutable webhook request."""
        if not prepared.idempotency_key:
            raise ValueError("prepared webhook idempotency_key must be non-empty")
        if not prepared.body:
            raise ValueError("prepared webhook body must be non-empty")
        if len(prepared.body) > _MAX_BODY_BYTES:
            raise ValueError(
                f"serialized webhook body is {len(prepared.body):,} bytes, over the "
                f"{_MAX_BODY_BYTES:,}-byte cap"
            )
        try:
            payload = json.loads(prepared.body)
        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
            raise ValueError("prepared webhook body must be a JSON object") from exc
        if not isinstance(payload, dict):
            raise ValueError("prepared webhook body must be a JSON object")
        if payload.get("idempotency_key") != prepared.idempotency_key:
            raise ValueError(
                "prepared webhook body idempotency_key does not match its immutable binding"
            )
        return await self._send_bytes(
            url=prepared.url,
            body=prepared.body,
            idempotency_key=prepared.idempotency_key,
            extra_headers=prepared.extra_headers or None,
        )

    async def send_revocation_notification(
        self,
        *,
        url: str,
        rights_id: str,
        brand_id: str,
        reason: str,
        effective_at: str,
        idempotency_key: str | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """POST a signed rights-revocation notification."""
        key = idempotency_key or generate_webhook_idempotency_key()
        payload: dict[str, Any] = {
            "idempotency_key": key,
            "rights_id": rights_id,
            "brand_id": brand_id,
            "reason": reason,
            "effective_at": effective_at,
        }
        return await self.send_raw(
            url=url, idempotency_key=key, payload=payload, extra_headers=extra_headers
        )

    async def send_artifact_webhook(
        self,
        *,
        url: str,
        media_buy_id: str,
        batch_id: str,
        timestamp: str,
        artifacts: list[dict[str, Any]],
        idempotency_key: str | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """POST a signed content-standards artifact webhook."""
        key = idempotency_key or generate_webhook_idempotency_key()
        payload: dict[str, Any] = {
            "idempotency_key": key,
            "media_buy_id": media_buy_id,
            "batch_id": batch_id,
            "timestamp": timestamp,
            "artifacts": artifacts,
        }
        return await self.send_raw(
            url=url, idempotency_key=key, payload=payload, extra_headers=extra_headers
        )

    async def send_collection_list_changed(
        self,
        *,
        url: str,
        list_id: str,
        resolved_at: str,
        signature: str,
        idempotency_key: str | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """POST a signed governance collection-list-changed webhook.

        ``signature`` is the payload-level signature field that predates 9421
        webhook transport signing — it remains required by the schema. The
        9421 signature this method adds protects the transport envelope.
        """
        key = idempotency_key or generate_webhook_idempotency_key()
        payload: dict[str, Any] = {
            "idempotency_key": key,
            "event": "collection_list_changed",
            "list_id": list_id,
            "resolved_at": resolved_at,
            "signature": signature,
        }
        return await self.send_raw(
            url=url, idempotency_key=key, payload=payload, extra_headers=extra_headers
        )

    async def send_property_list_changed(
        self,
        *,
        url: str,
        list_id: str,
        resolved_at: str,
        signature: str,
        idempotency_key: str | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """POST a signed governance property-list-changed webhook."""
        key = idempotency_key or generate_webhook_idempotency_key()
        payload: dict[str, Any] = {
            "idempotency_key": key,
            "event": "property_list_changed",
            "list_id": list_id,
            "resolved_at": resolved_at,
            "signature": signature,
        }
        return await self.send_raw(
            url=url, idempotency_key=key, payload=payload, extra_headers=extra_headers
        )

    async def send_wholesale_feed(
        self,
        *,
        url: str,
        subscriber_id: str,
        account_id: str,
        notification_type: str,
        wholesale_feed_version: str,
        cache_scope: str,
        event: WholesaleFeedEvent | Mapping[str, Any],
        previous_wholesale_feed_version: str | None = None,
        fired_at: datetime | None = None,
        idempotency_key: str | None = None,
        subscription_event_types: Sequence[Any] | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """POST a signed account-scoped wholesale feed notification.

        ``subscription_event_types`` is optional but recommended when the
        caller is sending to an ``accounts[].notification_configs[]`` entry:
        pass that entry's ``event_types`` to fail closed if the subscription
        did not request this notification type.
        """

        if not isinstance(subscriber_id, str) or not subscriber_id:
            raise ValueError("subscriber_id must be a non-empty string")
        if not isinstance(account_id, str) or not account_id:
            raise ValueError("account_id must be a non-empty string")
        if not isinstance(wholesale_feed_version, str) or not wholesale_feed_version:
            raise ValueError("wholesale_feed_version must be a non-empty string")

        event_model = event
        if not isinstance(event_model, WholesaleFeedEvent):
            event_model = WholesaleFeedEvent.model_validate(event_model)
        notification_type_value = _enum_value(notification_type)
        event_type = _enum_value(event_model.event_type)
        entity_type = _enum_value(event_model.entity_type)
        if notification_type_value != event_type:
            raise ValueError(
                "notification_type must match event.event_type "
                f"(got {notification_type_value!r}, event has {event_type!r})"
            )
        if subscription_event_types is not None:
            allowed_event_types = {_enum_value(item) for item in subscription_event_types}
        else:
            allowed_event_types = None
        if allowed_event_types is not None and notification_type_value not in allowed_event_types:
            raise ValueError(
                "notification_type is not present in the subscription's event_types; "
                "sellers must not silently widen account notification filters"
            )

        expected_entity_type = _entity_type_for_wholesale_notification(notification_type_value)
        if entity_type != expected_entity_type:
            raise ValueError(
                "event.entity_type does not match notification_type "
                f"(got {entity_type!r}, expected {expected_entity_type!r})"
            )

        cache_scope_value = _enum_value(cache_scope)
        applies_to = getattr(event_model.payload, "applies_to", None)
        applies_to_scope = _enum_value(getattr(applies_to, "scope", None))
        if applies_to_scope != cache_scope_value:
            raise ValueError(
                "cache_scope must match event.payload.applies_to.scope "
                f"(got {cache_scope_value!r}, event has {applies_to_scope!r})"
            )

        key = idempotency_key or generate_webhook_idempotency_key()
        timestamp = fired_at or datetime.now(timezone.utc)
        webhook = WholesaleFeedWebhook.model_validate(
            {
                "idempotency_key": key,
                "notification_id": event_model.event_id,
                "notification_type": notification_type_value,
                "fired_at": timestamp,
                "subscriber_id": subscriber_id,
                "account_id": account_id,
                "wholesale_feed_version": wholesale_feed_version,
                "previous_wholesale_feed_version": previous_wholesale_feed_version,
                "cache_scope": cache_scope_value,
                "event": event_model,
            }
        )
        return await self.send_raw(
            url=url,
            idempotency_key=key,
            payload=webhook.model_dump(mode="json", exclude_none=True),
            extra_headers=extra_headers,
        )

    async def send_wholesale_feed_to_subscription(
        self,
        *,
        subscription: NotificationConfig | Mapping[str, Any],
        account_id: str,
        notification_type: str,
        wholesale_feed_version: str,
        cache_scope: str,
        event: WholesaleFeedEvent | Mapping[str, Any],
        previous_wholesale_feed_version: str | None = None,
        fired_at: datetime | None = None,
        idempotency_key: str | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """POST a wholesale feed notification to a ``NotificationConfig``.

        This convenience wrapper keeps ``url``, ``subscriber_id``, and
        ``event_types`` coupled to the same persisted subscription entry.
        """

        config = (
            subscription
            if isinstance(subscription, NotificationConfig)
            else NotificationConfig.model_validate(subscription)
        )
        return await self.send_wholesale_feed(
            url=str(config.url),
            subscriber_id=config.subscriber_id,
            account_id=account_id,
            notification_type=notification_type,
            wholesale_feed_version=wholesale_feed_version,
            cache_scope=cache_scope,
            event=event,
            previous_wholesale_feed_version=previous_wholesale_feed_version,
            fired_at=fired_at,
            idempotency_key=idempotency_key,
            subscription_event_types=config.event_types,
            extra_headers=extra_headers,
        )

    async def send_webhook_challenge(
        self,
        *,
        url: str,
        account_id: str,
        subscriber_id: str,
        challenge: str | None = None,
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """POST a signed durable-subscription proof-of-control challenge.

        The body matches the durable ``notification_configs[]`` challenge
        shape and intentionally does not inject ``idempotency_key``:

        ``{"type":"webhook.challenge","challenge":"...", ...}``

        Pair this low-level sender method with
        :func:`adcp.webhooks.challenge_webhook_destination` when you also
        want URL validation and response echo checking in one call.
        """

        payload = create_webhook_challenge_payload(
            account_id=account_id,
            subscriber_id=subscriber_id,
            challenge=challenge,
        )
        challenge_value = str(payload["challenge"])
        body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        return await self._send_bytes(
            url=url,
            body=body,
            idempotency_key=challenge_value,
            extra_headers=extra_headers,
        )

    async def send_raw(
        self,
        *,
        url: str,
        idempotency_key: str,
        payload: dict[str, Any],
        extra_headers: Mapping[str, str] | None = None,
    ) -> WebhookDeliveryResult:
        """Low-level escape hatch: sign + POST an arbitrary payload.

        The ``idempotency_key`` kwarg is required and is injected into the
        payload before signing — the visible signature makes the contract
        impossible to forget, unlike a runtime dict check. If ``payload``
        already carries an ``idempotency_key``, the kwarg wins so the two
        cannot disagree.
        """
        if not isinstance(idempotency_key, str) or not idempotency_key:
            raise ValueError("idempotency_key must be a non-empty string")
        body_dict = {**payload, "idempotency_key": idempotency_key}
        # Byte-exact serialization — this is the ONLY representation that
        # gets signed AND posted. Do not allow an httpx `json=` path anywhere
        # in the stack because it would reserialize and break the digest.
        body = json.dumps(body_dict).encode("utf-8")
        if len(body) > _MAX_BODY_BYTES:
            raise ValueError(
                f"serialized webhook body is {len(body):,} bytes, over the "
                f"{_MAX_BODY_BYTES:,}-byte cap. Split into smaller webhooks "
                "or use batch-reporting endpoints."
            )
        return await self._send_bytes(
            url=url,
            body=body,
            idempotency_key=idempotency_key,
            extra_headers=extra_headers,
        )

    async def resend(self, result: WebhookDeliveryResult) -> WebhookDeliveryResult:
        """Replay an earlier delivery under a fresh signature.

        The bytes are identical (same ``idempotency_key``, same payload
        fields, same serialization) — only the Signature / Signature-Input /
        Content-Digest headers are regenerated. The receiver dedupes via
        ``idempotency_key``, so the replayed event is a spec-correct retry
        that won't cause double-processing.
        """
        if not result.sent_body:
            raise ValueError(
                "cannot resend: result has no captured sent_body (likely constructed "
                "externally). Call a send_* method on this sender first."
            )
        return await self._send_bytes(
            url=result.url,
            body=result.sent_body,
            idempotency_key=result.idempotency_key,
            extra_headers=result.sent_extra_headers or None,
        )

    async def _send_bytes(
        self,
        *,
        url: str,
        body: bytes,
        idempotency_key: str,
        extra_headers: Mapping[str, str] | None,
    ) -> WebhookDeliveryResult:
        """Sign + POST a pre-serialized body through an SSRF-validated transport.

        When the sender owns its httpx client (the default — ``client=None``
        was passed to ``__init__``), every delivery builds a per-request
        :class:`adcp.signing.ip_pinned_transport.AsyncIpPinnedTransport`
        that resolves the destination, runs the full SSRF range check
        (loopback / RFC 1918 / link-local / CGNAT / IPv6 ULA / multicast /
        cloud metadata), enforces the port allowlist, and pins the
        connection to the validated IP. This closes the DNS-rebinding
        TOCTOU between validate and connect.

        When the operator supplied their own client
        (``WebhookSender(client=...)`` — typically a vetted egress proxy
        with mTLS to a known buyer set, or an ASGI transport for testing),
        the sender trusts the operator's transport completely. Pin-and-bind
        is skipped; the operator's transport owns SSRF.

        On the owned-client path, SSRF validation runs **before** signing
        so a hostile URL is rejected without first generating an
        Ed25519/ES256 signature over the body. That signature would
        otherwise sit in process memory until the SSRF rejection —
        anything that snapshots locals on exception (faulthandler,
        custom logging) could capture it. Validate first, sign second.

        Transport hooks run before SSRF; the rewritten URL is what gets
        validated, signed, and POSTed. The signature covers the URL the
        request actually lands at, not the URL the caller typed —
        otherwise a receiver computing ``@target-uri`` from its observed
        Host header would see a different value and verification would
        fail. The hook output is bounded (hostname-only rewrite, scheme
        and port preserved), so this can't widen the destination space.
        """
        effective_url = apply_hooks(url, self._transport_hooks)

        # Build the pinned transport up-front for the owned-client path.
        # SSRF + port validation runs against the *post-hook* URL — the
        # one we'll actually connect to. A hostile URL raises
        # SSRFValidationError here and the body never gets signed (no
        # signature material to leak via faulthandler / custom logging
        # on exception).
        transport: AsyncIpPinnedTransport | None = None
        if self._owns_client:
            transport = build_async_ip_pinned_transport(
                effective_url,
                allow_private=self._allow_private_destinations,
                allowed_ports=self._allowed_destination_ports,
            )

        base_headers = {"Content-Type": "application/json", "Accept-Encoding": "identity"}
        auth_headers = self._auth.build_auth_headers(method="POST", url=effective_url, body=body)
        headers = merge_extra_headers(
            base={**base_headers, **auth_headers},
            extra=extra_headers,
            reserved=self._auth.reserved_headers(),
        )

        if transport is not None:
            # Owned-client path. ``trust_env=False`` prevents httpx from
            # routing the request through ``HTTPS_PROXY`` / ``HTTP_PROXY``
            # env vars — every other pinned-transport callsite in the
            # codebase sets this for the same reason (default_jwks_fetcher,
            # async_default_jwks_fetcher, revocation_fetcher). Without it,
            # an attacker who controls process env can route the signed
            # webhook through their endpoint, defeating the IP pin entirely.
            async with httpx.AsyncClient(
                transport=transport,
                timeout=self._timeout,
                follow_redirects=False,
                trust_env=False,
            ) as client:
                status_code, response_headers, response_body = await _post_with_bounded_response(
                    client,
                    effective_url,
                    body=body,
                    headers=headers,
                )
        else:
            # Operator-supplied client — they own the SSRF guarantees on
            # their transport (proxy allowlist, mTLS, etc.). Reachable as
            # None after aclose(); explicit raise survives ``python -O``
            # which would strip an assert.
            if self._client is None:
                raise RuntimeError(
                    "WebhookSender's operator-supplied client was already "
                    "closed. Construct a new sender or pass a fresh client."
                )
            status_code, response_headers, response_body = await _post_with_bounded_response(
                self._client,
                effective_url,
                body=body,
                headers=headers,
            )

        return WebhookDeliveryResult(
            status_code=status_code,
            idempotency_key=idempotency_key,
            url=effective_url,
            response_headers=response_headers,
            response_body=response_body,
            sent_body=body,
            sent_extra_headers=dict(extra_headers) if extra_headers else {},
        )

Outbound signed-webhook delivery client.

Owns one webhook-signing private key. Reuses a single :class:httpx.AsyncClient across requests for connection pooling — pass your own via client= if you want to share it with other SDK surfaces.

Thread/task safety: safe to call concurrent send_* from many asyncio tasks. The underlying httpx.AsyncClient manages its own pool.

Construct a sender wired to RFC 9421 JWK signing.

The HMAC and bearer modes are reached via :meth:from_bearer_token, :meth:from_adcp_legacy_hmac, and :meth:from_standard_webhooks_secret — those classmethods bypass this initializer through :meth:_from_strategy because their key material has different types (bytes / str rather than PrivateKey).

transport_hooks runs URL rewrites before SSRF validation — see :class:DockerLocalhostRewrite for the canonical use case. SSRF remains authoritative on the rewritten URL; hooks cannot punch through the range check.

Static methods

def from_adcp_legacy_hmac(secret: bytes,
*,
key_id: str,
client: httpx.AsyncClient | None = None,
timeout_seconds: float = 10.0,
allow_private_destinations: bool = False,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = ()) ‑> WebhookSender

Build a sender wired to AdCP-legacy HMAC-SHA256.

Wire format matches :func:verify_webhook_hmac(): X-AdCP-Signature: sha256=<hex> over f"{timestamp}.{body}", with X-AdCP-Timestamp set fresh per delivery (resends produce a new signature over the same body).

secret is the raw HMAC key — the AdCP-legacy scheme has no canonical encoding, so callers pass bytes directly. key_id is echoed in X-AdCP-Key-Id for receiver-side multi-key rotation; it is not used in the signature itself.

AdCP-legacy HMAC will be removed in AdCP 4.0 — operators SHOULD migrate to JWK signing (:meth:from_jwk) ahead of that boundary.

def from_bearer_token(token: str,
*,
client: httpx.AsyncClient | None = None,
timeout_seconds: float = 10.0,
allow_private_destinations: bool = False,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = ()) ‑> WebhookSender

Build a sender that POSTs with Authorization: Bearer <token>.

For buyers who authenticate the sender at the gateway and don't verify body signatures. The sender's marshaling guarantees still apply (byte-exact JSON, idempotency_key in body); body signing is skipped.

A buyer treating bearer tokens as the sole authenticity signal SHOULD also enforce TLS/mTLS at the transport layer — a stolen token is a complete forgery. Prefer JWK signing (:meth:from_jwk) for AdCP-conformant deliveries.

def from_jwk(jwk: Mapping[str, Any],
*,
d_field: str = 'd',
client: httpx.AsyncClient | None = None,
timeout_seconds: float = 10.0,
allow_private_destinations: bool = False,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = ()) ‑> WebhookSender

Construct from a JWK that includes the private scalar.

The JWK MUST have adcp_use == "webhook-signing" — the sender doesn't validate this (you're signing with your own key; validation happens at the receiver), but a key whose adcp_use is wrong will be rejected by every conformant verifier.

allow_private_destinations and allowed_destination_ports forward to :meth:__init__ — see that signature for semantics.

def from_pem(pem_path: str | Path | bytes,
*,
key_id: str,
alg: str = 'ed25519',
passphrase: bytes | None = None,
client: httpx.AsyncClient | None = None,
timeout_seconds: float = 10.0,
allow_private_destinations: bool = False,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = ()) ‑> WebhookSender

Load a private key from a PEM file and bind it as a webhook sender.

Companion to adcp-keygen --purpose webhook-signing, which writes the PEM and prints the public JWK. The JWK is published at your jwks_uri; the PEM holds the private key material. from_pem reads the PEM, constructs the right PrivateKey type for alg, and returns a sender ready to send.

Args
-----=
pem_path
Path to the PKCS#8 PEM, or the PEM bytes directly.
key_id
JWK kid claim — must match the published JWK.
alg
Signature algorithm. ed25519 (default) or es256. Also accepts the RFC 9421 form ecdsa-p256-sha256.
passphrase
Required if the PEM is encrypted (adcp-keygen --encrypt).
client
Optional pre-built :class:httpx.AsyncClient to share across the SDK; the sender owns its own client when omitted.
timeout_seconds
Per-request timeout for the owned client.
allow_private_destinations
Forwarded to :meth:__init__.
allowed_destination_ports
Forwarded to :meth:__init__.
Raises
-----=
ValueError
alg is not ed25519 / es256, or the PEM contains a key whose type doesn't match alg.
def from_standard_webhooks_secret(secret: str,
*,
key_id: str,
client: httpx.AsyncClient | None = None,
timeout_seconds: float = 10.0,
allow_private_destinations: bool = False,
allowed_destination_ports: frozenset[int] | None = None,
transport_hooks: tuple[TransportHook, ...] = ()) ‑> WebhookSender

Build a sender wired to standardwebhooks.com v1 (Svix/Resend interop).

secret is the canonical whsec_<base64> form distributed by buyers running Svix, Resend, or any other Standard Webhooks verifier. The constructor base64-decodes the prefix-stripped payload internally — passing the literal whsec_… to :meth:from_adcp_legacy_hmac would silently produce signatures Svix rejects, which is exactly the footgun this typed split prevents.

Wire format per spec: webhook-id / webhook-timestamp / webhook-signature: v1,<base64> over f"{webhook_id}.{webhook_timestamp}.{body}". Each delivery gets a fresh webhook-id so a receiver using webhook-id for its own replay cache doesn't false-positive on a legitimate retry — :meth:resend re-signs and gets a new id.

Instance variables

prop signs_with_rfc9421 : bool
Expand source code 
@property
def signs_with_rfc9421(self) -> bool:
    """``True`` iff this sender uses the RFC 9421 webhook-signing profile.

    Boot-time validators read this to enforce the
    ``webhook_signing.supported=true`` capability invariant:
    capabilities advertise RFC 9421 → wired sender must produce
    ``Signature`` / ``Signature-Input`` headers. ``from_bearer_token``,
    ``from_adcp_legacy_hmac``, and ``from_standard_webhooks_secret``
    senders return ``False``.
    """
    return isinstance(self._auth, JwkSignerStrategy)

True iff this sender uses the RFC 9421 webhook-signing profile.

Boot-time validators read this to enforce the webhook_signing.supported=true capability invariant: capabilities advertise RFC 9421 → wired sender must produce Signature / Signature-Input headers. from_bearer_token, from_adcp_legacy_hmac, and from_standard_webhooks_secret senders return False.

Methods

async def aclose(self) ‑> None
Expand source code 
async def aclose(self) -> None:
    """Close the internal httpx client if we own it."""
    if self._owns_client and self._client is not None:
        await self._client.aclose()
        self._client = None

Close the internal httpx client if we own it.

def prepare_mcp(self,
*,
url: str,
task_id: str,
status: GeneratedTaskStatus | str,
task_type: TaskType | str,
result: AdcpAsyncResponseData | dict[str, Any] | None = None,
timestamp: datetime | None = None,
operation_id: str,
notification_id: str | None = None,
message: str | None = None,
context_id: str | None = None,
protocol: AdcpProtocol | str | None = None,
idempotency_key: str | None = None,
token: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> PreparedWebhook
Expand source code 
def prepare_mcp(
    self,
    *,
    url: str,
    task_id: str,
    status: GeneratedTaskStatus | str,
    task_type: TaskType | str,
    result: AdcpAsyncResponseData | dict[str, Any] | None = None,
    timestamp: datetime | None = None,
    operation_id: str,
    notification_id: str | None = None,
    message: str | None = None,
    context_id: str | None = None,
    protocol: AdcpProtocol | str | None = None,
    idempotency_key: str | None = None,
    token: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> PreparedWebhook:
    """Prepare an MCP webhook without performing network I/O.

    Durable publishers call this inside (or immediately before) their
    outbox transaction and persist all returned fields verbatim. The
    returned body is exactly what :meth:`send_prepared` signs and posts.
    """
    payload = create_mcp_webhook_payload(
        task_id=task_id,
        status=status,
        task_type=task_type,
        result=result,
        timestamp=timestamp,
        operation_id=operation_id,
        notification_id=notification_id,
        message=message,
        context_id=context_id,
        protocol=protocol,
        idempotency_key=idempotency_key,
        token=token,
    )
    body_dict = {
        **to_wire_dict(payload),
        "idempotency_key": payload.idempotency_key,
    }
    body = json.dumps(body_dict).encode("utf-8")
    if len(body) > _MAX_BODY_BYTES:
        raise ValueError(
            f"serialized webhook body is {len(body):,} bytes, over the "
            f"{_MAX_BODY_BYTES:,}-byte cap. Split into smaller webhooks "
            "or use batch-reporting endpoints."
        )
    return PreparedWebhook(
        url=url,
        idempotency_key=payload.idempotency_key,
        body=body,
        extra_headers=dict(extra_headers) if extra_headers else {},
    )

Prepare an MCP webhook without performing network I/O.

Durable publishers call this inside (or immediately before) their outbox transaction and persist all returned fields verbatim. The returned body is exactly what :meth:send_prepared signs and posts.

async def resend(self,
result: WebhookDeliveryResult) ‑> WebhookDeliveryResult
Expand source code 
async def resend(self, result: WebhookDeliveryResult) -> WebhookDeliveryResult:
    """Replay an earlier delivery under a fresh signature.

    The bytes are identical (same ``idempotency_key``, same payload
    fields, same serialization) — only the Signature / Signature-Input /
    Content-Digest headers are regenerated. The receiver dedupes via
    ``idempotency_key``, so the replayed event is a spec-correct retry
    that won't cause double-processing.
    """
    if not result.sent_body:
        raise ValueError(
            "cannot resend: result has no captured sent_body (likely constructed "
            "externally). Call a send_* method on this sender first."
        )
    return await self._send_bytes(
        url=result.url,
        body=result.sent_body,
        idempotency_key=result.idempotency_key,
        extra_headers=result.sent_extra_headers or None,
    )

Replay an earlier delivery under a fresh signature.

The bytes are identical (same idempotency_key, same payload fields, same serialization) — only the Signature / Signature-Input / Content-Digest headers are regenerated. The receiver dedupes via idempotency_key, so the replayed event is a spec-correct retry that won't cause double-processing.

async def send_artifact_webhook(self,
*,
url: str,
media_buy_id: str,
batch_id: str,
timestamp: str,
artifacts: list[dict[str, Any]],
idempotency_key: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_artifact_webhook(
    self,
    *,
    url: str,
    media_buy_id: str,
    batch_id: str,
    timestamp: str,
    artifacts: list[dict[str, Any]],
    idempotency_key: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """POST a signed content-standards artifact webhook."""
    key = idempotency_key or generate_webhook_idempotency_key()
    payload: dict[str, Any] = {
        "idempotency_key": key,
        "media_buy_id": media_buy_id,
        "batch_id": batch_id,
        "timestamp": timestamp,
        "artifacts": artifacts,
    }
    return await self.send_raw(
        url=url, idempotency_key=key, payload=payload, extra_headers=extra_headers
    )

POST a signed content-standards artifact webhook.

async def send_collection_list_changed(self,
*,
url: str,
list_id: str,
resolved_at: str,
signature: str,
idempotency_key: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_collection_list_changed(
    self,
    *,
    url: str,
    list_id: str,
    resolved_at: str,
    signature: str,
    idempotency_key: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """POST a signed governance collection-list-changed webhook.

    ``signature`` is the payload-level signature field that predates 9421
    webhook transport signing — it remains required by the schema. The
    9421 signature this method adds protects the transport envelope.
    """
    key = idempotency_key or generate_webhook_idempotency_key()
    payload: dict[str, Any] = {
        "idempotency_key": key,
        "event": "collection_list_changed",
        "list_id": list_id,
        "resolved_at": resolved_at,
        "signature": signature,
    }
    return await self.send_raw(
        url=url, idempotency_key=key, payload=payload, extra_headers=extra_headers
    )

POST a signed governance collection-list-changed webhook.

signature is the payload-level signature field that predates 9421 webhook transport signing — it remains required by the schema. The 9421 signature this method adds protects the transport envelope.

async def send_mcp(self,
*,
url: str,
task_id: str,
status: GeneratedTaskStatus | str,
task_type: TaskType | str,
result: AdcpAsyncResponseData | dict[str, Any] | None = None,
timestamp: datetime | None = None,
operation_id: str,
notification_id: str | None = None,
message: str | None = None,
context_id: str | None = None,
protocol: AdcpProtocol | str | None = None,
idempotency_key: str | None = None,
token: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_mcp(
    self,
    *,
    url: str,
    task_id: str,
    status: GeneratedTaskStatus | str,
    task_type: TaskType | str,
    result: AdcpAsyncResponseData | dict[str, Any] | None = None,
    timestamp: datetime | None = None,
    operation_id: str,
    notification_id: str | None = None,
    message: str | None = None,
    context_id: str | None = None,
    protocol: AdcpProtocol | str | None = None,
    idempotency_key: str | None = None,
    token: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """POST a signed MCP-style task-status webhook.

    On retry, prefer :meth:`resend` over calling this again — ``resend``
    replays the exact same bytes, whereas re-invoking ``send_mcp`` with
    the "same" args would produce a fresh ``timestamp`` and potentially
    a different serialized body, which conflicts with the delivery key's
    immutable payload binding.

    :param token: Buyer-supplied token from
        ``push_notification_config.token`` echoed back on the
        payload's ``token`` field per spec
        (``schemas/cache/core/push_notification_config.json``: "Echoed
        back in webhook payload to validate request authenticity").
        Cross-language wire-parity with the JS implementation.
    """
    return await self.send_prepared(
        self.prepare_mcp(
            url=url,
            task_id=task_id,
            status=status,
            task_type=task_type,
            result=result,
            timestamp=timestamp,
            operation_id=operation_id,
            notification_id=notification_id,
            message=message,
            context_id=context_id,
            protocol=protocol,
            idempotency_key=idempotency_key,
            token=token,
            extra_headers=extra_headers,
        )
    )

POST a signed MCP-style task-status webhook.

On retry, prefer :meth:resend over calling this again — resend replays the exact same bytes, whereas re-invoking send_mcp with the "same" args would produce a fresh timestamp and potentially a different serialized body, which conflicts with the delivery key's immutable payload binding.

:param token: Buyer-supplied token from push_notification_config.token echoed back on the payload's token field per spec (schemas/cache/core/push_notification_config.json: "Echoed back in webhook payload to validate request authenticity"). Cross-language wire-parity with the JS implementation.

async def send_prepared(self,
prepared: PreparedWebhook) ‑> WebhookDeliveryResult
Expand source code 
async def send_prepared(self, prepared: PreparedWebhook) -> WebhookDeliveryResult:
    """Sign and post a previously prepared immutable webhook request."""
    if not prepared.idempotency_key:
        raise ValueError("prepared webhook idempotency_key must be non-empty")
    if not prepared.body:
        raise ValueError("prepared webhook body must be non-empty")
    if len(prepared.body) > _MAX_BODY_BYTES:
        raise ValueError(
            f"serialized webhook body is {len(prepared.body):,} bytes, over the "
            f"{_MAX_BODY_BYTES:,}-byte cap"
        )
    try:
        payload = json.loads(prepared.body)
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise ValueError("prepared webhook body must be a JSON object") from exc
    if not isinstance(payload, dict):
        raise ValueError("prepared webhook body must be a JSON object")
    if payload.get("idempotency_key") != prepared.idempotency_key:
        raise ValueError(
            "prepared webhook body idempotency_key does not match its immutable binding"
        )
    return await self._send_bytes(
        url=prepared.url,
        body=prepared.body,
        idempotency_key=prepared.idempotency_key,
        extra_headers=prepared.extra_headers or None,
    )

Sign and post a previously prepared immutable webhook request.

async def send_property_list_changed(self,
*,
url: str,
list_id: str,
resolved_at: str,
signature: str,
idempotency_key: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_property_list_changed(
    self,
    *,
    url: str,
    list_id: str,
    resolved_at: str,
    signature: str,
    idempotency_key: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """POST a signed governance property-list-changed webhook."""
    key = idempotency_key or generate_webhook_idempotency_key()
    payload: dict[str, Any] = {
        "idempotency_key": key,
        "event": "property_list_changed",
        "list_id": list_id,
        "resolved_at": resolved_at,
        "signature": signature,
    }
    return await self.send_raw(
        url=url, idempotency_key=key, payload=payload, extra_headers=extra_headers
    )

POST a signed governance property-list-changed webhook.

async def send_raw(self,
*,
url: str,
idempotency_key: str,
payload: dict[str, Any],
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_raw(
    self,
    *,
    url: str,
    idempotency_key: str,
    payload: dict[str, Any],
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """Low-level escape hatch: sign + POST an arbitrary payload.

    The ``idempotency_key`` kwarg is required and is injected into the
    payload before signing — the visible signature makes the contract
    impossible to forget, unlike a runtime dict check. If ``payload``
    already carries an ``idempotency_key``, the kwarg wins so the two
    cannot disagree.
    """
    if not isinstance(idempotency_key, str) or not idempotency_key:
        raise ValueError("idempotency_key must be a non-empty string")
    body_dict = {**payload, "idempotency_key": idempotency_key}
    # Byte-exact serialization — this is the ONLY representation that
    # gets signed AND posted. Do not allow an httpx `json=` path anywhere
    # in the stack because it would reserialize and break the digest.
    body = json.dumps(body_dict).encode("utf-8")
    if len(body) > _MAX_BODY_BYTES:
        raise ValueError(
            f"serialized webhook body is {len(body):,} bytes, over the "
            f"{_MAX_BODY_BYTES:,}-byte cap. Split into smaller webhooks "
            "or use batch-reporting endpoints."
        )
    return await self._send_bytes(
        url=url,
        body=body,
        idempotency_key=idempotency_key,
        extra_headers=extra_headers,
    )

Low-level escape hatch: sign + POST an arbitrary payload.

The idempotency_key kwarg is required and is injected into the payload before signing — the visible signature makes the contract impossible to forget, unlike a runtime dict check. If payload already carries an idempotency_key, the kwarg wins so the two cannot disagree.

async def send_revocation_notification(self,
*,
url: str,
rights_id: str,
brand_id: str,
reason: str,
effective_at: str,
idempotency_key: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_revocation_notification(
    self,
    *,
    url: str,
    rights_id: str,
    brand_id: str,
    reason: str,
    effective_at: str,
    idempotency_key: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """POST a signed rights-revocation notification."""
    key = idempotency_key or generate_webhook_idempotency_key()
    payload: dict[str, Any] = {
        "idempotency_key": key,
        "rights_id": rights_id,
        "brand_id": brand_id,
        "reason": reason,
        "effective_at": effective_at,
    }
    return await self.send_raw(
        url=url, idempotency_key=key, payload=payload, extra_headers=extra_headers
    )

POST a signed rights-revocation notification.

async def send_webhook_challenge(self,
*,
url: str,
account_id: str,
subscriber_id: str,
challenge: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_webhook_challenge(
    self,
    *,
    url: str,
    account_id: str,
    subscriber_id: str,
    challenge: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """POST a signed durable-subscription proof-of-control challenge.

    The body matches the durable ``notification_configs[]`` challenge
    shape and intentionally does not inject ``idempotency_key``:

    ``{"type":"webhook.challenge","challenge":"...", ...}``

    Pair this low-level sender method with
    :func:`adcp.webhooks.challenge_webhook_destination` when you also
    want URL validation and response echo checking in one call.
    """

    payload = create_webhook_challenge_payload(
        account_id=account_id,
        subscriber_id=subscriber_id,
        challenge=challenge,
    )
    challenge_value = str(payload["challenge"])
    body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
    return await self._send_bytes(
        url=url,
        body=body,
        idempotency_key=challenge_value,
        extra_headers=extra_headers,
    )

POST a signed durable-subscription proof-of-control challenge.

The body matches the durable notification_configs[] challenge shape and intentionally does not inject idempotency_key:

{"type":"webhook.challenge","challenge":"...", ...}

Pair this low-level sender method with :func:challenge_webhook_destination() when you also want URL validation and response echo checking in one call.

async def send_wholesale_feed(self,
*,
url: str,
subscriber_id: str,
account_id: str,
notification_type: str,
wholesale_feed_version: str,
cache_scope: str,
event: WholesaleFeedEvent | Mapping[str, Any],
previous_wholesale_feed_version: str | None = None,
fired_at: datetime | None = None,
idempotency_key: str | None = None,
subscription_event_types: Sequence[Any] | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_wholesale_feed(
    self,
    *,
    url: str,
    subscriber_id: str,
    account_id: str,
    notification_type: str,
    wholesale_feed_version: str,
    cache_scope: str,
    event: WholesaleFeedEvent | Mapping[str, Any],
    previous_wholesale_feed_version: str | None = None,
    fired_at: datetime | None = None,
    idempotency_key: str | None = None,
    subscription_event_types: Sequence[Any] | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """POST a signed account-scoped wholesale feed notification.

    ``subscription_event_types`` is optional but recommended when the
    caller is sending to an ``accounts[].notification_configs[]`` entry:
    pass that entry's ``event_types`` to fail closed if the subscription
    did not request this notification type.
    """

    if not isinstance(subscriber_id, str) or not subscriber_id:
        raise ValueError("subscriber_id must be a non-empty string")
    if not isinstance(account_id, str) or not account_id:
        raise ValueError("account_id must be a non-empty string")
    if not isinstance(wholesale_feed_version, str) or not wholesale_feed_version:
        raise ValueError("wholesale_feed_version must be a non-empty string")

    event_model = event
    if not isinstance(event_model, WholesaleFeedEvent):
        event_model = WholesaleFeedEvent.model_validate(event_model)
    notification_type_value = _enum_value(notification_type)
    event_type = _enum_value(event_model.event_type)
    entity_type = _enum_value(event_model.entity_type)
    if notification_type_value != event_type:
        raise ValueError(
            "notification_type must match event.event_type "
            f"(got {notification_type_value!r}, event has {event_type!r})"
        )
    if subscription_event_types is not None:
        allowed_event_types = {_enum_value(item) for item in subscription_event_types}
    else:
        allowed_event_types = None
    if allowed_event_types is not None and notification_type_value not in allowed_event_types:
        raise ValueError(
            "notification_type is not present in the subscription's event_types; "
            "sellers must not silently widen account notification filters"
        )

    expected_entity_type = _entity_type_for_wholesale_notification(notification_type_value)
    if entity_type != expected_entity_type:
        raise ValueError(
            "event.entity_type does not match notification_type "
            f"(got {entity_type!r}, expected {expected_entity_type!r})"
        )

    cache_scope_value = _enum_value(cache_scope)
    applies_to = getattr(event_model.payload, "applies_to", None)
    applies_to_scope = _enum_value(getattr(applies_to, "scope", None))
    if applies_to_scope != cache_scope_value:
        raise ValueError(
            "cache_scope must match event.payload.applies_to.scope "
            f"(got {cache_scope_value!r}, event has {applies_to_scope!r})"
        )

    key = idempotency_key or generate_webhook_idempotency_key()
    timestamp = fired_at or datetime.now(timezone.utc)
    webhook = WholesaleFeedWebhook.model_validate(
        {
            "idempotency_key": key,
            "notification_id": event_model.event_id,
            "notification_type": notification_type_value,
            "fired_at": timestamp,
            "subscriber_id": subscriber_id,
            "account_id": account_id,
            "wholesale_feed_version": wholesale_feed_version,
            "previous_wholesale_feed_version": previous_wholesale_feed_version,
            "cache_scope": cache_scope_value,
            "event": event_model,
        }
    )
    return await self.send_raw(
        url=url,
        idempotency_key=key,
        payload=webhook.model_dump(mode="json", exclude_none=True),
        extra_headers=extra_headers,
    )

POST a signed account-scoped wholesale feed notification.

subscription_event_types is optional but recommended when the caller is sending to an accounts[].notification_configs[] entry: pass that entry's event_types to fail closed if the subscription did not request this notification type.

async def send_wholesale_feed_to_subscription(self,
*,
subscription: NotificationConfig | Mapping[str, Any],
account_id: str,
notification_type: str,
wholesale_feed_version: str,
cache_scope: str,
event: WholesaleFeedEvent | Mapping[str, Any],
previous_wholesale_feed_version: str | None = None,
fired_at: datetime | None = None,
idempotency_key: str | None = None,
extra_headers: Mapping[str, str] | None = None) ‑> WebhookDeliveryResult
Expand source code 
async def send_wholesale_feed_to_subscription(
    self,
    *,
    subscription: NotificationConfig | Mapping[str, Any],
    account_id: str,
    notification_type: str,
    wholesale_feed_version: str,
    cache_scope: str,
    event: WholesaleFeedEvent | Mapping[str, Any],
    previous_wholesale_feed_version: str | None = None,
    fired_at: datetime | None = None,
    idempotency_key: str | None = None,
    extra_headers: Mapping[str, str] | None = None,
) -> WebhookDeliveryResult:
    """POST a wholesale feed notification to a ``NotificationConfig``.

    This convenience wrapper keeps ``url``, ``subscriber_id``, and
    ``event_types`` coupled to the same persisted subscription entry.
    """

    config = (
        subscription
        if isinstance(subscription, NotificationConfig)
        else NotificationConfig.model_validate(subscription)
    )
    return await self.send_wholesale_feed(
        url=str(config.url),
        subscriber_id=config.subscriber_id,
        account_id=account_id,
        notification_type=notification_type,
        wholesale_feed_version=wholesale_feed_version,
        cache_scope=cache_scope,
        event=event,
        previous_wholesale_feed_version=previous_wholesale_feed_version,
        fired_at=fired_at,
        idempotency_key=idempotency_key,
        subscription_event_types=config.event_types,
        extra_headers=extra_headers,
    )

POST a wholesale feed notification to a NotificationConfig.

This convenience wrapper keeps url, subscriber_id, and event_types coupled to the same persisted subscription entry.

class WebhookVerifyOptions (*,
jwks_resolver: JwksResolver,
replay_store: ReplayStore | None = <factory>,
revocation_checker: RevocationChecker | None = None,
revocation_list: RevocationList | None = None,
max_skew_seconds: int = 60,
max_window_seconds: int = 300,
label: str = 'sig1',
allowed_algs: frozenset[str] = frozenset({'ed25519', 'ecdsa-p256-sha256'}),
sender_url: str | None = None,
expected_key_origins: Mapping[str, str] | None = None,
posture: str | None = None,
clock: Callable[[], float] = <built-in function time>)
Expand source code 
@dataclass(frozen=True, kw_only=True)
class WebhookVerifyOptions:
    """Options for the webhook verifier.

    Subset of :class:`VerifyOptions` — several fields are pinned (tag, adcp_use,
    content-digest policy) because the webhook profile doesn't leave them as
    caller choices.

    Unlike the request verifier, there is no ``now`` field — the webhook
    verifier stamps time-of-check itself, so the same :class:`WebhookVerifyOptions`
    instance can live for the lifetime of your receiver without a factory
    closure around it. Override via ``clock=`` for deterministic tests.

    ``replay_store`` defaults to a per-options in-memory store so captured
    signatures are rejected without extra configuration. Multi-process
    receivers should supply a shared store. Passing ``None`` explicitly is
    the opt-out for specialized tests or externally enforced replay policy.
    """

    jwks_resolver: JwksResolver
    replay_store: ReplayStore | None = field(default_factory=InMemoryReplayStore)
    revocation_checker: RevocationChecker | None = None
    revocation_list: RevocationList | None = None
    max_skew_seconds: int = DEFAULT_SKEW_SECONDS
    max_window_seconds: int = MAX_WINDOW_SECONDS
    label: str = SIG_LABEL_DEFAULT
    allowed_algs: frozenset[str] = ALLOWED_ALGS
    sender_url: str | None = None
    expected_key_origins: Mapping[str, str] | None = None
    posture: str | None = None
    clock: Callable[[], float] = time.time

Options for the webhook verifier.

Subset of :class:VerifyOptions — several fields are pinned (tag, adcp_use, content-digest policy) because the webhook profile doesn't leave them as caller choices.

Unlike the request verifier, there is no now field — the webhook verifier stamps time-of-check itself, so the same :class:WebhookVerifyOptions instance can live for the lifetime of your receiver without a factory closure around it. Override via clock= for deterministic tests.

replay_store defaults to a per-options in-memory store so captured signatures are rejected without extra configuration. Multi-process receivers should supply a shared store. Passing None explicitly is the opt-out for specialized tests or externally enforced replay policy.

Instance variables

var allowed_algs : frozenset[str]
var expected_key_origins : collections.abc.Mapping[str, str] | None
var jwks_resolver : JwksResolver
var label : str
var max_skew_seconds : int
var max_window_seconds : int
var posture : str | None
var replay_store : ReplayStore | None
var revocation_checker : RevocationChecker | None
var revocation_list : RevocationList | None
var sender_url : str | None

Methods

def clock(...) ‑> Callable[[], float]

time() -> floating point number

Return the current time in seconds since the Epoch. Fractions of a second may be present if the system clock provides them.