Module `adcp.types.base`

Global variables

var SchemaInt: The annotation for a schema's type: integer. Accepts an int and a float with no fractional part; refuses a fractional float, a bool and a numeric string, which is exactly what the bundled validator does for the same field. The generator marks every such field and scripts/post_generate_fixes.py points the marker here.
var WireUrl: A format: uri string whose bytes are part of an identity. Validated as a URL, carried as the str the wire delivered. scripts/post_generate_fixes.py points the format-reference agent_url here.

Classes

class AdCPBaseModel (**data: Any)

Expand source code

class AdCPBaseModel(BaseModel):
    """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.
    """

    # ``defer_build=True`` skips building each model's pydantic-core
    # validator/serializer at class-definition time. With ~700 generated model
    # modules, eager builds dominate ``import adcp`` memory; deferring means each
    # model's core schema is built lazily on first validate/serialize, so only
    # the handful of models actually used are paid for.
    model_config = ConfigDict(extra=_EXTRA_POLICY, defer_build=True)

    # Pydantic derives serialization JSON Schema from the model fields only when
    # this serializer's return is unannotated. Even ``-> Any`` erases the shape.
    @model_serializer(mode="wrap")
    def _notification_config_wire_defaults(  # type: ignore[no-untyped-def]
        self, handler: SerializerFunctionWrapHandler
    ):
        """Retain the unrelated product default only for product subscriptions.

        Reporting capability defaults and tier validation live in their own
        generated models; this wrapper does not mask reporting attributes.
        """
        value = handler(self)
        if not isinstance(value, dict):
            return value
        # Bundled schemas can rename or inline the same wire shape. Match its
        # declared fields rather than generated class/module names; supplied
        # values still come exclusively from this instance's fields-set.
        fields = type(self).model_fields
        if {"subscriber_id", "url", "event_types", "product_payload_view"} <= fields.keys():
            events = getattr(self, "event_types", ())
            if "product_payload_view" not in self.model_fields_set and not any(
                str(getattr(e, "value", e)).startswith("product.") for e in events
            ):
                value.pop("product_payload_view", None)
        return value

    def model_dump(self, **kwargs: Any) -> dict[str, Any]:
        # ``serialize_as_any=True`` makes Pydantic dispatch on the runtime type of
        # nested values rather than the declared schema, so subclass
        # ``@model_serializer`` overrides fire from a base-typed parent field. Combined
        # with ``Field(exclude=True)`` on internal fields (which already works at every
        # nesting depth), this removes the parent-side ``model_dump`` boilerplate that
        # adopters previously needed to write per response type. See
        # docs/extending-types.md.
        if "exclude_none" not in kwargs:
            kwargs["exclude_none"] = True
        if "serialize_as_any" not in kwargs:
            kwargs["serialize_as_any"] = True
        try:
            return super().model_dump(**kwargs)
        except (TypeError, PydanticSerializationError) as exc:
            if "MockValSer" not in str(exc):
                raise
            _build_deferred_serializers(self, set())
            return super().model_dump(**kwargs)

    def model_dump_json(self, **kwargs: Any) -> str:
        if "exclude_none" not in kwargs:
            kwargs["exclude_none"] = True
        if "serialize_as_any" not in kwargs:
            kwargs["serialize_as_any"] = True
        try:
            return super().model_dump_json(**kwargs)
        except (TypeError, PydanticSerializationError) as exc:
            if "MockValSer" not in str(exc):
                raise
            _build_deferred_serializers(self, set())
            return super().model_dump_json(**kwargs)

    def model_summary(self) -> str:
        """Human-readable summary for protocol responses.

        Returns a standardized human-readable message suitable for MCP tool
        results, A2A task communications, and REST API responses.

        For types without a registered formatter, returns a generic message
        with the class name.
        """
        formatter = _RESPONSE_MESSAGE_REGISTRY.get(self.__class__.__name__)
        if formatter:
            return formatter(self)
        return f"{self.__class__.__name__} response"

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

pydantic.main.BaseModel

Subclasses

Class variables

var model_config

Methods

def model_dump(self, **kwargs: Any) ‑> dict[str, typing.Any]

Expand source code

def model_dump(self, **kwargs: Any) -> dict[str, Any]:
    # ``serialize_as_any=True`` makes Pydantic dispatch on the runtime type of
    # nested values rather than the declared schema, so subclass
    # ``@model_serializer`` overrides fire from a base-typed parent field. Combined
    # with ``Field(exclude=True)`` on internal fields (which already works at every
    # nesting depth), this removes the parent-side ``model_dump`` boilerplate that
    # adopters previously needed to write per response type. See
    # docs/extending-types.md.
    if "exclude_none" not in kwargs:
        kwargs["exclude_none"] = True
    if "serialize_as_any" not in kwargs:
        kwargs["serialize_as_any"] = True
    try:
        return super().model_dump(**kwargs)
    except (TypeError, PydanticSerializationError) as exc:
        if "MockValSer" not in str(exc):
            raise
        _build_deferred_serializers(self, set())
        return super().model_dump(**kwargs)

Usage Documentation

model_dump

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Args
-----=
mode: The mode in which to_python should run. If mode is 'json', the output will only contain JSON serializable types. If mode is 'python', the output may contain non-JSON-serializable Python objects.
include: A set of fields to include in the output.
exclude: A set of fields to exclude from the output.
context: Additional context to pass to the serializer.
by_alias: Whether to use the field's alias in the dictionary key if defined.
exclude_unset: Whether to exclude fields that have not been explicitly set.
exclude_defaults: Whether to exclude fields that are set to their default value.
exclude_none: Whether to exclude fields that have a value of None.
exclude_computed_fields: Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.
round_trip: If True, dumped values should be valid as input for non-idempotent types such as Json[T].
warnings: How to handle serialization errors. False/"none" ignores them, True/"warn" logs errors, "error" raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].
fallback: A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.
serialize_as_any: Whether to serialize fields with duck-typing serialization behavior.
polymorphic_serialization: Whether to use model and dataclass polymorphic serialization for this call.

Returns -----= A dictionary representation of the model.

def model_dump_json(self, **kwargs: Any) ‑> str

Expand source code

def model_dump_json(self, **kwargs: Any) -> str:
    if "exclude_none" not in kwargs:
        kwargs["exclude_none"] = True
    if "serialize_as_any" not in kwargs:
        kwargs["serialize_as_any"] = True
    try:
        return super().model_dump_json(**kwargs)
    except (TypeError, PydanticSerializationError) as exc:
        if "MockValSer" not in str(exc):
            raise
        _build_deferred_serializers(self, set())
        return super().model_dump_json(**kwargs)

Usage Documentation

model_dump_json

Generates a JSON representation of the model using Pydantic's to_json method.

Args
-----=
indent: Indentation to use in the JSON output. If None is passed, the output will be compact.
ensure_ascii: If True, the output is guaranteed to have all incoming non-ASCII characters escaped. If False (the default), these characters will be output as-is.
include: Field(s) to include in the JSON output.
exclude: Field(s) to exclude from the JSON output.
context: Additional context to pass to the serializer.
by_alias: Whether to serialize using field aliases.
exclude_unset: Whether to exclude fields that have not been explicitly set.
exclude_defaults: Whether to exclude fields that are set to their default value.
exclude_none: Whether to exclude fields that have a value of None.
exclude_computed_fields: Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.
round_trip: If True, dumped values should be valid as input for non-idempotent types such as Json[T].
warnings: How to handle serialization errors. False/"none" ignores them, True/"warn" logs errors, "error" raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].
fallback: A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.
serialize_as_any: Whether to serialize fields with duck-typing serialization behavior.
polymorphic_serialization: Whether to use model and dataclass polymorphic serialization for this call.

Returns -----= A JSON string representation of the model.

def model_summary(self) ‑> str

Expand source code

def model_summary(self) -> str:
    """Human-readable summary for protocol responses.

    Returns a standardized human-readable message suitable for MCP tool
    results, A2A task communications, and REST API responses.

    For types without a registered formatter, returns a generic message
    with the class name.
    """
    formatter = _RESPONSE_MESSAGE_REGISTRY.get(self.__class__.__name__)
    if formatter:
        return formatter(self)
    return f"{self.__class__.__name__} response"

Human-readable summary for protocol responses.

Returns a standardized human-readable message suitable for MCP tool results, A2A task communications, and REST API responses.

For types without a registered formatter, returns a generic message with the class name.

class AdcpRequest

Expand source code

class AdcpRequest(_AdcpMessage):
    """The request message of a task in the pinned bundle's task registry.

    A consumer holding one can resolve its account, decide at-most-once, echo its
    context and negotiate version -- the whole transport-boundary job -- before knowing
    which tool it is. ``issubclass(model, AdcpRequest)`` is the registration-time proof
    that a model is spec-derived rather than a hand-written parallel: a field test passes
    for a forged model, descent does not.

    Each accessor returns the field's value, or ``None`` when this tool's schema declares
    no such field. Only 49 of the 87 request schemas declare an ``account`` and only 43 an
    ``idempotency_key``, so asking the request is what replaces
    ``getattr(req, "account", None)`` against ``Any`` at the boundary.
    """

    def get_account(
        self,
    ) -> AccountReference1 | AccountReference2 | CanonicalAccountReference | None:
        """The account this request names, or None when its schema declares none.

        The union is what the generated ``account`` fields actually hold, measured: 44
        request classes type it as the two ``core/account-ref.json`` arms directly (the
        ``AccountReference`` RootModel is unwrapped at the field by
        ``expose_account_reference_union_fields``, so naming the wrapper here would be a
        type no value ever has), and 8 as the ``CanonicalAccountReference`` root. The one
        inline declarer, ``compliance/comply-test-controller-request.json``, answers with
        its own generated ``Account`` through the same instance-dict read.
        """
        return self.__dict__.get("account")

    def get_idempotency_key(self) -> str | None:
        """The at-most-once key this request carries, or None when its schema declares none.

        Presence is the honest structural signal for write-versus-read: a read is
        idempotent by construction and takes no key. ``x-mutates-state`` is not a
        substitute -- it covers 44 of 87 schemas and contradicts the key in both
        directions on 2 schemas each way.
        """
        return self.__dict__.get("idempotency_key")

    def get_context(self) -> ContextObject | None:
        """The buyer's opaque ``context``, echoed unchanged onto whatever leaves."""
        return self.__dict__.get("context")

    def get_push_notification_config(self) -> PushNotificationConfig | None:
        """The webhook configuration this request asks for, or None (19 of 87 declare it)."""
        return self.__dict__.get("push_notification_config")

    def get_adcp_version(self) -> str | None:
        """The release this buyer pins, or None when it pinned none.

        Answers from the instance dict, so it still answers for a schema that inlines the
        field without composing the envelope -- the one case ``version_fields()`` reports
        empty.
        """
        return self.__dict__.get("adcp_version")

    def get_adcp_major_version(self) -> int | None:
        """The major this buyer pins (deprecated through 3.x), or None."""
        return self.__dict__.get("adcp_major_version")

The request message of a task in the pinned bundle's task registry.

A consumer holding one can resolve its account, decide at-most-once, echo its context and negotiate version – the whole transport-boundary job – before knowing which tool it is. issubclass(model, AdcpRequest) is the registration-time proof that a model is spec-derived rather than a hand-written parallel: a field test passes for a forged model, descent does not.

Each accessor returns the field's value, or None when this tool's schema declares no such field. Only 49 of the 87 request schemas declare an account and only 43 an idempotency_key, so asking the request is what replaces getattr(req, "account", None) against Any at the boundary.

Ancestors

adcp.types.base._AdcpMessage

Subclasses

Methods

def get_account(self) ‑> AccountReference1 | AccountReference2 | CanonicalAccountReference | None

Expand source code

def get_account(
    self,
) -> AccountReference1 | AccountReference2 | CanonicalAccountReference | None:
    """The account this request names, or None when its schema declares none.

    The union is what the generated ``account`` fields actually hold, measured: 44
    request classes type it as the two ``core/account-ref.json`` arms directly (the
    ``AccountReference`` RootModel is unwrapped at the field by
    ``expose_account_reference_union_fields``, so naming the wrapper here would be a
    type no value ever has), and 8 as the ``CanonicalAccountReference`` root. The one
    inline declarer, ``compliance/comply-test-controller-request.json``, answers with
    its own generated ``Account`` through the same instance-dict read.
    """
    return self.__dict__.get("account")

The account this request names, or None when its schema declares none.

The union is what the generated account fields actually hold, measured: 44 request classes type it as the two core/account-ref.json arms directly (the AccountReference RootModel is unwrapped at the field by expose_account_reference_union_fields, so naming the wrapper here would be a type no value ever has), and 8 as the CanonicalAccountReference root. The one inline declarer, compliance/comply-test-controller-request.json, answers with its own generated Account through the same instance-dict read.

def get_adcp_major_version(self) ‑> int | None

Expand source code

def get_adcp_major_version(self) -> int | None:
    """The major this buyer pins (deprecated through 3.x), or None."""
    return self.__dict__.get("adcp_major_version")

The major this buyer pins (deprecated through 3.x), or None.

def get_adcp_version(self) ‑> str | None

Expand source code

def get_adcp_version(self) -> str | None:
    """The release this buyer pins, or None when it pinned none.

    Answers from the instance dict, so it still answers for a schema that inlines the
    field without composing the envelope -- the one case ``version_fields()`` reports
    empty.
    """
    return self.__dict__.get("adcp_version")

The release this buyer pins, or None when it pinned none.

Answers from the instance dict, so it still answers for a schema that inlines the field without composing the envelope – the one case version_fields() reports empty.

def get_context(self) ‑> ContextObject | None

Expand source code

def get_context(self) -> ContextObject | None:
    """The buyer's opaque ``context``, echoed unchanged onto whatever leaves."""
    return self.__dict__.get("context")

The buyer's opaque context, echoed unchanged onto whatever leaves.

def get_idempotency_key(self) ‑> str | None

Expand source code

def get_idempotency_key(self) -> str | None:
    """The at-most-once key this request carries, or None when its schema declares none.

    Presence is the honest structural signal for write-versus-read: a read is
    idempotent by construction and takes no key. ``x-mutates-state`` is not a
    substitute -- it covers 44 of 87 schemas and contradicts the key in both
    directions on 2 schemas each way.
    """
    return self.__dict__.get("idempotency_key")

The at-most-once key this request carries, or None when its schema declares none.

Presence is the honest structural signal for write-versus-read: a read is idempotent by construction and takes no key. x-mutates-state is not a substitute – it covers 44 of 87 schemas and contradicts the key in both directions on 2 schemas each way.

def get_push_notification_config(self) ‑> PushNotificationConfig | None

Expand source code

def get_push_notification_config(self) -> PushNotificationConfig | None:
    """The webhook configuration this request asks for, or None (19 of 87 declare it)."""
    return self.__dict__.get("push_notification_config")

The webhook configuration this request asks for, or None (19 of 87 declare it).

class AdcpResponse

Expand source code

class AdcpResponse(_AdcpMessage):
    """The response message of a task in the pinned bundle's task registry.

    A consumer holding one can route on task state, pick up an async ``task_id``, split
    envelope from payload and log uniformly, before knowing which tool answered. Which
    makes one generic poll-to-terminal loop possible for all 77 tasks, where today each
    arm has no common type at all.
    """

    def get_status(self) -> TaskStatus | None:
        """The AdCP task state the seller asserted, or None.

        None only for the 10 task responses whose schemas do not compose
        ``core/protocol-envelope.json``, against the envelope's own "REQUIRED on every
        task response envelope" -- a residue this SDK cannot invent its way out of.
        """
        return self.__dict__.get("status")

    def get_task_id(self) -> str | None:
        """The async operation identifier, present when the task needs polling."""
        return self.__dict__.get("task_id")

    def get_context(self) -> ContextObject | None:
        """The caller's ``context``, echoed back byte-for-byte."""
        return self.__dict__.get("context")

    def get_adcp_error(self) -> Error | None:
        """The envelope-level typed error for a fatal task failure, or None.

        The payload's ``errors[]`` array is a different field by design and is read from
        the concrete arm; the two MUST be treated as distinct by name.
        """
        return self.__dict__.get("adcp_error")

    def get_message(self) -> str | None:
        """The human-readable summary of the result, or None."""
        return self.__dict__.get("message")

    def get_replayed(self) -> bool | None:
        """True when this answer came from the idempotency cache.

        ``False`` when the response declares the field and was executed fresh, ``None``
        when the tool's schema does not declare it -- absence stays distinguishable from
        a negative answer.
        """
        return self.__dict__.get("replayed")

The response message of a task in the pinned bundle's task registry.

A consumer holding one can route on task state, pick up an async task_id, split envelope from payload and log uniformly, before knowing which tool answered. Which makes one generic poll-to-terminal loop possible for all 77 tasks, where today each arm has no common type at all.

Ancestors

adcp.types.base._AdcpMessage

Subclasses

Methods

def get_adcp_error(self) ‑> Error | None

Expand source code

def get_adcp_error(self) -> Error | None:
    """The envelope-level typed error for a fatal task failure, or None.

    The payload's ``errors[]`` array is a different field by design and is read from
    the concrete arm; the two MUST be treated as distinct by name.
    """
    return self.__dict__.get("adcp_error")

The envelope-level typed error for a fatal task failure, or None.

The payload's errors[] array is a different field by design and is read from the concrete arm; the two MUST be treated as distinct by name.

def get_context(self) ‑> ContextObject | None

Expand source code

def get_context(self) -> ContextObject | None:
    """The caller's ``context``, echoed back byte-for-byte."""
    return self.__dict__.get("context")

The caller's context, echoed back byte-for-byte.

def get_message(self) ‑> str | None

Expand source code

def get_message(self) -> str | None:
    """The human-readable summary of the result, or None."""
    return self.__dict__.get("message")

The human-readable summary of the result, or None.

def get_replayed(self) ‑> bool | None

Expand source code

def get_replayed(self) -> bool | None:
    """True when this answer came from the idempotency cache.

    ``False`` when the response declares the field and was executed fresh, ``None``
    when the tool's schema does not declare it -- absence stays distinguishable from
    a negative answer.
    """
    return self.__dict__.get("replayed")

True when this answer came from the idempotency cache.

False when the response declares the field and was executed fresh, None when the tool's schema does not declare it – absence stays distinguishable from a negative answer.

def get_status(self) ‑> TaskStatus | None

Expand source code

def get_status(self) -> TaskStatus | None:
    """The AdCP task state the seller asserted, or None.

    None only for the 10 task responses whose schemas do not compose
    ``core/protocol-envelope.json``, against the envelope's own "REQUIRED on every
    task response envelope" -- a residue this SDK cannot invent its way out of.
    """
    return self.__dict__.get("status")

The AdCP task state the seller asserted, or None.

None only for the 10 task responses whose schemas do not compose core/protocol-envelope.json, against the envelope's own "REQUIRED on every task response envelope" – a residue this SDK cannot invent its way out of.

def get_task_id(self) ‑> str | None

Expand source code

def get_task_id(self) -> str | None:
    """The async operation identifier, present when the task needs polling."""
    return self.__dict__.get("task_id")

The async operation identifier, present when the task needs polling.

class RegistryBaseModel (**data: Any)

Expand source code

class RegistryBaseModel(BaseModel):
    """Base model for registry API types.

    Uses ``extra='allow'`` so that new fields from the registry API
    are preserved rather than dropped. This differs from AdCPBaseModel
    which defaults to ``extra='ignore'`` for protocol types.
    """

    model_config = ConfigDict(extra="allow")

Base model for registry API types.

Uses extra='allow' so that new fields from the registry API are preserved rather than dropped. This differs from AdCPBaseModel which defaults to extra='ignore' for protocol types.

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

pydantic.main.BaseModel

Subclasses

Class variables

var model_config