Module adcp.compat.legacy

Per-tool adapters for buyers on legacy AdCP wire shapes.

The dispatcher consults :func:get_legacy_adapter() when the buyer's adcp_version / adcp_major_version resolves into :data:LEGACY_ADAPTER_VERSIONS. If an adapter is registered for the (version, tool) pair, the request is translated to the current wire shape before validation + handler dispatch. If no adapter is registered for that tool at that version, the dispatcher surfaces INVALID_REQUEST — the legacy version doesn't expose the tool.

Architecturally this replaces the heuristic :func:spec_compat_hooks() model. Each adapter is hand-written (or, in time, codegen'd from declarative wire-delta specs) and tested end-to-end so the translation is auditable rather than implicit.

Adapters register themselves at import time via :func:register_adapter(). Importing :mod:adcp.compat.legacy.v2_5 populates the v2.5 registry; see that submodule's docstring for the list of supported tools.

Mirrors src/lib/adapters/legacy/v2-5/ in the TypeScript SDK (getV25Adapter / listV25AdapterTools).

Sub-modules

adcp.compat.legacy.errors

Error policy for transport-facing legacy adapter failures.

adcp.compat.legacy.types

AdapterPair — the typed contract for legacy-version translators …

adcp.compat.legacy.v2_5

Adapters for buyers on the AdCP 2.5 wire shape …

Global variables

var LEGACY_ADAPTER_VERSIONS : Final[tuple[str, ...]]

Iteration order is probe-precedence order. When the dispatcher runs shape-based detection (Stage 6, is_legacy_shape), it walks this tuple and breaks on the first match. List versions highest-to-lowest priority — newer legacy versions before older — so an ambiguous payload routes to the most-recent matching adapter.

Functions

def coerce_legacy_asset(asset_key: str, asset: Mapping[str, typing.Any]) ‑> dict[str, typing.Any]

Expand source code

def coerce_legacy_asset(asset_key: str, asset: Mapping[str, Any]) -> dict[str, Any]:
    """Copy an asset, infer missing asset_type, and demote dimensionless images.

    An image with a URL but missing either dimension becomes a url asset;
    any remaining width/height is removed. Existing discriminators otherwise
    survive unchanged, including unknown values. This normalizes rather than
    validates: malformed or ambiguous assets still fail normal model validation.
    Nested values are preserved; neither the input mapping nor its values are
    modified.
    """
    out = dict(asset)
    if "asset_type" not in out:
        inferred = infer_asset_type(asset_key, out)
        if inferred is not None:
            out["asset_type"] = inferred
    if out.get("asset_type") == "image" and not ("width" in out and "height" in out):
        if "url" in out:
            out.pop("width", None)
            out.pop("height", None)
            out["asset_type"] = "url"
    return out

Copy an asset, infer missing asset_type, and demote dimensionless images.

An image with a URL but missing either dimension becomes a url asset; any remaining width/height is removed. Existing discriminators otherwise survive unchanged, including unknown values. This normalizes rather than validates: malformed or ambiguous assets still fail normal model validation. Nested values are preserved; neither the input mapping nor its values are modified.

def coerce_legacy_assets(assets: Mapping[str, typing.Any]) ‑> dict[str, typing.Any]

Expand source code

def coerce_legacy_assets(assets: Mapping[str, Any]) -> dict[str, Any]:
    """Copy an asset mapping and normalize each dictionary with legacy rules.

    Non-mapping values and unknown keys are retained for model validation.
    Every mapping value is copied, so repeated normalization is idempotent
    without modifying stored rows or caller-owned dictionaries.
    """
    return {
        key: coerce_legacy_asset(key, value) if isinstance(value, Mapping) else value
        for key, value in assets.items()
    }

Copy an asset mapping and normalize each dictionary with legacy rules.

Non-mapping values and unknown keys are retained for model validation. Every mapping value is copied, so repeated normalization is idempotent without modifying stored rows or caller-owned dictionaries.

def get_legacy_adapter(version: str, tool_name: str) ‑> AdapterPair | None

Expand source code

def get_legacy_adapter(version: str, tool_name: str) -> AdapterPair | None:
    """Return the adapter for ``(version, tool_name)`` or ``None``.

    ``None`` means "no translation registered for this tool at this
    version" — the dispatcher converts that into ``INVALID_REQUEST``
    because the buyer claimed a legacy version this seller doesn't
    serve the tool on.
    """
    _ensure_loaded(version)
    return _REGISTRY.get((version, tool_name))

Return the adapter for (version, tool_name) or None.

None means "no translation registered for this tool at this version" — the dispatcher converts that into INVALID_REQUEST because the buyer claimed a legacy version this seller doesn't serve the tool on.

def infer_asset_type(asset_key: str, asset: Mapping[str, typing.Any]) ‑> str | None

Expand source code

def infer_asset_type(asset_key: str, asset: Mapping[str, Any]) -> str | None:
    """Infer a missing discriminator from an exact key, then field presence.

    Type-name keys take precedence over fields. Otherwise URL plus both
    dimensions implies image, URL alone implies url, and content implies text.
    Unknown keys without those fields return None for validation to diagnose.
    This function does not inspect or replace an existing discriminator.
    """
    if asset_key in _KNOWN_ASSET_TYPES:
        return asset_key
    if "url" in asset:
        return "image" if "width" in asset and "height" in asset else "url"
    if "content" in asset:
        return "text"
    return None

Infer a missing discriminator from an exact key, then field presence.

Type-name keys take precedence over fields. Otherwise URL plus both dimensions implies image, URL alone implies url, and content implies text. Unknown keys without those fields return None for validation to diagnose. This function does not inspect or replace an existing discriminator.

def list_legacy_adapter_tools(version: str) ‑> list[str]

Expand source code

def list_legacy_adapter_tools(version: str) -> list[str]:
    """Tools with a registered adapter at this legacy version."""
    _ensure_loaded(version)
    return sorted(tool for (v, tool) in _REGISTRY if v == version)

Tools with a registered adapter at this legacy version.

def register_adapter(version: str,
adapter: AdapterPair) ‑> None

Expand source code

def register_adapter(version: str, adapter: AdapterPair) -> None:
    """Register an :class:`AdapterPair` under ``(version, tool_name)``.

    Idempotent — re-registering the same pair (same callables) is a
    no-op. Re-registering a *different* pair for the same key raises
    :class:`ValueError`; tests should call :func:`_reset_registry_for_tests`
    if they need to swap an adapter mid-suite.

    Adapters self-register at module import time; the framework imports
    :mod:`adcp.compat.legacy.v2_5` lazily on first dispatch so adopters
    that don't speak legacy don't pay the cost.
    """
    if version not in LEGACY_ADAPTER_VERSIONS:
        raise ValueError(
            f"register_adapter: version {version!r} is not in "
            f"LEGACY_ADAPTER_VERSIONS={list(LEGACY_ADAPTER_VERSIONS)}. "
            "Add the version to the constant first."
        )
    key = (version, adapter.tool_name)
    existing = _REGISTRY.get(key)
    if existing is not None and existing is not adapter:
        raise ValueError(
            f"register_adapter: an adapter is already registered for "
            f"{key!r} ({existing!r}); refusing to overwrite with "
            f"{adapter!r}."
        )
    _REGISTRY[key] = adapter

Register an :class:AdapterPair under (version, tool_name).

Idempotent — re-registering the same pair (same callables) is a no-op. Re-registering a different pair for the same key raises :class:ValueError; tests should call :func:_reset_registry_for_tests if they need to swap an adapter mid-suite.

Adapters self-register at module import time; the framework imports :mod:adcp.compat.legacy.v2_5 lazily on first dispatch so adopters that don't speak legacy don't pay the cost.

Classes

class AdapterPair (tool_name: str,
adapt_request: Callable[[dict[str, Any]], dict[str, Any]],
normalize_response: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
is_legacy_shape: Callable[[dict[str, Any]], bool] | None = None)

Expand source code

@dataclass(frozen=True)
class AdapterPair:
    """Translation pair for one tool at one legacy version.

    Adapters live under ``adcp.compat.legacy.{version_key}.{tool_name}``
    and register themselves via :func:`adcp.compat.legacy.register_adapter`
    at import time. The dispatcher looks them up by
    ``(version_key, tool_name)`` once per request.

    Contract every adapter must hold:

    * **Sync + pure.** Both callables run synchronously and produce a
      new dict — they MUST NOT mutate their input (callers rely on the
      original being intact for retries, logging, and idempotency
      tracking). Tests in
      ``tests/test_legacy_adapter_registry.py::test_v2_5_adapter_does_not_mutate_input``
      assert this for shipped adapters; new adapters should add the
      equivalent check.
    * **No I/O.** Heavier work (resolving format references, calling
      upstream services) belongs in handlers, not adapters.
    * **Exception mapping.** An explicit
      :class:`adcp.compat.legacy.LegacyAdapterValidationError` from
      ``adapt_request`` surfaces its buyer-safe message as ``INVALID_REQUEST``.
      Every other request exception uses a fixed public message while retaining
      that code. A raise inside ``normalize_response`` similarly uses a fixed
      public message and surfaces as ``INTERNAL_ERROR`` (the handler produced a
      valid response that the adapter can't rewrite — SDK bug, not buyer bug).
      Unexpected exceptions and tracebacks are written to operator logs only.
    """

    tool_name: str
    adapt_request: Callable[[dict[str, Any]], dict[str, Any]]
    normalize_response: Callable[[dict[str, Any]], dict[str, Any]] | None = None
    # Optional shape probe used by the dispatcher when the buyer didn't
    # send an ``adcp_version`` / ``adcp_major_version`` envelope (real v2.5
    # buyers can't — the field didn't exist in the v2.5 schema). The
    # probe should return ``True`` only on strong, unambiguous v2.5
    # markers — fields that exist in v2.5 but NOT in v3 (e.g.
    # ``brand_manifest``, ``creative_ids`` in packages, bare-string
    # ``format_id``). False positives downgrade a real v3 buyer to v2.5
    # validation, which is the worst outcome; bias conservatively. Tools
    # with pass-through requests (``list_creative_formats``,
    # ``preview_creative``) leave this ``None`` because their request
    # shape is identical across versions.
    is_legacy_shape: Callable[[dict[str, Any]], bool] | None = None

Translation pair for one tool at one legacy version.

Adapters live under adcp.compat.legacy.{version_key}.{tool_name} and register themselves via :func:register_adapter() at import time. The dispatcher looks them up by (version_key, tool_name) once per request.

Contract every adapter must hold:

• Sync + pure. Both callables run synchronously and produce a new dict — they MUST NOT mutate their input (callers rely on the original being intact for retries, logging, and idempotency tracking). Tests in tests/test_legacy_adapter_registry.py::test_v2_5_adapter_does_not_mutate_input assert this for shipped adapters; new adapters should add the equivalent check.
• No I/O. Heavier work (resolving format references, calling upstream services) belongs in handlers, not adapters.
• Exception mapping. An explicit :class:LegacyAdapterValidationError from adapt_request surfaces its buyer-safe message as INVALID_REQUEST. Every other request exception uses a fixed public message while retaining that code. A raise inside normalize_response similarly uses a fixed public message and surfaces as INTERNAL_ERROR (the handler produced a valid response that the adapter can't rewrite — SDK bug, not buyer bug). Unexpected exceptions and tracebacks are written to operator logs only.

Instance variables

var adapt_request : Callable[[dict[str, typing.Any]], dict[str, typing.Any]]

var is_legacy_shape : collections.abc.Callable[[dict[str, typing.Any]], bool] | None

var normalize_response : collections.abc.Callable[[dict[str, typing.Any]], dict[str, typing.Any]] | None

var tool_name : str

class LegacyAdapterValidationError (*args, **kwargs)

Expand source code

class LegacyAdapterValidationError(ValueError):
    """Buyer-correctable request error whose message is safe for the wire.

    Legacy request adapters may raise this exception when the failure is caused
    by caller input and the exception message was deliberately written for the
    buyer. Arbitrary exceptions, including ordinary ``ValueError`` instances,
    are treated as private implementation failures.
    """

Buyer-correctable request error whose message is safe for the wire.

Legacy request adapters may raise this exception when the failure is caused by caller input and the exception message was deliberately written for the buyer. Arbitrary exceptions, including ordinary ValueError instances, are treated as private implementation failures.

Ancestors

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