Module adcp.canonical_formats

Pythonic v1↔v2 canonical-formats projection layer.

AdCP 3.1 introduced canonical formats as the v2 catalog-side vocabulary that replaces v1's per-publisher format proliferation. A seller publishes Product.format_options[] carrying ProductFormatDeclaration entries (one per accepted canonical kind); buyer agents reason about creative compatibility against the closed canonical set rather than the open v1 format_ids namespace.

During the migration window both wire shapes coexist: 3.0-era buyers read Product.format_ids[] (v1) and 3.1-aware buyers read Product.format_options[] (v2). This module supplies the projection layer the SDK needs to bridge them without adopters hand-rolling translation per integration.

Public surface

Resolution-order semantics for v2 → v1 follow registries/v1-canonical-mapping.json:

  1. canonical_formats_only=True or format_kind=custom → no v1 emit, no advisory.
  2. v1_format_ref[] set → emit those refs; if params.sizes[] count exceeds v1_format_ref[] count, emit FORMAT_DECLARATION_V1_LOSSY_MULTI_SIZE.
  3. Canonical's v1_translatable=False (agent_placement, sponsored_placement, responsive_creative, image_carousel) → no v1 emit, no advisory — the canonical is structurally v1-unreachable by design.
  4. Canonical's v1_translatable=True but no v1_format_ref[] → emit FORMAT_DECLARATION_V1_AMBIGUOUS. SDKs MUST NOT synthesize a v1 format_id from registry structural matches; the registry is authoritative for v1→v2 projection only.

Sub-modules

adcp.canonical_formats.advisory

SDK-source errors[] advisory construction …

adcp.canonical_formats.compat_helpers

Compatibility helpers for legacy and parameterized creative format IDs …

adcp.canonical_formats.dialect

Negotiated creative dialect selection for AdCP 3.x.

adcp.canonical_formats.fixtures

Public access to the vendored canonical-formats reference fixtures …

adcp.canonical_formats.format_options

Closed-set adcp.canonical_formats.format_options[] validation …

adcp.canonical_formats.identity

Format identity normalization helpers.

adcp.canonical_formats.narrowing

FORMAT_DECLARATION_DIVERGENT narrowing check …

adcp.canonical_formats.pixel_tracker

Bidirectional adcp.canonical_formats.pixel_tracker ↔ v1 url asset projection …

adcp.canonical_formats.projection

TypeScript RC3-compatible legacy-to-canonical creative projection.

adcp.canonical_formats.references

Fetch and cache immutable canonical-format reference documents …

adcp.canonical_formats.registry

v1↔v2 canonical mapping registry loader + matchers …

adcp.canonical_formats.v1_to_v2

v1 → v2 canonical-format projection …

adcp.canonical_formats.v2_to_v1

v2 → v1 canonical-format projection …

Functions

def build_catalog_index(entries: Iterable[Mapping[str, Any]]) ‑> CatalogIndex
Expand source code 
def build_catalog_index(entries: Iterable[Mapping[str, Any]]) -> CatalogIndex:
    """Build exact-owner and collision-aware bare-ID indexes."""

    by_owner_and_id: dict[tuple[str, str], dict[str, Any] | None] = {}
    by_unique_id: dict[str, dict[str, Any] | None] = {}
    for raw in entries:
        entry = dict(raw)
        ref = entry.get("format_id")
        if not isinstance(ref, Mapping):
            continue
        owner = _catalog_owner(ref.get("agent_url"))
        identifier = ref.get("id")
        if owner is None or not isinstance(identifier, str) or not identifier:
            continue
        owner_key = (owner, identifier)
        if owner_key in by_owner_and_id:
            by_owner_and_id[owner_key] = None
        else:
            by_owner_and_id[owner_key] = entry
        if identifier in by_unique_id:
            by_unique_id[identifier] = None
        else:
            by_unique_id[identifier] = entry
    return CatalogIndex(by_owner_and_id=by_owner_and_id, by_unique_id=by_unique_id)

Build exact-owner and collision-aware bare-ID indexes.

def canonical_creatives_capability(capabilities: Any) ‑> bool | None
Expand source code 
def canonical_creatives_capability(capabilities: Any) -> bool | None:
    """Read ``media_buy.features.canonical_creatives`` without guessing."""

    root = _as_mapping(capabilities)
    if root is None:
        return None
    media_buy = _as_mapping(root.get("media_buy"))
    features = _as_mapping(media_buy.get("features")) if media_buy else None
    value = features.get("canonical_creatives") if features else None
    return value if isinstance(value, bool) else None

Read media_buy.features.canonical_creatives without guessing.

def canonical_format_legacy_resolver_from_catalog_snapshots(snapshots: Iterable[Mapping[str, Any]]) ‑> Callable[[CanonicalFormatLegacyResolutionContext], collections.abc.Sequence[LegacyFormatId] | None]
Expand source code 
def canonical_format_legacy_resolver_from_catalog_snapshots(
    snapshots: Iterable[Mapping[str, Any]],
) -> CanonicalFormatLegacyResolver:
    """Compile owner-scoped, one-to-one durable reverse routes."""

    routes: dict[
        tuple[str | None, str, str],
        tuple[int, tuple[dict[str, Any], list[LegacyFormatId]] | None],
    ] = {}
    for snapshot in snapshots:
        priority = _snapshot_priority(snapshot)
        for raw in _snapshot_formats(snapshot):
            if raw.get("canonical_formats_only") is True:
                continue
            option_id = raw.get("format_option_id")
            kind = raw.get("format_kind")
            refs = raw.get("v1_format_ref")
            if (
                not isinstance(option_id, str)
                or not isinstance(kind, str)
                or kind in _CANONICAL_ONLY_FORMAT_KINDS
                or not isinstance(refs, list)
            ):
                continue
            try:
                parsed = [LegacyFormatId.model_validate(ref) for ref in refs]
            except ValidationError:
                continue
            if not parsed:
                continue
            publisher = _normalized_publisher_domain(
                raw.get("publisher_domain", snapshot.get("publisher_domain"))
            )
            key = (publisher or None, option_id, kind)
            candidate = (deepcopy(raw.get("params") or {}), parsed)
            existing = routes.get(key)
            if existing is None or priority < existing[0]:
                routes[key] = (priority, candidate)
            elif priority == existing[0]:
                routes[key] = (priority, None)

    def resolve(context: CanonicalFormatLegacyResolutionContext) -> Sequence[LegacyFormatId] | None:
        declaration = context.declaration
        if not declaration.format_option_id:
            return None
        ranked = routes.get(
            (
                _normalized_publisher_domain(declaration.publisher_domain) or None,
                declaration.format_option_id,
                declaration.format_kind.value,
            )
        )
        if ranked is not None and ranked[1] is None:
            raise CanonicalFormatLegacyResolutionError(
                "projection catalog contains ambiguous canonical format option aliases"
            )
        route = ranked[1] if ranked else None
        if not route:
            return None
        expected_params, refs = route
        if declaration.params != expected_params:
            return None
        return tuple(refs)

    return resolve

Compile owner-scoped, one-to-one durable reverse routes.

def check_narrows(v2_params: dict[str, Any] | Any, v1_requirements: dict[str, Any] | Any) ‑> list[Divergence]
Expand source code 
def check_narrows(
    v2_params: dict[str, Any] | Any,
    v1_requirements: dict[str, Any] | Any,
) -> list[Divergence]:
    """Compare ``v2_params`` against ``v1_requirements`` and return divergences.

    Returns an empty list when ``v2_params`` narrows ``v1_requirements``
    (the spec-conformant case). Returns a list of :class:`Divergence`
    records when divergent — one per diverging field. Call
    :meth:`Divergence.to_dict` to project a record onto the wire shape
    used in advisory ``details.divergences``.
    """
    v2 = _as_dict(v2_params)
    v1 = _as_dict(v1_requirements)
    if not v2 or not v1:
        return []

    divergences: list[Divergence] = []

    for field_name in _MAX_FIELDS:
        v1_max = v1.get(field_name)
        if not _is_numeric(v1_max):
            continue
        # v2 may carry the cap directly OR the value being capped (e.g.,
        # v1 declares ``max_width`` and v2 declares ``width``).
        v2_value = v2.get(field_name)
        if v2_value is None:
            v2_value = v2.get(field_name.removeprefix("max_"))
        if v2_value is None:
            continue
        # The value-being-capped form: v2 ``width`` against v1 ``max_width``
        # is a "v2 value MUST be ≤ v1 cap" check.
        if _is_numeric(v2_value) and v2_value > v1_max:
            divergences.append(
                Divergence(field=field_name, kind="exceeds_max", cap=v1_max, value=v2_value)
            )

    for field_name in _MIN_FIELDS:
        v1_min = v1.get(field_name)
        if not _is_numeric(v1_min):
            continue
        v2_value = v2.get(field_name)
        if v2_value is None:
            v2_value = v2.get(field_name.removeprefix("min_"))
        if v2_value is None:
            continue
        if _is_numeric(v2_value) and v2_value < v1_min:
            divergences.append(
                Divergence(field=field_name, kind="below_min", cap=v1_min, value=v2_value)
            )

    for field_name in _ENUM_SUBSET_FIELDS:
        v1_set = v1.get(field_name)
        v2_set = v2.get(field_name)
        if v1_set is None or v2_set is None:
            continue
        if not _is_subset(v2_set, v1_set):
            divergences.append(
                Divergence(
                    field=field_name,
                    kind="not_subset",
                    cap=_echo_set(v1_set),
                    value=_echo_set(v2_set),
                )
            )

    for field_name in _EXACT_FIELDS:
        v1_value = v1.get(field_name)
        v2_value = v2.get(field_name)
        if v1_value is None or v2_value is None:
            continue
        if v1_value != v2_value:
            divergences.append(
                Divergence(field=field_name, kind="not_equal", cap=v1_value, value=v2_value)
            )

    return divergences

Compare v2_params against v1_requirements and return divergences.

Returns an empty list when v2_params narrows v1_requirements (the spec-conformant case). Returns a list of :class:Divergence records when divergent — one per diverging field. Call :meth:Divergence.to_dict() to project a record onto the wire shape used in advisory details.divergences.

def downgrade_pixel_tracker(pixel: PixelTrackerAsset, *, field_path: str | None = None) ‑> PixelTrackerDowngrade
Expand source code 
def downgrade_pixel_tracker(
    pixel: PixelTrackerAsset,
    *,
    field_path: str | None = None,
) -> PixelTrackerDowngrade:
    """Project a single :class:`PixelTrackerAsset` onto v1 wire shape.

    Lossy when the source pixel carries a viewability variant, the
    ``custom`` event, or ``method=js`` — those don't fit the single v1
    slot they collapse onto. The advisory carries the source
    ``event``, ``method``, and (when present) ``custom_event_name``
    under ``details`` so downstream consumers can reason about what
    was lost.

    Args:
        pixel: The v2 ``PixelTrackerAsset`` to downgrade.
        field_path: Optional JSONPath-lite pointer for the emitted
            advisory's ``field`` (e.g.,
            ``"creative_manifest.assets[2]"``).
    """
    event = _coerce_event(pixel.event)
    method = _coerce_method(pixel.method)
    url = str(pixel.url)
    js = method is PixelTrackerMethod.js
    custom_name = pixel.custom_event_name if hasattr(pixel, "custom_event_name") else None

    v1 = V1UrlTracker(asset_id=_downgrade_slot(event), url=url, js_method=js)

    # Determine whether this downgrade is lossy per the spec table.
    is_lossy_event = event in _VIEWABILITY_EVENTS or event is PixelTrackerEvent.custom
    is_lossy = is_lossy_event or js

    if not is_lossy:
        return PixelTrackerDowngrade(v1=v1, advisory=None)

    details: dict[str, Any] = {
        "source_event": event.value if event is not None else None,
        "source_method": method.value,
        "v1_asset_id": v1.asset_id,
    }
    if custom_name is not None:
        # ``custom_event_name`` is buyer-controlled and unbounded at the
        # Pydantic level — cap + scrub before echoing into multi-hop
        # ``errors[]`` per the half-1 ``_echo_identifier`` pattern.
        details["source_custom_event_name"] = _echo_identifier(custom_name)

    lost_axes: list[str] = []
    if is_lossy_event:
        lost_axes.append("event")
    if js:
        lost_axes.append("method_js_execution")
    details["lost"] = lost_axes

    advisory = make_sdk_advisory(
        code="PIXEL_TRACKER_LOSSY_DOWNGRADE",
        message=(
            f"Pixel tracker (event={event.value if event else 'impression'!r}, "
            f"method={method.value!r}) downgrades to v1 url-tracker slot "
            f"{v1.asset_id!r} with loss on {', '.join(lost_axes)!r}."
        ),
        field=field_path,
        details=details,
        suggestion=(
            "v1-only buyers will see the URL fire but cannot distinguish "
            "the original event variant or execute the JS body. Keep the "
            "v2 manifest in flight for 3.1+ buyers."
        ),
    )
    return PixelTrackerDowngrade(v1=v1, advisory=advisory)

Project a single :class:PixelTrackerAsset onto v1 wire shape.

Lossy when the source pixel carries a viewability variant, the custom event, or method=js — those don't fit the single v1 slot they collapse onto. The advisory carries the source event, method, and (when present) custom_event_name under details so downstream consumers can reason about what was lost.

Args
-----=
pixel
The v2 PixelTrackerAsset to downgrade.
field_path
Optional JSONPath-lite pointer for the emitted advisory's field (e.g., "creative_manifest.assets[2]").
def downgrade_pixel_trackers(pixels: list[PixelTrackerAsset], *, field_path_prefix: str | None = None) ‑> PixelTrackerBatchResult
Expand source code 
def downgrade_pixel_trackers(
    pixels: list[PixelTrackerAsset],
    *,
    field_path_prefix: str | None = None,
) -> PixelTrackerBatchResult:
    """Apply :func:`downgrade_pixel_tracker` across a list.

    Returns the projected v1 trackers + a deduplicated list of
    advisories. Advisories are deduplicated on
    ``(code, source_event, source_method, source_custom_event_name)``
    so a manifest with many viewability pixels surfaces ONE advisory
    per kind. Distinct custom events keep distinct advisories because
    losing their ``custom_event_name`` is exactly the information
    consumers need to act on.
    """
    out = PixelTrackerBatchResult()
    seen: set[tuple[str, str | None, str, str | None]] = set()
    for i, pt in enumerate(pixels):
        prefix = f"{field_path_prefix}[{i}]" if field_path_prefix else None
        result = downgrade_pixel_tracker(pt, field_path=prefix)
        out.items.append(result.v1)
        if result.advisory is not None:
            details = result.advisory.details or {}
            key = (
                result.advisory.code,
                details.get("source_event"),
                details.get("source_method", "img"),
                details.get("source_custom_event_name"),
            )
            if key not in seen:
                seen.add(key)
                out.advisories.append(result.advisory)
    return out

Apply :func:downgrade_pixel_tracker() across a list.

Returns the projected v1 trackers + a deduplicated list of advisories. Advisories are deduplicated on (code, source_event, source_method, source_custom_event_name) so a manifest with many viewability pixels surfaces ONE advisory per kind. Distinct custom events keep distinct advisories because losing their custom_event_name is exactly the information consumers need to act on.

def find_declaration_by_kind(format_kind: str | CanonicalFormatKind,
format_options: Iterable[ProductFormatDeclaration],
*,
format_option_id: str | None = None) ‑> Format | None
Expand source code 
def find_declaration_by_kind(
    format_kind: str | CanonicalFormatKind,
    format_options: Iterable[ProductFormatDeclaration],
    *,
    format_option_id: str | None = None,
) -> ProductFormatDeclaration | None:
    """Look up the declaration in ``format_options[]`` matching the kind.

    Disambiguates with ``format_option_id`` when the closed set carries
    multiple declarations sharing the same ``format_kind`` (the case
    where ``format_option_id`` is required by the canonical contract.

    Args:
        format_kind: The kind to match. Accepts string or enum.
        format_options: The product's ``format_options[]``.
        format_option_id: When provided, only declarations whose
            ``format_option_id`` equals this value are considered a match.
            When omitted, the first kind match wins; this is unambiguous
            only when every declaration of that kind shares the same
            ``format_option_id``.

    Returns:
        The matching declaration, or ``None`` when no declaration in the
        closed set satisfies the query.
    """
    wanted = _coerce_kind(format_kind)
    for d in format_options:
        if _coerce_kind(d.format_kind) != wanted:
            continue
        if format_option_id is not None and d.format_option_id != format_option_id:
            continue
        return d
    return None

Look up the declaration in adcp.canonical_formats.format_options[] matching the kind.

Disambiguates with format_option_id when the closed set carries multiple declarations sharing the same format_kind (the case where format_option_id is required by the canonical contract.

Args
-----=
format_kind
The kind to match. Accepts string or enum.
format_options
The product's adcp.canonical_formats.format_options[].
format_option_id
When provided, only declarations whose format_option_id equals this value are considered a match. When omitted, the first kind match wins; this is unambiguous only when every declaration of that kind shares the same format_option_id.

Returns -----= The matching declaration, or None when no declaration in the closed set satisfies the query.

def find_declaration_by_v1_format_id(format_id: FormatId, format_options: Iterable[ProductFormatDeclaration]) ‑> Format | None
Expand source code 
def find_declaration_by_v1_format_id(
    format_id: FormatId,
    format_options: Iterable[ProductFormatDeclaration],
) -> ProductFormatDeclaration | None:
    """Look up the declaration whose ``v1_format_ref[]`` includes ``format_id``.

    Seller-side helper for processing v1 ``create_media_buy`` requests
    against a product publishing v2 ``format_options[]``. A buyer
    targeting a v1 ``format_id`` lands here: the SDK walks the closed
    set looking for the declaration that asserted this v1 ref.

    Matches on both ``agent_url`` and ``id`` — a v1 format identity is
    the ``(agent_url, id)`` pair, not the id alone. Returns the first
    declaration whose ``v1_format_ref[]`` contains a structurally equal
    entry.

    Args:
        format_id: The v1 ``FormatId`` the buyer's manifest targets.
        format_options: The product's ``format_options[]`` closed set.

    Returns:
        The matching declaration, or ``None`` when no declaration in the
        closed set asserts this v1 ref. ``None`` means the request
        should be rejected with ``UNSUPPORTED_FEATURE`` — the v1
        ``format_id`` is not a recognised entry for this product. If a
        caller expected a legacy ID such as ``display_300x250`` to match a
        parameterized canonical ref, use ``formats_are_equivalent`` or
        ``format_is_supported`` from :mod:`adcp.canonical_formats` instead of
        this closed-set v1 reference lookup.
    """
    target_url = canonicalize_agent_url(format_id.agent_url)
    target_id = format_id.id
    for decl in format_options:
        refs = decl.legacy_format_refs
        for ref in refs:
            ref_url = canonicalize_agent_url(ref.agent_url)
            if ref_url == target_url and ref.id == target_id:
                return decl
    return None

Look up the declaration whose v1_format_ref[] includes format_id.

Seller-side helper for processing v1 create_media_buy requests against a product publishing v2 adcp.canonical_formats.format_options[]. A buyer targeting a v1 format_id lands here: the SDK walks the closed set looking for the declaration that asserted this v1 ref.

Matches on both agent_url and id — a v1 format identity is the (agent_url, id) pair, not the id alone. Returns the first declaration whose v1_format_ref[] contains a structurally equal entry.

Args
-----=
format_id
The v1 FormatId the buyer's manifest targets.
format_options
The product's adcp.canonical_formats.format_options[] closed set.

Returns -----= The matching declaration, or None when no declaration in the closed set asserts this v1 ref. None means the request should be rejected with UNSUPPORTED_FEATURE — the v1 format_id is not a recognised entry for this product. If a caller expected a legacy ID such as display_300x250 to match a parameterized canonical ref, use formats_are_equivalent() or format_is_supported() from :mod:adcp.canonical_formats instead of this closed-set v1 reference lookup.

def format_is_supported(requested: str | FormatId | Mapping[str, Any],
supported: str | FormatId | Mapping[str, Any],
*,
default_agent_url: str = 'https://creative.adcontextprotocol.org') ‑> bool
Expand source code 
def format_is_supported(
    requested: str | FormatId | Mapping[str, Any],
    supported: str | FormatId | Mapping[str, Any],
    *,
    default_agent_url: str = CANONICAL_CREATIVE_AGENT_URL,
) -> bool:
    """Return true when ``requested`` is acceptable for ``supported``.

    This is intentionally stricter than :func:`formats_are_equivalent`.
    A broad supported format such as ``display_image`` accepts a specific
    request such as ``display_image`` 300x250, but a fixed supported product
    format requires the request to provide and match every fixed parameter
    (``width``, ``height``, and ``duration_ms``).
    """
    req = upgrade_legacy_format_id(requested, default_agent_url=default_agent_url)
    sup = upgrade_legacy_format_id(supported, default_agent_url=default_agent_url)
    if not formats_are_equivalent(req, sup, default_agent_url=default_agent_url):
        return False

    for field in ("width", "height", "duration_ms"):
        supported_value = getattr(sup, field)
        if supported_value is None:
            continue
        if getattr(req, field) != supported_value:
            return False
    return True

Return true when requested is acceptable for supported.

This is intentionally stricter than :func:formats_are_equivalent(). A broad supported format such as display_image accepts a specific request such as display_image 300x250, but a fixed supported product format requires the request to provide and match every fixed parameter (width, height, and duration_ms).

def formats_are_equivalent(a: str | FormatId | Mapping[str, Any],
b: str | FormatId | Mapping[str, Any],
*,
default_agent_url: str = 'https://creative.adcontextprotocol.org') ‑> bool
Expand source code 
def formats_are_equivalent(
    a: str | FormatId | Mapping[str, Any],
    b: str | FormatId | Mapping[str, Any],
    *,
    default_agent_url: str = CANONICAL_CREATIVE_AGENT_URL,
) -> bool:
    """Return true when two format IDs identify the same canonical family.

    Both inputs are first passed through :func:`upgrade_legacy_format_id`.
    Declared parameters must not conflict, but an omitted parameter on either
    side is treated as unspecified rather than a mismatch. Use
    :func:`format_is_supported` for product/capability gating where a
    supported fixed size or duration requires the request to state that value.
    """
    left = upgrade_legacy_format_id(a, default_agent_url=default_agent_url)
    right = upgrade_legacy_format_id(b, default_agent_url=default_agent_url)
    if canonicalize_agent_url(left.agent_url) != canonicalize_agent_url(right.agent_url):
        return False
    if left.id != right.id:
        return False

    for field in ("width", "height", "duration_ms"):
        left_value = getattr(left, field)
        right_value = getattr(right, field)
        if left_value is not None and right_value is not None and left_value != right_value:
            return False
    return True

Return true when two format IDs identify the same canonical family.

Both inputs are first passed through :func:upgrade_legacy_format_id(). Declared parameters must not conflict, but an omitted parameter on either side is treated as unspecified rather than a mismatch. Use :func:format_is_supported() for product/capability gating where a supported fixed size or duration requires the request to state that value.

def glob_match(value: str, pattern: str) ‑> bool
Expand source code 
def glob_match(value: str, pattern: str) -> bool:
    """Glob-match ``value`` against a registry ``format_id_glob`` pattern.

    Per the registry schema: ``*`` matches any segment. Patterns are
    compared against the v1 ``format_id.id`` (NOT the ``{agent_url, id}``
    pair — the registry mantra is family identification, not full
    namespace resolution).

    Treats ``*`` as a permissive wildcard (any chars including ``_``).
    Other regex metacharacters are escaped — the pattern language is
    glob, not regex.

    .. note::
       **Experimental.** This helper is unused on the v2 → v1 path (the
       registry is consulted only on the v1 → v2 inbound path, which
       lands in #741's second PR). The signature may shift when that
       path consumes it; adopters SHOULD pin to the SDK release they
       integrate against.
    """
    if pattern == "*":
        return True
    regex = "^" + re.escape(pattern).replace(r"\*", ".*") + "$"
    return re.fullmatch(regex, value) is not None

Glob-match value against a registry format_id_glob pattern.

Per the registry schema: * matches any segment. Patterns are compared against the v1 format_id.id (NOT the {agent_url, id} pair — the registry mantra is family identification, not full namespace resolution).

Treats * as a permissive wildcard (any chars including _). Other regex metacharacters are escaped — the pattern language is glob, not regex.

Note

Experimental. This helper is unused on the v2 → v1 path (the registry is consulted only on the v1 → v2 inbound path, which lands in #741's second PR). The signature may shift when that path consumes it; adopters SHOULD pin to the SDK release they integrate against.

def group_declarations_by_product(declarations: list[ProductFormatDeclaration], mapping: dict[str, list[str]]) ‑> dict[str, list[Format]]
Expand source code 
def group_declarations_by_product(
    declarations: list[ProductFormatDeclaration],
    mapping: dict[str, list[str]],
) -> dict[str, list[ProductFormatDeclaration]]:
    """Group projected v2 declarations into products by ``v1_format_ref`` id.

    A buyer-side adopter porting a v1 catalog onto v2 frequently has a
    pre-existing mapping of which v1 format ids belong to which product
    (e.g., from the seller's published product catalog or an internal
    routing table). After running
    :func:`project_v1_catalog_to_v2` over the flat v1 format list, this
    helper buckets the resulting declarations into per-product
    ``format_options[]`` lists.

    Args:
        declarations: Output of :func:`project_v1_catalog_to_v2` —
            every entry MUST carry a non-empty ``v1_format_ref[]``.
            Declarations whose ``v1_format_ref[0].id`` doesn't appear
            in ``mapping`` are silently skipped; pass-through is the
            adopter's choice.
        mapping: ``{product_id: [v1_format_id, ...]}`` describing
            which v1 format ids belong to which product.

    Returns:
        ``{product_id: [ProductFormatDeclaration, ...]}`` ready to
        drop into ``Product.format_options[]`` for each product.
        Order within each product preserves the input declaration
        order. Products with no matching declarations are omitted.

    Example::

        catalog = project_v1_catalog_to_v2(v1_formats)
        per_product = group_declarations_by_product(
            catalog.declarations,
            mapping={
                "homepage_mrec": ["display_300x250_image"],
                "homepage_billboard": ["display_970x250_image"],
            },
        )
        product = Product(
            product_id="homepage_mrec",
            format_options=per_product["homepage_mrec"],
            ...
        )
    """
    # Build a reverse index ``v1_id -> product_id`` once so the per-
    # declaration lookup is O(1). Earlier-occurring product_ids win on
    # collision — adopters with overlapping mappings get deterministic
    # behaviour matching the dict iteration order they passed in.
    reverse: dict[str, str] = {}
    for product_id, v1_ids in mapping.items():
        for v1_id in v1_ids:
            reverse.setdefault(v1_id, product_id)

    out: dict[str, list[ProductFormatDeclaration]] = {}
    for declaration in declarations:
        refs = declaration.legacy_format_refs
        if not refs:
            continue
        # A declaration may carry multiple v1 refs (multi-size fan-out
        # of a single canonical onto several v1 sizes). The first ref
        # determines product membership; this matches the half-1
        # ``find_declaration_by_v1_format_id`` lookup semantics.
        matched_product = reverse.get(refs[0].id)
        if matched_product is None:
            continue
        out.setdefault(matched_product, []).append(declaration)
    return out

Group projected v2 declarations into products by v1_format_ref id.

A buyer-side adopter porting a v1 catalog onto v2 frequently has a pre-existing mapping of which v1 format ids belong to which product (e.g., from the seller's published product catalog or an internal routing table). After running :func:project_v1_catalog_to_v2() over the flat v1 format list, this helper buckets the resulting declarations into per-product adcp.canonical_formats.format_options[] lists.

Args
-----=
declarations
Output of :func:project_v1_catalog_to_v2() — every entry MUST carry a non-empty v1_format_ref[]. Declarations whose v1_format_ref[0].id doesn't appear in mapping are silently skipped; pass-through is the adopter's choice.
mapping
{product_id: [v1_format_id, ...]} describing which v1 format ids belong to which product.

Returns -----= {product_id: [ProductFormatDeclaration, ...]} ready to drop into Product.format_options[] for each product. Order within each product preserves the input declaration order. Products with no matching declarations are omitted. Example::

catalog = project_v1_catalog_to_v2(v1_formats)
per_product = group_declarations_by_product(
    catalog.declarations,
    mapping={
        "homepage_mrec": ["display_300x250_image"],
        "homepage_billboard": ["display_970x250_image"],
    },
)
product = Product(
    product_id="homepage_mrec",
    format_options=per_product["homepage_mrec"],
    ...
)
def legacy_format_converter_from_catalog_snapshots(snapshots: Iterable[Mapping[str, Any]]) ‑> Callable[[LegacyFormatConversionContext], Format | collections.abc.Mapping[str, typing.Any] | None]
Expand source code 
def legacy_format_converter_from_catalog_snapshots(
    snapshots: Iterable[Mapping[str, Any]],
) -> LegacyFormatConverter:
    """Compile exact owner+ID forward routes from catalog snapshots."""

    routes: dict[
        tuple[str, str, int | None, int | None, float | None],
        tuple[int, dict[str, Any] | None],
    ] = {}
    for snapshot in snapshots:
        priority = _snapshot_priority(snapshot)
        for raw in _snapshot_formats(snapshot):
            if raw.get("canonical_formats_only") is True:
                continue
            if raw.get("format_kind") in _CANONICAL_ONLY_FORMAT_KINDS:
                continue
            refs = raw.get("v1_format_ref")
            if not isinstance(refs, list):
                continue
            canonical = deepcopy(
                {key: value for key, value in raw.items() if key != "v1_format_ref"}
            )
            canonical.setdefault("publisher_domain", snapshot.get("publisher_domain"))
            for item in refs:
                try:
                    ref = LegacyFormatId.model_validate(item)
                except ValidationError:
                    continue
                owner = _catalog_owner(ref.agent_url)
                if owner is None or not _is_safe_public_https_owner(ref.agent_url):
                    continue
                key = (owner, ref.id, ref.width, ref.height, ref.duration_ms)
                existing = routes.get(key)
                if existing is None or priority < existing[0]:
                    routes[key] = (priority, canonical)
                elif priority == existing[0]:
                    routes[key] = (priority, None)

    def convert(context: LegacyFormatConversionContext) -> Mapping[str, Any] | None:
        ref = context.format_id
        owner = _catalog_owner(ref.agent_url)
        if owner is None:
            return None
        ranked = routes.get((owner, ref.id, ref.width, ref.height, ref.duration_ms))
        if ranked is not None and ranked[1] is None:
            raise LegacyCreativeProjectionError(
                "projection catalog contains ambiguous legacy aliases"
            )
        route = ranked[1] if ranked else None
        return deepcopy(route) if route else None

    # Catalog snapshots are pre-resolved, validated sources with protocol
    # precedence above the bundled AAO fallback. The private wrapper makes the
    # trusted channel structurally distinct from an ordinary adopter callback.
    return _SnapshotLegacyFormatConverter(convert)

Compile exact owner+ID forward routes from catalog snapshots.

def load_default_registry() ‑> adcp.types.generated_poc.registries.v1_canonical_mapping.V1V2CanonicalFormatMappingRegistry
Expand source code 
def load_default_registry() -> V1V2CanonicalFormatMappingRegistry:
    """Load and parse the AAO-published v1↔v2 mapping registry.

    Returns a fresh deep copy of the cached parsed registry — callers
    can safely mutate the returned instance without affecting other
    callers in the same process. The underlying parsed registry is
    cached per process (the registry is immutable for a given SDK
    build, keyed by ``ADCP_VERSION``).

    .. note::
       **Experimental.** The registry is consulted only on the v1 → v2
       inbound path (lands in #741 part 2). Adopters using this helper
       to inspect the bundled mappings SHOULD pin to the SDK release
       they integrate against; the return shape may sharpen when the
       inbound consumer lands.

    Raises:
        RegistryLoadError: when the bundle is missing, malformed JSON,
            or fails schema validation.
    """
    return _load_registry_uncopied().model_copy(deep=True)

Load and parse the AAO-published v1↔v2 mapping registry.

Returns a fresh deep copy of the cached parsed registry — callers can safely mutate the returned instance without affecting other callers in the same process. The underlying parsed registry is cached per process (the registry is immutable for a given SDK build, keyed by ADCP_VERSION).

Note

Experimental. The registry is consulted only on the v1 → v2 inbound path (lands in #741 part 2). Adopters using this helper to inspect the bundled mappings SHOULD pin to the SDK release they integrate against; the return shape may sharpen when the inbound consumer lands.

Raises
-----=
RegistryLoadError
when the bundle is missing, malformed JSON, or fails schema validation.
def load_rc3_catalog_index() ‑> CatalogIndex
Expand source code 
@lru_cache(maxsize=1)
def load_rc3_catalog_index() -> CatalogIndex:
    """Load the vendored RC3 AAO catalog used by the compatibility fallback."""

    index = build_catalog_index(load_v1_reference_catalog())
    return CatalogIndex(
        by_owner_and_id=index.by_owner_and_id,
        by_unique_id=index.by_unique_id,
        _allow_bare_id_fallback=True,
    )

Load the vendored RC3 AAO catalog used by the compatibility fallback.

def make_sdk_advisory(*,
code: str,
message: str,
field: str | None = None,
details: dict[str, Any] | None = None,
recovery: Recovery = correctable,
suggestion: str | None = None) ‑> adcp.types.generated_poc.core.error.Error
Expand source code 
def make_sdk_advisory(
    *,
    code: str,
    message: str,
    field: str | None = None,
    details: dict[str, Any] | None = None,
    recovery: Recovery = Recovery.correctable,
    suggestion: str | None = None,
) -> Error:
    """Build an SDK-source advisory entry for ``errors[]`` augmentation.

    Sets ``source=sdk`` and ``sdk_id=<package>@<version>`` per the
    multi-hop propagation contract in ``core/error.json``. Consumers
    receiving this entry MUST treat it as advisory — the response stays
    success on the v1 path; only the v2 projection is degraded.

    Args:
        code: AdCP error code (e.g., ``FORMAT_DECLARATION_V1_AMBIGUOUS``).
            Must be ≤64 chars per the wire schema.
        message: Human-readable description.
        field: JSONPath-lite pointer to the offending field
            (e.g., ``products[0].format_options[2]``).
        details: Code-specific structured payload.
        recovery: Recovery classification — defaults to ``correctable``
            because canonical-projection advisories tell the seller what
            to fix (add ``v1_format_ref``, file a registry PR, etc.).
        suggestion: Optional one-line fix hint surfaced to operators.
    """
    return Error(
        code=code,
        message=message,
        field=field,
        details=details,
        recovery=recovery,
        source=Source.sdk,
        sdk_id=_resolve_sdk_id(),
        suggestion=suggestion,
    )

Build an SDK-source advisory entry for errors[] augmentation.

Sets source=sdk and sdk_id=<package>@<version> per the multi-hop propagation contract in core/error.json. Consumers receiving this entry MUST treat it as advisory — the response stays success on the v1 path; only the v2 projection is degraded.

Args
-----=
code
AdCP error code (e.g., FORMAT_DECLARATION_V1_AMBIGUOUS). Must be ≤64 chars per the wire schema.
message
Human-readable description.
field
JSONPath-lite pointer to the offending field (e.g., products[0].adcp.canonical_formats.format_options[2]).
details
Code-specific structured payload.
recovery
Recovery classification — defaults to correctable because canonical-projection advisories tell the seller what to fix (add v1_format_ref, file a registry PR, etc.).
suggestion
Optional one-line fix hint surfaced to operators.
def migrated_format_option_id(format_id: LegacyFormatId | Mapping[str, Any]) ‑> str
Expand source code 
def migrated_format_option_id(format_id: LegacyFormatId | Mapping[str, Any]) -> str:
    """Return the normative stable option ID from the complete legacy tuple."""

    ref = (
        format_id
        if isinstance(format_id, LegacyFormatId)
        else LegacyFormatId.model_validate(format_id)
    )
    duration: int | float | None = ref.duration_ms
    if duration is not None and float(duration).is_integer():
        # JSON.stringify renders JavaScript's sole Number type without a
        # trailing .0. Pydantic stores duration_ms as float, so normalize the
        # representation before hashing to preserve byte-for-byte RC3 parity.
        duration = int(duration)
    identity = json.dumps(
        [
            str(ref.agent_url),
            ref.id,
            ref.width,
            ref.height,
            duration,
        ],
        ensure_ascii=False,
        separators=(",", ":"),
    )
    # This empty-key HMAC is the cross-SDK deterministic ID algorithm, not an
    # integrity check or trust boundary.
    digest = hmac.new(b"", identity.encode("utf-8"), hashlib.sha256).hexdigest()[:32]
    return f"migrated_{digest}"

Return the normative stable option ID from the complete legacy tuple.

def narrowing_advisory(declaration: ProductFormatDeclaration,
*,
v1_requirements: dict[str, Any],
v1_format_id: str,
field_path: str = 'format_options[]') ‑> adcp.types.generated_poc.core.error.Error | None
Expand source code 
def narrowing_advisory(
    declaration: ProductFormatDeclaration,
    *,
    v1_requirements: dict[str, Any],
    v1_format_id: str,
    field_path: str = "format_options[]",
) -> Error | None:
    """Build the ``FORMAT_DECLARATION_DIVERGENT`` advisory for a single pairing.

    Returns ``None`` when ``declaration.params`` narrows ``v1_requirements``
    (no divergence to report). Returns an :class:`Error` with
    ``details.divergences`` listing the failing fields when divergent.

    Args:
        declaration: The v2 ``ProductFormatDeclaration`` carrying
            ``v1_format_ref[]``.
        v1_requirements: The referenced v1 format's ``requirements``
            object (dict or Pydantic model).
        v1_format_id: The v1 format identifier (``id`` portion of the
            ``FormatId``) — surfaced in advisory details so adopters can
            locate the divergent pair when a declaration carries many
            refs.
        field_path: JSONPath-lite pointer for the advisory's ``field``.
    """
    divs = check_narrows(declaration.params, v1_requirements)
    if not divs:
        return None
    safe_id = _echo_identifier(v1_format_id)
    return make_sdk_advisory(
        code="FORMAT_DECLARATION_DIVERGENT",
        message=(
            f"v2 declaration (format_kind={declaration.format_kind.value!r}) "
            f"params do not narrow v1 format {safe_id!r} requirements: "
            f"{len(divs)} divergence(s)."
        ),
        field=field_path,
        details={
            "format_kind": declaration.format_kind.value,
            "v1_format_id": safe_id,
            "divergences": [d.to_dict() for d in divs],
        },
        suggestion=(
            "Reconcile the v2 params against the referenced v1 format's "
            "requirements: lower the v2 cap, expand the v1 allowed set, "
            "or drop the v1_format_ref entry if the formats genuinely "
            "differ in shape."
        ),
    )

Build the FORMAT_DECLARATION_DIVERGENT advisory for a single pairing.

Returns None when declaration.params narrows v1_requirements (no divergence to report). Returns an :class:Error with details.divergences listing the failing fields when divergent.

Args
-----=
declaration
The v2 ProductFormatDeclaration carrying v1_format_ref[].
v1_requirements
The referenced v1 format's requirements object (dict or Pydantic model).
v1_format_id
The v1 format identifier (id portion of the FormatId) — surfaced in advisory details so adopters can locate the divergent pair when a declaration carries many refs.
field_path
JSONPath-lite pointer for the advisory's field.
def normalize_legacy_creative_request(value: Mapping[str, Any],
*,
legacy_format_converter: LegacyFormatConverter | None = None,
projection_sources: list[Any] | None = None) ‑> dict[str, typing.Any]
Expand source code 
def normalize_legacy_creative_request(
    value: Mapping[str, Any],
    *,
    legacy_format_converter: LegacyFormatConverter | None = None,
    projection_sources: list[Any] | None = None,
) -> dict[str, Any]:
    """Upgrade legacy selectors before a primary server handler runs.

    This projector performs no network I/O. Unmappable selectors reject the
    request rather than silently broadening its creative scope.
    """

    def retain_routes(declarations: Sequence[Format], product_id: object) -> None:
        """Keep exact tuples beside, never inside, canonical handler input."""

        if projection_sources is None or not declarations:
            return
        projection_sources.append(
            {
                "product_id": product_id if isinstance(product_id, str) else None,
                "format_options": list(declarations),
            }
        )

    def visit(item: Any, field_path: str) -> Any:
        if isinstance(item, list):
            return [visit(child, f"{field_path}[{index}]") for index, child in enumerate(item)]
        if not isinstance(item, Mapping):
            return item

        # Context and extension bags are application-owned opaque values. They
        # are neither creative-dialect evidence nor protocol selectors, so the
        # projector must preserve them verbatim rather than interpreting a
        # coincidental ``format_id``/``format_ids`` key inside the bag.
        result = {
            key: (
                deepcopy(child)
                if key in {"context", "ext"}
                else visit(child, f"{field_path}.{key}")
            )
            for key, child in item.items()
        }
        fields = result.get("fields")
        if isinstance(fields, list):
            result["fields"] = list(
                dict.fromkeys(
                    "format_options" if field in {"format_id", "format_ids"} else field
                    for field in fields
                )
            )
        format_ids = result.pop("format_ids", None)
        if format_ids is not None:
            declarations: list[Format] = []
            for index, legacy_id in enumerate(format_ids):
                projected = project_legacy_format_id(
                    legacy_id,
                    product_id=str(result.get("product_id") or ""),
                    field=f"{field_path}.format_ids[{index}]",
                    legacy_format_converter=legacy_format_converter,
                )
                if projected.declaration is None:
                    reason = (
                        projected.diagnostic.resolution_failure
                        if projected.diagnostic is not None
                        else "no_match"
                    )
                    raise LegacyCreativeProjectionError(
                        f"{field_path}.format_ids[{index}] cannot be projected ({reason})"
                    )
                declarations.append(projected.declaration)
            retain_routes(declarations, result.get("product_id"))
            if "product_id" in result:
                result["format_option_refs"] = [
                    {
                        "scope": "product",
                        "format_option_id": declaration.format_option_id,
                    }
                    for declaration in declarations
                ]
            else:
                result["format_options"] = declarations

        legacy_id = result.pop("format_id", None)
        if legacy_id is not None:
            projected = project_legacy_format_id(
                legacy_id,
                product_id=str(result.get("product_id") or ""),
                field=f"{field_path}.format_id",
                legacy_format_converter=legacy_format_converter,
            )
            if projected.declaration is None:
                reason = (
                    projected.diagnostic.resolution_failure
                    if projected.diagnostic is not None
                    else "no_match"
                )
                raise LegacyCreativeProjectionError(
                    f"{field_path}.format_id cannot be projected ({reason})"
                )
            declaration = projected.declaration
            retain_routes([declaration], result.get("product_id"))
            result["format_kind"] = declaration.format_kind.value
            if declaration.format_option_id:
                result["format_option_ref"] = {
                    "scope": "product",
                    "format_option_id": declaration.format_option_id,
                }
        return result

    return cast(dict[str, Any], visit(value, "request"))

Upgrade legacy selectors before a primary server handler runs.

This projector performs no network I/O. Unmappable selectors reject the request rather than silently broadening its creative scope.

def parse_canonical_reference(reference: CanonicalReference | Mapping[str, Any] | str | Any) ‑> tuple[CanonicalReference | None, CanonicalReferenceResult | None]
Expand source code 
def parse_canonical_reference(
    reference: CanonicalReference | Mapping[str, Any] | str | Any,
) -> tuple[CanonicalReference | None, CanonicalReferenceResult | None]:
    """Parse supported reference inputs into a normalized dataclass."""

    if isinstance(reference, CanonicalReference):
        parsed = reference
    elif isinstance(reference, str):
        if _REFERENCE_MARKER not in reference:
            return None, _invalid_reference("reference must use uri@sha256:<digest> form")
        compact_uri, digest_hex = reference.rsplit(_REFERENCE_MARKER, 1)
        parsed = CanonicalReference(uri=compact_uri, digest=f"{_DIGEST_PREFIX}{digest_hex}")
    elif isinstance(reference, Mapping):
        mapped_uri = reference.get("uri")
        mapped_digest = reference.get("digest")
        if mapped_uri is None or mapped_digest is None:
            return None, _invalid_reference("reference mapping requires uri and digest")
        parsed = CanonicalReference(uri=str(mapped_uri), digest=str(mapped_digest))
    elif hasattr(reference, "uri") and hasattr(reference, "digest"):
        raw_uri = getattr(reference, "uri")
        raw_digest = getattr(reference, "digest")
        if raw_uri is None or raw_digest is None:
            return None, _invalid_reference("reference object requires uri and digest")
        parsed = CanonicalReference(uri=str(raw_uri), digest=str(raw_digest))
    else:
        return None, _invalid_reference("unsupported canonical reference type")

    if not _is_valid_digest(parsed.digest):
        return None, _invalid_reference("digest must be sha256:<64 lowercase hex characters>")
    if not parsed.uri:
        return None, _invalid_reference("reference URI is empty")
    return parsed, None

Parse supported reference inputs into a normalized dataclass.

def project_canonical_response_to_legacy(value: Any,
*,
resolver: CanonicalFormatLegacyResolver | None = None,
sources: Sequence[Any] = ()) ‑> Any
Expand source code 
def project_canonical_response_to_legacy(
    value: Any,
    *,
    resolver: CanonicalFormatLegacyResolver | None = None,
    sources: Sequence[Any] = (),
) -> Any:
    """Project a canonical server result to a captured legacy caller dialect.

    Private same-process routes are honored. Once serialization erased them,
    only the explicit durable resolver may authorize legacy delivery.
    """

    declaration_routes: dict[tuple[str, str | None, str], Format | None] = {}

    def declaration_fingerprint(declaration: Format) -> tuple[str, tuple[str, ...]]:
        canonical = json.dumps(
            declaration.model_dump(mode="json", exclude_none=True),
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        )
        legacy = tuple(
            json.dumps(
                ref.model_dump(mode="json", exclude_none=True),
                ensure_ascii=False,
                sort_keys=True,
                separators=(",", ":"),
            )
            for ref in declaration.legacy_format_refs
        )
        return canonical, legacy

    def register_declaration(
        scope: str,
        owner: str | None,
        option_id: str,
        declaration: Format,
    ) -> None:
        if scope == "publisher":
            owner = _normalized_publisher_domain(owner) or None
        key = (scope, owner, option_id)
        if key not in declaration_routes:
            declaration_routes[key] = declaration
            return
        existing = declaration_routes[key]
        if existing is None or declaration_fingerprint(existing) != declaration_fingerprint(
            declaration
        ):
            declaration_routes[key] = None

    def collect(item: Any, product_id: str | None = None) -> None:
        if isinstance(item, Format):
            if item.format_option_id:
                register_declaration("product", product_id, item.format_option_id, item)
                if item.publisher_domain:
                    register_declaration(
                        "publisher",
                        item.publisher_domain,
                        item.format_option_id,
                        item,
                    )
            return
        if hasattr(item.__class__, "model_fields"):
            current_product = getattr(item, "product_id", None) or product_id
            for name in (
                "format_options",
                "products",
                "placements",
                "packages",
                "affected_packages",
                "creatives",
                "media_buys",
                "variants",
                "manifest",
            ):
                if name in item.__class__.model_fields:
                    collect(getattr(item, name, None), current_product)
            return
        if isinstance(item, Mapping):
            current_product = item.get("product_id") or product_id
            for source in getattr(item, "_canonical_sources", ()):
                collect(source, current_product)
            for child in item.values():
                collect(child, current_product)
            return
        if isinstance(item, (list, tuple)):
            for child in item:
                collect(child, product_id)

    collect(value)
    for source in sources:
        collect(source)

    def declaration_for_ref(
        ref: Mapping[str, Any],
        raw: Mapping[str, Any],
        field_path: str,
        product_id: str | None,
    ) -> Format:
        option_id = ref.get("format_option_id")
        if not isinstance(option_id, str):
            raise CanonicalFormatLegacyResolutionError(
                f"{field_path} has an invalid format_option_id"
            )
        scope = ref.get("scope")
        scope = getattr(scope, "value", scope)
        if scope == "product":
            key = ("product", product_id, option_id)
        elif scope == "publisher":
            publisher_domain = ref.get("publisher_domain")
            if not isinstance(publisher_domain, str) or not publisher_domain:
                raise CanonicalFormatLegacyResolutionError(
                    f"{field_path} publisher reference has no publisher_domain"
                )
            key = (
                "publisher",
                _normalized_publisher_domain(publisher_domain),
                option_id,
            )
        else:
            raise CanonicalFormatLegacyResolutionError(
                f"{field_path} has unsupported format option scope {scope!r}"
            )
        if key in declaration_routes:
            declaration = declaration_routes[key]
            if declaration is None:
                raise CanonicalFormatLegacyResolutionError(
                    f"{field_path} has conflicting declarations for {scope} route {option_id!r}"
                )
            return declaration
        kind = raw.get("format_kind")
        if not isinstance(kind, str):
            raise CanonicalFormatLegacyResolutionError(
                f"{field_path} has no discovered declaration; configure a durable resolver"
            )
        params = raw.get("params")
        return Format(
            format_option_id=option_id,
            publisher_domain=(key[1] if scope == "publisher" else None),
            format_kind=kind,
            params=params if isinstance(params, dict) else {},
        )

    def visit(item: Any, field_path: str, product_id: str | None = None) -> Any:
        if isinstance(item, Format):
            return item
        if hasattr(item, "model_dump"):
            raw = item.model_dump(mode="json", exclude_none=True)
            # Keep the actual nested objects long enough to read private
            # same-process legacy routes; model_dump intentionally erases them.
            model_options = getattr(item, "format_options", None)
            if model_options is not None:
                raw["format_options"] = model_options
            for collection in (
                "products",
                "packages",
                "affected_packages",
                "creatives",
                "media_buys",
            ):
                if collection in item.__class__.model_fields:
                    model_items = getattr(item, collection, None)
                    if model_items is not None:
                        raw[collection] = model_items
        elif isinstance(item, Mapping):
            raw = dict(item)
        elif isinstance(item, (list, tuple)):
            return [
                visit(child, f"{field_path}[{index}]", product_id)
                for index, child in enumerate(item)
            ]
        else:
            return item

        raw_product_id = raw.get("product_id")
        current_product_id = raw_product_id if isinstance(raw_product_id, str) else product_id

        options = raw.get("format_options")
        if isinstance(options, list):
            legacy_ids: list[dict[str, Any]] = []
            for index, option in enumerate(options):
                parsed_declaration = (
                    option if isinstance(option, Format) else Format.model_validate(option)
                )
                option_id = parsed_declaration.format_option_id
                declaration = parsed_declaration
                if option_id is not None:
                    key = ("product", current_product_id, option_id)
                    if key in declaration_routes:
                        registered = declaration_routes[key]
                        if registered is None:
                            raise CanonicalFormatLegacyResolutionError(
                                f"{field_path}.format_options[{index}] has conflicting "
                                f"product declarations for {option_id!r}"
                            )
                        declaration = registered
                legacy_ids.extend(
                    ref.model_dump(mode="json")
                    for ref in resolve_legacy_format_refs(
                        declaration,
                        resolver=resolver,
                        product_id=current_product_id,
                        field=f"{field_path}.format_options[{index}]",
                    )
                )
            raw.pop("format_options", None)
            raw["format_ids"] = legacy_ids

        option_refs = raw.pop("format_option_refs", None)
        if isinstance(option_refs, list):
            legacy_ids = []
            for index, option_ref in enumerate(option_refs):
                if not isinstance(option_ref, Mapping):
                    raise CanonicalFormatLegacyResolutionError(
                        f"{field_path}.format_option_refs[{index}] is invalid"
                    )
                declaration = declaration_for_ref(option_ref, raw, field_path, current_product_id)
                legacy_ids.extend(
                    ref.model_dump(mode="json")
                    for ref in resolve_legacy_format_refs(
                        declaration,
                        resolver=resolver,
                        product_id=current_product_id,
                        field=f"{field_path}.format_option_refs[{index}]",
                    )
                )
            raw["format_ids"] = legacy_ids

        option_ref = raw.pop("format_option_ref", None)
        if isinstance(option_ref, Mapping):
            declaration = declaration_for_ref(option_ref, raw, field_path, current_product_id)
            creative_legacy_refs = resolve_legacy_format_refs(
                declaration,
                resolver=resolver,
                product_id=current_product_id,
                field=f"{field_path}.format_option_ref",
            )
            if len(creative_legacy_refs) != 1:
                raise CanonicalFormatLegacyResolutionError(
                    f"{field_path}.format_option_ref resolves to "
                    f"{len(creative_legacy_refs)} legacy "
                    "formats; a creative requires exactly one"
                )
            raw["format_id"] = creative_legacy_refs[0].model_dump(mode="json")
            raw.pop("format_kind", None)
            raw.pop("params", None)
        elif isinstance(raw.get("format_kind"), str) and (
            "creative_id" in raw or "package_id" in raw
        ):
            declaration = Format(
                format_kind=raw["format_kind"],
                params=raw.get("params") if isinstance(raw.get("params"), dict) else {},
            )
            inferred_refs = resolve_legacy_format_refs(
                declaration,
                resolver=resolver,
                product_id=current_product_id,
                field=f"{field_path}.format_kind",
            )
            raw.pop("format_kind", None)
            raw.pop("params", None)
            if "creative_id" in raw:
                if len(inferred_refs) != 1:
                    raise CanonicalFormatLegacyResolutionError(
                        f"{field_path}.format_kind resolves to {len(inferred_refs)} legacy "
                        "formats; a creative requires exactly one"
                    )
                raw["format_id"] = inferred_refs[0].model_dump(mode="json")
            else:
                raw["format_ids"] = [ref.model_dump(mode="json") for ref in inferred_refs]

        for key, child in list(raw.items()):
            raw[key] = visit(child, f"{field_path}.{key}", current_product_id)
        return raw

    return visit(value, "response")

Project a canonical server result to a captured legacy caller dialect.

Private same-process routes are honored. Once serialization erased them, only the explicit durable resolver may authorize legacy delivery.

def project_declaration_to_v1(declaration: ProductFormatDeclaration,
*,
field_path: str = 'format_options[]',
product_id: str | None = None) ‑> V2ToV1Projection
Expand source code 
def project_declaration_to_v1(
    declaration: ProductFormatDeclaration,
    *,
    field_path: str = "format_options[]",
    product_id: str | None = None,
) -> V2ToV1Projection:
    """Project a single declaration to v1, emitting advisories per the
    resolution order documented at module level.

    Args:
        declaration: The v2 ``ProductFormatDeclaration`` to project.
        field_path: JSONPath-lite pointer surfaced on emitted advisories
            (e.g., ``products[0].format_options[2]``). The default points
            at the seller-published declaration without product context;
            callers wrapping a ``Product`` should pass the indexed form.
        product_id: Optional product identifier — surfaced in advisory
            ``details.product_id`` for buyer-side correlation.

    Returns:
        :class:`V2ToV1Projection` with the projected refs and any
        advisories the resolution order emitted.
    """
    kind = declaration.format_kind
    refs = list(declaration.legacy_format_refs)

    # Step 1: seller has explicitly opted out of v1 projection.
    # ``ProductFormatDeclaration`` enforces this is mutually exclusive
    # with ``v1_format_ref[]``, so we can't reach step 2 from here.
    if declaration.canonical_formats_only:
        return V2ToV1Projection()

    # Step 2: seller-asserted v1 link — emit refs, check multi-size fan-out.
    if refs:
        advisories: list[Error] = []
        sizes_n = _params_sizes_count(declaration)
        if sizes_n > len(refs):
            details: dict[str, Any] = {
                "format_kind": kind.value,
                "v1_format_ref_count": len(refs),
                "sizes_count": sizes_n,
            }
            if product_id is not None:
                details["product_id"] = _echo_identifier(product_id)
            advisories.append(
                make_sdk_advisory(
                    code="FORMAT_DECLARATION_V1_LOSSY_MULTI_SIZE",
                    message=(
                        f"v1_format_ref[] has {len(refs)} entries but params.sizes[] "
                        f"declares {sizes_n} sizes — the partial v1 emission covers "
                        f"only the referenced sizes. Seller SHOULD author one "
                        f"v1_format_ref entry per size."
                    ),
                    field=field_path,
                    details=details,
                    suggestion=(
                        "Add per-size v1_format_ref[] entries (one per params.sizes "
                        "entry) to give v1-only buyers full size coverage."
                    ),
                )
            )
        return V2ToV1Projection(format_ids=refs, advisories=advisories)

    # Step 3: canonical is not v1-translatable — silent.
    if not V1_TRANSLATABLE.get(kind, True):
        return V2ToV1Projection()

    # Step 4: canonical IS v1-translatable but seller didn't author refs.
    details = {
        "format_kind": kind.value,
        "reason": "no_v1_format_ref",
    }
    if product_id is not None:
        details["product_id"] = _echo_identifier(product_id)
    return V2ToV1Projection(
        advisories=[
            make_sdk_advisory(
                code="FORMAT_DECLARATION_V1_AMBIGUOUS",
                message=(
                    f"Canonical '{kind.value}' is normally v1-translatable but the "
                    f"declaration carries no v1_format_ref[] — SDK cannot synthesize "
                    f"a v1 format_id without seller assertion."
                ),
                field=field_path,
                details=details,
                suggestion=(
                    "Add v1_format_ref[] pointing at the v1 named format(s) this "
                    "declaration projects to (e.g., AAO-hosted formats at "
                    "https://creative.adcontextprotocol.org or a platform-published "
                    "adagents.json formats[] entry)."
                ),
            )
        ]
    )

Project a single declaration to v1, emitting advisories per the resolution order documented at module level.

Args
-----=
declaration
The v2 ProductFormatDeclaration to project.
field_path
JSONPath-lite pointer surfaced on emitted advisories (e.g., products[0].adcp.canonical_formats.format_options[2]). The default points at the seller-published declaration without product context; callers wrapping a Product should pass the indexed form.
product_id
Optional product identifier — surfaced in advisory details.product_id for buyer-side correlation.

Returns -----= :class:V2ToV1Projection with the projected refs and any advisories the resolution order emitted.

def project_legacy_format_id(format_id: LegacyFormatId | Mapping[str, Any],
*,
product_id: str,
field: str,
legacy_format_converter: LegacyFormatConverter | None = None,
catalog: CatalogIndex | None = None) ‑> ProjectedFormat
Expand source code 
def project_legacy_format_id(
    format_id: LegacyFormatId | Mapping[str, Any],
    *,
    product_id: str,
    field: str,
    legacy_format_converter: LegacyFormatConverter | None = None,
    catalog: CatalogIndex | None = None,
) -> ProjectedFormat:
    """Project one legacy tuple using RC3 precedence and safety semantics."""

    try:
        ref = (
            format_id
            if isinstance(format_id, LegacyFormatId)
            else LegacyFormatId.model_validate(format_id)
        )
    except ValidationError:
        return ProjectedFormat(
            diagnostic=ProjectionDiagnostic(
                code="FORMAT_PROJECTION_FAILED",
                field=field,
                product_id=product_id,
                resolution_failure="invalid_format_id_parameters",
            )
        )
    if not _validate_inline_parameters(ref):
        return ProjectedFormat(
            diagnostic=ProjectionDiagnostic(
                code="FORMAT_PROJECTION_FAILED",
                field=field,
                product_id=product_id,
                resolution_failure="invalid_format_id_parameters",
            )
        )

    index = catalog or load_rc3_catalog_index()
    owner = _catalog_owner(ref.agent_url)
    if owner is None or not _is_safe_public_https_owner(ref.agent_url):
        return ProjectedFormat(
            diagnostic=ProjectionDiagnostic(
                code="FORMAT_PROJECTION_FAILED",
                field=field,
                product_id=product_id,
                resolution_failure="no_match",
            )
        )
    snapshot_converter = (
        legacy_format_converter
        if isinstance(legacy_format_converter, _SnapshotLegacyFormatConverter)
        else None
    )
    if snapshot_converter is not None:
        # Pre-resolved publisher/community snapshots are validated catalog
        # sources, not an arbitrary compatibility callback. Their declared
        # precedence is publisher -> approved mirror -> configured/agent, all
        # ahead of the SDK's bundled AAO fallback.
        converted = _converted_format(
            snapshot_converter,
            LegacyFormatConversionContext(ref, product_id, field),
        )
        if converted is not None:
            return converted
    exact_key = (owner, ref.id)
    exact = index.by_owner_and_id.get(exact_key)
    if exact_key in index.by_owner_and_id and exact is None:
        return ProjectedFormat(
            diagnostic=ProjectionDiagnostic(
                code="FORMAT_PROJECTION_FAILED",
                field=field,
                product_id=product_id,
                resolution_failure="catalog_collision",
            )
        )
    entry = exact

    if exact is None:
        unique = index.by_unique_id.get(ref.id) if index._allow_bare_id_fallback else None
        if legacy_format_converter is not None and snapshot_converter is None:
            converted = _converted_format(
                legacy_format_converter,
                LegacyFormatConversionContext(ref, product_id, field),
            )
            if converted is not None:
                return converted
        historical_size = _HISTORICAL_DISPLAY_SIZE.fullmatch(ref.id)
        if owner == AAO_CANONICAL_AGENT_URL and historical_size is not None:
            return ProjectedFormat(
                declaration=Format(
                    format_option_id=migrated_format_option_id(ref),
                    format_kind="image",
                    params={
                        "width": int(historical_size.group(1)),
                        "height": int(historical_size.group(2)),
                    },
                    v1_format_ref=[ref],
                )
            )
        if unique is None:
            return ProjectedFormat(
                diagnostic=ProjectionDiagnostic(
                    code="FORMAT_PROJECTION_FAILED",
                    field=field,
                    product_id=product_id,
                    resolution_failure="no_match",
                )
            )
        entry = unique

    canonical = entry.get("canonical") if isinstance(entry, Mapping) else None
    if not isinstance(canonical, Mapping) or not isinstance(canonical.get("kind"), str):
        if legacy_format_converter is not None and snapshot_converter is None:
            converted = _converted_format(
                legacy_format_converter,
                LegacyFormatConversionContext(ref, product_id, field),
            )
            if converted is not None:
                return converted
        return ProjectedFormat(
            diagnostic=ProjectionDiagnostic(
                code="FORMAT_PROJECTION_FAILED",
                field=field,
                product_id=product_id,
                resolution_failure="catalog_lacks_canonical_annotation",
            )
        )

    assert isinstance(entry, Mapping)
    try:
        params = _fixed_catalog_params(ref, entry)
    except _CatalogConflictError:
        return ProjectedFormat(
            diagnostic=ProjectionDiagnostic(
                code="FORMAT_PROJECTION_FAILED",
                field=field,
                product_id=product_id,
                format_kind=canonical["kind"],
                resolution_failure="catalog_requirement_conflict",
            )
        )
    if canonical.get("asset_source"):
        params["asset_source"] = canonical["asset_source"]
    if canonical.get("slots_override"):
        params["slots"] = canonical["slots_override"]
    try:
        declaration = Format(
            format_option_id=migrated_format_option_id(ref),
            format_kind=canonical["kind"],
            params=params,
            v1_format_ref=[ref],
        )
    except ValidationError:
        return ProjectedFormat(
            diagnostic=ProjectionDiagnostic(
                code="FORMAT_PROJECTION_FAILED",
                field=field,
                product_id=product_id,
                resolution_failure="catalog_requirement_conflict",
            )
        )
    return ProjectedFormat(declaration=declaration)

Project one legacy tuple using RC3 precedence and safety semantics.

def project_legacy_product(product: LegacyProduct | Mapping[str, Any],
*,
legacy_format_converter: LegacyFormatConverter | None = None,
catalog: CatalogIndex | None = None) ‑> CanonicalProductProjection
Expand source code 
def project_legacy_product(
    product: LegacyProduct | Mapping[str, Any],
    *,
    legacy_format_converter: LegacyFormatConverter | None = None,
    catalog: CatalogIndex | None = None,
) -> CanonicalProductProjection:
    """Project a raw product, omitting it when no format can map."""

    raw = product.model_dump(mode="json") if isinstance(product, LegacyProduct) else dict(product)
    product_id = str(raw.get("product_id", ""))
    declarations: list[Format] = []
    diagnostics: list[ProjectionDiagnostic] = []

    for option in raw.get("format_options") or []:
        try:
            declarations.append(Format.model_validate(option))
        except ValidationError:
            diagnostics.append(
                ProjectionDiagnostic(
                    code="FORMAT_PROJECTION_FAILED",
                    field=f"products[{product_id}].format_options",
                    product_id=product_id,
                    resolution_failure="invalid_canonical_declaration",
                )
            )

    had_legacy_format_ids = "format_ids" in raw
    legacy_format_ids = raw.get("format_ids")
    for index, ref in enumerate(legacy_format_ids or []):
        result = project_legacy_format_id(
            ref,
            product_id=product_id,
            field=f"products[{product_id}].format_ids[{index}]",
            legacy_format_converter=legacy_format_converter,
            catalog=catalog,
        )
        if result.declaration is not None:
            declarations.append(result.declaration)
        if result.diagnostic is not None:
            diagnostics.append(result.diagnostic)

    projected_placements: list[dict[str, Any]] = []
    for placement_index, placement_value in enumerate(raw.get("placements") or []):
        placement = (
            placement_value.model_dump(mode="json")
            if hasattr(placement_value, "model_dump")
            else dict(placement_value)
        )
        placement_options: list[Format] = []
        for option in placement.pop("format_options", None) or []:
            try:
                placement_options.append(Format.model_validate(option))
            except ValidationError:
                diagnostics.append(
                    ProjectionDiagnostic(
                        code="FORMAT_PROJECTION_FAILED",
                        field=(
                            f"products[{product_id}].placements[{placement_index}]"
                            ".format_options"
                        ),
                        product_id=product_id,
                        resolution_failure="invalid_canonical_declaration",
                    )
                )
        placement_ids = placement.pop("format_ids", None)
        if placement_ids == [] and not placement_options:
            diagnostics.append(
                ProjectionDiagnostic(
                    code="CANONICAL_PRODUCT_FORMATS_UNAVAILABLE",
                    field=f"products[{product_id}].placements[{placement_index}].format_options",
                    product_id=product_id,
                    reason="nested_placement_format_list_empty",
                )
            )
        for ref_index, ref in enumerate(placement_ids or []):
            result = project_legacy_format_id(
                ref,
                product_id=product_id,
                field=(
                    f"products[{product_id}].placements[{placement_index}]"
                    f".format_ids[{ref_index}]"
                ),
                legacy_format_converter=legacy_format_converter,
                catalog=catalog,
            )
            if result.declaration is not None:
                placement_options.append(result.declaration)
                declarations.append(result.declaration)
            if result.diagnostic is not None:
                diagnostics.append(result.diagnostic)
        if placement_options:
            placement["format_options"] = placement_options
        projected_placements.append(placement)

    raw.pop("format_ids", None)
    if projected_placements:
        raw["placements"] = projected_placements
    raw["format_options"] = declarations
    if not declarations:
        diagnostics.append(
            ProjectionDiagnostic(
                code="CANONICAL_PRODUCT_FORMATS_UNAVAILABLE",
                field=f"products[{product_id}].format_options",
                product_id=product_id,
                reason=(
                    "legacy_format_list_empty"
                    if had_legacy_format_ids and not legacy_format_ids
                    else (
                        "legacy_format_projection_failed"
                        if legacy_format_ids
                        else "missing_format_declaration"
                    )
                ),
            )
        )
        return CanonicalProductProjection(product=None, diagnostics=diagnostics)
    return CanonicalProductProjection(
        product=Product.model_validate(raw),
        diagnostics=diagnostics,
    )

Project a raw product, omitting it when no format can map.

def project_product_to_v1(product: Any, *, product_index: int | None = None) ‑> V2ToV1Projection
Expand source code 
def project_product_to_v1(
    product: Any,
    *,
    product_index: int | None = None,
) -> V2ToV1Projection:
    """Project every ``format_options[]`` entry on a ``Product`` to v1.

    Walks the product's declarations, applies
    :func:`project_declaration_to_v1` to each, and accumulates the
    aggregated refs + advisories. The product's existing v1 ``format_ids``
    field is preserved by the caller — this helper produces the *additive*
    set that the seller publishes alongside seller-declared v1 ids.

    Args:
        product: A ``Product`` instance carrying ``format_options[]``.
            Duck-typed so the helper works against the wire response,
            adopter-typed wrappers, or in-progress builders.
        product_index: Optional zero-based index of the product within the
            enclosing ``Products[]`` array. When provided, advisories
            carry the indexed field path (``products[N].format_options[K]``)
            so multi-product responses don't collapse to ambiguous pointers.

    Returns:
        :class:`V2ToV1Projection` with the union of per-declaration results.
    """
    declarations = getattr(product, "format_options", None) or []
    product_id = getattr(product, "product_id", None) or getattr(product, "id", None)

    out = V2ToV1Projection()
    for i, decl in enumerate(declarations):
        prefix = (
            f"products[{product_index}].format_options[{i}]"
            if product_index is not None
            else f"format_options[{i}]"
        )
        result = project_declaration_to_v1(
            decl,
            field_path=prefix,
            product_id=product_id,
        )
        out.format_ids.extend(result.format_ids)
        out.advisories.extend(result.advisories)
    return out

Project every adcp.canonical_formats.format_options[] entry on a Product to v1.

Walks the product's declarations, applies :func:project_declaration_to_v1() to each, and accumulates the aggregated refs + advisories. The product's existing v1 format_ids field is preserved by the caller — this helper produces the additive set that the seller publishes alongside seller-declared v1 ids.

Args
-----=
product
A Product instance carrying adcp.canonical_formats.format_options[]. Duck-typed so the helper works against the wire response, adopter-typed wrappers, or in-progress builders.
product_index
Optional zero-based index of the product within the enclosing Products[] array. When provided, advisories carry the indexed field path (products[N].adcp.canonical_formats.format_options[K]) so multi-product responses don't collapse to ambiguous pointers.

Returns -----= :class:V2ToV1Projection with the union of per-declaration results.

def project_v1_catalog_to_v2(v1_formats: list[Any], *, field_path_prefix: str = 'formats') ‑> V1CatalogProjection
Expand source code 
def project_v1_catalog_to_v2(
    v1_formats: list[Any],
    *,
    field_path_prefix: str = "formats",
) -> V1CatalogProjection:
    """Project a list of v1 named formats (a catalog) to v2 declarations.

    Aggregates per-format projection results — failed-closed entries
    contribute their advisory but no declaration. Useful for migrating
    an entire v1 ``reference-formats.json``-style catalog in one call.
    """
    out = V1CatalogProjection()
    for i, v1_format in enumerate(v1_formats):
        result = project_v1_format_to_declaration(
            v1_format,
            field_path=f"{field_path_prefix}[{i}]",
        )
        if result.declaration is not None:
            out.declarations.append(result.declaration)
        out.advisories.extend(result.advisories)
    return out

Project a list of v1 named formats (a catalog) to v2 declarations.

Aggregates per-format projection results — failed-closed entries contribute their advisory but no declaration. Useful for migrating an entire v1 reference-formats.json-style catalog in one call.

def project_v1_format_to_declaration(v1_format: Any, *, field_path: str = 'formats[]') ‑> V1ToV2Projection
Expand source code 
def project_v1_format_to_declaration(
    v1_format: Any,
    *,
    field_path: str = "formats[]",
) -> V1ToV2Projection:
    """Project a single v1 named format to a v2 ``ProductFormatDeclaration``.

    Walks the resolution order documented at module level. Tolerates
    both raw dicts (the common case when reading a v1 catalog from
    JSON) and Pydantic-validated v1 ``Format`` instances.

    Args:
        v1_format: The v1 format declaration to project. Dict or
            duck-typed object with ``format_id``, ``canonical``,
            ``assets`` accessors.
        field_path: JSONPath-lite pointer for emitted advisories
            (e.g., ``"formats[2]"``).

    Returns:
        :class:`V1ToV2Projection` carrying the declaration (when
        projection succeeded) and any advisories the resolution order
        emitted.
    """
    fid = _v1_format_id(v1_format)
    if fid is None:
        return V1ToV2Projection(
            advisories=[
                make_sdk_advisory(
                    code="FORMAT_PROJECTION_FAILED",
                    message="v1 format declaration carries no parseable format_id.",
                    field=field_path,
                    details={"resolution_failure": "missing_format_id"},
                )
            ]
        )

    registry = load_default_registry()

    # Look up the registry's matching glob mapping (if any) so partial
    # seller annotations can still pick up registry-published default
    # ``parameters`` without forcing the seller to restate them on the v1
    # file. Step 1 (seller annotation) overrides ``kind`` /
    # ``slots_override`` / ``asset_source``. Neutral registry parameters
    # such as dimensions still fill omitted values, but seller-authored
    # declarations must not inherit registry slot or pixel-density
    # contracts.
    registry_params: dict[str, Any] = {}
    registry_kind: CanonicalFormatKind | None = None
    for mapping in registry.mappings:
        pattern = mapping.v1_pattern
        glob = getattr(pattern, "format_id_glob", None)
        if isinstance(glob, str) and glob_match(fid.id, glob):
            registry_params = dict(mapping.v2.parameters or {})
            registry_kind = mapping.v2.canonical
            break

    # --- Step 1 (registry resolution-order step 2): seller-asserted
    # ``canonical`` annotation on the v1 file. Annotation wins on
    # ``kind`` + ``asset_source`` + ``slots_override``; neutral registry
    # parameters fill omitted values without importing registry slot or
    # pixel-density contracts.
    annotation = _v1_canonical_annotation(v1_format)
    if annotation is not None:
        seller_params = {
            key: value
            for key, value in registry_params.items()
            if key not in _SELLER_AUTHORITATIVE_PARAM_KEYS
        }
        return V1ToV2Projection(
            declaration=_build_declaration(
                kind=annotation.kind,
                v1_format_id=fid,
                params=seller_params,
                canonical_ref=annotation,
            )
        )

    # --- Step 2 (registry resolution-order step 3): registry glob hit
    # without a seller annotation — emit the registry's pairing.
    if registry_kind is not None:
        return V1ToV2Projection(
            declaration=_build_declaration(
                kind=registry_kind,
                v1_format_id=fid,
                params=registry_params,
            )
        )

    # No literal-glob hit — try structural fallback.
    asset_types = _v1_asset_types(v1_format)
    vast_versions = _v1_version_constraints(v1_format, keys=("vast_version", "vast_versions"))
    daast_versions = _v1_version_constraints(v1_format, keys=("daast_version", "daast_versions"))

    structural_hits: list[Any] = []
    for mapping in registry.mappings:
        pattern = mapping.v1_pattern
        # The discriminated union distinguishes structural (``V1Pattern1``) from
        # glob (``V1Pattern``); only the structural branch is consultable here.
        structural = getattr(pattern, "structural", None)
        if structural is None:
            continue
        if structural_match(
            asset_types=asset_types,
            vast_versions=vast_versions or None,
            daast_versions=daast_versions or None,
            pattern=structural,
        ):
            structural_hits.append(mapping)

    if structural_hits:
        # Step 4: family-level match — emit AMBIGUOUS advisory but still
        # produce a usable declaration with the matched canonical so
        # consumers have a typed shape to work against.
        first = structural_hits[0]
        declaration = _build_declaration(
            kind=first.v2.canonical,
            v1_format_id=fid,
            params=dict(first.v2.parameters or {}),
        )
        advisory = make_sdk_advisory(
            code="FORMAT_DECLARATION_V1_AMBIGUOUS",
            message=(
                f"v1 format {fid.id!r} structurally matched the "
                f"{first.v2.canonical.value!r} family but the registry "
                f"entry is pure-structural — the projection is a "
                f"family-level guess. Seller SHOULD add an explicit "
                f"``canonical`` annotation on the v1 format."
            ),
            field=field_path,
            details={
                "v1_format_id": _echo_identifier(fid.id),
                "matched_canonical": first.v2.canonical.value,
                "match_kind": "structural_family",
                "candidate_count": len(structural_hits),
            },
            suggestion=(
                "Add a ``canonical: { kind: ..., asset_source?: ..., "
                "slots_override?: [...] }`` annotation on the v1 format "
                "file so the projection is seller-declared rather than "
                "family-inferred."
            ),
        )
        return V1ToV2Projection(declaration=declaration, advisories=[advisory])

    # --- Step 5: fail closed ---
    return V1ToV2Projection(
        advisories=[
            make_sdk_advisory(
                code="FORMAT_PROJECTION_FAILED",
                message=(
                    f"v1 format {fid.id!r} has no ``canonical`` annotation and "
                    f"no registry match — SDK cannot project it onto a v2 "
                    f"canonical."
                ),
                field=field_path,
                details={
                    "v1_format_id": _echo_identifier(fid.id),
                    "resolution_failure": "no_registry_match",
                    "asset_types": asset_types,
                },
                suggestion=(
                    "Add a ``canonical`` annotation to the v1 format file, "
                    "or file a registry PR adding a structural pattern "
                    "covering this format's shape."
                ),
            )
        ]
    )

Project a single v1 named format to a v2 ProductFormatDeclaration.

Walks the resolution order documented at module level. Tolerates both raw dicts (the common case when reading a v1 catalog from JSON) and Pydantic-validated v1 Format instances.

Args
-----=
v1_format
The v1 format declaration to project. Dict or duck-typed object with format_id, canonical, assets accessors.
field_path
JSONPath-lite pointer for emitted advisories (e.g., "formats[2]").

Returns -----= :class:V1ToV2Projection carrying the declaration (when projection succeeded) and any advisories the resolution order emitted.

def projection_adapters_from_catalog_snapshots(snapshots: Iterable[Mapping[str, Any]]) ‑> ProjectionCatalogAdapters
Expand source code 
def projection_adapters_from_catalog_snapshots(
    snapshots: Iterable[Mapping[str, Any]],
) -> ProjectionCatalogAdapters:
    """Build symmetric forward and durable reverse routes from one corpus."""

    materialized = list(snapshots)
    _assert_bidirectional_projection_catalogs(materialized)
    return ProjectionCatalogAdapters(
        legacy_format_converter=legacy_format_converter_from_catalog_snapshots(materialized),
        canonical_format_legacy_resolver=canonical_format_legacy_resolver_from_catalog_snapshots(
            materialized
        ),
    )

Build symmetric forward and durable reverse routes from one corpus.

def resolve_creative_dialect(adcp_version: str,
*,
capabilities: Any = None,
request: Any = None,
legacy_projection_available: bool = False) ‑> CreativeDialect
Expand source code 
def resolve_creative_dialect(
    adcp_version: str,
    *,
    capabilities: Any = None,
    request: Any = None,
    legacy_projection_available: bool = False,
) -> CreativeDialect:
    """Apply the normative 3.0/3.1/3.2 canonical-creatives matrix.

    AdCP 3.1 is deliberately evidence-driven. Contradictory or absent evidence
    fails closed, except when the caller has already established an unambiguous
    legacy projection route.
    """

    normalized = normalize_to_release_precision(adcp_version)
    release = normalized.split("-", 1)[0]
    major, minor = (int(part) for part in release.split(".", 1))
    if major != 3:
        raise CreativeDialectError(
            f"canonical creative negotiation only supports AdCP 3.x, got {adcp_version!r}"
        )

    capability = canonical_creatives_capability(capabilities)
    if minor == 0:
        return CreativeDialect.LEGACY
    if minor >= 2:
        if capability is False:
            raise CreativeDialectError(
                "AdCP 3.2+ requires canonical creatives, but the seller advertised "
                "canonical_creatives=false"
            )
        return CreativeDialect.CANONICAL

    if capability is True:
        return CreativeDialect.CANONICAL
    if capability is False:
        return CreativeDialect.LEGACY

    canonical, legacy = _schema_evidence(request)
    # During the 3.1 transition a sender may dual-emit legacy format IDs next
    # to canonical option references. Canonical evidence is authoritative;
    # the legacy fields are compatibility data rather than a contradiction.
    if canonical:
        return CreativeDialect.CANONICAL
    if legacy and not canonical:
        return CreativeDialect.LEGACY
    if legacy_projection_available:
        return CreativeDialect.LEGACY
    raise CreativeDialectError(
        "AdCP 3.1 does not establish a creative dialect: advertise "
        "media_buy.features.canonical_creatives or provide unambiguous "
        "request-local schema evidence"
    )

Apply the normative 3.0/3.1/3.2 canonical-creatives matrix.

AdCP 3.1 is deliberately evidence-driven. Contradictory or absent evidence fails closed, except when the caller has already established an unambiguous legacy projection route.

def resolve_legacy_format_refs(declaration: Format,
*,
resolver: CanonicalFormatLegacyResolver | None = None,
product_id: str | None = None,
field: str = 'format_options[]') ‑> list[LegacyFormatId]
Expand source code 
def resolve_legacy_format_refs(
    declaration: Format,
    *,
    resolver: CanonicalFormatLegacyResolver | None = None,
    product_id: str | None = None,
    field: str = "format_options[]",
) -> list[LegacyFormatId]:
    """Resolve a canonical declaration without ever reverse-guessing.

    Same-process projections retain the original tuple in private model state.
    JSON/process boundaries deliberately erase that state; callers must then
    provide a durable resolver backed by adopter storage or catalog snapshots.
    """

    if declaration.legacy_format_refs:
        return list(declaration.legacy_format_refs)
    if resolver is None:
        raise CanonicalFormatLegacyResolutionError(
            "canonical creative crossed a process boundary without a durable "
            "canonical-to-legacy resolver; rediscover the product or configure a resolver"
        )
    resolved = resolver(
        CanonicalFormatLegacyResolutionContext(
            declaration=declaration,
            product_id=product_id,
            field=field,
        )
    )
    if not resolved:
        raise CanonicalFormatLegacyResolutionError(
            f"no durable legacy route for {field}; refusing to reverse-guess"
        )
    return [
        item if isinstance(item, LegacyFormatId) else LegacyFormatId.model_validate(item)
        for item in resolved
    ]

Resolve a canonical declaration without ever reverse-guessing.

Same-process projections retain the original tuple in private model state. JSON/process boundaries deliberately erase that state; callers must then provide a durable resolver backed by adopter storage or catalog snapshots.

def structural_match(*,
asset_types: list[str],
vast_versions: list[str] | None = None,
daast_versions: list[str] | None = None,
width: int | None = None,
height: int | None = None,
pattern: Any) ‑> bool
Expand source code 
def structural_match(
    *,
    asset_types: list[str],
    vast_versions: list[str] | None = None,
    daast_versions: list[str] | None = None,
    width: int | None = None,
    height: int | None = None,
    pattern: Any,
) -> bool:
    """Check whether a v1 format's structural shape matches a registry entry.

    ``pattern`` is the registry entry's ``structural`` block (a
    :class:`adcp.types.V1CanonicalStructural` or equivalent dict). All
    constraints declared on the pattern MUST match; constraints absent
    from the pattern do not narrow the match.

    Args:
        asset_types: Asset types appearing in the v1 format's slots.
            The pattern's ``asset_types`` is a *subset* requirement —
            every type the pattern lists must be present in ``asset_types``.
        vast_versions: VAST version(s) declared on the v1 format (typically
            a single value like ``"4.2"``). Each must satisfy at least one
            constraint in the pattern's ``vast_versions`` list.
        daast_versions: DAAST version(s); same matching semantics as VAST.
        width: Slot dimension width (pixels), if applicable.
        height: Slot dimension height (pixels), if applicable.
        pattern: The registry entry's ``structural`` block.

    Returns:
        ``True`` iff every constraint declared on ``pattern`` is satisfied
        by the v1 format's structural shape.

    .. note::
       **Experimental.** Same caveat as :func:`glob_match` — the v1 → v2
       inbound consumer lands in the second half of #741.
    """
    if hasattr(pattern, "model_dump"):
        p = pattern.model_dump(exclude_none=True)
    else:
        p = dict(pattern) if pattern else {}

    want_types = p.get("asset_types")
    if want_types:
        for t in want_types:
            if t not in asset_types:
                return False

    want_vast = p.get("vast_versions")
    if want_vast:
        if not vast_versions:
            return False
        if not any(_versions_overlap(v, want_vast) for v in vast_versions):
            return False

    want_daast = p.get("daast_versions")
    if want_daast:
        if not daast_versions:
            return False
        if not any(_versions_overlap(v, want_daast) for v in daast_versions):
            return False

    want_dims = p.get("dimensions") or {}
    if want_dims.get("width") is not None and want_dims["width"] != width:
        return False
    if want_dims.get("height") is not None and want_dims["height"] != height:
        return False

    return True

Check whether a v1 format's structural shape matches a registry entry.

pattern is the registry entry's structural block (a :class:adcp.types.V1CanonicalStructural or equivalent dict). All constraints declared on the pattern MUST match; constraints absent from the pattern do not narrow the match.

Args
-----=
asset_types
Asset types appearing in the v1 format's slots. The pattern's asset_types is a subset requirement — every type the pattern lists must be present in asset_types.
vast_versions
VAST version(s) declared on the v1 format (typically a single value like "4.2"). Each must satisfy at least one constraint in the pattern's vast_versions list.
daast_versions
DAAST version(s); same matching semantics as VAST.
width
Slot dimension width (pixels), if applicable.
height
Slot dimension height (pixels), if applicable.
pattern
The registry entry's structural block.

Returns -----= True iff every constraint declared on pattern is satisfied by the v1 format's structural shape.

Note

Experimental. Same caveat as :func:glob_match() — the v1 → v2 inbound consumer lands in the second half of #741.

def upgrade_legacy_format_id(value: str | FormatId | Mapping[str, Any],
*,
default_agent_url: str = 'https://creative.adcontextprotocol.org') ‑> LegacyFormatId
Expand source code 
def upgrade_legacy_format_id(
    value: str | FormatId | Mapping[str, Any],
    *,
    default_agent_url: str = CANONICAL_CREATIVE_AGENT_URL,
) -> FormatId:
    """Return ``value`` as a canonical, parameterized ``FormatId`` when known.

    The current canonical upgrade maps legacy display size IDs such as
    ``display_300x250`` and ``display_300x250_image`` to
    ``display_image`` with ``width=300`` and ``height=250``. Unknown IDs are
    still returned as structured ``FormatId`` values so callers can compare
    them consistently.
    """
    is_bare_legacy_id = isinstance(value, str)
    fid = _coerce_format_id(value, default_agent_url=default_agent_url)
    match = _DISPLAY_SIZE_RE.fullmatch(fid.id)
    if match is None:
        return fid
    default_fid = _coerce_format_id("__default__", default_agent_url=default_agent_url)
    if not is_bare_legacy_id and canonicalize_agent_url(fid.agent_url) != canonicalize_agent_url(
        default_fid.agent_url
    ):
        return fid

    return FormatId.model_validate(
        {
            "agent_url": str(fid.agent_url),
            "id": "display_image",
            "width": int(match.group("width")),
            "height": int(match.group("height")),
            "duration_ms": fid.duration_ms,
        }
    )

Return value as a canonical, parameterized FormatId when known.

The current canonical upgrade maps legacy display size IDs such as display_300x250 and display_300x250_image to display_image with width=300 and height=250. Unknown IDs are still returned as structured FormatId values so callers can compare them consistently.

def upgrade_v1_tracker(*, asset_id: str, url: str, field_path: str | None = None) ‑> PixelTrackerUpgrade
Expand source code 
def upgrade_v1_tracker(
    *,
    asset_id: str,
    url: str,
    field_path: str | None = None,
) -> PixelTrackerUpgrade:
    """Project a v1 ``{asset_type: url, url_type: tracker_pixel}`` to v2.

    ALWAYS emits ``PIXEL_TRACKER_UPGRADE_INFERRED`` for accepted entries
    — the v1 wire shape carries no explicit event/method, so the
    inferred values are an SDK convention, not a wire fact. Consumers
    reading the advisory can decide whether to trust the convention or
    treat the pixel as opaque.

    Rejects URLs whose scheme is outside the SDK's allowlist
    (currently ``http``/``https``). Disallowed schemes get a
    ``pixel_tracker=None`` result + an advisory carrying the rejected
    scheme; callers MUST drop the source entry rather than substitute
    a value.

    Args:
        asset_id: v1 ``asset_id`` of the tracker slot (e.g.,
            ``"impression_tracker"``). Drives the inference.
        url: The tracker URL.
        field_path: Optional JSONPath-lite pointer for the emitted
            advisory's ``field``.
    """
    # --- URL scheme gate ---
    scheme = _url_scheme(url)
    if scheme not in _PIXEL_TRACKER_URL_ALLOWED_SCHEMES:
        return PixelTrackerUpgrade(
            pixel_tracker=None,
            advisory=make_sdk_advisory(
                code="PIXEL_TRACKER_UPGRADE_INFERRED",
                message=(
                    f"v1 url-tracker asset_id={_echo_identifier(asset_id)!r} "
                    f"REJECTED: scheme {scheme!r} not in allowed set "
                    f"{sorted(_PIXEL_TRACKER_URL_ALLOWED_SCHEMES)!r}."
                ),
                field=field_path,
                details={
                    "source_asset_id": _echo_identifier(asset_id),
                    "rejected_scheme": scheme,
                    "inference_basis": "rejected_disallowed_scheme",
                },
                suggestion=(
                    "Renderer-fired tracker URLs MUST use ``http`` or "
                    "``https``. Source v1 catalogs carrying javascript:, "
                    "file:, or data: schemes in tracker slots are operator-"
                    "poisoning vectors; drop them or replace with an "
                    "https tracker endpoint."
                ),
            ),
        )

    inferred = _UPGRADE_TABLE.get(asset_id)
    if inferred is None:
        # Fallback: preserve the original asset_id as the custom event
        # name so a downstream consumer who knows the seller's
        # convention can still bucket events correctly.
        event, method = PixelTrackerEvent.custom, PixelTrackerMethod.img
        custom_name: str | None = asset_id
        basis = "fallback_custom_event"
    else:
        event, method = inferred
        custom_name = None
        basis = "asset_id_convention"

    if custom_name is not None:
        pixel = PixelTrackerAsset(
            asset_type="pixel_tracker",
            event=event,
            method=method,
            url=url,
            custom_event_name=custom_name,
        )
    else:
        pixel = PixelTrackerAsset(
            asset_type="pixel_tracker",
            event=event,
            method=method,
            url=url,
        )

    # ``asset_id`` and ``custom_event_name`` are seller-controlled and
    # unbounded at the v1 wire level — cap + scrub before echoing into
    # multi-hop ``errors[]`` per the half-1 ``_echo_identifier`` pattern.
    details: dict[str, Any] = {
        "source_asset_id": _echo_identifier(asset_id),
        "inferred_event": event.value,
        "inferred_method": method.value,
        "inference_basis": basis,
    }
    if custom_name is not None:
        details["inferred_custom_event_name"] = _echo_identifier(custom_name)

    advisory = make_sdk_advisory(
        code="PIXEL_TRACKER_UPGRADE_INFERRED",
        message=(
            f"v1 url-tracker asset_id={_echo_identifier(asset_id)!r} "
            f"upgraded to v2 pixel_tracker(event={event.value!r}, "
            f"method={method.value!r}) by {basis}."
        ),
        field=field_path,
        details=details,
        suggestion=(
            "Sellers SHOULD migrate v1 catalogs to v2 pixel_tracker so event "
            "/ method are declared on the wire rather than inferred from "
            "asset_id naming convention."
        ),
    )
    return PixelTrackerUpgrade(pixel_tracker=pixel, advisory=advisory)

Project a v1 {asset_type: url, url_type: tracker_pixel} to v2.

ALWAYS emits PIXEL_TRACKER_UPGRADE_INFERRED for accepted entries — the v1 wire shape carries no explicit event/method, so the inferred values are an SDK convention, not a wire fact. Consumers reading the advisory can decide whether to trust the convention or treat the pixel as opaque.

Rejects URLs whose scheme is outside the SDK's allowlist (currently http/https). Disallowed schemes get a pixel_tracker=None result + an advisory carrying the rejected scheme; callers MUST drop the source entry rather than substitute a value.

Args
-----=
asset_id
v1 asset_id of the tracker slot (e.g., "impression_tracker"). Drives the inference.
url
The tracker URL.
field_path
Optional JSONPath-lite pointer for the emitted advisory's field.
def upgrade_v1_trackers(v1_trackers: list[dict[str, Any]],
*,
field_path_prefix: str | None = None,
quiet_inference: bool = False) ‑> PixelTrackerBatchResult
Expand source code 
def upgrade_v1_trackers(
    v1_trackers: list[dict[str, Any]],
    *,
    field_path_prefix: str | None = None,
    quiet_inference: bool = False,
) -> PixelTrackerBatchResult:
    """Apply :func:`upgrade_v1_tracker` across a list of v1 url-tracker dicts.

    Each input MUST be a dict with ``asset_id`` + ``url`` keys (the v1
    wire shape). Advisories are deduplicated on ``(code, asset_id)``
    so many trackers under the same slot surface ONE advisory.

    The ``PIXEL_TRACKER_UPGRADE_INFERRED`` advisory fires on every
    accepted entry by default. Pass ``quiet_inference=True`` to
    suppress the advisory when the inference is unambiguous
    (``impression_tracker``, ``click_tracker``, ``viewability_tracker``
    via convention) — useful for high-volume buyer-side adopters
    reading v1 manifests at scale. The scheme-rejection advisory
    fires regardless of this flag (it carries a security signal).
    Entries the scheme gate rejects are NOT added to ``items`` —
    callers MUST drop them from the upgraded manifest.
    """
    out = PixelTrackerBatchResult()
    # Dedup includes ``inference_basis`` so a rejected upgrade and an
    # accepted upgrade on the same ``asset_id`` don't collapse — they
    # carry distinct semantics for the consumer.
    seen: set[tuple[str, str, str | None]] = set()
    for i, v1 in enumerate(v1_trackers):
        prefix = f"{field_path_prefix}[{i}]" if field_path_prefix else None
        asset_id = v1.get("asset_id")
        url = v1.get("url")
        if not isinstance(asset_id, str) or not isinstance(url, str):
            continue
        result = upgrade_v1_tracker(asset_id=asset_id, url=url, field_path=prefix)
        details = result.advisory.details or {}
        basis = details.get("inference_basis")
        is_rejection = basis == "rejected_disallowed_scheme"
        is_quietable = quiet_inference and not is_rejection and basis == "asset_id_convention"
        if result.pixel_tracker is not None:
            out.items.append(result.pixel_tracker)
        if is_quietable:
            continue
        key = (result.advisory.code, asset_id, basis)
        if key not in seen:
            seen.add(key)
            out.advisories.append(result.advisory)
    return out

Apply :func:upgrade_v1_tracker() across a list of v1 url-tracker dicts.

Each input MUST be a dict with asset_id + url keys (the v1 wire shape). Advisories are deduplicated on (code, asset_id) so many trackers under the same slot surface ONE advisory.

The PIXEL_TRACKER_UPGRADE_INFERRED advisory fires on every accepted entry by default. Pass quiet_inference=True to suppress the advisory when the inference is unambiguous (impression_tracker, click_tracker, viewability_tracker via convention) — useful for high-volume buyer-side adopters reading v1 manifests at scale. The scheme-rejection advisory fires regardless of this flag (it carries a security signal). Entries the scheme gate rejects are NOT added to items — callers MUST drop them from the upgraded manifest.

def validate_format_kind_in_options(format_kind: str | CanonicalFormatKind,
format_options: Iterable[ProductFormatDeclaration]) ‑> None
Expand source code 
def validate_format_kind_in_options(
    format_kind: str | CanonicalFormatKind,
    format_options: Iterable[ProductFormatDeclaration],
) -> None:
    """Raise if ``format_kind`` isn't published in ``format_options[]``.

    Args:
        format_kind: The kind a buyer's manifest targets. Accepts both
            the wire-string form (``"image"``) and the typed enum form
            (``CanonicalFormatKind.image``).
        format_options: The product's closed set of accepted format
            declarations.

    Raises:
        FormatKindNotInClosedSetError: when no declaration in the closed
            set carries that ``format_kind``. The seller MUST surface
            ``UNSUPPORTED_FEATURE`` on the response.
    """
    wanted = _coerce_kind(format_kind)
    accepted = [_coerce_kind(d.format_kind) for d in format_options]
    if wanted not in accepted:
        raise FormatKindNotInClosedSetError(wanted, accepted)

Raise if format_kind isn't published in adcp.canonical_formats.format_options[].

Args
-----=
format_kind
The kind a buyer's manifest targets. Accepts both the wire-string form ("image") and the typed enum form (CanonicalFormatKind.image).
format_options
The product's closed set of accepted format declarations.
Raises
-----=
FormatKindNotInClosedSetError
when no declaration in the closed set carries that format_kind. The seller MUST surface UNSUPPORTED_FEATURE on the response.

Classes

class CanonicalFormatLegacyResolutionContext (declaration: Format,
product_id: str | None = None,
field: str = 'format_options[]')
Expand source code 
@dataclass(frozen=True)
class CanonicalFormatLegacyResolutionContext:
    declaration: Format
    product_id: str | None = None
    field: str = "format_options[]"

CanonicalFormatLegacyResolutionContext(declaration: 'Format', product_id: 'str | None' = None, field: 'str' = 'format_options[]')

Instance variables

var declaration : Format
var field : str
var product_id : str | None
class CanonicalFormatLegacyResolutionError (*args, **kwargs)
Expand source code 
class CanonicalFormatLegacyResolutionError(ValueError):
    """Raised when persisted canonical state has no explicit reverse route."""

Raised when persisted canonical state has no explicit reverse route.

Ancestors

  • builtins.ValueError
  • builtins.Exception
  • builtins.BaseException
class CanonicalProductProjection (product: Product | None,
diagnostics: list[ProjectionDiagnostic] = <factory>)
Expand source code 
@dataclass
class CanonicalProductProjection:
    product: Product | None
    diagnostics: list[ProjectionDiagnostic] = field(default_factory=list)

CanonicalProductProjection(product: 'Product | None', diagnostics: 'list[ProjectionDiagnostic]' = )

Instance variables

var diagnostics : list[ProjectionDiagnostic]
var product : Product | None
class CanonicalReference (uri: str, digest: str)
Expand source code 
@dataclass(frozen=True)
class CanonicalReference:
    """Normalized immutable reference.

    Callers may pass either this dataclass, a mapping with ``uri`` and
    ``digest`` keys, or a compact string in ``uri@sha256:<digest>`` form.
    """

    uri: str
    digest: str

    @property
    def cache_key(self) -> str:
        return f"{self.uri}@{self.digest}"

Normalized immutable reference.

Callers may pass either this dataclass, a mapping with uri and digest keys, or a compact string in uri@sha256:<digest> form.

Instance variables

prop cache_key : str
Expand source code 
@property
def cache_key(self) -> str:
    return f"{self.uri}@{self.digest}"
var digest : str
var uri : str
class CanonicalReferenceResolver (*,
timeout: float = 5.0,
max_body_bytes: int = 1048576,
max_schema_refs: int = 256,
max_schema_ids: int = 256,
max_ref_depth: int = 8,
max_schema_keywords: int = 10000,
max_schema_depth: int = 128,
transport_factory: TransportFactory | None = None)
Expand source code 
class CanonicalReferenceResolver:
    """Resolve immutable ``format_schema`` and ``platform_extensions`` refs.

    The resolver owns an in-memory cache keyed by ``uri@digest``. Construct one
    per adopter component or request context; no process-global mutable
    configuration is used.
    """

    def __init__(
        self,
        *,
        timeout: float = DEFAULT_REFERENCE_TIMEOUT_SECONDS,
        max_body_bytes: int = DEFAULT_REFERENCE_BODY_LIMIT_BYTES,
        max_schema_refs: int = DEFAULT_MAX_SCHEMA_REFS,
        max_schema_ids: int = DEFAULT_MAX_SCHEMA_IDS,
        max_ref_depth: int = DEFAULT_MAX_REF_DEPTH,
        max_schema_keywords: int = DEFAULT_MAX_SCHEMA_KEYWORDS,
        max_schema_depth: int = DEFAULT_MAX_SCHEMA_DEPTH,
        transport_factory: TransportFactory | None = None,
    ) -> None:
        self._timeout = timeout
        self._max_body_bytes = max_body_bytes
        self._max_schema_refs = max_schema_refs
        self._max_schema_ids = max_schema_ids
        self._max_ref_depth = max_ref_depth
        self._max_schema_keywords = max_schema_keywords
        self._max_schema_depth = max_schema_depth
        self._transport_factory = transport_factory
        self._cache: dict[str, CanonicalReferenceResult] = {}

    @property
    def cache(self) -> Mapping[str, CanonicalReferenceResult]:
        """Read-only view of resolved immutable references."""

        return MappingProxyType({key: _copy_result(value) for key, value in self._cache.items()})

    def resolve_platform_extension(
        self,
        reference: CanonicalReference | Mapping[str, Any] | str,
    ) -> CanonicalReferenceResult:
        """Fetch and digest-verify a ``platform_extensions`` reference."""

        parsed, error = parse_canonical_reference(reference)
        if error is not None:
            return error
        assert parsed is not None
        cached = self._cache.get(parsed.cache_key)
        if cached is not None:
            return _copy_result(cached, from_cache=True)

        fetch_result = self._fetch(parsed)
        if fetch_result.status is CanonicalReferenceStatus.RESOLVED:
            self._cache[parsed.cache_key] = _copy_result(fetch_result)
        return fetch_result

    def resolve_format_schema(
        self,
        reference: CanonicalReference | Mapping[str, Any] | str,
    ) -> CanonicalReferenceResult:
        """Fetch, digest-verify, and validate a ``format_schema`` document."""

        parsed, error = parse_canonical_reference(reference)
        if error is not None:
            return error
        assert parsed is not None
        cached = self._cache.get(parsed.cache_key)
        if cached is not None:
            if cached.document is None and cached.body is not None:
                schema_result = self._validate_schema(parsed, cached.body)
                if schema_result.status is CanonicalReferenceStatus.RESOLVED:
                    self._cache[parsed.cache_key] = _copy_result(schema_result)
                return _copy_result(schema_result, from_cache=True)
            return _copy_result(cached, from_cache=True)

        fetch_result = self._fetch(parsed)
        if fetch_result.status is not CanonicalReferenceStatus.RESOLVED:
            return fetch_result
        assert fetch_result.body is not None
        self._cache[parsed.cache_key] = _copy_result(fetch_result)

        schema_result = self._validate_schema(parsed, fetch_result.body)
        if schema_result.status is CanonicalReferenceStatus.RESOLVED:
            self._cache[parsed.cache_key] = _copy_result(schema_result)
        return schema_result

    def _fetch(self, reference: CanonicalReference) -> CanonicalReferenceResult:
        resolved, unsafe = _resolve_public_https(reference.uri)
        if unsafe is not None:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.BLOCKED_UNSAFE_URL,
                reference=reference,
                message=unsafe,
            )
        assert resolved is not None

        transport = (
            self._transport_factory(resolved.host, resolved.resolved_ip)
            if self._transport_factory is not None
            else IpPinnedTransport(hostname=resolved.host, resolved_ip=resolved.resolved_ip)
        )
        try:
            with httpx.Client(
                transport=transport,
                timeout=self._timeout,
                follow_redirects=False,
                trust_env=False,
            ) as client:
                with client.stream(
                    "GET",
                    reference.uri,
                    headers={"Accept": "application/schema+json, application/json"},
                ) as response:
                    if 300 <= response.status_code < 400:
                        return CanonicalReferenceResult(
                            status=CanonicalReferenceStatus.BLOCKED_UNSAFE_URL,
                            reference=reference,
                            message="redirects are not allowed for canonical references",
                        )
                    if response.status_code != 200:
                        return CanonicalReferenceResult(
                            status=CanonicalReferenceStatus.NETWORK_ERROR,
                            reference=reference,
                            message="reference fetch failed",
                        )
                    body = _read_capped_body(response, self._max_body_bytes)
        except _BodyTooLargeError:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.BODY_TOO_LARGE,
                reference=reference,
                message="reference body exceeded configured size cap",
            )
        except httpx.HTTPError:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.NETWORK_ERROR,
                reference=reference,
                message="reference fetch failed",
            )

        actual_digest = _DIGEST_PREFIX + hashlib.sha256(body).hexdigest()
        if actual_digest != reference.digest:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.DIGEST_MISMATCH,
                reference=reference,
                message="reference digest mismatch",
            )
        return CanonicalReferenceResult(
            status=CanonicalReferenceStatus.RESOLVED,
            reference=reference,
            body=body,
        )

    def _validate_schema(
        self,
        reference: CanonicalReference,
        body: bytes,
    ) -> CanonicalReferenceResult:
        try:
            document = json.loads(body.decode("utf-8"))
        except (UnicodeDecodeError, ValueError):
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message="format_schema is not valid UTF-8 JSON",
            )
        if not isinstance(document, dict):
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message="format_schema root must be a JSON object",
            )

        ref_error = _validate_schema_refs(
            document,
            base_uri=reference.uri,
            max_refs=self._max_schema_refs,
            max_ids=self._max_schema_ids,
            max_ref_depth=self._max_ref_depth,
            max_keywords=self._max_schema_keywords,
            max_depth=self._max_schema_depth,
        )
        if ref_error is not None:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message=ref_error,
            )

        validator_class, draft_error = _validator_class_for_schema(document)
        if draft_error is not None:
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message=draft_error,
            )
        try:
            validator_class.check_schema(document)
        except (jsonschema.exceptions.SchemaError, RecursionError):
            return CanonicalReferenceResult(
                status=CanonicalReferenceStatus.INVALID_SCHEMA,
                reference=reference,
                body=body,
                message="format_schema failed JSON Schema validation",
            )

        return CanonicalReferenceResult(
            status=CanonicalReferenceStatus.RESOLVED,
            reference=reference,
            body=body,
            document=document,
        )

Resolve immutable format_schema and platform_extensions refs.

The resolver owns an in-memory cache keyed by uri@digest. Construct one per adopter component or request context; no process-global mutable configuration is used.

Instance variables

prop cache : Mapping[str, CanonicalReferenceResult]
Expand source code 
@property
def cache(self) -> Mapping[str, CanonicalReferenceResult]:
    """Read-only view of resolved immutable references."""

    return MappingProxyType({key: _copy_result(value) for key, value in self._cache.items()})

Read-only view of resolved immutable references.

Methods

def resolve_format_schema(self,
reference: CanonicalReference | Mapping[str, Any] | str) ‑> CanonicalReferenceResult
Expand source code 
def resolve_format_schema(
    self,
    reference: CanonicalReference | Mapping[str, Any] | str,
) -> CanonicalReferenceResult:
    """Fetch, digest-verify, and validate a ``format_schema`` document."""

    parsed, error = parse_canonical_reference(reference)
    if error is not None:
        return error
    assert parsed is not None
    cached = self._cache.get(parsed.cache_key)
    if cached is not None:
        if cached.document is None and cached.body is not None:
            schema_result = self._validate_schema(parsed, cached.body)
            if schema_result.status is CanonicalReferenceStatus.RESOLVED:
                self._cache[parsed.cache_key] = _copy_result(schema_result)
            return _copy_result(schema_result, from_cache=True)
        return _copy_result(cached, from_cache=True)

    fetch_result = self._fetch(parsed)
    if fetch_result.status is not CanonicalReferenceStatus.RESOLVED:
        return fetch_result
    assert fetch_result.body is not None
    self._cache[parsed.cache_key] = _copy_result(fetch_result)

    schema_result = self._validate_schema(parsed, fetch_result.body)
    if schema_result.status is CanonicalReferenceStatus.RESOLVED:
        self._cache[parsed.cache_key] = _copy_result(schema_result)
    return schema_result

Fetch, digest-verify, and validate a format_schema document.

def resolve_platform_extension(self,
reference: CanonicalReference | Mapping[str, Any] | str) ‑> CanonicalReferenceResult
Expand source code 
def resolve_platform_extension(
    self,
    reference: CanonicalReference | Mapping[str, Any] | str,
) -> CanonicalReferenceResult:
    """Fetch and digest-verify a ``platform_extensions`` reference."""

    parsed, error = parse_canonical_reference(reference)
    if error is not None:
        return error
    assert parsed is not None
    cached = self._cache.get(parsed.cache_key)
    if cached is not None:
        return _copy_result(cached, from_cache=True)

    fetch_result = self._fetch(parsed)
    if fetch_result.status is CanonicalReferenceStatus.RESOLVED:
        self._cache[parsed.cache_key] = _copy_result(fetch_result)
    return fetch_result

Fetch and digest-verify a platform_extensions reference.

class CanonicalReferenceResult (status: CanonicalReferenceStatus,
reference: CanonicalReference | None = None,
body: bytes | None = None,
document: Any | None = None,
message: str | None = None,
from_cache: bool = False)
Expand source code 
@dataclass(frozen=True)
class CanonicalReferenceResult:
    """Structured resolver result.

    ``body`` is populated only on successful fetches. ``document`` is populated
    for successful ``format_schema`` resolution after JSON parsing. ``message``
    is intentionally coarse and should be safe to show in diagnostics.
    """

    status: CanonicalReferenceStatus
    reference: CanonicalReference | None = None
    body: bytes | None = None
    document: Any | None = None
    message: str | None = None
    from_cache: bool = False

    @property
    def resolved(self) -> bool:
        return self.status is CanonicalReferenceStatus.RESOLVED

Structured resolver result.

body is populated only on successful fetches. document is populated for successful format_schema resolution after JSON parsing. message is intentionally coarse and should be safe to show in diagnostics.

Instance variables

var body : bytes | None
var document : typing.Any | None
var from_cache : bool
var message : str | None
var reference : CanonicalReference | None
prop resolved : bool
Expand source code 
@property
def resolved(self) -> bool:
    return self.status is CanonicalReferenceStatus.RESOLVED
var status : CanonicalReferenceStatus
class CanonicalReferenceStatus (*args, **kwds)
Expand source code 
class CanonicalReferenceStatus(str, Enum):
    """Stable resolver outcomes for conformance diagnostics."""

    RESOLVED = "resolved"
    INVALID_REFERENCE = "invalid_reference"
    BLOCKED_UNSAFE_URL = "blocked_unsafe_url"
    NETWORK_ERROR = "network_error"
    BODY_TOO_LARGE = "body_too_large"
    DIGEST_MISMATCH = "digest_mismatch"
    INVALID_SCHEMA = "invalid_schema"

Stable resolver outcomes for conformance diagnostics.

Ancestors

  • builtins.str
  • enum.Enum

Class variables

var BLOCKED_UNSAFE_URL
var BODY_TOO_LARGE
var DIGEST_MISMATCH
var INVALID_REFERENCE
var INVALID_SCHEMA
var NETWORK_ERROR
var RESOLVED
class CatalogIndex (by_owner_and_id: dict[tuple[str, str], dict[str, Any] | None],
by_unique_id: dict[str, dict[str, Any] | None])
Expand source code 
@dataclass(frozen=True)
class CatalogIndex:
    by_owner_and_id: dict[tuple[str, str], dict[str, Any] | None]
    by_unique_id: dict[str, dict[str, Any] | None]
    _allow_bare_id_fallback: bool = False

CatalogIndex(by_owner_and_id: 'dict[tuple[str, str], dict[str, Any] | None]', by_unique_id: 'dict[str, dict[str, Any] | None]', _allow_bare_id_fallback: 'bool' = False)

Instance variables

var by_owner_and_id : dict[tuple[str, str], dict[str, typing.Any] | None]
var by_unique_id : dict[str, dict[str, typing.Any] | None]
class CreativeDialect (*args, **kwds)
Expand source code 
class CreativeDialect(str, Enum):
    """Creative identity shape expected at a protocol boundary."""

    LEGACY = "legacy"
    CANONICAL = "canonical"

Creative identity shape expected at a protocol boundary.

Ancestors

  • builtins.str
  • enum.Enum

Class variables

var CANONICAL
var LEGACY
class CreativeDialectError (*args, **kwargs)
Expand source code 
class CreativeDialectError(ValueError):
    """Raised when negotiation cannot select a safe creative dialect."""

Raised when negotiation cannot select a safe creative dialect.

Ancestors

  • builtins.ValueError
  • builtins.Exception
  • builtins.BaseException
class Divergence (field: str, kind: DivergenceKind, cap: Any, value: Any)
Expand source code 
@dataclass
class Divergence:
    """One narrowing-check failure between v2 ``params`` and v1 ``requirements``.

    Returned by :func:`check_narrows` and folded into the advisory's
    ``details.divergences`` list by :func:`narrowing_advisory`. The
    typed form lets callers route on ``kind`` without parsing dicts;
    the ``to_dict`` projection is what lands on the wire.

    Field semantics:

    * ``exceeds_max`` — v2 declared a value above v1's published cap.
      ``cap`` is the v1 maximum; ``value`` is the v2 over-cap declaration.
    * ``below_min`` — v2 declared a value below v1's published floor.
      ``cap`` is the v1 minimum; ``value`` is the v2 under-floor declaration.
    * ``not_subset`` — v2 declared an enum-typed set with values v1
      doesn't allow. ``cap`` is the v1 allowed set; ``value`` is the v2
      declared set.
    * ``not_equal`` — v2 disagreed on an exact-equal scalar.
      ``cap`` is the v1 value; ``value`` is the v2 value.

    The bound names (``cap``/``value``) trade some clarity for a single
    discriminated-union shape that's easier for adopters to switch over
    than four kind-specific records.
    """

    field: str
    kind: DivergenceKind
    cap: Any
    value: Any

    def to_dict(self) -> dict[str, Any]:
        """Wire-shape projection used in advisory ``details.divergences``.

        Preserves the field-name vocabulary the half-2 implementation
        emits (``v1_max`` / ``v1_min`` / ``v1_allowed`` / ``v1_value``
        on one side, ``v2_value`` / ``v2_declared`` on the other) so
        existing buyer-side parsers don't break on the typed switch.
        """
        v1_key, v2_key = {
            "exceeds_max": ("v1_max", "v2_value"),
            "below_min": ("v1_min", "v2_value"),
            "not_subset": ("v1_allowed", "v2_declared"),
            "not_equal": ("v1_value", "v2_value"),
        }[self.kind]
        return {
            "field": self.field,
            "kind": self.kind,
            v1_key: self.cap,
            v2_key: self.value,
        }

One narrowing-check failure between v2 params and v1 requirements.

Returned by :func:check_narrows() and folded into the advisory's details.divergences list by :func:narrowing_advisory(). The typed form lets callers route on kind without parsing dicts; the to_dict projection is what lands on the wire.

Field semantics:

  • exceeds_max — v2 declared a value above v1's published cap. cap is the v1 maximum; value is the v2 over-cap declaration.
  • below_min — v2 declared a value below v1's published floor. cap is the v1 minimum; value is the v2 under-floor declaration.
  • not_subset — v2 declared an enum-typed set with values v1 doesn't allow. cap is the v1 allowed set; value is the v2 declared set.
  • not_equal — v2 disagreed on an exact-equal scalar. cap is the v1 value; value is the v2 value.

The bound names (cap/value) trade some clarity for a single discriminated-union shape that's easier for adopters to switch over than four kind-specific records.

Instance variables

var cap : Any
var field : str
var kind : Literal['exceeds_max', 'below_min', 'not_subset', 'not_equal']
var value : Any

Methods

def to_dict(self) ‑> dict[str, typing.Any]
Expand source code 
def to_dict(self) -> dict[str, Any]:
    """Wire-shape projection used in advisory ``details.divergences``.

    Preserves the field-name vocabulary the half-2 implementation
    emits (``v1_max`` / ``v1_min`` / ``v1_allowed`` / ``v1_value``
    on one side, ``v2_value`` / ``v2_declared`` on the other) so
    existing buyer-side parsers don't break on the typed switch.
    """
    v1_key, v2_key = {
        "exceeds_max": ("v1_max", "v2_value"),
        "below_min": ("v1_min", "v2_value"),
        "not_subset": ("v1_allowed", "v2_declared"),
        "not_equal": ("v1_value", "v2_value"),
    }[self.kind]
    return {
        "field": self.field,
        "kind": self.kind,
        v1_key: self.cap,
        v2_key: self.value,
    }

Wire-shape projection used in advisory details.divergences.

Preserves the field-name vocabulary the half-2 implementation emits (v1_max / v1_min / v1_allowed / v1_value on one side, v2_value / v2_declared on the other) so existing buyer-side parsers don't break on the typed switch.

class SdkAdvisory (**data: Any)
Expand source code 
class Error(AdCPBaseModel):
    model_config = ConfigDict(
        extra='allow',
    )
    code: Annotated[
        str,
        Field(
            description='Error code for programmatic handling. The error-code vocabulary is open: `error.code` is wire-typed `string` (not a closed enum), the standard codes published in `enums/error-code.json` are documentary, and senders MAY emit codes outside that set (platform-specific codes, or codes introduced in a later AdCP version). Receivers MUST decode unknown codes — treat the response as well-formed, read `error.recovery` for the recovery classification, and fall back to `transient` when `recovery` is absent. See `error-handling.mdx#forward-compatible-decoding-normative` for the full forward-compat contract — this rule is what lets future maintenance lines ship new codes additively.',
            max_length=64,
            min_length=1,
        ),
    ]
    message: Annotated[str, Field(description='Human-readable error message')]
    field: Annotated[
        str | None,
        Field(
            description="Field path associated with the error in JSONPath-lite format (e.g., 'packages[0].targeting'). When `issues[]` is also present, sellers MUST set this to `issues[0].pointer` translated from RFC 6901 to JSONPath-lite (e.g., '/packages/0/targeting' → 'packages[0].targeting') so pre-3.1 consumers reading `field` only get deterministic behavior. Will be deprecated in a future major version in favor of `issues[].pointer`."
        ),
    ] = None
    suggestion: Annotated[str | None, Field(description='Suggested fix for the error')] = None
    retry_after: Annotated[
        float | None,
        Field(
            description='Seconds to wait before retrying the operation. AdCP 3.2 producers MUST emit an integer from 1 through 3600. The 3.x schema continues to accept finite fractional values for backward compatibility with earlier producers; consumers receiving one MUST round up to the next whole second before clamping so the retry is never scheduled earlier than intended. Non-finite values are treated as absent.',
            ge=1.0,
            le=3600.0,
        ),
    ] = None
    issues: Annotated[
        list[Issue] | None,
        Field(
            description='Structured list of validation failures. Primary use is `VALIDATION_ERROR`, where multi-field rejections are common and `field` (singular) cannot carry the full pointer map. MAY appear on other error codes that reject multiple fields at once. When `issues` is present, sellers MUST also populate `field` from `issues[0]` for backward compatibility with pre-3.1 consumers that read `field` only — translating the RFC 6901 `pointer` format to the JSONPath-lite format `field` uses (e.g., `/packages/0/targeting` → `packages[0].targeting`). MUST (not SHOULD) so consumers reading `field` get deterministic behavior across sellers — the cost is one line of dual-write per seller; the cost of SHOULD is a long tail of seller-A-vs-seller-B inconsistency. Future major versions will deprecate `field` in favor of `issues[].pointer`.'
        ),
    ] = None
    details: Annotated[
        dict[str, Any] | None,
        Field(
            description='Additional task-specific error details. Sellers MAY mirror `issues[]` here as `details.issues` for backward compatibility with pre-3.1 consumers reading from `details`; new consumers SHOULD prefer the top-level `issues` field.\n\n**Canonical rejection-set shape (3.1+).** When the error reports a rejected value against a closed set of accepted values (e.g., enum mismatch, unsupported pricing option, invalid signal id), sellers SHOULD use the canonical key `accepted_values: <array>` under `details` rather than seller-specific variants observed in the wild (`available`, `allowed`, `accepted_values` at the error root, etc.). The canonical shape:\n\n```json\n{\n  "code": "INVALID_PRICING_OPTION",\n  "message": "Pricing option not found: po_prism_abandoner_cpm",\n  "field": "pricing_option_id",\n  "details": {\n    "rejected_value": "po_prism_abandoner_cpm",\n    "accepted_values": ["po_prism_cart_cpm", "po_prism_view_cpm"]\n  }\n}\n```\n\n- `rejected_value` (optional): the offending value the buyer supplied, echoed for buyer-side diagnostic clarity (especially when the offending field is nested or transformed before validation).\n- `accepted_values` (optional): the closed set the seller would have accepted at this field on this call. Sellers MUST NOT enumerate the full ecosystem-wide accepted set if it differs from what\'s accepted for *this caller in this context* (account, brand, scope) — leaking ecosystem-wide accepted sets to a per-caller rejection turns the error into an enumeration oracle.\n\nThis is **SHOULD-level guidance**, not MUST: `details` remains `additionalProperties: true` and pre-3.1 sellers using `available` / `allowed` / `accepted_values` at the error root remain conformant. The canonical shape lets buyer-side diagnostic tooling (SDK runner hints, dashboards, error classifiers) reliably surface the accepted-set without per-seller pattern matching. SDKs SHOULD accept any of the legacy variants and normalize on read; the canonical shape is what new sellers and 3.1+ adopters should emit going forward.'
        ),
    ] = None
    recovery: Annotated[
        Recovery | None,
        Field(
            description='Agent recovery classification. transient: retry after delay (rate limit, service unavailable, timeout). correctable: fix the request and resend (invalid field, budget too low, creative rejected). terminal: requires human action (account suspended, payment required, account not found). Senders SHOULD populate `recovery` on every error from 3.1 onward — it is the normative carrier of recovery semantics across version skew. A receiver that does not recognize `error.code` (a newer code, or a platform-specific code) MUST still be able to classify the error from `recovery`. The `enumMetadata.recovery` block in `enums/error-code.json` is the documentary mirror for known codes; `error.recovery` on the wire is authoritative.'
        ),
    ] = None
    source: Annotated[
        Source | None,
        Field(
            description='Who emitted this error entry. `producer` (default when absent): emitted by the response\'s authoring agent (the seller for `get_products`, the creative agent for `build_creative`, etc.). `sdk`: augmented by a consuming SDK that detected a non-fatal advisory condition on consumption (e.g., `FORMAT_PROJECTION_FAILED` when the buyer SDK couldn\'t project a v1 format to a canonical, or `FORMAT_DECLARATION_DIVERGENT` when the SDK detected a producer bug on read). SDK-augmented entries SHOULD also set `sdk_id` so downstream consumers can identify which intermediate processor inserted the entry.\n\n**Multi-hop propagation (normative).** AdCP is a federated agent network — responses commonly traverse multiple SDKs (e.g., sales agent → interchange → DSP → buyer). When an SDK augments `errors[]` with a consumption-detected entry, the augmented response carries the entry forward to subsequent hops. Each hop that detects the same condition independently SHOULD deduplicate by `(code, field)` rather than re-emit; the existing entry\'s `sdk_id` identifies which earlier processor saw it first. Producer entries (those without `source: "sdk"`) are authoritative for what the response\'s authoring agent self-detected; SDK entries are observations made on top.\n\n**Replay/audit safety.** Persisted or replayed responses carry `source` and `sdk_id` so the audit trail can distinguish seller-emitted entries from SDK-augmented ones. Without `source`, a downstream consumer can\'t tell whether a code came from the seller or an intermediate SDK, which corrupts attribution.'
        ),
    ] = None
    sdk_id: Annotated[
        str | None,
        Field(
            description='Optional identifier for the SDK that augmented this error entry. Format: `<sdk_package_name>@<version>` (e.g., `@adcontextprotocol/adcp@7.3.0`, `adcontextprotocol-adcp-python@1.2.0`). MUST be set when `source: "sdk"`; MUST be absent when `source: "producer"` or absent. Lets downstream consumers identify which intermediate processor inserted the entry, useful for debugging cross-SDK divergence (e.g., one SDK detects a projection failure that another SDK\'s registry version doesn\'t).'
        ),
    ] = None

Base model for AdCP types with spec-compliant serialization.

Defaults to extra='ignore' so unknown fields from newer spec versions are silently dropped rather than causing validation errors. Generated types whose schemas set additionalProperties: true override this with extra='allow' in their own model_config.

Set ADCP_STRICT_VALIDATION=1 in the environment ("1", "true", "yes", "on" are accepted) to flip the default to extra='forbid'. Use this during spec upgrades to catch silently-dropped renamed fields in tests. See :func:_resolve_extra_policy.

Important

The env var is resolved once at module import time. Set it in your shell or CI environment before import adcp runs — mutating os.environ["ADCP_STRICT_VALIDATION"] after the first adcp import has no effect on already-imported model classes (they captured the policy at class-body evaluation).

Consumers who want per-model strict validation can override model_config on their subclass.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Subclasses

  • adcp.types.generated_poc.core.catalog_item_availability_error.CatalogItemAvailabilityError

Class variables

var code : str
var details : dict[str, typing.Any] | None
var field : str | None
var issues : list[adcp.types.generated_poc.core.error.Issue] | None
var message : str
var model_config
var recovery : adcp.types.generated_poc.core.error.Recovery | None
var retry_after : float | None
var sdk_id : str | None
var source : adcp.types.generated_poc.core.error.Source | None
var suggestion : str | None

Inherited members

class FormatKindNotInClosedSetError (format_kind: str, accepted_kinds: list[str])
Expand source code 
class FormatKindNotInClosedSetError(ValueError):
    """Raised when a ``format_kind`` is not in the product's ``format_options[]``.

    Carries the rejected kind plus the closed set on the exception
    instance so handlers can surface them on the wire response (e.g.,
    via ``error.details.accepted_values``). Use :meth:`to_wire_error`
    to construct the response ``Error`` directly.
    """

    def __init__(
        self,
        format_kind: str,
        accepted_kinds: list[str],
    ) -> None:
        self.format_kind = format_kind
        self.accepted_kinds = accepted_kinds
        super().__init__(
            f"format_kind={format_kind!r} is not in the product's format_options[] "
            f"closed set (accepted: {sorted(set(accepted_kinds))!r})."
        )

    def to_wire_error(
        self,
        *,
        field: str = "manifest.format_kind",
        message: str | None = None,
    ) -> Error:
        """Build the wire-correct ``UNSUPPORTED_FEATURE`` ``Error`` for the response.

        Per ``error.json``, closed-set rejections SHOULD use
        ``details.rejected_value`` + ``details.accepted_values`` so
        buyer-side diagnostic tooling can surface the accepted set
        without per-seller pattern matching.

        Args:
            field: JSONPath-lite pointer to the rejected field on the
                buyer's request (default ``"manifest.format_kind"`` —
                the typical ``create_media_buy`` location).
            message: Override the default human-readable message.
        """
        return Error(
            code="UNSUPPORTED_FEATURE",
            message=message or str(self),
            field=field,
            details={
                "rejected_value": self.format_kind,
                "accepted_values": sorted(set(self.accepted_kinds)),
            },
        )

Raised when a format_kind is not in the product's adcp.canonical_formats.format_options[].

Carries the rejected kind plus the closed set on the exception instance so handlers can surface them on the wire response (e.g., via error.details.accepted_values). Use :meth:to_wire_error to construct the response Error directly.

Ancestors

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

Methods

def to_wire_error(self, *, field: str = 'manifest.format_kind', message: str | None = None) ‑> adcp.types.generated_poc.core.error.Error
Expand source code 
def to_wire_error(
    self,
    *,
    field: str = "manifest.format_kind",
    message: str | None = None,
) -> Error:
    """Build the wire-correct ``UNSUPPORTED_FEATURE`` ``Error`` for the response.

    Per ``error.json``, closed-set rejections SHOULD use
    ``details.rejected_value`` + ``details.accepted_values`` so
    buyer-side diagnostic tooling can surface the accepted set
    without per-seller pattern matching.

    Args:
        field: JSONPath-lite pointer to the rejected field on the
            buyer's request (default ``"manifest.format_kind"`` —
            the typical ``create_media_buy`` location).
        message: Override the default human-readable message.
    """
    return Error(
        code="UNSUPPORTED_FEATURE",
        message=message or str(self),
        field=field,
        details={
            "rejected_value": self.format_kind,
            "accepted_values": sorted(set(self.accepted_kinds)),
        },
    )

Build the wire-correct UNSUPPORTED_FEATURE Error for the response.

Per error.json, closed-set rejections SHOULD use details.rejected_value + details.accepted_values so buyer-side diagnostic tooling can surface the accepted set without per-seller pattern matching.

Args
-----=
field
JSONPath-lite pointer to the rejected field on the buyer's request (default "manifest.format_kind" — the typical create_media_buy location).
message
Override the default human-readable message.
class LegacyCreativeProjectionError (*args, **kwargs)
Expand source code 
class LegacyCreativeProjectionError(ValueError):
    """Raised when a legacy inbound request cannot reach a canonical handler."""

Raised when a legacy inbound request cannot reach a canonical handler.

Ancestors

  • builtins.ValueError
  • builtins.Exception
  • builtins.BaseException
class LegacyFormatConversionContext (format_id: LegacyFormatId, product_id: str, field: str)
Expand source code 
@dataclass(frozen=True)
class LegacyFormatConversionContext:
    format_id: LegacyFormatId
    product_id: str
    field: str

LegacyFormatConversionContext(format_id: 'LegacyFormatId', product_id: 'str', field: 'str')

Instance variables

var field : str
var format_id : LegacyFormatId
var product_id : str
class PixelTrackerBatchResult (items: list[Any] = <factory>,
advisories: list[Error] = <factory>)
Expand source code 
@dataclass
class PixelTrackerBatchResult:
    """Aggregate downgrade or upgrade across a list of trackers."""

    items: list[Any] = field(default_factory=list)
    advisories: list[Error] = field(default_factory=list)

Aggregate downgrade or upgrade across a list of trackers.

Instance variables

var advisories : list[adcp.types.generated_poc.core.error.Error]
var items : list[typing.Any]
class PixelTrackerDowngrade (v1: V1UrlTracker,
advisory: Error | None = None)
Expand source code 
@dataclass
class PixelTrackerDowngrade:
    """Result of downgrading one ``PixelTrackerAsset`` to v1 wire shape."""

    v1: V1UrlTracker
    advisory: Error | None = None

Result of downgrading one PixelTrackerAsset to v1 wire shape.

Instance variables

var advisory : adcp.types.generated_poc.core.error.Error | None
var v1 : V1UrlTracker
class PixelTrackerUpgrade (pixel_tracker: PixelTrackerAsset | None,
advisory: Error)
Expand source code 
@dataclass
class PixelTrackerUpgrade:
    """Result of upgrading one v1 url-tracker asset to v2 ``PixelTrackerAsset``.

    The upgrade ALWAYS carries an advisory per the spec — event/method
    are inferred, not declared.

    ``pixel_tracker`` is ``None`` when the upgrade was rejected (e.g.,
    the v1 URL used a disallowed scheme like ``javascript:`` or
    ``file:``). The advisory carries the rejection reason and the
    rejected scheme; consumers MUST treat the v1 entry as opaque and
    drop it from the upgraded manifest.
    """

    pixel_tracker: PixelTrackerAsset | None
    advisory: Error

Result of upgrading one v1 url-tracker asset to v2 PixelTrackerAsset.

The upgrade ALWAYS carries an advisory per the spec — event/method are inferred, not declared.

adcp.canonical_formats.pixel_tracker is None when the upgrade was rejected (e.g., the v1 URL used a disallowed scheme like javascript: or file:). The advisory carries the rejection reason and the rejected scheme; consumers MUST treat the v1 entry as opaque and drop it from the upgraded manifest.

Instance variables

var advisory : adcp.types.generated_poc.core.error.Error
var pixel_tracker : adcp.types.generated_poc.core.assets.pixel_tracker_asset.PixelTrackerAsset | None
class ProjectedFormat (declaration: Format | None = None,
diagnostic: ProjectionDiagnostic | None = None)
Expand source code 
@dataclass
class ProjectedFormat:
    declaration: Format | None = None
    diagnostic: ProjectionDiagnostic | None = None

ProjectedFormat(declaration: 'Format | None' = None, diagnostic: 'ProjectionDiagnostic | None' = None)

Instance variables

var declaration : Format | None
var diagnostic : ProjectionDiagnostic | None
class ProjectionCatalogAdapters (legacy_format_converter: LegacyFormatConverter,
canonical_format_legacy_resolver: CanonicalFormatLegacyResolver)
Expand source code 
@dataclass(frozen=True)
class ProjectionCatalogAdapters:
    legacy_format_converter: LegacyFormatConverter
    canonical_format_legacy_resolver: CanonicalFormatLegacyResolver

ProjectionCatalogAdapters(legacy_format_converter: 'LegacyFormatConverter', canonical_format_legacy_resolver: 'CanonicalFormatLegacyResolver')

Instance variables

var canonical_format_legacy_resolver : Callable[[CanonicalFormatLegacyResolutionContext], collections.abc.Sequence[LegacyFormatId] | None]
var legacy_format_converter : Callable[[LegacyFormatConversionContext], Format | collections.abc.Mapping[str, typing.Any] | None]
class ProjectionDiagnostic (code: str,
field: str,
product_id: str,
resolution_failure: str | None = None,
format_kind: str = 'custom',
source: str = 'sdk',
reason: str | None = None)
Expand source code 
@dataclass(frozen=True)
class ProjectionDiagnostic:
    code: str
    field: str
    product_id: str
    resolution_failure: str | None = None
    format_kind: str = "custom"
    source: str = "sdk"
    reason: str | None = None

    def model_dump(self) -> dict[str, Any]:
        details = (
            {"product_id": self.product_id, "reason": self.reason}
            if self.code == "CANONICAL_PRODUCT_FORMATS_UNAVAILABLE"
            else {
                "format_kind": self.format_kind,
                "product_id": self.product_id,
                "resolution_failure": self.resolution_failure,
            }
        )
        return {
            "source": self.source,
            "sdk_id": SDK_ID,
            "field": self.field,
            "code": self.code,
            "error": {"details": details},
        }

ProjectionDiagnostic(code: 'str', field: 'str', product_id: 'str', resolution_failure: 'str | None' = None, format_kind: 'str' = 'custom', source: 'str' = 'sdk', reason: 'str | None' = None)

Instance variables

var code : str
var field : str
var format_kind : str
var product_id : str
var reason : str | None
var resolution_failure : str | None
var source : str

Methods

def model_dump(self) ‑> dict[str, typing.Any]
Expand source code 
def model_dump(self) -> dict[str, Any]:
    details = (
        {"product_id": self.product_id, "reason": self.reason}
        if self.code == "CANONICAL_PRODUCT_FORMATS_UNAVAILABLE"
        else {
            "format_kind": self.format_kind,
            "product_id": self.product_id,
            "resolution_failure": self.resolution_failure,
        }
    )
    return {
        "source": self.source,
        "sdk_id": SDK_ID,
        "field": self.field,
        "code": self.code,
        "error": {"details": details},
    }
class RegistryLoadError (*args, **kwargs)
Expand source code 
class RegistryLoadError(RuntimeError):
    """Raised when the bundled v1↔v2 registry cannot be loaded or parsed.

    Wraps the underlying :class:`FileNotFoundError`,
    :class:`json.JSONDecodeError`, or :class:`pydantic.ValidationError`
    with a contextual message naming the registry path + ADCP version so
    adopters can diagnose a corrupt bundle.
    """

Raised when the bundled v1↔v2 registry cannot be loaded or parsed.

Wraps the underlying :class:FileNotFoundError, :class:json.JSONDecodeError, or :class:pydantic.ValidationError with a contextual message naming the registry path + ADCP version so adopters can diagnose a corrupt bundle.

Ancestors

  • builtins.RuntimeError
  • builtins.Exception
  • builtins.BaseException
class V1CatalogProjection (declarations: list[ProductFormatDeclaration] = <factory>,
advisories: list[Error] = <factory>)
Expand source code 
@dataclass
class V1CatalogProjection:
    """Aggregate result of projecting a list of v1 formats to v2 declarations."""

    declarations: list[ProductFormatDeclaration] = field(default_factory=list)
    advisories: list[Error] = field(default_factory=list)

Aggregate result of projecting a list of v1 formats to v2 declarations.

Instance variables

var advisories : list[adcp.types.generated_poc.core.error.Error]
var declarations : list[Format]
class V1ToV2Projection (declaration: ProductFormatDeclaration | None = None,
advisories: list[Error] = <factory>)
Expand source code 
@dataclass
class V1ToV2Projection:
    """Result of projecting one v1 named format to a v2 declaration.

    Attributes:
        declaration: The projected ``ProductFormatDeclaration``, or
            ``None`` when projection failed closed (see step 5 above).
            When non-``None`` the declaration carries ``v1_format_ref``
            pointing back at the source v1 format.
        advisories: SDK-source ``errors[]`` entries the resolution
            order emitted. May include
            ``FORMAT_DECLARATION_V1_AMBIGUOUS`` (family-only structural
            match) or ``FORMAT_PROJECTION_FAILED`` (no match).
    """

    declaration: ProductFormatDeclaration | None = None
    advisories: list[Error] = field(default_factory=list)

Result of projecting one v1 named format to a v2 declaration.

Attributes
-----=
declaration
The projected ProductFormatDeclaration, or None when projection failed closed (see step 5 above). When non-None the declaration carries v1_format_ref pointing back at the source v1 format.
advisories
SDK-source errors[] entries the resolution order emitted. May include FORMAT_DECLARATION_V1_AMBIGUOUS (family-only structural match) or FORMAT_PROJECTION_FAILED (no match).

Instance variables

var advisories : list[adcp.types.generated_poc.core.error.Error]
var declaration : Format | None
class V1UrlTracker (asset_id: str, url: str, js_method: bool = False)
Expand source code 
@dataclass
class V1UrlTracker:
    """v1 wire-shape projection of a single ``pixel_tracker``.

    Carries the projected ``asset_id`` + ``url`` plus a flag for whether
    the source pixel was a JS include (``method=js``). Adopters
    assembling a v1 ``assets[]`` array consume this directly:

    .. code-block:: python

        v1 = downgrade_pixel_tracker(pt).v1
        v1_asset = {
            "asset_type": "url",
            "url_type": "tracker_pixel",
            "asset_id": v1.asset_id,
            "url": v1.url,
        }

    The ``js_method`` flag is exposed so adopters with v1 catalogs that
    track a separate JS-tracker slot can still distinguish — the
    spec collapses both onto the same ``url_type`` on the wire, but
    nothing prevents an adopter from tracking the source method.
    """

    asset_id: str
    url: str
    js_method: bool = False

v1 wire-shape projection of a single adcp.canonical_formats.pixel_tracker.

Carries the projected asset_id + url plus a flag for whether the source pixel was a JS include (method=js). Adopters assembling a v1 assets[] array consume this directly:

.. code-block:: python

v1 = downgrade_pixel_tracker(pt).v1
v1_asset = {
    "asset_type": "url",
    "url_type": "tracker_pixel",
    "asset_id": v1.asset_id,
    "url": v1.url,
}

The js_method flag is exposed so adopters with v1 catalogs that track a separate JS-tracker slot can still distinguish — the spec collapses both onto the same url_type on the wire, but nothing prevents an adopter from tracking the source method.

Instance variables

var asset_id : str
var js_method : bool
var url : str
class V2ToV1Projection (format_ids: list[FormatId] = <factory>,
advisories: list[Error] = <factory>)
Expand source code 
@dataclass
class V2ToV1Projection:
    """Result of projecting one or more ``ProductFormatDeclaration``s to v1.

    Attributes:
        format_ids: v1 ``format_ids[]`` entries to dual-emit alongside the
            v2 ``format_options[]``. Empty when the declarations are all
            v1-unreachable (custom / canonical_formats_only / non-translatable
            canonicals).
        advisories: SDK-source ``errors[]`` entries to augment the response
            with. Each carries ``source="sdk"`` and ``sdk_id=<this SDK>``.
    """

    format_ids: list[FormatId] = field(default_factory=list)
    advisories: list[Error] = field(default_factory=list)

Result of projecting one or more ProductFormatDeclarations to v1.

Attributes
-----=
format_ids
v1 format_ids[] entries to dual-emit alongside the v2 adcp.canonical_formats.format_options[]. Empty when the declarations are all v1-unreachable (custom / canonical_formats_only / non-translatable canonicals).
advisories
SDK-source errors[] entries to augment the response with. Each carries source="sdk" and sdk_id=<this SDK>.

Instance variables

var advisories : list[adcp.types.generated_poc.core.error.Error]
var format_ids : list[LegacyFormatId]