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

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.** A raise inside ``adapt_request`` surfaces
      to the buyer as :class:`adcp.exceptions.ADCPTaskError` with code
      ``INVALID_REQUEST`` (translation = buyer-correctable, per spec).
      A raise inside ``normalize_response`` surfaces as
      ``INTERNAL_ERROR`` (the handler produced a valid response that
      the adapter can't rewrite — SDK bug, not buyer bug).
    """

    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. A raise inside adapt_request surfaces to the buyer as :class:ADCPTaskError with code INVALID_REQUEST (translation = buyer-correctable, per spec). A raise inside normalize_response surfaces as INTERNAL_ERROR (the handler produced a valid response that the adapter can't rewrite — SDK bug, not buyer bug).

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