Module adcp.decisioning.resolve

Async framework-mediated resource resolver for :class:RequestContext.

Defines:

• :class:ResourceResolver — Protocol for async fetches of framework-validated resources (property lists, collection lists, creative formats). The framework owns the cache + validation; platform methods get pre-validated typed results.
• :class:_NotYetWiredResolver — v6.0 stub. Raises :class:NotImplementedError on every call with a pointer to the v6.1 follow-up. Asymmetry vs. the state stub (which returns empty + warns) is deliberate: an empty :class:PropertyListReference in v6.0 vs. a real one in v6.1 is divergence the framework cannot silently paper over. See docs/proposals/decisioning-platform-dispatch-design.md#d15.

The :class:Format and :class:PropertyListReference types are re-exported from :mod:adcp.types.domains so adopters import once from :mod:adcp.decisioning. :class:PropertyListReference and :class:CollectionList use the spec-defined wire shapes; the resolver returns the same Pydantic models adopters would construct themselves.

Classes

class CollectionList (**data: Any)

Expand source code

class CollectionList(AdCPBaseModel):
    model_config = ConfigDict(
        extra='forbid',
    )
    list_id: Annotated[str, Field(description='Unique identifier for this collection list')]
    name: Annotated[str, Field(description='Human-readable name for the list')]
    description: Annotated[str | None, Field(description="Description of the list's purpose")] = (
        None
    )
    account: Annotated[
        account_ref.AccountReference | None,
        Field(
            description='Account that owns this list. Returned as account_id form (seller-assigned identifier).'
        ),
    ] = None
    base_collections: Annotated[
        list[base_collection_source.BaseCollectionSource] | None,
        Field(
            description="Array of collection sources to evaluate. Each entry is a discriminated union: distribution_ids (platform-independent identifiers), publisher_collections (publisher_domain + collection_ids), or publisher_genres (publisher_domain + genres). If omitted, queries the agent's entire collection database."
        ),
    ] = None
    filters: Annotated[
        collection_list_filters.CollectionListFilters | None,
        Field(description='Dynamic filters applied when resolving the list'),
    ] = None
    brand: Annotated[
        brand_ref.BrandReference | None,
        Field(
            description='Brand reference used to automatically apply appropriate rules. Resolved to full brand identity at execution time.'
        ),
    ] = None
    webhook_url: Annotated[
        AnyUrl | None,
        Field(description='URL to receive notifications when the resolved list changes'),
    ] = None
    cache_duration_hours: Annotated[
        SchemaInt | None,
        Field(
            description='Recommended cache duration for resolved list. Consumers should re-fetch after this period. Defaults to 168 (one week) because collection metadata changes less frequently than property metadata.',
            ge=1,
        ),
    ] = 168
    created_at: Annotated[AwareDatetime | None, Field(description='When the list was created')] = (
        None
    )
    updated_at: Annotated[
        AwareDatetime | None, Field(description='When the list was last modified')
    ] = None
    collection_count: Annotated[
        SchemaInt | None,
        Field(
            description='Number of collections in the resolved list (at time of last resolution)'
        ),
    ] = 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 account : AccountReference1 | AccountReference2 | None

var base_collections : list[BaseCollectionSource1 | BaseCollectionSource2 | BaseCollectionSource3] | None

var brand : BrandReference | None

var cache_duration_hours : int | None

var collection_count : int | None

var created_at : pydantic.types.AwareDatetime | None

var description : str | None

var filters : CollectionListFilters | None

var list_id : str

var model_config

var name : str

var updated_at : pydantic.types.AwareDatetime | None

var webhook_url : pydantic.networks.AnyUrl | None

Inherited members

• AdCPBaseModel:
  • model_dump
  • model_dump_json
  • model_summary

class Format (**data: Any)

Expand source code

class Format(CanonicalBoundaryModel):
    """Canonical format declaration exposed as ``adcp.Format``."""

    format_option_id: str | None = Field(
        default=None,
        description="Stable option identifier within the product or publisher namespace.",
    )
    publisher_domain: str | None = Field(
        default=None,
        pattern=r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)*$",
    )
    display_name: str | None = None
    applies_to_channels: list[MediaChannel] | None = None
    seller_preference: SellerPreference | None = None
    canonical_formats_only: bool | None = None
    experimental: bool | None = None
    format_shape: str | None = None
    format_schema: PlatformExtensionReference | None = None
    format_kind: str
    params: dict[str, Any]

    _legacy_format_refs: list[LegacyFormatId] = PrivateAttr(default_factory=list)

    @model_validator(mode="before")
    @classmethod
    def _reject_legacy_conflicts_and_credentials(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        if data.get("canonical_formats_only") is True and data.get("v1_format_ref"):
            raise ValueError(
                "canonical_formats_only=True is mutually exclusive with legacy v1_format_ref"
            )
        for bag_name, bag in (
            ("params", data.get("params")),
            (
                "extras",
                {
                    key: value
                    for key, value in data.items()
                    if key not in cls.model_fields and key != "v1_format_ref"
                },
            ),
        ):
            found = _walk_for_credential_keys(bag, path=bag_name)
            if found is not None:
                raise ValueError(
                    f"{found!r} matches a credential-shaped key suffix and cannot "
                    "be stored in a canonical format declaration"
                )
        return data

    def __init__(self, **data: Any) -> None:
        refs = data.get("v1_format_ref")
        if "capability_id" in data and "format_option_id" not in data:
            data["format_option_id"] = data.pop("capability_id")
        super().__init__(**data)
        if self.__pydantic_extra__ is not None:
            self.__pydantic_extra__.pop("v1_format_ref", None)
        if refs:
            self._legacy_format_refs = [
                LegacyFormatId.model_validate(copy.deepcopy(ref)) for ref in refs
            ]

    @property
    def legacy_format_refs(self) -> tuple[LegacyFormatId, ...]:
        """Original tuples retained only for an explicit compatibility adapter."""

        return tuple(copy.deepcopy(ref) for ref in self._legacy_format_refs)

    def params_as(self, canonical_type: type[_CanonicalParamsT]) -> _CanonicalParamsT:
        """Validate the open parameter bag against a typed canonical model."""

        return canonical_type.model_validate(self.params)

    @model_validator(mode="after")
    def _validate_custom_shape(self) -> Format:
        if self.format_kind == CanonicalFormatKind.custom.value:
            if not self.format_shape:
                raise ValueError("custom formats require format_shape")
            if self.format_schema is None:
                raise ValueError("custom formats require format_schema")
        elif self.format_shape is not None or self.format_schema is not None:
            raise ValueError("format_shape and format_schema are only valid for custom formats")
        return self

Canonical format declaration exposed as Format.

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

• CanonicalBoundaryModel
• AdCPBaseModel
• pydantic.main.BaseModel

Class variables

var applies_to_channels : list[MediaChannel] | None

var canonical_formats_only : bool | None

var display_name : str | None

var experimental : bool | None

var format_kind : str

var format_option_id : str | None

var format_schema : PlatformExtensionReference | None

var format_shape : str | None

var model_config

var params : dict[str, typing.Any]

var publisher_domain : str | None

var seller_preference : SellerPreference | None

Instance variables

prop legacy_format_refs : tuple[FormatReferenceStructuredObject, ...]

Expand source code

@property
def legacy_format_refs(self) -> tuple[LegacyFormatId, ...]:
    """Original tuples retained only for an explicit compatibility adapter."""

    return tuple(copy.deepcopy(ref) for ref in self._legacy_format_refs)

Original tuples retained only for an explicit compatibility adapter.

Methods

def model_post_init(self: BaseModel, context: Any, /) ‑> None

Expand source code

def init_private_attributes(self: BaseModel, context: Any, /) -> None:
    """This function is meant to behave like a BaseModel method to initialize private attributes.

    It takes context as an argument since that's what pydantic-core passes when calling it.

    Args:
        self: The BaseModel instance.
        context: The context.
    """
    if getattr(self, '__pydantic_private__', None) is None:
        pydantic_private = {}
        for name, private_attr in self.__private_attributes__.items():
            # Avoid needlessly creating a new dict for the validated data:
            if private_attr.default_factory_takes_validated_data:
                default = private_attr.get_default(
                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
                )
            else:
                default = private_attr.get_default(call_default_factory=True)
            if default is not PydanticUndefined:
                pydantic_private[name] = default
        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args

-----=

self

The BaseModel instance.

context

The context.

def params_as(self, canonical_type: type[_CanonicalParamsT]) ‑> ~_CanonicalParamsT

Expand source code

def params_as(self, canonical_type: type[_CanonicalParamsT]) -> _CanonicalParamsT:
    """Validate the open parameter bag against a typed canonical model."""

    return canonical_type.model_validate(self.params)

Validate the open parameter bag against a typed canonical model.

Inherited members

• CanonicalBoundaryModel:
  • model_dump
  • model_dump_json
  • model_json_schema
  • model_summary

class LegacyFormatId (**data: Any)

Expand source code

class FormatReferenceStructuredObject(AdCPBaseModel):
    model_config = ConfigDict(
        extra='allow',
    )
    agent_url: Annotated[
        WireUrl,
        Field(
            description="URL of the agent that defines this format (e.g., 'https://creative.adcontextprotocol.org' for standard formats, or 'https://publisher.com/.well-known/adcp/sales' for custom formats). Callers comparing two `format-id` values MUST canonicalize `agent_url` per the AdCP URL canonicalization rules before treating two formats as the same. See docs/reference/url-canonicalization."
        ),
    ]
    id: Annotated[
        str,
        Field(
            description="Format identifier within the agent's namespace (e.g., 'display_static', 'video_hosted', 'audio_standard'). When used alone, references a template format. When combined with dimension/duration fields, creates a parameterized format ID for a specific variant.",
            pattern='^[a-zA-Z0-9_-]+$',
        ),
    ]
    width: Annotated[
        SchemaInt | None,
        Field(
            description='Width in pixels for visual formats. When specified, height must also be specified. Both fields together create a parameterized format ID for dimension-specific variants.',
            ge=1,
        ),
    ] = None
    height: Annotated[
        SchemaInt | None,
        Field(
            description='Height in pixels for visual formats. When specified, width must also be specified. Both fields together create a parameterized format ID for dimension-specific variants.',
            ge=1,
        ),
    ] = None
    duration_ms: Annotated[
        StrictFloat | None,
        Field(
            description='Duration in milliseconds for time-based formats (video, audio). When specified, creates a parameterized format ID. Omit to reference a template format without parameters.',
            ge=1.0,
        ),
    ] = None
    pixel_ratio: Annotated[
        StrictFloat | None,
        Field(
            description='Required intrinsic-pixel density for a parameterized visual format, expressed as intrinsic pixels per logical pixel. Requires `width` and `height`. Example: `{id: "display_image", width: 300, height: 250, pixel_ratio: 2}` identifies a 300×250 logical render supplied by a 600×500 image. Omit for the backward-compatible 1x variant.',
            gt=0.0,
        ),
    ] = 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 agent_url : str

var duration_ms : float | None

var height : int | None

var id : str

var model_config

var pixel_ratio : float | None

var width : int | None

Inherited members

• AdCPBaseModel:
  • model_dump
  • model_dump_json
  • model_summary

class PropertyList (**data: Any)

Expand source code

class PropertyListReference(AdCPBaseModel):
    model_config = ConfigDict(
        extra='forbid',
    )
    agent_url: Annotated[AnyUrl, Field(description='URL of the agent managing the property list')]
    list_id: Annotated[
        str, Field(description='Identifier for the property list within the agent', min_length=1)
    ]
    auth_token: Annotated[
        str | None,
        Field(
            description='JWT or other authorization token for accessing the list. Optional if the list is public or caller has implicit access.'
        ),
    ] = 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 agent_url : pydantic.networks.AnyUrl

var auth_token : str | None

var list_id : str

var model_config

class PropertyListReference (**data: Any)

Expand source code

class PropertyListReference(AdCPBaseModel):
    model_config = ConfigDict(
        extra='forbid',
    )
    agent_url: Annotated[AnyUrl, Field(description='URL of the agent managing the property list')]
    list_id: Annotated[
        str, Field(description='Identifier for the property list within the agent', min_length=1)
    ]
    auth_token: Annotated[
        str | None,
        Field(
            description='JWT or other authorization token for accessing the list. Optional if the list is public or caller has implicit access.'
        ),
    ] = 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 agent_url : pydantic.networks.AnyUrl

var auth_token : str | None

var list_id : str

var model_config

Inherited members

• AdCPBaseModel:
  • model_dump
  • model_dump_json
  • model_summary

class ResourceResolver (*args, **kwargs)

Expand source code

@runtime_checkable
class ResourceResolver(Protocol):
    """Async fetches of framework-mediated resources.

    Platforms call ``ctx.resolve.property_list(list_id)`` instead of
    fetching from their own DB; the framework returns a validated
    typed result. The resolver routes through
    ``capabilities.creative_agents`` for creative-format reads, hits
    the framework's local ``CreativePlatform.list_formats`` for
    self-hosted formats, and reads the seller's declared property /
    collection lists with id-validation built in.

    Framework-supplied; never constructed by adopter code. The
    ``RequestContext.resolve`` field is populated by the dispatch
    hydration helper. Adopters substituting test doubles use
    :func:`dataclasses.replace` on the context, not direct
    construction.

    Mirrors the TS-side ``ResourceResolver`` interface in
    ``src/lib/server/decisioning/context.ts``. v6.0 ships the contract
    + the no-op stub (raises ``NotImplementedError`` on every call);
    v6.1 lands the backing fetchers.

    .. note::
       :class:`runtime_checkable` Protocols only check attribute
       *presence*. Whether a method is ``async def`` is irrelevant to
       the runtime ``isinstance`` check — a sync method named
       ``property_list`` would pass the structural check but fail at
       ``await`` time. Use mypy to enforce ``async def`` signatures
       across adopter impls.
    """

    async def property_list(self, list_id: str) -> PropertyList:
        """Fetch a property list by id. Framework validates the id
        exists in the seller's declared lists before returning;
        consumers can trust the result."""
        ...

    async def collection_list(self, list_id: str) -> CollectionList:
        """Fetch a collection list by id. Same id-validation
        guarantee as :meth:`property_list`."""
        ...

    async def creative_format(
        self,
        format_id: LegacyFormatId,
        *,
        revalidate: bool = False,
    ) -> Format:
        """Fetch a creative format definition.

        Routes through ``capabilities.creative_agents`` declaration
        with a framework-managed cache; self-hosted formats hit the
        local ``CreativePlatform.list_formats``. Returns the resolved
        :class:`Format` with full asset slot definitions.

        :param revalidate: When ``True``, bypasses the framework cache
            and re-fetches from the upstream creative-agent. Adopters
            with freshness needs (e.g., creative submission validating
            against the latest format spec) pass ``revalidate=True``;
            most reads use the default (``False``) to amortize the
            agent round-trip.

        Cache TTL is implementation detail (defaults to 1h on the
        reference impl); adopters who need stricter freshness use
        ``revalidate=True`` rather than depending on the TTL value.
        """
        ...

Async fetches of framework-mediated resources.

Platforms call ctx.resolve.property_list(list_id) instead of fetching from their own DB; the framework returns a validated typed result. The resolver routes through capabilities.creative_agents for creative-format reads, hits the framework's local CreativePlatform.list_formats for self-hosted formats, and reads the seller's declared property / collection lists with id-validation built in.

Framework-supplied; never constructed by adopter code. The RequestContext.resolve field is populated by the dispatch hydration helper. Adopters substituting test doubles use :func:dataclasses.replace on the context, not direct construction.

Mirrors the TS-side ResourceResolver interface in src/lib/server/decisioning/context.ts. v6.0 ships the contract + the no-op stub (raises NotImplementedError on every call); v6.1 lands the backing fetchers.

Note

:class:runtime_checkable Protocols only check attribute presence. Whether a method is async def is irrelevant to the runtime isinstance check — a sync method named property_list would pass the structural check but fail at await time. Use mypy to enforce async def signatures across adopter impls.

Ancestors

• typing.Protocol
• typing.Generic

Methods

async def collection_list(self, list_id: str) ‑> CollectionList

Expand source code

async def collection_list(self, list_id: str) -> CollectionList:
    """Fetch a collection list by id. Same id-validation
    guarantee as :meth:`property_list`."""
    ...

Fetch a collection list by id. Same id-validation guarantee as :meth:property_list.

async def creative_format(self,
format_id: FormatReferenceStructuredObject,
*,
revalidate: bool = False) ‑> Format

Expand source code

async def creative_format(
    self,
    format_id: LegacyFormatId,
    *,
    revalidate: bool = False,
) -> Format:
    """Fetch a creative format definition.

    Routes through ``capabilities.creative_agents`` declaration
    with a framework-managed cache; self-hosted formats hit the
    local ``CreativePlatform.list_formats``. Returns the resolved
    :class:`Format` with full asset slot definitions.

    :param revalidate: When ``True``, bypasses the framework cache
        and re-fetches from the upstream creative-agent. Adopters
        with freshness needs (e.g., creative submission validating
        against the latest format spec) pass ``revalidate=True``;
        most reads use the default (``False``) to amortize the
        agent round-trip.

    Cache TTL is implementation detail (defaults to 1h on the
    reference impl); adopters who need stricter freshness use
    ``revalidate=True`` rather than depending on the TTL value.
    """
    ...

Fetch a creative format definition.

Routes through capabilities.creative_agents declaration with a framework-managed cache; self-hosted formats hit the local CreativePlatform.list_formats. Returns the resolved :class:Format with full asset slot definitions.

:param revalidate: When True, bypasses the framework cache and re-fetches from the upstream creative-agent. Adopters with freshness needs (e.g., creative submission validating against the latest format spec) pass revalidate=True; most reads use the default (False) to amortize the agent round-trip.

Cache TTL is implementation detail (defaults to 1h on the reference impl); adopters who need stricter freshness use revalidate=True rather than depending on the TTL value.

async def property_list(self, list_id: str) ‑> PropertyListReference

Expand source code

async def property_list(self, list_id: str) -> PropertyList:
    """Fetch a property list by id. Framework validates the id
    exists in the seller's declared lists before returning;
    consumers can trust the result."""
    ...

Fetch a property list by id. Framework validates the id exists in the seller's declared lists before returning; consumers can trust the result.