Module adcp.webhook_sender

One-call outbound webhook delivery for AdCP senders.

A seller that wants to emit a signed webhook today has to do six steps by hand — construct payload, JSON-serialize to bytes, sign, merge headers, POST with content= (not json=, which reserializes and breaks the signature), and remember to reuse idempotency_key on retry. Each step is a footgun.

:class:WebhookSender packages all of them::

from adcp.webhooks import WebhookSender

sender = WebhookSender.from_jwk(webhook_signing_jwk_with_private_d)

async with sender:
    result = await sender.send_mcp(
        url="https://buyer.example.com/webhooks/adcp/create_media_buy/op_abc",
        task_id="task_456",
        task_type="create_media_buy",
        status="completed",
        result={"media_buy_id": "mb_1"},
    )
    if not result.ok:
        # Retry replays the exact same bytes under a fresh signature,
        # preserving idempotency_key so the receiver dedupes.
        retry = await sender.resend(result)

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: str = "host.docker.internal"

    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.

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 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 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 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 | 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 the receiver would dedupe but
        with different observed payload data.

        :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.
        """
        payload = create_mcp_webhook_payload(
            task_id=task_id,
            status=status,
            task_type=task_type,
            result=result,
            timestamp=timestamp,
            operation_id=operation_id,
            message=message,
            context_id=context_id,
            protocol=protocol,
            idempotency_key=idempotency_key,
            token=token,
        )
        return await self.send_raw(
            url=url,
            idempotency_key=payload.idempotency_key,
            payload=to_wire_dict(payload),
            extra_headers=extra_headers,
        )

    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"}
        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:
                response = await client.post(effective_url, content=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."
                )
            response = await self._client.post(effective_url, content=body, headers=headers)

        return WebhookDeliveryResult(
            status_code=response.status_code,
            idempotency_key=idempotency_key,
            url=effective_url,
            response_headers=dict(response.headers),
            response_body=response.content,
            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.

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 | 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 | 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 the receiver would dedupe but
    with different observed payload data.

    :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.
    """
    payload = create_mcp_webhook_payload(
        task_id=task_id,
        status=status,
        task_type=task_type,
        result=result,
        timestamp=timestamp,
        operation_id=operation_id,
        message=message,
        context_id=context_id,
        protocol=protocol,
        idempotency_key=idempotency_key,
        token=token,
    )
    return await self.send_raw(
        url=url,
        idempotency_key=payload.idempotency_key,
        payload=to_wire_dict(payload),
        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 the receiver would dedupe but with different observed payload data.

: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_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.