Module adcp.capabilities

Feature capability resolution for AdCP.

Shared logic for resolving feature support from a capabilities response. Used by both the client (buyer-side validation) and server (seller-side validation).

Functions

def build_synthetic_capabilities(supported_protocols: list[str], *, major_versions: list[int] | None = None) ‑> dict[str, typing.Any]

Expand source code

def build_synthetic_capabilities(
    supported_protocols: list[str],
    *,
    major_versions: list[int] | None = None,
) -> dict[str, Any]:
    """Build a synthetic capabilities response for pre-v3 sellers.

    Use this when connecting to a seller that doesn't support
    ``get_adcp_capabilities`` (pre-v3 sellers). The returned dict
    can be passed to ``FeatureResolver`` after wrapping with the
    appropriate Pydantic model.

    Args:
        supported_protocols: List of protocol domains the seller supports
            (e.g., ``["media_buy"]``).
        major_versions: ADCP major versions the seller supports.
            Defaults to ``[2]``.

    Returns:
        A dict matching the GetAdcpCapabilitiesResponse shape.
    """
    return {
        "adcp": {"major_versions": major_versions or [2]},
        "supported_protocols": supported_protocols,
    }

Build a synthetic capabilities response for pre-v3 sellers.

Use this when connecting to a seller that doesn't support get_adcp_capabilities (pre-v3 sellers). The returned dict can be passed to FeatureResolver after wrapping with the appropriate Pydantic model.

Args

supported_protocols

List of protocol domains the seller supports (e.g., ["media_buy"]).

major_versions

ADCP major versions the seller supports. Defaults to [2].

Returns

A dict matching the GetAdcpCapabilitiesResponse shape.

def looks_like_v3_capabilities(data: Any) ‑> bool

Expand source code

def looks_like_v3_capabilities(data: Any) -> bool:
    """Heuristic: does this ``get_adcp_capabilities`` response look v3-shaped?

    Used by ``ADCPClient.refresh_capabilities`` when the response fails strict
    schema validation but is structurally non-empty. The question the heuristic
    answers is "is this a v3 agent with a wire-shape bug, or a v2 agent that
    happens to advertise the tool?". Falling back to v2 in the former case
    masks the original bug behind cascading v2.5-schema-not-found errors;
    treating it as v3 surfaces the wire-shape bug at its source.

    Affirmative v3 signals (any one is enough):

    - ``adcp`` block (only v3 servers carry the
      ``{ major_versions, idempotency, ... }`` envelope)
    - ``supported_protocols`` array (v3-only top-level field)
    - any v3 protocol-level capability block (``account``, ``media_buy``,
      ``signals``, ``creative``, ``brand``, ``governance``,
      ``sponsored_intelligence``, ``compliance_testing``)

    v2 servers don't expose ``get_adcp_capabilities`` at all (the tool itself
    is a v3-only addition), so reaching this function with a non-empty payload
    already strongly implies v3 — but the structural check belt-and-suspenders
    against genuinely empty / null responses.

    Args:
        data: Raw response payload (typically a dict, but accepts any value
            so callers don't have to narrow before calling).

    Returns:
        True if any v3 signal is present; False for empty, null, non-dict,
        or shape-mismatched inputs.
    """
    if not _is_plain_object(data):
        return False
    if _is_plain_object(data.get("adcp")):
        return True
    if isinstance(data.get("supported_protocols"), list):
        return True
    v3_blocks = (
        "account",
        "media_buy",
        "signals",
        "creative",
        "brand",
        "governance",
        "sponsored_intelligence",
        "compliance_testing",
    )
    return any(_is_plain_object(data.get(block)) for block in v3_blocks)

Heuristic: does this get_adcp_capabilities response look v3-shaped?

Used by ADCPClient.refresh_capabilities when the response fails strict schema validation but is structurally non-empty. The question the heuristic answers is "is this a v3 agent with a wire-shape bug, or a v2 agent that happens to advertise the tool?". Falling back to v2 in the former case masks the original bug behind cascading v2.5-schema-not-found errors; treating it as v3 surfaces the wire-shape bug at its source.

Affirmative v3 signals (any one is enough):

• adcp block (only v3 servers carry the { major_versions, idempotency, ... } envelope)
• supported_protocols array (v3-only top-level field)
• any v3 protocol-level capability block (account, media_buy, signals, creative, brand, governance, sponsored_intelligence, compliance_testing)

v2 servers don't expose get_adcp_capabilities at all (the tool itself is a v3-only addition), so reaching this function with a non-empty payload already strongly implies v3 — but the structural check belt-and-suspenders against genuinely empty / null responses.

Args

data

Raw response payload (typically a dict, but accepts any value so callers don't have to narrow before calling).

Returns

True if any v3 signal is present; False for empty, null, non-dict, or shape-mismatched inputs.

def validate_capabilities(handler: Any, capabilities: GetAdcpCapabilitiesResponse) ‑> list[str]

Expand source code

def validate_capabilities(
    handler: Any,
    capabilities: GetAdcpCapabilitiesResponse,
) -> list[str]:
    """Check that a handler implements the methods required by its declared features.

    Compares the features declared in a capabilities response against the handler's
    method implementations. Returns warnings for features that are declared but
    whose corresponding handler methods are not overridden from the base class.

    This is a development-time check — call it at startup to catch misconfigurations.

    Args:
        handler: An ADCPHandler instance (or any object with handler methods).
        capabilities: The capabilities response the handler will serve.

    Returns:
        List of warning strings. Empty if everything is consistent.
    """
    # Late import to avoid circular dependency: server.base imports from adcp.types
    # which may transitively import from this module.
    from adcp.server.base import ADCPHandler

    resolver = FeatureResolver(capabilities)
    warnings: list[str] = []

    for feature, handler_methods in FEATURE_HANDLER_MAP.items():
        if not resolver.supports(feature):
            continue

        for method_name in handler_methods:
            if not hasattr(handler, method_name):
                warnings.append(
                    f"Feature '{feature}' is declared but handler has no " f"'{method_name}' method"
                )
                continue

            # Walk MRO to check if any class between the leaf and ADCPHandler
            # overrides the method (handles mixin / intermediate-class patterns).
            if isinstance(handler, ADCPHandler):
                overridden = any(
                    method_name in cls.__dict__
                    for cls in type(handler).__mro__
                    if cls is not ADCPHandler and not issubclass(ADCPHandler, cls)
                )
                if not overridden:
                    warnings.append(
                        f"Feature '{feature}' is declared but '{method_name}' "
                        f"is not overridden from ADCPHandler"
                    )

    return warnings

Check that a handler implements the methods required by its declared features.

Compares the features declared in a capabilities response against the handler's method implementations. Returns warnings for features that are declared but whose corresponding handler methods are not overridden from the base class.

This is a development-time check — call it at startup to catch misconfigurations.

Args

handler

An ADCPHandler instance (or any object with handler methods).

capabilities

The capabilities response the handler will serve.

Returns

List of warning strings. Empty if everything is consistent.

Classes

class FeatureResolver (capabilities: GetAdcpCapabilitiesResponse)

Expand source code

class FeatureResolver:
    """Resolves feature support from a GetAdcpCapabilitiesResponse.

    Supports multiple feature namespaces:

    - Protocol support: ``"media_buy"`` checks ``supported_protocols``
    - Extension support: ``"ext:scope3"`` checks ``extensions_supported``
    - Targeting: ``"targeting.geo_countries"`` checks
      ``media_buy.execution.targeting``
    - Media buy features: ``"inline_creative_management"`` checks
      ``media_buy.features``
    - Signals features: ``"catalog_signals"`` checks
      ``signals.features``
    """

    def __init__(self, capabilities: GetAdcpCapabilitiesResponse) -> None:
        self._caps = capabilities

        # Pre-compute the set of valid protocol names so supports() doesn't
        # need a runtime import on every call.
        from adcp.types.generated_poc.protocol.get_adcp_capabilities_response import (
            SupportedProtocol,
        )

        self._valid_protocols = {p.value for p in SupportedProtocol}
        self._declared_protocols = {p.value for p in capabilities.supported_protocols}

    @property
    def capabilities(self) -> GetAdcpCapabilitiesResponse:
        return self._caps

    def supports_v3(self) -> bool:
        """Check if the seller supports ADCP v3.

        Returns:
            True if major_versions includes 3.
        """
        for v in self._caps.adcp.major_versions:
            if (v.root if hasattr(v, "root") else v) == 3:
                return True
        return False

    def supports(self, feature: str) -> bool:
        """Check if a feature is supported."""
        caps = self._caps

        # Extension check: "ext:scope3"
        if feature.startswith("ext:"):
            ext_name = feature[4:]
            if caps.extensions_supported is None:
                return False
            return any(item.root == ext_name for item in caps.extensions_supported)

        # Targeting check: "targeting.geo_countries"
        if feature.startswith("targeting."):
            attr_name = feature[len("targeting.") :]
            if caps.media_buy is None or caps.media_buy.execution is None:
                return False
            targeting = caps.media_buy.execution.targeting
            if targeting is None:
                return False
            if attr_name not in type(targeting).model_fields:
                return False
            val = getattr(targeting, attr_name, None)
            # For bool fields, check truthiness. For object fields (like geo_metros),
            # presence means supported.
            return val is not None and val is not False

        # Protocol check: if the string is a known protocol name, resolve it
        # against supported_protocols and stop — don't fall through to features.
        if feature in self._declared_protocols:
            return True
        if feature in self._valid_protocols:
            return False

        # Media buy features check
        if caps.media_buy is not None and caps.media_buy.features is not None:
            if feature in type(caps.media_buy.features).model_fields:
                val = getattr(caps.media_buy.features, feature, None)
                if val is True:
                    return True

        # Signals features check
        if caps.signals is not None and caps.signals.features is not None:
            if feature in type(caps.signals.features).model_fields:
                val = getattr(caps.signals.features, feature, None)
                if val is True:
                    return True

        return False

    def require(
        self,
        *features: str,
        agent_id: str | None = None,
        agent_uri: str | None = None,
    ) -> None:
        """Assert that all listed features are supported.

        Args:
            *features: Feature identifiers to require.
            agent_id: Optional agent ID for error context.
            agent_uri: Optional agent URI for error context.

        Raises:
            ADCPFeatureUnsupportedError: If any features are not supported.
        """
        unsupported = [f for f in features if not self.supports(f)]
        if not unsupported:
            return

        declared = self.get_declared_features()

        raise ADCPFeatureUnsupportedError(
            unsupported_features=unsupported,
            declared_features=declared,
            agent_id=agent_id,
            agent_uri=agent_uri,
        )

    def get_declared_features(self) -> list[str]:
        """Collect all features the response declares as supported."""
        caps = self._caps
        declared: list[str] = []

        # Supported protocols
        for p in caps.supported_protocols:
            declared.append(p.value)

        # Media buy features
        if caps.media_buy is not None and caps.media_buy.features is not None:
            for field_name in type(caps.media_buy.features).model_fields:
                if getattr(caps.media_buy.features, field_name, None) is True:
                    declared.append(field_name)

        # Signals features
        if caps.signals is not None and caps.signals.features is not None:
            for field_name in type(caps.signals.features).model_fields:
                if getattr(caps.signals.features, field_name, None) is True:
                    declared.append(field_name)

        # Targeting features
        if caps.media_buy is not None and caps.media_buy.execution is not None:
            targeting = caps.media_buy.execution.targeting
            if targeting is not None:
                for field_name in type(targeting).model_fields:
                    val = getattr(targeting, field_name, None)
                    if val is not None and val is not False:
                        declared.append(f"targeting.{field_name}")

        # Extensions
        if caps.extensions_supported is not None:
            for item in caps.extensions_supported:
                declared.append(f"ext:{item.root}")

        return declared

Resolves feature support from a GetAdcpCapabilitiesResponse.

Supports multiple feature namespaces:

• Protocol support: "media_buy" checks supported_protocols
• Extension support: "ext:scope3" checks extensions_supported
• Targeting: "targeting.geo_countries" checks media_buy.execution.targeting
• Media buy features: "inline_creative_management" checks media_buy.features
• Signals features: "catalog_signals" checks signals.features

Instance variables

prop capabilities : GetAdcpCapabilitiesResponse

Expand source code

@property
def capabilities(self) -> GetAdcpCapabilitiesResponse:
    return self._caps

Methods

def get_declared_features(self) ‑> list[str]

Expand source code

def get_declared_features(self) -> list[str]:
    """Collect all features the response declares as supported."""
    caps = self._caps
    declared: list[str] = []

    # Supported protocols
    for p in caps.supported_protocols:
        declared.append(p.value)

    # Media buy features
    if caps.media_buy is not None and caps.media_buy.features is not None:
        for field_name in type(caps.media_buy.features).model_fields:
            if getattr(caps.media_buy.features, field_name, None) is True:
                declared.append(field_name)

    # Signals features
    if caps.signals is not None and caps.signals.features is not None:
        for field_name in type(caps.signals.features).model_fields:
            if getattr(caps.signals.features, field_name, None) is True:
                declared.append(field_name)

    # Targeting features
    if caps.media_buy is not None and caps.media_buy.execution is not None:
        targeting = caps.media_buy.execution.targeting
        if targeting is not None:
            for field_name in type(targeting).model_fields:
                val = getattr(targeting, field_name, None)
                if val is not None and val is not False:
                    declared.append(f"targeting.{field_name}")

    # Extensions
    if caps.extensions_supported is not None:
        for item in caps.extensions_supported:
            declared.append(f"ext:{item.root}")

    return declared

Collect all features the response declares as supported.

def require(self, *features: str, agent_id: str | None = None, agent_uri: str | None = None) ‑> None

Expand source code

def require(
    self,
    *features: str,
    agent_id: str | None = None,
    agent_uri: str | None = None,
) -> None:
    """Assert that all listed features are supported.

    Args:
        *features: Feature identifiers to require.
        agent_id: Optional agent ID for error context.
        agent_uri: Optional agent URI for error context.

    Raises:
        ADCPFeatureUnsupportedError: If any features are not supported.
    """
    unsupported = [f for f in features if not self.supports(f)]
    if not unsupported:
        return

    declared = self.get_declared_features()

    raise ADCPFeatureUnsupportedError(
        unsupported_features=unsupported,
        declared_features=declared,
        agent_id=agent_id,
        agent_uri=agent_uri,
    )

Assert that all listed features are supported.

Args

*features

Feature identifiers to require.

agent_id

Optional agent ID for error context.

agent_uri

Optional agent URI for error context.

Raises

ADCPFeatureUnsupportedError

If any features are not supported.

def supports(self, feature: str) ‑> bool

Expand source code

def supports(self, feature: str) -> bool:
    """Check if a feature is supported."""
    caps = self._caps

    # Extension check: "ext:scope3"
    if feature.startswith("ext:"):
        ext_name = feature[4:]
        if caps.extensions_supported is None:
            return False
        return any(item.root == ext_name for item in caps.extensions_supported)

    # Targeting check: "targeting.geo_countries"
    if feature.startswith("targeting."):
        attr_name = feature[len("targeting.") :]
        if caps.media_buy is None or caps.media_buy.execution is None:
            return False
        targeting = caps.media_buy.execution.targeting
        if targeting is None:
            return False
        if attr_name not in type(targeting).model_fields:
            return False
        val = getattr(targeting, attr_name, None)
        # For bool fields, check truthiness. For object fields (like geo_metros),
        # presence means supported.
        return val is not None and val is not False

    # Protocol check: if the string is a known protocol name, resolve it
    # against supported_protocols and stop — don't fall through to features.
    if feature in self._declared_protocols:
        return True
    if feature in self._valid_protocols:
        return False

    # Media buy features check
    if caps.media_buy is not None and caps.media_buy.features is not None:
        if feature in type(caps.media_buy.features).model_fields:
            val = getattr(caps.media_buy.features, feature, None)
            if val is True:
                return True

    # Signals features check
    if caps.signals is not None and caps.signals.features is not None:
        if feature in type(caps.signals.features).model_fields:
            val = getattr(caps.signals.features, feature, None)
            if val is True:
                return True

    return False

Check if a feature is supported.

def supports_v3(self) ‑> bool

Expand source code

def supports_v3(self) -> bool:
    """Check if the seller supports ADCP v3.

    Returns:
        True if major_versions includes 3.
    """
    for v in self._caps.adcp.major_versions:
        if (v.root if hasattr(v, "root") else v) == 3:
            return True
    return False

Check if the seller supports ADCP v3.

Returns

True if major_versions includes 3.