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

• :func:project_declaration_to_v1() — single ProductFormatDeclaration → format_ids[] (with advisory errors[] emission on ambiguity / multi-size lossy fan-out).
• :func:project_product_to_v1() — fan-out helper across a product's adcp.canonical_formats.format_options[], accumulating refs and advisories.
• :func:validate_format_kind_in_options() — closed-set guard: rejects a format_kind that isn't published in the seller's adcp.canonical_formats.format_options[]. Sellers call this before accepting a create_media_buy.
• :func:find_declaration_by_kind() — looks up the matching declaration (with optional capability_id disambiguation).
• :func:upgrade_legacy_format_id() — upgrades common legacy named format IDs such as display_300x250 to parameterized canonical FormatId values.
• :func:formats_are_equivalent() — family/equivalence comparison after legacy upgrade.
• :func:format_is_supported() — stricter product/capability gating comparison after legacy upgrade.
• :class:CanonicalReferenceResolver — hardened fetch/cache helper for immutable format_schema and platform_extensions references.
• :func:load_default_registry() — loads the AAO-published v1↔v2 mapping registry from the bundled schema cache.
• :class:Error — typed wrapper around the SDK-source Error entries the projection emits on errors[].

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.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.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 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],
*,
capability_id: str | None = None) ‑> ProductFormatDeclaration | None

Expand source code

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

    Disambiguates with ``capability_id`` when the closed set carries
    multiple declarations sharing the same ``format_kind`` (the case
    where ``capability_id`` is REQUIRED per
    ``ProductFormatDeclaration.capability_id``).

    Args:
        format_kind: The kind to match. Accepts string or enum.
        format_options: The product's ``format_options[]``.
        capability_id: When provided, only declarations whose
            ``capability_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
            ``capability_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 capability_id is not None and d.capability_id != capability_id:
            continue
        return d
    return None

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

Disambiguates with capability_id when the closed set carries multiple declarations sharing the same format_kind (the case where capability_id is REQUIRED per ProductFormatDeclaration.capability_id).

Args

format_kind

The kind to match. Accepts string or enum.

format_options

The product's adcp.canonical_formats.format_options[].

capability_id

When provided, only declarations whose capability_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 capability_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]) ‑> ProductFormatDeclaration | 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.v1_format_ref or []
        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[ProductFormatDeclaration]]

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.v1_format_ref or []
        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 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 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 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 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_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.v1_format_ref or [])

    # 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_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`` but does NOT clobber the
    # registry's parametric defaults — that would lose every parameter
    # a registry glob carries (e.g., ``vast_version``, dimensions) when
    # a seller annotates only ``{kind: video_vast}``.
    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``; registry
    # parameters fill in anything the seller didn't restate.
    annotation = _v1_canonical_annotation(v1_format)
    if annotation is not None:
        return V1ToV2Projection(
            declaration=_build_declaration(
                kind=annotation.kind,
                v1_format_id=fid,
                params=registry_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 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') ‑> adcp.types.generated_poc.core.format_id.FormatReferenceStructuredObject

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 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_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_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_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_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 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. Sellers MUST return values between 1 and 3600. Clients MUST clamp values outside this range.',
            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_MODEL",\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

• AdCPBaseModel
• pydantic.main.BaseModel

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

• AdCPBaseModel:
  • model_dump
  • model_dump_json
  • model_summary

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 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 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[ProductFormatDeclaration]

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 : ProductFormatDeclaration | 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[adcp.types.generated_poc.core.format_id.FormatReferenceStructuredObject]