Module adcp.server.base

Base classes for ADCP server implementations.

Defines the ADCPHandler base class and utilities for building ADCP-compliant agents.

Global variables

var TContext

TypeVar bound to :class:ToolContext for parameterising :class:ADCPHandler over a caller-defined context subclass.

Multi-tenant agents typically subclass :class:ToolContext to carry typed tenant/adapter/testing fields the base doesn't name. Declaring class MyAgent(ADCPHandler[MyContext]) makes that subclass visible to every handler method signature — callers get the typed subclass on the context parameter without casting::

@dataclass
class MyContext(ToolContext):
    adapter: MyPlatformAdapter

class MyAgent(ADCPHandler[MyContext]):
    async def get_products(self, params, context: MyContext | None = None):
        if context is not None:
            adapter = context.adapter  # typed, no cast

Handlers that don't subclass ToolContext can still write class MyAgent(ADCPHandler) — unparameterised Generic resolves to ADCPHandler[ToolContext] at runtime (the TypeVar bound), so existing subclasses keep working without edits.

Functions

def not_supported(reason: str = 'This operation is not supported by this agent') ‑> NotImplementedResponse

Expand source code

def not_supported(
    reason: str = "This operation is not supported by this agent",
) -> NotImplementedResponse:
    """Create a standard 'not supported' response.

    Use this to return from operations that your agent does not implement.

    Args:
        reason: Human-readable explanation of why the operation is not supported

    Returns:
        NotImplementedResponse with supported=False
    """
    return NotImplementedResponse(
        supported=False,
        reason=reason,
        error=Error(
            code="NOT_SUPPORTED",
            message=reason,
        ),
    )

Create a standard 'not supported' response.

Use this to return from operations that your agent does not implement.

Args

reason

Human-readable explanation of why the operation is not supported

Returns

NotImplementedResponse with supported=False

Classes

class ADCPHandler

Expand source code

class ADCPHandler(ABC, Generic[TContext]):
    """Base class for ADCP operation handlers.

    Subclass this to implement ADCP operations. All operations have default
    implementations that return 'not supported', allowing you to implement
    only the operations your agent supports.

    Parameterise over a :class:`ToolContext` subclass — ``class MyAgent(ADCPHandler[MyContext])``
    — to get typed ``context`` arguments on every method signature. See
    :data:`TContext` for the pattern.

    For protocol-specific handlers, use:
    - ContentStandardsHandler: For content standards agents
    - SponsoredIntelligenceHandler: For sponsored intelligence agents
    - GovernanceHandler: For governance agents

    **Tool advertisement** (`advertised_tools` class attribute):

    A subclass that introduces a new specialism — i.e., a custom base
    that needs its own ``tools/list`` filter rather than inheriting one
    from a built-in handler — declares the tool set on the class body::

        class PlatformHandler(ADCPHandler):
            advertised_tools: ClassVar[set[str]] = {
                "get_products",
                "create_media_buy",
                ...
            }

    The framework registers ``PlatformHandler -> advertised_tools`` with
    :func:`adcp.server.mcp_tools.register_handler_tools` at class
    definition time. Subclasses that DON'T introduce a new specialism
    (a custom ``MyContentAgent(ContentStandardsHandler)``, for example)
    inherit their parent's tool set unchanged — no class attr needed.

    Hand-written equivalent (no ``advertised_tools`` declaration)::

        from adcp.server.mcp_tools import register_handler_tools
        register_handler_tools("PlatformHandler", {...})

    Either path is fine; codegen targets emit the class attribute so the
    declaration sits next to the class definition.
    """

    _agent_type: str = "this agent"

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Auto-register subclass-declared tool advertisement.

        Reads ``cls.__dict__["advertised_tools"]`` (subclass-defined-only
        — inherited values don't trigger re-registration) and routes
        through :func:`adcp.server.mcp_tools.register_handler_tools`.
        Only fires when the subclass declares the attribute on its own
        class body; intermediate subclasses (multi-level hierarchy)
        register at the level that introduces the attribute.

        The lazy import avoids a base.py ↔ mcp_tools.py circular —
        mcp_tools imports ADCPHandler at module load, so register is
        looked up only when a subclass is actually being created.
        """
        super().__init_subclass__(**kwargs)
        if "advertised_tools" in cls.__dict__:
            from adcp.server.mcp_tools import register_handler_tools

            register_handler_tools(cls.__name__, cls.__dict__["advertised_tools"])

    def _not_supported(self, operation: str) -> NotImplementedResponse:
        """Create a not-supported response that includes the agent type."""
        return not_supported(f"{operation} is not supported by {self._agent_type}")

    # ========================================================================
    # Core Catalog Operations
    # ========================================================================

    async def get_products(
        self, params: GetProductsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Get advertising products.

        Override this to provide product catalog functionality.
        """
        return self._not_supported("get_products")

    async def list_creative_formats(
        self,
        params: ListCreativeFormatsRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """List supported creative formats.

        Override this to provide creative format information.
        """
        return self._not_supported("list_creative_formats")

    # ========================================================================
    # Creative Operations
    # ========================================================================

    async def sync_creatives(
        self, params: SyncCreativesRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Sync creatives.

        Override this to handle creative synchronization.
        """
        return self._not_supported("sync_creatives")

    async def list_creatives(
        self, params: ListCreativesRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """List creatives.

        Override this to list synced creatives.
        """
        return self._not_supported("list_creatives")

    async def build_creative(
        self, params: BuildCreativeRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Build a creative.

        Override this to build creatives from assets.
        """
        return self._not_supported("build_creative")

    async def preview_creative(
        self, params: PreviewCreativeRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Preview a creative rendering.

        Override this to provide creative preview functionality.
        """
        return self._not_supported("preview_creative")

    async def get_creative_delivery(
        self,
        params: GetCreativeDeliveryRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Get creative delivery metrics.

        Override this to provide functionality.
        """
        return self._not_supported("get_creative_delivery")

    async def list_transformers(
        self,
        params: ListTransformersRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """List creative transformers.

        Override this to advertise available creative transformation options.
        """
        return self._not_supported("list_transformers")

    # ========================================================================
    # Media Buy Operations
    # ========================================================================

    async def create_media_buy(
        self, params: CreateMediaBuyRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Create a media buy.

        Override this to handle media buy creation.
        """
        return self._not_supported("create_media_buy")

    async def update_media_buy(
        self, params: UpdateMediaBuyRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Update a media buy.

        Override this to handle media buy updates.
        """
        return self._not_supported("update_media_buy")

    async def get_media_buy_delivery(
        self,
        params: GetMediaBuyDeliveryRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Get media buy delivery metrics.

        Override this to provide delivery reporting.
        """
        return self._not_supported("get_media_buy_delivery")

    async def get_media_buys(
        self, params: GetMediaBuysRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Get media buys with status and optional delivery snapshots.

        Override this to provide media buy listing functionality.
        """
        return self._not_supported("get_media_buys")

    # ========================================================================
    # Signal Operations
    # ========================================================================

    async def get_signals(
        self, params: GetSignalsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Get available signals.

        Override this to provide signal catalog.
        """
        return self._not_supported("get_signals")

    async def activate_signal(
        self, params: ActivateSignalRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Activate a signal.

        Override this to handle signal activation.
        """
        return self._not_supported("activate_signal")

    # ========================================================================
    # Feedback Operations
    # ========================================================================

    async def provide_performance_feedback(
        self,
        params: ProvidePerformanceFeedbackRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Provide performance feedback.

        Override this to handle performance feedback ingestion.
        """
        return self._not_supported("provide_performance_feedback")

    # ========================================================================
    # Account Operations
    # ========================================================================

    async def list_accounts(
        self, params: ListAccountsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """List accounts.

        Override this to provide functionality.
        """
        return self._not_supported("list_accounts")

    async def sync_accounts(
        self, params: SyncAccountsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Sync accounts.

        Override this to provide functionality.
        """
        return self._not_supported("sync_accounts")

    async def get_account_financials(
        self,
        params: GetAccountFinancialsRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Get account financials.

        Override this to provide account financial reporting.
        """
        return self._not_supported("get_account_financials")

    async def report_usage(
        self, params: ReportUsageRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Report account usage.

        Override this to ingest account usage.
        """
        return self._not_supported("report_usage")

    # ========================================================================
    # Event Operations
    # ========================================================================

    async def log_event(
        self, params: LogEventRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Log event.

        Override this to provide functionality.
        """
        return self._not_supported("log_event")

    async def sync_event_sources(
        self, params: SyncEventSourcesRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Sync event sources.

        Override this to provide functionality.
        """
        return self._not_supported("sync_event_sources")

    async def sync_audiences(
        self, params: SyncAudiencesRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Sync audiences.

        Override this to provide audience synchronization.
        """
        return self._not_supported("sync_audiences")

    async def sync_governance(
        self, params: SyncGovernanceRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Sync governance agents for accounts.

        Override this to handle governance agent registration.
        """
        return self._not_supported("sync_governance")

    async def sync_catalogs(
        self, params: SyncCatalogsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Sync catalogs.

        Override this to provide catalog synchronization.
        """
        return self._not_supported("sync_catalogs")

    # ========================================================================
    # V3 Protocol Discovery
    # ========================================================================

    async def get_adcp_capabilities(
        self,
        params: GetAdcpCapabilitiesRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Get ADCP capabilities.

        Override this to advertise your agent's capabilities.
        """
        return self._not_supported("get_adcp_capabilities")

    async def get_task_status(
        self,
        params: GetTaskStatusRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Get task status.

        Override this to expose persisted async task status.
        """
        return self._not_supported("get_task_status")

    async def list_tasks(
        self,
        params: ListTasksRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """List tasks.

        Override this to expose persisted async tasks.
        """
        return self._not_supported("list_tasks")

    # ========================================================================
    # V3 Content Standards Operations
    # ========================================================================

    async def create_content_standards(
        self,
        params: CreateContentStandardsRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Create content standards configuration.

        Override this in ContentStandardsHandler subclasses.
        """
        return self._not_supported("create_content_standards")

    async def get_content_standards(
        self,
        params: GetContentStandardsRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Get content standards configuration.

        Override this in ContentStandardsHandler subclasses.
        """
        return self._not_supported("get_content_standards")

    async def list_content_standards(
        self,
        params: ListContentStandardsRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """List content standards configurations.

        Override this in ContentStandardsHandler subclasses.
        """
        return self._not_supported("list_content_standards")

    async def update_content_standards(
        self,
        params: UpdateContentStandardsRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Update content standards configuration.

        Override this in ContentStandardsHandler subclasses.
        """
        return self._not_supported("update_content_standards")

    async def calibrate_content(
        self, params: CalibrateContentRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Calibrate content against standards.

        Override this in ContentStandardsHandler subclasses.
        """
        return self._not_supported("calibrate_content")

    async def validate_content_delivery(
        self,
        params: ValidateContentDeliveryRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Validate content delivery against standards.

        Override this in ContentStandardsHandler subclasses.
        """
        return self._not_supported("validate_content_delivery")

    async def get_media_buy_artifacts(
        self,
        params: GetMediaBuyArtifactsRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Get artifacts associated with a media buy.

        Override this in ContentStandardsHandler subclasses.
        """
        return self._not_supported("get_media_buy_artifacts")

    # ========================================================================
    # V3 Sponsored Intelligence Operations
    # ========================================================================

    async def si_get_offering(
        self, params: SiGetOfferingRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Get sponsored intelligence offering.

        Override this in SponsoredIntelligenceHandler subclasses.
        """
        return self._not_supported("si_get_offering")

    async def si_initiate_session(
        self, params: SiInitiateSessionRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Initiate sponsored intelligence session.

        Override this in SponsoredIntelligenceHandler subclasses.
        """
        return self._not_supported("si_initiate_session")

    async def si_send_message(
        self, params: SiSendMessageRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Send message in sponsored intelligence session.

        Override this in SponsoredIntelligenceHandler subclasses.
        """
        return self._not_supported("si_send_message")

    async def si_terminate_session(
        self, params: SiTerminateSessionRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Terminate sponsored intelligence session.

        Override this in SponsoredIntelligenceHandler subclasses.
        """
        return self._not_supported("si_terminate_session")

    # ========================================================================
    # V3 Governance Operations
    # ========================================================================

    async def get_creative_features(
        self,
        params: GetCreativeFeaturesRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Evaluate governance features for a creative.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("get_creative_features")

    async def sync_plans(
        self, params: SyncPlansRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Sync campaign governance plans.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("sync_plans")

    async def check_governance(
        self, params: CheckGovernanceRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Check an action against campaign governance.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("check_governance")

    async def report_plan_outcome(
        self, params: ReportPlanOutcomeRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Report the outcome of a governed action.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("report_plan_outcome")

    async def get_plan_audit_logs(
        self, params: GetPlanAuditLogsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Retrieve governance audit logs for plans.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("get_plan_audit_logs")

    async def create_property_list(
        self, params: CreatePropertyListRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Create a property list for governance filtering.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("create_property_list")

    async def get_property_list(
        self, params: GetPropertyListRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Get a property list with optional resolution.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("get_property_list")

    async def list_property_lists(
        self, params: ListPropertyListsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """List property lists.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("list_property_lists")

    async def update_property_list(
        self, params: UpdatePropertyListRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Update a property list.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("update_property_list")

    async def delete_property_list(
        self, params: DeletePropertyListRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Delete a property list.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("delete_property_list")

    # ========================================================================
    # V3 Governance (Collection Lists) Operations
    # ========================================================================

    async def create_collection_list(
        self,
        params: CreateCollectionListRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Create a collection list for governance filtering.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("create_collection_list")

    async def get_collection_list(
        self, params: GetCollectionListRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Get a collection list with optional resolution.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("get_collection_list")

    async def list_collection_lists(
        self,
        params: ListCollectionListsRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """List collection lists.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("list_collection_lists")

    async def update_collection_list(
        self,
        params: UpdateCollectionListRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Update a collection list.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("update_collection_list")

    async def delete_collection_list(
        self,
        params: DeleteCollectionListRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Delete a collection list.

        Override this in GovernanceHandler subclasses.
        """
        return self._not_supported("delete_collection_list")

    # ========================================================================
    # V3 TMP Operations
    # ========================================================================

    async def context_match(
        self, params: ContextMatchRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Match ad context to buyer packages.

        Override this to provide TMP context matching.
        """
        return self._not_supported("context_match")

    async def identity_match(
        self, params: IdentityMatchRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Match user identity for package eligibility.

        Override this to provide TMP identity matching.
        """
        return self._not_supported("identity_match")

    # ========================================================================
    # V3 Brand Rights Operations
    # ========================================================================

    async def get_brand_identity(
        self, params: GetBrandIdentityRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Get brand identity information.

        Override this in BrandHandler subclasses.
        """
        return self._not_supported("get_brand_identity")

    async def get_rights(
        self, params: GetRightsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Get available rights for licensing.

        Override this in BrandHandler subclasses.
        """
        return self._not_supported("get_rights")

    async def acquire_rights(
        self, params: AcquireRightsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Acquire rights for brand content usage.

        Override this in BrandHandler subclasses.
        """
        return self._not_supported("acquire_rights")

    async def update_rights(
        self, params: UpdateRightsRequest | dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Update terms of an existing rights acquisition.

        Override this in BrandHandler subclasses. Partial update: the
        request carries ``rights_id`` plus any subset of the mutable fields
        (``end_date``, ``impression_cap``, ``pricing_option_id``, ``paused``).

        Seller responsibilities you own when implementing this:

        * Reject updates on expired or revoked acquisitions with an
          appropriate error code — do not partial-commit.
        * Reject ``pricing_option_id`` swaps to incompatible options — the
          new option's terms must be a strict superset of the original.
        * Apply all accepted fields atomically — callers should never
          observe a half-applied update on failure.
        """
        return self._not_supported("update_rights")

    async def validate_input(self, params: dict[str, Any], context: TContext | None = None) -> Any:
        """Validate creative input."""
        return self._not_supported("validate_input")

    async def verify_brand_claim(
        self, params: dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Verify a single brand claim."""
        return self._not_supported("verify_brand_claim")

    async def verify_brand_claims(
        self, params: dict[str, Any], context: TContext | None = None
    ) -> Any:
        """Verify multiple brand claims."""
        return self._not_supported("verify_brand_claims")

    # ========================================================================
    # V3 Compliance Operations
    # ========================================================================

    async def comply_test_controller(
        self,
        params: ComplyTestControllerRequest | dict[str, Any],
        context: TContext | None = None,
    ) -> Any:
        """Compliance test controller (sandbox only).

        Override this in ComplianceHandler subclasses.
        """
        return self._not_supported("comply_test_controller")

Base class for ADCP operation handlers.

Subclass this to implement ADCP operations. All operations have default implementations that return 'not supported', allowing you to implement only the operations your agent supports.

Parameterise over a :class:ToolContext subclass — class MyAgent(ADCPHandler[MyContext]) — to get typed context arguments on every method signature. See :data:TContext for the pattern.

For protocol-specific handlers, use: - ContentStandardsHandler: For content standards agents - SponsoredIntelligenceHandler: For sponsored intelligence agents - GovernanceHandler: For governance agents

Tool advertisement (advertised_tools class attribute):

A subclass that introduces a new specialism — i.e., a custom base that needs its own tools/list filter rather than inheriting one from a built-in handler — declares the tool set on the class body::

class PlatformHandler(ADCPHandler):
    advertised_tools: ClassVar[set[str]] = {
        "get_products",
        "create_media_buy",
        ...
    }

The framework registers PlatformHandler -> advertised_tools with :func:register_handler_tools() at class definition time. Subclasses that DON'T introduce a new specialism (a custom MyContentAgent(ContentStandardsHandler), for example) inherit their parent's tool set unchanged — no class attr needed.

Hand-written equivalent (no advertised_tools declaration)::

from adcp.server.mcp_tools import register_handler_tools
register_handler_tools("PlatformHandler", {...})

Either path is fine; codegen targets emit the class attribute so the declaration sits next to the class definition.

Ancestors

• abc.ABC
• typing.Generic

Subclasses

• PlatformHandler
• BrandHandler
• ComplianceHandler
• ContentStandardsHandler
• GovernanceHandler
• SponsoredIntelligenceHandler
• TmpHandler

Methods

async def acquire_rights(self,
params: AcquireRightsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def acquire_rights(
    self, params: AcquireRightsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Acquire rights for brand content usage.

    Override this in BrandHandler subclasses.
    """
    return self._not_supported("acquire_rights")

Acquire rights for brand content usage.

Override this in BrandHandler subclasses.

async def activate_signal(self,
params: ActivateSignalRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def activate_signal(
    self, params: ActivateSignalRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Activate a signal.

    Override this to handle signal activation.
    """
    return self._not_supported("activate_signal")

Activate a signal.

Override this to handle signal activation.

async def build_creative(self,
params: BuildCreativeRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def build_creative(
    self, params: BuildCreativeRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Build a creative.

    Override this to build creatives from assets.
    """
    return self._not_supported("build_creative")

Build a creative.

Override this to build creatives from assets.

async def calibrate_content(self,
params: CalibrateContentRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def calibrate_content(
    self, params: CalibrateContentRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Calibrate content against standards.

    Override this in ContentStandardsHandler subclasses.
    """
    return self._not_supported("calibrate_content")

Calibrate content against standards.

Override this in ContentStandardsHandler subclasses.

async def check_governance(self,
params: CheckGovernanceRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def check_governance(
    self, params: CheckGovernanceRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Check an action against campaign governance.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("check_governance")

Check an action against campaign governance.

Override this in GovernanceHandler subclasses.

async def comply_test_controller(self,
params: ComplyTestControllerRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def comply_test_controller(
    self,
    params: ComplyTestControllerRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Compliance test controller (sandbox only).

    Override this in ComplianceHandler subclasses.
    """
    return self._not_supported("comply_test_controller")

Compliance test controller (sandbox only).

Override this in ComplianceHandler subclasses.

async def context_match(self,
params: ContextMatchRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def context_match(
    self, params: ContextMatchRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Match ad context to buyer packages.

    Override this to provide TMP context matching.
    """
    return self._not_supported("context_match")

Match ad context to buyer packages.

Override this to provide TMP context matching.

async def create_collection_list(self,
params: CreateCollectionListRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def create_collection_list(
    self,
    params: CreateCollectionListRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Create a collection list for governance filtering.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("create_collection_list")

Create a collection list for governance filtering.

Override this in GovernanceHandler subclasses.

async def create_content_standards(self,
params: CreateContentStandardsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def create_content_standards(
    self,
    params: CreateContentStandardsRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Create content standards configuration.

    Override this in ContentStandardsHandler subclasses.
    """
    return self._not_supported("create_content_standards")

Create content standards configuration.

Override this in ContentStandardsHandler subclasses.

async def create_media_buy(self,
params: CreateMediaBuyRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def create_media_buy(
    self, params: CreateMediaBuyRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Create a media buy.

    Override this to handle media buy creation.
    """
    return self._not_supported("create_media_buy")

Create a media buy.

Override this to handle media buy creation.

async def create_property_list(self,
params: CreatePropertyListRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def create_property_list(
    self, params: CreatePropertyListRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Create a property list for governance filtering.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("create_property_list")

Create a property list for governance filtering.

Override this in GovernanceHandler subclasses.

async def delete_collection_list(self,
params: DeleteCollectionListRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def delete_collection_list(
    self,
    params: DeleteCollectionListRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Delete a collection list.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("delete_collection_list")

Delete a collection list.

Override this in GovernanceHandler subclasses.

async def delete_property_list(self,
params: DeletePropertyListRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def delete_property_list(
    self, params: DeletePropertyListRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Delete a property list.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("delete_property_list")

Delete a property list.

Override this in GovernanceHandler subclasses.

async def get_account_financials(self,
params: GetAccountFinancialsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_account_financials(
    self,
    params: GetAccountFinancialsRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Get account financials.

    Override this to provide account financial reporting.
    """
    return self._not_supported("get_account_financials")

Get account financials.

Override this to provide account financial reporting.

async def get_adcp_capabilities(self,
params: GetAdcpCapabilitiesRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_adcp_capabilities(
    self,
    params: GetAdcpCapabilitiesRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Get ADCP capabilities.

    Override this to advertise your agent's capabilities.
    """
    return self._not_supported("get_adcp_capabilities")

Get ADCP capabilities.

Override this to advertise your agent's capabilities.

async def get_brand_identity(self,
params: GetBrandIdentityRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_brand_identity(
    self, params: GetBrandIdentityRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Get brand identity information.

    Override this in BrandHandler subclasses.
    """
    return self._not_supported("get_brand_identity")

Get brand identity information.

Override this in BrandHandler subclasses.

async def get_collection_list(self,
params: GetCollectionListRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_collection_list(
    self, params: GetCollectionListRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Get a collection list with optional resolution.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("get_collection_list")

Get a collection list with optional resolution.

Override this in GovernanceHandler subclasses.

async def get_content_standards(self,
params: GetContentStandardsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_content_standards(
    self,
    params: GetContentStandardsRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Get content standards configuration.

    Override this in ContentStandardsHandler subclasses.
    """
    return self._not_supported("get_content_standards")

Get content standards configuration.

Override this in ContentStandardsHandler subclasses.

async def get_creative_delivery(self,
params: GetCreativeDeliveryRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_creative_delivery(
    self,
    params: GetCreativeDeliveryRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Get creative delivery metrics.

    Override this to provide functionality.
    """
    return self._not_supported("get_creative_delivery")

Get creative delivery metrics.

Override this to provide functionality.

async def get_creative_features(self,
params: GetCreativeFeaturesRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_creative_features(
    self,
    params: GetCreativeFeaturesRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Evaluate governance features for a creative.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("get_creative_features")

Evaluate governance features for a creative.

Override this in GovernanceHandler subclasses.

async def get_media_buy_artifacts(self,
params: GetMediaBuyArtifactsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_media_buy_artifacts(
    self,
    params: GetMediaBuyArtifactsRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Get artifacts associated with a media buy.

    Override this in ContentStandardsHandler subclasses.
    """
    return self._not_supported("get_media_buy_artifacts")

Get artifacts associated with a media buy.

Override this in ContentStandardsHandler subclasses.

async def get_media_buy_delivery(self,
params: GetMediaBuyDeliveryRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_media_buy_delivery(
    self,
    params: GetMediaBuyDeliveryRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Get media buy delivery metrics.

    Override this to provide delivery reporting.
    """
    return self._not_supported("get_media_buy_delivery")

Get media buy delivery metrics.

Override this to provide delivery reporting.

async def get_media_buys(self,
params: GetMediaBuysRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_media_buys(
    self, params: GetMediaBuysRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Get media buys with status and optional delivery snapshots.

    Override this to provide media buy listing functionality.
    """
    return self._not_supported("get_media_buys")

Get media buys with status and optional delivery snapshots.

Override this to provide media buy listing functionality.

async def get_plan_audit_logs(self,
params: GetPlanAuditLogsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_plan_audit_logs(
    self, params: GetPlanAuditLogsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Retrieve governance audit logs for plans.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("get_plan_audit_logs")

Retrieve governance audit logs for plans.

Override this in GovernanceHandler subclasses.

async def get_products(self,
params: GetProductsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_products(
    self, params: GetProductsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Get advertising products.

    Override this to provide product catalog functionality.
    """
    return self._not_supported("get_products")

Get advertising products.

Override this to provide product catalog functionality.

async def get_property_list(self,
params: GetPropertyListRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_property_list(
    self, params: GetPropertyListRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Get a property list with optional resolution.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("get_property_list")

Get a property list with optional resolution.

Override this in GovernanceHandler subclasses.

async def get_rights(self,
params: GetRightsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_rights(
    self, params: GetRightsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Get available rights for licensing.

    Override this in BrandHandler subclasses.
    """
    return self._not_supported("get_rights")

Get available rights for licensing.

Override this in BrandHandler subclasses.

async def get_signals(self,
params: GetSignalsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_signals(
    self, params: GetSignalsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Get available signals.

    Override this to provide signal catalog.
    """
    return self._not_supported("get_signals")

Get available signals.

Override this to provide signal catalog.

async def get_task_status(self,
params: GetTaskStatusRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def get_task_status(
    self,
    params: GetTaskStatusRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Get task status.

    Override this to expose persisted async task status.
    """
    return self._not_supported("get_task_status")

Get task status.

Override this to expose persisted async task status.

async def identity_match(self,
params: IdentityMatchRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def identity_match(
    self, params: IdentityMatchRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Match user identity for package eligibility.

    Override this to provide TMP identity matching.
    """
    return self._not_supported("identity_match")

Match user identity for package eligibility.

Override this to provide TMP identity matching.

async def list_accounts(self,
params: ListAccountsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def list_accounts(
    self, params: ListAccountsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """List accounts.

    Override this to provide functionality.
    """
    return self._not_supported("list_accounts")

List accounts.

Override this to provide functionality.

async def list_collection_lists(self,
params: ListCollectionListsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def list_collection_lists(
    self,
    params: ListCollectionListsRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """List collection lists.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("list_collection_lists")

List collection lists.

Override this in GovernanceHandler subclasses.

async def list_content_standards(self,
params: ListContentStandardsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def list_content_standards(
    self,
    params: ListContentStandardsRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """List content standards configurations.

    Override this in ContentStandardsHandler subclasses.
    """
    return self._not_supported("list_content_standards")

List content standards configurations.

Override this in ContentStandardsHandler subclasses.

async def list_creative_formats(self,
params: ListCreativeFormatsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def list_creative_formats(
    self,
    params: ListCreativeFormatsRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """List supported creative formats.

    Override this to provide creative format information.
    """
    return self._not_supported("list_creative_formats")

List supported creative formats.

Override this to provide creative format information.

async def list_creatives(self,
params: ListCreativesRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def list_creatives(
    self, params: ListCreativesRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """List creatives.

    Override this to list synced creatives.
    """
    return self._not_supported("list_creatives")

List creatives.

Override this to list synced creatives.

async def list_property_lists(self,
params: ListPropertyListsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def list_property_lists(
    self, params: ListPropertyListsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """List property lists.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("list_property_lists")

List property lists.

Override this in GovernanceHandler subclasses.

async def list_tasks(self,
params: ListTasksRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def list_tasks(
    self,
    params: ListTasksRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """List tasks.

    Override this to expose persisted async tasks.
    """
    return self._not_supported("list_tasks")

List tasks.

Override this to expose persisted async tasks.

async def list_transformers(self,
params: ListTransformersRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def list_transformers(
    self,
    params: ListTransformersRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """List creative transformers.

    Override this to advertise available creative transformation options.
    """
    return self._not_supported("list_transformers")

List creative transformers.

Override this to advertise available creative transformation options.

async def log_event(self,
params: LogEventRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def log_event(
    self, params: LogEventRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Log event.

    Override this to provide functionality.
    """
    return self._not_supported("log_event")

Log event.

Override this to provide functionality.

async def preview_creative(self,
params: PreviewCreativeRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def preview_creative(
    self, params: PreviewCreativeRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Preview a creative rendering.

    Override this to provide creative preview functionality.
    """
    return self._not_supported("preview_creative")

Preview a creative rendering.

Override this to provide creative preview functionality.

async def provide_performance_feedback(self,
params: ProvidePerformanceFeedbackRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def provide_performance_feedback(
    self,
    params: ProvidePerformanceFeedbackRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Provide performance feedback.

    Override this to handle performance feedback ingestion.
    """
    return self._not_supported("provide_performance_feedback")

Provide performance feedback.

Override this to handle performance feedback ingestion.

async def report_plan_outcome(self,
params: ReportPlanOutcomeRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def report_plan_outcome(
    self, params: ReportPlanOutcomeRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Report the outcome of a governed action.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("report_plan_outcome")

Report the outcome of a governed action.

Override this in GovernanceHandler subclasses.

async def report_usage(self,
params: ReportUsageRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def report_usage(
    self, params: ReportUsageRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Report account usage.

    Override this to ingest account usage.
    """
    return self._not_supported("report_usage")

Report account usage.

Override this to ingest account usage.

async def si_get_offering(self,
params: SiGetOfferingRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def si_get_offering(
    self, params: SiGetOfferingRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Get sponsored intelligence offering.

    Override this in SponsoredIntelligenceHandler subclasses.
    """
    return self._not_supported("si_get_offering")

Get sponsored intelligence offering.

Override this in SponsoredIntelligenceHandler subclasses.

async def si_initiate_session(self,
params: SiInitiateSessionRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def si_initiate_session(
    self, params: SiInitiateSessionRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Initiate sponsored intelligence session.

    Override this in SponsoredIntelligenceHandler subclasses.
    """
    return self._not_supported("si_initiate_session")

Initiate sponsored intelligence session.

Override this in SponsoredIntelligenceHandler subclasses.

async def si_send_message(self,
params: SiSendMessageRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def si_send_message(
    self, params: SiSendMessageRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Send message in sponsored intelligence session.

    Override this in SponsoredIntelligenceHandler subclasses.
    """
    return self._not_supported("si_send_message")

Send message in sponsored intelligence session.

Override this in SponsoredIntelligenceHandler subclasses.

async def si_terminate_session(self,
params: SiTerminateSessionRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def si_terminate_session(
    self, params: SiTerminateSessionRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Terminate sponsored intelligence session.

    Override this in SponsoredIntelligenceHandler subclasses.
    """
    return self._not_supported("si_terminate_session")

Terminate sponsored intelligence session.

Override this in SponsoredIntelligenceHandler subclasses.

async def sync_accounts(self,
params: SyncAccountsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def sync_accounts(
    self, params: SyncAccountsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Sync accounts.

    Override this to provide functionality.
    """
    return self._not_supported("sync_accounts")

Sync accounts.

Override this to provide functionality.

async def sync_audiences(self,
params: SyncAudiencesRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def sync_audiences(
    self, params: SyncAudiencesRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Sync audiences.

    Override this to provide audience synchronization.
    """
    return self._not_supported("sync_audiences")

Sync audiences.

Override this to provide audience synchronization.

async def sync_catalogs(self,
params: SyncCatalogsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def sync_catalogs(
    self, params: SyncCatalogsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Sync catalogs.

    Override this to provide catalog synchronization.
    """
    return self._not_supported("sync_catalogs")

Sync catalogs.

Override this to provide catalog synchronization.

async def sync_creatives(self,
params: SyncCreativesRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def sync_creatives(
    self, params: SyncCreativesRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Sync creatives.

    Override this to handle creative synchronization.
    """
    return self._not_supported("sync_creatives")

Sync creatives.

Override this to handle creative synchronization.

async def sync_event_sources(self,
params: SyncEventSourcesRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def sync_event_sources(
    self, params: SyncEventSourcesRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Sync event sources.

    Override this to provide functionality.
    """
    return self._not_supported("sync_event_sources")

Sync event sources.

Override this to provide functionality.

async def sync_governance(self,
params: SyncGovernanceRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def sync_governance(
    self, params: SyncGovernanceRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Sync governance agents for accounts.

    Override this to handle governance agent registration.
    """
    return self._not_supported("sync_governance")

Sync governance agents for accounts.

Override this to handle governance agent registration.

async def sync_plans(self,
params: SyncPlansRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def sync_plans(
    self, params: SyncPlansRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Sync campaign governance plans.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("sync_plans")

Sync campaign governance plans.

Override this in GovernanceHandler subclasses.

async def update_collection_list(self,
params: UpdateCollectionListRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def update_collection_list(
    self,
    params: UpdateCollectionListRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Update a collection list.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("update_collection_list")

Update a collection list.

Override this in GovernanceHandler subclasses.

async def update_content_standards(self,
params: UpdateContentStandardsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def update_content_standards(
    self,
    params: UpdateContentStandardsRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Update content standards configuration.

    Override this in ContentStandardsHandler subclasses.
    """
    return self._not_supported("update_content_standards")

Update content standards configuration.

Override this in ContentStandardsHandler subclasses.

async def update_media_buy(self,
params: UpdateMediaBuyRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def update_media_buy(
    self, params: UpdateMediaBuyRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Update a media buy.

    Override this to handle media buy updates.
    """
    return self._not_supported("update_media_buy")

Update a media buy.

Override this to handle media buy updates.

async def update_property_list(self,
params: UpdatePropertyListRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def update_property_list(
    self, params: UpdatePropertyListRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Update a property list.

    Override this in GovernanceHandler subclasses.
    """
    return self._not_supported("update_property_list")

Update a property list.

Override this in GovernanceHandler subclasses.

async def update_rights(self,
params: UpdateRightsRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def update_rights(
    self, params: UpdateRightsRequest | dict[str, Any], context: TContext | None = None
) -> Any:
    """Update terms of an existing rights acquisition.

    Override this in BrandHandler subclasses. Partial update: the
    request carries ``rights_id`` plus any subset of the mutable fields
    (``end_date``, ``impression_cap``, ``pricing_option_id``, ``paused``).

    Seller responsibilities you own when implementing this:

    * Reject updates on expired or revoked acquisitions with an
      appropriate error code — do not partial-commit.
    * Reject ``pricing_option_id`` swaps to incompatible options — the
      new option's terms must be a strict superset of the original.
    * Apply all accepted fields atomically — callers should never
      observe a half-applied update on failure.
    """
    return self._not_supported("update_rights")

Update terms of an existing rights acquisition.

Override this in BrandHandler subclasses. Partial update: the request carries rights_id plus any subset of the mutable fields (end_date, impression_cap, pricing_option_id, paused).

Seller responsibilities you own when implementing this:

• Reject updates on expired or revoked acquisitions with an appropriate error code — do not partial-commit.
• Reject pricing_option_id swaps to incompatible options — the new option's terms must be a strict superset of the original.
• Apply all accepted fields atomically — callers should never observe a half-applied update on failure.

async def validate_content_delivery(self,
params: ValidateContentDeliveryRequest | dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def validate_content_delivery(
    self,
    params: ValidateContentDeliveryRequest | dict[str, Any],
    context: TContext | None = None,
) -> Any:
    """Validate content delivery against standards.

    Override this in ContentStandardsHandler subclasses.
    """
    return self._not_supported("validate_content_delivery")

Validate content delivery against standards.

Override this in ContentStandardsHandler subclasses.

async def validate_input(self,
params: dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def validate_input(self, params: dict[str, Any], context: TContext | None = None) -> Any:
    """Validate creative input."""
    return self._not_supported("validate_input")

Validate creative input.

async def verify_brand_claim(self,
params: dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def verify_brand_claim(
    self, params: dict[str, Any], context: TContext | None = None
) -> Any:
    """Verify a single brand claim."""
    return self._not_supported("verify_brand_claim")

Verify a single brand claim.

async def verify_brand_claims(self,
params: dict[str, Any],
context: TContext | None = None) ‑> Any

Expand source code

async def verify_brand_claims(
    self, params: dict[str, Any], context: TContext | None = None
) -> Any:
    """Verify multiple brand claims."""
    return self._not_supported("verify_brand_claims")

Verify multiple brand claims.

class AccountAwareToolContext (request_id: str | None = None,
caller_identity: str | None = None,
tenant_id: str | None = None,
metadata: dict[str, Any] = <factory>,
resolved_adcp_version: str | None = None,
account_id: str | None = None,
account: Any | None = None)

Expand source code

@dataclass
class AccountAwareToolContext(ToolContext):
    """ToolContext subclass carrying a resolved account scope.

    AdCP is account-aware: many operations accept an ``account`` field
    (:class:`~adcp.types.AccountReference`) that the seller resolves to
    a concrete account before executing the request. Handlers that need
    ``account_id`` throughout their business logic shouldn't have to
    re-derive it on every call — this subclass carries the resolved
    result on the context itself.

    The typical flow::

        class MyAgent(ADCPHandler[AccountAwareToolContext]):
            async def get_products(self, params, context=None):
                err = await resolve_account_into_context(
                    params, context, my_resolver,
                )
                if err:
                    return err  # ACCOUNT_NOT_FOUND / SUSPENDED / etc.
                # context.account_id is now populated
                return products_response(self.catalog.for_account(context.account_id))

    Sellers whose account scope is fixed by the authenticated principal
    (e.g. per-tenant API keys that map 1:1 to an account) can populate
    ``account_id`` directly in their ``context_factory`` and skip the
    per-call resolution entirely.

    :param account_id: The resolved, stable account identifier. Safe to
        use as a cache key, audit log field, or authorization scope.
    :param account: The resolver's opaque account object — whatever the
        seller's :func:`resolve_account` resolver returned. Typed as
        ``Any`` so sellers aren't forced to match the SDK's shape.
    """

    account_id: str | None = None
    account: Any | None = None

ToolContext subclass carrying a resolved account scope.

AdCP is account-aware: many operations accept an account field (:class:~adcp.types.AccountReference) that the seller resolves to a concrete account before executing the request. Handlers that need account_id throughout their business logic shouldn't have to re-derive it on every call — this subclass carries the resolved result on the context itself.

The typical flow::

class MyAgent(ADCPHandler[AccountAwareToolContext]):
    async def get_products(self, params, context=None):
        err = await resolve_account_into_context(
            params, context, my_resolver,
        )
        if err:
            return err  # ACCOUNT_NOT_FOUND / SUSPENDED / etc.
        # context.account_id is now populated
        return products_response(self.catalog.for_account(context.account_id))

Sellers whose account scope is fixed by the authenticated principal (e.g. per-tenant API keys that map 1:1 to an account) can populate account_id directly in their context_factory and skip the per-call resolution entirely.

:param account_id: The resolved, stable account identifier. Safe to use as a cache key, audit log field, or authorization scope. :param account: The resolver's opaque account object — whatever the seller's :func:resolve_account resolver returned. Typed as Any so sellers aren't forced to match the SDK's shape.

Ancestors

• ToolContext

Instance variables

var account : typing.Any | None

var account_id : str | None

class NotImplementedResponse (**data: Any)

Expand source code

class NotImplementedResponse(BaseModel):
    """Standard response for operations not supported by this handler."""

    supported: bool = False
    reason: str = "This operation is not supported by this agent"
    error: Error | None = None

Standard response for operations not supported by this handler.

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

Class variables

var error : adcp.types.generated_poc.core.error.Error | None

var model_config

var reason : str

var supported : bool

class ToolContext (request_id: str | None = None,
caller_identity: str | None = None,
tenant_id: str | None = None,
metadata: dict[str, Any] = <factory>,
resolved_adcp_version: str | None = None)

Expand source code

@dataclass
class ToolContext:
    """Context passed to tool handlers.

    Contains metadata about the current request that may be useful
    for logging, authorization, or other cross-cutting concerns.

    Subclassing is supported. Multi-tenant agents commonly define a
    subclass carrying typed tenant + adapter fields (see
    ``docs/handler-authoring.md``) and populate it from a
    ``context_factory`` passed to :func:`create_mcp_server`.

    :param caller_identity: The authenticated principal making the request.
        **MUST** be a stable, globally-unique identifier within the seller's
        tenant — never an email, display name, or any other mutable handle.
        The server-side idempotency middleware keys its cache by
        ``(caller_identity, idempotency_key)`` — reuse of the same string for
        two distinct principals (e.g. email reuse after account deletion)
        causes cross-principal replay (confidentiality leak). Populated by
        the transport layer (A2A: ``ServerCallContext.user.user_name``; MCP:
        seller's FastMCP auth middleware).
    :param tenant_id: Multi-tenant agents may populate this with the tenant
        the request is scoped to. Typed as a first-class field so
        multi-tenant handlers don't have to smuggle it through ``metadata``.
        The server-side idempotency middleware composes the cache scope key
        from ``(tenant_id, caller_identity)`` when ``tenant_id`` is set —
        sellers whose principal IDs are only unique *within* a tenant (Okta
        group-scoped, SCIM per-tenant, seller-internal employee IDs) **MUST**
        populate this so cross-tenant response replay can't happen. When
        unset, the scope collapses to ``caller_identity`` alone (safe for
        single-tenant deployments).
    :param metadata: Open extension point for transport-specific or
        agent-specific fields (e.g. adapter instance handles, request
        headers, testing hooks). Downstream agents may subclass
        :class:`ToolContext` for typed fields; ``metadata`` is the escape
        hatch when subclassing isn't worth it.
    :param resolved_adcp_version: Release-precision AdCP version resolved by
        the transport dispatcher for this request. MCP unversioned native
        traffic is pinned to ``"3.0"`` for compatibility; A2A unversioned
        traffic leaves this as ``None`` and is served as the current SDK
        shape. Handlers should read this only when business logic truly
        depends on the buyer's wire contract.
    """

    request_id: str | None = None
    caller_identity: str | None = None
    tenant_id: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    resolved_adcp_version: str | None = None

Context passed to tool handlers.

Contains metadata about the current request that may be useful for logging, authorization, or other cross-cutting concerns.

Subclassing is supported. Multi-tenant agents commonly define a subclass carrying typed tenant + adapter fields (see docs/handler-authoring.md) and populate it from a context_factory passed to :func:create_mcp_server.

:param caller_identity: The authenticated principal making the request. MUST be a stable, globally-unique identifier within the seller's tenant — never an email, display name, or any other mutable handle. The server-side idempotency middleware keys its cache by (caller_identity, idempotency_key) — reuse of the same string for two distinct principals (e.g. email reuse after account deletion) causes cross-principal replay (confidentiality leak). Populated by the transport layer (A2A: ServerCallContext.user.user_name; MCP: seller's FastMCP auth middleware). :param tenant_id: Multi-tenant agents may populate this with the tenant the request is scoped to. Typed as a first-class field so multi-tenant handlers don't have to smuggle it through metadata. The server-side idempotency middleware composes the cache scope key from (tenant_id, caller_identity) when tenant_id is set — sellers whose principal IDs are only unique within a tenant (Okta group-scoped, SCIM per-tenant, seller-internal employee IDs) MUST populate this so cross-tenant response replay can't happen. When unset, the scope collapses to caller_identity alone (safe for single-tenant deployments). :param metadata: Open extension point for transport-specific or agent-specific fields (e.g. adapter instance handles, request headers, testing hooks). Downstream agents may subclass :class:ToolContext for typed fields; metadata is the escape hatch when subclassing isn't worth it. :param resolved_adcp_version: Release-precision AdCP version resolved by the transport dispatcher for this request. MCP unversioned native traffic is pinned to "3.0" for compatibility; A2A unversioned traffic leaves this as None and is served as the current SDK shape. Handlers should read this only when business logic truly depends on the buyer's wire contract.

Subclasses

• RequestContext
• AccountAwareToolContext

Instance variables

var caller_identity : str | None

var metadata : dict[str, typing.Any]

var request_id : str | None

var resolved_adcp_version : str | None

var tenant_id : str | None