ADCP Server Framework.

Provides base classes and adapters for building ADCP-compliant servers/agents. Supports selective protocol implementation via protocol-specific adapters.

Examples

Content Standards agent (stubs media buy operations)

from adcp.server import ContentStandardsHandler, create_mcp_tools

class MyContentHandler(ContentStandardsHandler): async def create_content_standards(self, request): # Implement your logic pass

Register with MCP server

tools = create_mcp_tools(MyContentHandler())

Sub-modules

adcp.server.base

Base classes for ADCP server implementations …

adcp.server.content_standards

Content Standards protocol handler …

adcp.server.governance

Governance protocol handler …

adcp.server.mcp_tools

MCP server integration helpers …

adcp.server.proposal

Proposal generation helpers …

adcp.server.sponsored_intelligence

Sponsored Intelligence protocol handler …

Functions

def create_mcp_tools(handler: ADCPHandler) ‑> MCPToolSet

Expand source code

def create_mcp_tools(handler: ADCPHandler) -> MCPToolSet:
    """Create MCP tools from an ADCP handler.

    This is the main entry point for MCP server integration.

    Example with mcp library:
        from mcp.server import Server
        from adcp.server import ContentStandardsHandler, create_mcp_tools

        class MyHandler(ContentStandardsHandler):
            # ... implement methods

        handler = MyHandler()
        tools = create_mcp_tools(handler)

        server = Server("my-content-agent")

        @server.list_tools()
        async def list_tools():
            return tools.tool_definitions

        @server.call_tool()
        async def call_tool(name: str, arguments: dict):
            return await tools.call_tool(name, arguments)

    Args:
        handler: ADCP handler instance

    Returns:
        MCPToolSet with tool definitions and handlers
    """
    return MCPToolSet(handler)

Create MCP tools from an ADCP handler.

This is the main entry point for MCP server integration.

Example with mcp library: from mcp.server import Server from adcp.server import ContentStandardsHandler, create_mcp_tools

class MyHandler(ContentStandardsHandler):
    # ... implement methods

handler = MyHandler()
tools = create_mcp_tools(handler)

server = Server("my-content-agent")

@server.list_tools()
async def list_tools():
    return tools.tool_definitions

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    return await tools.call_tool(name, arguments)

Args

handler

ADCP handler instance

Returns

MCPToolSet with tool definitions and handlers

def get_tools_for_handler(handler: ADCPHandler | type[ADCPHandler]) ‑> list[dict[str, typing.Any]]

Expand source code

def get_tools_for_handler(handler: ADCPHandler | type[ADCPHandler]) -> list[dict[str, Any]]:
    """Return tool definitions filtered by handler type.

    Walks the MRO to find the matching handler base class, so subclasses
    (e.g. MyGovernanceAgent(GovernanceHandler)) get the correct tool set.
    ADCPHandler gets all tools. Unknown handlers get only protocol discovery
    (minimum privilege).

    Args:
        handler: The handler instance or class

    Returns:
        Filtered list of tool definitions
    """
    cls = handler if isinstance(handler, type) else type(handler)
    for base in cls.__mro__:
        if base.__name__ in _HANDLER_TOOLS:
            allowed = _HANDLER_TOOLS[base.__name__] | _PROTOCOL_TOOLS
            return [tool for tool in ADCP_TOOL_DEFINITIONS if tool["name"] in allowed]

    return [tool for tool in ADCP_TOOL_DEFINITIONS if tool["name"] in _PROTOCOL_TOOLS]

Return tool definitions filtered by handler type.

Walks the MRO to find the matching handler base class, so subclasses (e.g. MyGovernanceAgent(GovernanceHandler)) get the correct tool set. ADCPHandler gets all tools. Unknown handlers get only protocol discovery (minimum privilege).

Args

handler

The handler instance or class

Returns

Filtered list of tool definitions

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

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

Expand source code

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

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

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

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

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

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

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

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

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

    return warnings

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

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

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

Args

handler

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

capabilities

The capabilities response the handler will serve.

Returns

List of warning strings. Empty if everything is consistent.

Classes

class ADCPHandler

Expand source code

class ADCPHandler(ABC):
    """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.

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

    _agent_type: str = "this agent"

    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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | None = None
    ) -> Any:
        """Sync creatives.

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

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

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

    async def build_creative(
        self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | None = None
    ) -> Any:
        """Get creative delivery metrics.

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

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

    async def create_media_buy(
        self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | None = None) -> Any:
        """Get available signals.

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

    async def activate_signal(
        self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | None = None
    ) -> Any:
        """List accounts.

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

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

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

    async def get_account_financials(
        self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | None = None) -> Any:
        """Log event.

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

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

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

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

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

    async def sync_catalogs(
        self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | None = None
    ) -> Any:
        """Get ADCP capabilities.

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

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

    async def create_content_standards(
        self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | None = None
    ) -> Any:
        """Sync campaign governance plans.

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

    async def check_governance(
        self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | 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: dict[str, Any], context: ToolContext | None = None
    ) -> Any:
        """Delete a property list.

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

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.

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

Ancestors

• abc.ABC

Subclasses

• ContentStandardsHandler
• GovernanceHandler
• SponsoredIntelligenceHandler

Methods

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

Expand source code

async def activate_signal(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def build_creative(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def calibrate_content(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def check_governance(
    self, params: dict[str, Any], context: ToolContext | 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 create_content_standards(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def create_content_standards(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def create_media_buy(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def create_property_list(
    self, params: dict[str, Any], context: ToolContext | 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_property_list(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def delete_property_list(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_account_financials(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_adcp_capabilities(
    self, params: dict[str, Any], context: ToolContext | 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_content_standards(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_content_standards(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_creative_delivery(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_creative_features(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_media_buy_artifacts(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_media_buy_delivery(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_media_buys(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_plan_audit_logs(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_products(self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_property_list(
    self, params: dict[str, Any], context: ToolContext | 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_signals(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def get_signals(self, params: dict[str, Any], context: ToolContext | 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 list_accounts(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def list_accounts(
    self, params: dict[str, Any], context: ToolContext | 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_content_standards(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def list_content_standards(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def list_creative_formats(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def list_creatives(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def list_property_lists(
    self, params: dict[str, Any], context: ToolContext | 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 log_event(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def log_event(self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def preview_creative(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def provide_performance_feedback(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def report_plan_outcome(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def report_usage(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def si_get_offering(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def si_initiate_session(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def si_send_message(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def si_terminate_session(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def sync_accounts(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def sync_audiences(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def sync_catalogs(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def sync_creatives(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def sync_event_sources(
    self, params: dict[str, Any], context: ToolContext | 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_plans(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def sync_plans(
    self, params: dict[str, Any], context: ToolContext | 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_content_standards(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def update_content_standards(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def update_media_buy(
    self, params: dict[str, Any], context: ToolContext | 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: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def update_property_list(
    self, params: dict[str, Any], context: ToolContext | 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 validate_content_delivery(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> Any

Expand source code

async def validate_content_delivery(
    self, params: dict[str, Any], context: ToolContext | 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.

class ContentStandardsHandler

Expand source code

class ContentStandardsHandler(ADCPHandler):
    """Handler for Content Standards protocol.

    Subclass this to implement a Content Standards agent. All Content Standards
    operations must be implemented via the handle_* methods.
    The public methods (create_content_standards, etc.) handle validation and
    error handling automatically.

    Non-Content-Standards operations (get_products, create_media_buy, etc.)
    return 'not supported' via the base class.

    Example:
        class MyContentStandardsHandler(ContentStandardsHandler):
            async def handle_create_content_standards(
                self,
                request: CreateContentStandardsRequest,
                context: ToolContext | None = None
            ) -> CreateContentStandardsResponse:
                # Your implementation
                return CreateContentStandardsResponse(...)
    """

    _agent_type: str = "Content Standards agents"

    # ========================================================================
    # Content Standards Operations - Override base class with validation
    # ========================================================================

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

        Validates params and delegates to handle_create_content_standards.
        """
        try:
            request = CreateContentStandardsRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_create_content_standards(request, context)

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

        Validates params and delegates to handle_get_content_standards.
        """
        try:
            request = GetContentStandardsRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_get_content_standards(request, context)

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

        Validates params and delegates to handle_list_content_standards.
        """
        try:
            request = ListContentStandardsRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_list_content_standards(request, context)

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

        Validates params and delegates to handle_update_content_standards.
        """
        try:
            request = UpdateContentStandardsRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_update_content_standards(request, context)

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

        Validates params and delegates to handle_calibrate_content.
        """
        try:
            request = CalibrateContentRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_calibrate_content(request, context)

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

        Validates params and delegates to handle_validate_content_delivery.
        """
        try:
            request = ValidateContentDeliveryRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_validate_content_delivery(request, context)

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

        Validates params and delegates to handle_get_media_buy_artifacts.
        """
        try:
            request = GetMediaBuyArtifactsRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_get_media_buy_artifacts(request, context)

    # ========================================================================
    # Abstract handlers - Implement these in subclasses
    # ========================================================================

    @abstractmethod
    async def handle_create_content_standards(
        self,
        request: CreateContentStandardsRequest,
        context: ToolContext | None = None,
    ) -> CreateContentStandardsResponse:
        """Handle create content standards request."""
        ...

    @abstractmethod
    async def handle_get_content_standards(
        self,
        request: GetContentStandardsRequest,
        context: ToolContext | None = None,
    ) -> GetContentStandardsResponse:
        """Handle get content standards request."""
        ...

    @abstractmethod
    async def handle_list_content_standards(
        self,
        request: ListContentStandardsRequest,
        context: ToolContext | None = None,
    ) -> ListContentStandardsResponse:
        """Handle list content standards request."""
        ...

    @abstractmethod
    async def handle_update_content_standards(
        self,
        request: UpdateContentStandardsRequest,
        context: ToolContext | None = None,
    ) -> UpdateContentStandardsResponse:
        """Handle update content standards request."""
        ...

    @abstractmethod
    async def handle_calibrate_content(
        self,
        request: CalibrateContentRequest,
        context: ToolContext | None = None,
    ) -> CalibrateContentResponse:
        """Handle calibrate content request."""
        ...

    @abstractmethod
    async def handle_validate_content_delivery(
        self,
        request: ValidateContentDeliveryRequest,
        context: ToolContext | None = None,
    ) -> ValidateContentDeliveryResponse:
        """Handle validate content delivery request."""
        ...

    @abstractmethod
    async def handle_get_media_buy_artifacts(
        self,
        request: GetMediaBuyArtifactsRequest,
        context: ToolContext | None = None,
    ) -> GetMediaBuyArtifactsResponse:
        """Handle get media buy artifacts request."""
        ...

Handler for Content Standards protocol.

Subclass this to implement a Content Standards agent. All Content Standards operations must be implemented via the handle_* methods. The public methods (create_content_standards, etc.) handle validation and error handling automatically.

Non-Content-Standards operations (get_products, create_media_buy, etc.) return 'not supported' via the base class.

Example

class MyContentStandardsHandler(ContentStandardsHandler): async def handle_create_content_standards( self, request: CreateContentStandardsRequest, context: ToolContext | None = None ) -> CreateContentStandardsResponse: # Your implementation return CreateContentStandardsResponse(…)

Ancestors

• ADCPHandler
• abc.ABC

Methods

async def calibrate_content(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> CalibrateContentResponse1 | CalibrateContentResponse2 | NotImplementedResponse

Expand source code

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

    Validates params and delegates to handle_calibrate_content.
    """
    try:
        request = CalibrateContentRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_calibrate_content(request, context)

Calibrate content against standards.

Validates params and delegates to handle_calibrate_content.

async def create_content_standards(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> CreateContentStandardsResponse1 | CreateContentStandardsResponse2 | NotImplementedResponse

Expand source code

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

    Validates params and delegates to handle_create_content_standards.
    """
    try:
        request = CreateContentStandardsRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_create_content_standards(request, context)

Create content standards configuration.

Validates params and delegates to handle_create_content_standards.

async def get_content_standards(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> GetContentStandardsResponse1 | GetContentStandardsResponse2 | NotImplementedResponse

Expand source code

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

    Validates params and delegates to handle_get_content_standards.
    """
    try:
        request = GetContentStandardsRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_get_content_standards(request, context)

Get content standards configuration.

Validates params and delegates to handle_get_content_standards.

async def get_media_buy_artifacts(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> GetMediaBuyArtifactsResponse1 | GetMediaBuyArtifactsResponse2 | NotImplementedResponse

Expand source code

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

    Validates params and delegates to handle_get_media_buy_artifacts.
    """
    try:
        request = GetMediaBuyArtifactsRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_get_media_buy_artifacts(request, context)

Get artifacts associated with a media buy.

Validates params and delegates to handle_get_media_buy_artifacts.

async def handle_calibrate_content(self,
request: CalibrateContentRequest,
context: ToolContext | None = None) ‑> CalibrateContentResponse1 | CalibrateContentResponse2

Expand source code

@abstractmethod
async def handle_calibrate_content(
    self,
    request: CalibrateContentRequest,
    context: ToolContext | None = None,
) -> CalibrateContentResponse:
    """Handle calibrate content request."""
    ...

Handle calibrate content request.

async def handle_create_content_standards(self,
request: CreateContentStandardsRequest,
context: ToolContext | None = None) ‑> CreateContentStandardsResponse1 | CreateContentStandardsResponse2

Expand source code

@abstractmethod
async def handle_create_content_standards(
    self,
    request: CreateContentStandardsRequest,
    context: ToolContext | None = None,
) -> CreateContentStandardsResponse:
    """Handle create content standards request."""
    ...

Handle create content standards request.

async def handle_get_content_standards(self,
request: GetContentStandardsRequest,
context: ToolContext | None = None) ‑> GetContentStandardsResponse1 | GetContentStandardsResponse2

Expand source code

@abstractmethod
async def handle_get_content_standards(
    self,
    request: GetContentStandardsRequest,
    context: ToolContext | None = None,
) -> GetContentStandardsResponse:
    """Handle get content standards request."""
    ...

Handle get content standards request.

async def handle_get_media_buy_artifacts(self,
request: GetMediaBuyArtifactsRequest,
context: ToolContext | None = None) ‑> GetMediaBuyArtifactsResponse1 | GetMediaBuyArtifactsResponse2

Expand source code

@abstractmethod
async def handle_get_media_buy_artifacts(
    self,
    request: GetMediaBuyArtifactsRequest,
    context: ToolContext | None = None,
) -> GetMediaBuyArtifactsResponse:
    """Handle get media buy artifacts request."""
    ...

Handle get media buy artifacts request.

async def handle_list_content_standards(self,
request: ListContentStandardsRequest,
context: ToolContext | None = None) ‑> ListContentStandardsResponse1 | ListContentStandardsResponse2

Expand source code

@abstractmethod
async def handle_list_content_standards(
    self,
    request: ListContentStandardsRequest,
    context: ToolContext | None = None,
) -> ListContentStandardsResponse:
    """Handle list content standards request."""
    ...

Handle list content standards request.

async def handle_update_content_standards(self,
request: UpdateContentStandardsRequest,
context: ToolContext | None = None) ‑> UpdateContentStandardsResponse1 | UpdateContentStandardsResponse2

Expand source code

@abstractmethod
async def handle_update_content_standards(
    self,
    request: UpdateContentStandardsRequest,
    context: ToolContext | None = None,
) -> UpdateContentStandardsResponse:
    """Handle update content standards request."""
    ...

Handle update content standards request.

async def handle_validate_content_delivery(self,
request: ValidateContentDeliveryRequest,
context: ToolContext | None = None) ‑> ValidateContentDeliveryResponse1 | ValidateContentDeliveryResponse2

Expand source code

@abstractmethod
async def handle_validate_content_delivery(
    self,
    request: ValidateContentDeliveryRequest,
    context: ToolContext | None = None,
) -> ValidateContentDeliveryResponse:
    """Handle validate content delivery request."""
    ...

Handle validate content delivery request.

async def list_content_standards(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> ListContentStandardsResponse1 | ListContentStandardsResponse2 | NotImplementedResponse

Expand source code

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

    Validates params and delegates to handle_list_content_standards.
    """
    try:
        request = ListContentStandardsRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_list_content_standards(request, context)

List content standards configurations.

Validates params and delegates to handle_list_content_standards.

async def update_content_standards(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> UpdateContentStandardsResponse1 | UpdateContentStandardsResponse2 | NotImplementedResponse

Expand source code

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

    Validates params and delegates to handle_update_content_standards.
    """
    try:
        request = UpdateContentStandardsRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_update_content_standards(request, context)

Update content standards configuration.

Validates params and delegates to handle_update_content_standards.

async def validate_content_delivery(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> ValidateContentDeliveryResponse1 | ValidateContentDeliveryResponse2 | NotImplementedResponse

Expand source code

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

    Validates params and delegates to handle_validate_content_delivery.
    """
    try:
        request = ValidateContentDeliveryRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_validate_content_delivery(request, context)

Validate content delivery against standards.

Validates params and delegates to handle_validate_content_delivery.

Inherited members

• ADCPHandler:
  • activate_signal
  • build_creative
  • check_governance
  • create_media_buy
  • create_property_list
  • delete_property_list
  • get_account_financials
  • get_adcp_capabilities
  • get_creative_delivery
  • get_creative_features
  • get_media_buy_delivery
  • get_media_buys
  • get_plan_audit_logs
  • get_products
  • get_property_list
  • get_signals
  • list_accounts
  • list_creative_formats
  • list_creatives
  • list_property_lists
  • log_event
  • preview_creative
  • provide_performance_feedback
  • report_plan_outcome
  • report_usage
  • si_get_offering
  • si_initiate_session
  • si_send_message
  • si_terminate_session
  • sync_accounts
  • sync_audiences
  • sync_catalogs
  • sync_creatives
  • sync_event_sources
  • sync_plans
  • update_media_buy
  • update_property_list

class GovernanceHandler

Expand source code

class GovernanceHandler(ADCPHandler):
    """Handler for Governance protocol (Property Lists).

    Subclass this to implement a Governance agent that manages property lists
    for brand safety, compliance scoring, and quality filtering.

    All property list operations must be implemented via the handle_* methods.
    The public methods (create_property_list, etc.) handle validation and
    error handling automatically.

    Non-governance operations (get_products, create_media_buy, etc.)
    return 'not supported' via the base class.

    Example:
        class MyGovernanceHandler(GovernanceHandler):
            async def handle_create_property_list(
                self,
                request: CreatePropertyListRequest,
                context: ToolContext | None = None
            ) -> CreatePropertyListResponse:
                # Store the list definition
                list_id = generate_id()
                # ...
                return CreatePropertyListResponse(list=PropertyList(...))
    """

    _agent_type: str = "Governance agents"

    # ========================================================================
    # Governance Operations - Override base class with validation
    # ========================================================================

    async def get_creative_features(
        self,
        params: dict[str, Any],
        context: ToolContext | None = None,
    ) -> GetCreativeFeaturesResponse | NotImplementedResponse:
        """Evaluate governance features for a creative manifest."""
        try:
            request = GetCreativeFeaturesRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_get_creative_features(request, context)

    async def sync_plans(
        self,
        params: dict[str, Any],
        context: ToolContext | None = None,
    ) -> SyncPlansResponse | NotImplementedResponse:
        """Sync campaign governance plans to the agent."""
        try:
            request = SyncPlansRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_sync_plans(request, context)

    async def check_governance(
        self,
        params: dict[str, Any],
        context: ToolContext | None = None,
    ) -> CheckGovernanceResponse | NotImplementedResponse:
        """Check whether a proposed or committed action complies with plan governance."""
        try:
            request = CheckGovernanceRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_check_governance(request, context)

    async def report_plan_outcome(
        self,
        params: dict[str, Any],
        context: ToolContext | None = None,
    ) -> ReportPlanOutcomeResponse | NotImplementedResponse:
        """Report the outcome of a previously governed action."""
        try:
            request = ReportPlanOutcomeRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_report_plan_outcome(request, context)

    async def get_plan_audit_logs(
        self,
        params: dict[str, Any],
        context: ToolContext | None = None,
    ) -> GetPlanAuditLogsResponse | NotImplementedResponse:
        """Retrieve governance audit logs for one or more plans."""
        try:
            request = GetPlanAuditLogsRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_get_plan_audit_logs(request, context)

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

        Validates params and delegates to handle_create_property_list.
        """
        try:
            request = CreatePropertyListRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_create_property_list(request, context)

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

        Validates params and delegates to handle_get_property_list.
        """
        try:
            request = GetPropertyListRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_get_property_list(request, context)

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

        Validates params and delegates to handle_list_property_lists.
        """
        try:
            request = ListPropertyListsRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_list_property_lists(request, context)

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

        Validates params and delegates to handle_update_property_list.
        """
        try:
            request = UpdatePropertyListRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_update_property_list(request, context)

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

        Validates params and delegates to handle_delete_property_list.
        """
        try:
            request = DeletePropertyListRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_delete_property_list(request, context)

    # ========================================================================
    # Abstract handlers - Implement these in subclasses
    # ========================================================================

    @abstractmethod
    async def handle_get_creative_features(
        self,
        request: GetCreativeFeaturesRequest,
        context: ToolContext | None = None,
    ) -> GetCreativeFeaturesResponse:
        """Handle creative feature evaluation."""
        ...

    @abstractmethod
    async def handle_sync_plans(
        self,
        request: SyncPlansRequest,
        context: ToolContext | None = None,
    ) -> SyncPlansResponse:
        """Handle campaign governance plan sync."""
        ...

    @abstractmethod
    async def handle_check_governance(
        self,
        request: CheckGovernanceRequest,
        context: ToolContext | None = None,
    ) -> CheckGovernanceResponse:
        """Handle a governance check request."""
        ...

    @abstractmethod
    async def handle_report_plan_outcome(
        self,
        request: ReportPlanOutcomeRequest,
        context: ToolContext | None = None,
    ) -> ReportPlanOutcomeResponse:
        """Handle reporting of a governed action outcome."""
        ...

    @abstractmethod
    async def handle_get_plan_audit_logs(
        self,
        request: GetPlanAuditLogsRequest,
        context: ToolContext | None = None,
    ) -> GetPlanAuditLogsResponse:
        """Handle retrieval of governance audit logs."""
        ...

    @abstractmethod
    async def handle_create_property_list(
        self,
        request: CreatePropertyListRequest,
        context: ToolContext | None = None,
    ) -> CreatePropertyListResponse:
        """Handle create property list request."""
        ...

    @abstractmethod
    async def handle_get_property_list(
        self,
        request: GetPropertyListRequest,
        context: ToolContext | None = None,
    ) -> GetPropertyListResponse:
        """Handle get property list request."""
        ...

    @abstractmethod
    async def handle_list_property_lists(
        self,
        request: ListPropertyListsRequest,
        context: ToolContext | None = None,
    ) -> ListPropertyListsResponse:
        """Handle list property lists request."""
        ...

    @abstractmethod
    async def handle_update_property_list(
        self,
        request: UpdatePropertyListRequest,
        context: ToolContext | None = None,
    ) -> UpdatePropertyListResponse:
        """Handle update property list request."""
        ...

    @abstractmethod
    async def handle_delete_property_list(
        self,
        request: DeletePropertyListRequest,
        context: ToolContext | None = None,
    ) -> DeletePropertyListResponse:
        """Handle delete property list request."""
        ...

Handler for Governance protocol (Property Lists).

Subclass this to implement a Governance agent that manages property lists for brand safety, compliance scoring, and quality filtering.

All property list operations must be implemented via the handle_* methods. The public methods (create_property_list, etc.) handle validation and error handling automatically.

Non-governance operations (get_products, create_media_buy, etc.) return 'not supported' via the base class.

Example

class MyGovernanceHandler(GovernanceHandler): async def handle_create_property_list( self, request: CreatePropertyListRequest, context: ToolContext | None = None ) -> CreatePropertyListResponse: # Store the list definition list_id = generate_id() # … return CreatePropertyListResponse(list=PropertyList(…))

Ancestors

• ADCPHandler
• abc.ABC

Methods

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

Expand source code

async def check_governance(
    self,
    params: dict[str, Any],
    context: ToolContext | None = None,
) -> CheckGovernanceResponse | NotImplementedResponse:
    """Check whether a proposed or committed action complies with plan governance."""
    try:
        request = CheckGovernanceRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_check_governance(request, context)

Check whether a proposed or committed action complies with plan governance.

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

Expand source code

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

    Validates params and delegates to handle_create_property_list.
    """
    try:
        request = CreatePropertyListRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_create_property_list(request, context)

Create a property list for governance filtering.

Validates params and delegates to handle_create_property_list.

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

Expand source code

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

    Validates params and delegates to handle_delete_property_list.
    """
    try:
        request = DeletePropertyListRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_delete_property_list(request, context)

Delete a property list.

Validates params and delegates to handle_delete_property_list.

async def get_creative_features(self,
params: dict[str, Any],
context: ToolContext | None = None) ‑> GetCreativeFeaturesResponse1 | GetCreativeFeaturesResponse2 | NotImplementedResponse

Expand source code

async def get_creative_features(
    self,
    params: dict[str, Any],
    context: ToolContext | None = None,
) -> GetCreativeFeaturesResponse | NotImplementedResponse:
    """Evaluate governance features for a creative manifest."""
    try:
        request = GetCreativeFeaturesRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_get_creative_features(request, context)

Evaluate governance features for a creative manifest.

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

Expand source code

async def get_plan_audit_logs(
    self,
    params: dict[str, Any],
    context: ToolContext | None = None,
) -> GetPlanAuditLogsResponse | NotImplementedResponse:
    """Retrieve governance audit logs for one or more plans."""
    try:
        request = GetPlanAuditLogsRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_get_plan_audit_logs(request, context)

Retrieve governance audit logs for one or more plans.

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

Expand source code

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

    Validates params and delegates to handle_get_property_list.
    """
    try:
        request = GetPropertyListRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_get_property_list(request, context)

Get a property list with optional resolution.

Validates params and delegates to handle_get_property_list.

async def handle_check_governance(self,
request: CheckGovernanceRequest,
context: ToolContext | None = None) ‑> CheckGovernanceResponse

Expand source code

@abstractmethod
async def handle_check_governance(
    self,
    request: CheckGovernanceRequest,
    context: ToolContext | None = None,
) -> CheckGovernanceResponse:
    """Handle a governance check request."""
    ...

Handle a governance check request.

async def handle_create_property_list(self,
request: CreatePropertyListRequest,
context: ToolContext | None = None) ‑> CreatePropertyListResponse

Expand source code

@abstractmethod
async def handle_create_property_list(
    self,
    request: CreatePropertyListRequest,
    context: ToolContext | None = None,
) -> CreatePropertyListResponse:
    """Handle create property list request."""
    ...

Handle create property list request.

async def handle_delete_property_list(self,
request: DeletePropertyListRequest,
context: ToolContext | None = None) ‑> DeletePropertyListResponse

Expand source code

@abstractmethod
async def handle_delete_property_list(
    self,
    request: DeletePropertyListRequest,
    context: ToolContext | None = None,
) -> DeletePropertyListResponse:
    """Handle delete property list request."""
    ...

Handle delete property list request.

async def handle_get_creative_features(self,
request: GetCreativeFeaturesRequest,
context: ToolContext | None = None) ‑> GetCreativeFeaturesResponse1 | GetCreativeFeaturesResponse2

Expand source code

@abstractmethod
async def handle_get_creative_features(
    self,
    request: GetCreativeFeaturesRequest,
    context: ToolContext | None = None,
) -> GetCreativeFeaturesResponse:
    """Handle creative feature evaluation."""
    ...

Handle creative feature evaluation.

async def handle_get_plan_audit_logs(self,
request: GetPlanAuditLogsRequest,
context: ToolContext | None = None) ‑> GetPlanAuditLogsResponse

Expand source code

@abstractmethod
async def handle_get_plan_audit_logs(
    self,
    request: GetPlanAuditLogsRequest,
    context: ToolContext | None = None,
) -> GetPlanAuditLogsResponse:
    """Handle retrieval of governance audit logs."""
    ...

Handle retrieval of governance audit logs.

async def handle_get_property_list(self,
request: GetPropertyListRequest,
context: ToolContext | None = None) ‑> GetPropertyListResponse

Expand source code

@abstractmethod
async def handle_get_property_list(
    self,
    request: GetPropertyListRequest,
    context: ToolContext | None = None,
) -> GetPropertyListResponse:
    """Handle get property list request."""
    ...

Handle get property list request.

async def handle_list_property_lists(self,
request: ListPropertyListsRequest,
context: ToolContext | None = None) ‑> ListPropertyListsResponse

Expand source code

@abstractmethod
async def handle_list_property_lists(
    self,
    request: ListPropertyListsRequest,
    context: ToolContext | None = None,
) -> ListPropertyListsResponse:
    """Handle list property lists request."""
    ...

Handle list property lists request.

async def handle_report_plan_outcome(self,
request: ReportPlanOutcomeRequest,
context: ToolContext | None = None) ‑> ReportPlanOutcomeResponse

Expand source code

@abstractmethod
async def handle_report_plan_outcome(
    self,
    request: ReportPlanOutcomeRequest,
    context: ToolContext | None = None,
) -> ReportPlanOutcomeResponse:
    """Handle reporting of a governed action outcome."""
    ...

Handle reporting of a governed action outcome.

async def handle_sync_plans(self,
request: SyncPlansRequest,
context: ToolContext | None = None) ‑> SyncPlansResponse

Expand source code

@abstractmethod
async def handle_sync_plans(
    self,
    request: SyncPlansRequest,
    context: ToolContext | None = None,
) -> SyncPlansResponse:
    """Handle campaign governance plan sync."""
    ...

Handle campaign governance plan sync.

async def handle_update_property_list(self,
request: UpdatePropertyListRequest,
context: ToolContext | None = None) ‑> UpdatePropertyListResponse

Expand source code

@abstractmethod
async def handle_update_property_list(
    self,
    request: UpdatePropertyListRequest,
    context: ToolContext | None = None,
) -> UpdatePropertyListResponse:
    """Handle update property list request."""
    ...

Handle update property list request.

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

Expand source code

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

    Validates params and delegates to handle_list_property_lists.
    """
    try:
        request = ListPropertyListsRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_list_property_lists(request, context)

List property lists.

Validates params and delegates to handle_list_property_lists.

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

Expand source code

async def report_plan_outcome(
    self,
    params: dict[str, Any],
    context: ToolContext | None = None,
) -> ReportPlanOutcomeResponse | NotImplementedResponse:
    """Report the outcome of a previously governed action."""
    try:
        request = ReportPlanOutcomeRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_report_plan_outcome(request, context)

Report the outcome of a previously governed action.

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

Expand source code

async def sync_plans(
    self,
    params: dict[str, Any],
    context: ToolContext | None = None,
) -> SyncPlansResponse | NotImplementedResponse:
    """Sync campaign governance plans to the agent."""
    try:
        request = SyncPlansRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_sync_plans(request, context)

Sync campaign governance plans to the agent.

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

Expand source code

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

    Validates params and delegates to handle_update_property_list.
    """
    try:
        request = UpdatePropertyListRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_update_property_list(request, context)

Update a property list.

Validates params and delegates to handle_update_property_list.

Inherited members

• ADCPHandler:
  • activate_signal
  • build_creative
  • calibrate_content
  • create_content_standards
  • create_media_buy
  • get_account_financials
  • get_adcp_capabilities
  • get_content_standards
  • get_creative_delivery
  • get_media_buy_artifacts
  • get_media_buy_delivery
  • get_media_buys
  • get_products
  • get_signals
  • list_accounts
  • list_content_standards
  • list_creative_formats
  • list_creatives
  • log_event
  • preview_creative
  • provide_performance_feedback
  • report_usage
  • si_get_offering
  • si_initiate_session
  • si_send_message
  • si_terminate_session
  • sync_accounts
  • sync_audiences
  • sync_catalogs
  • sync_creatives
  • sync_event_sources
  • update_content_standards
  • update_media_buy
  • validate_content_delivery

class MCPToolSet (handler: ADCPHandler)

Expand source code

class MCPToolSet:
    """Collection of MCP tools from an ADCP handler.

    Provides tool definitions and handlers for registering with an MCP server.
    """

    def __init__(self, handler: ADCPHandler):
        """Create tool set from handler.

        Args:
            handler: ADCP handler instance
        """
        self.handler = handler
        self._filtered_definitions = get_tools_for_handler(handler)
        self._tools: dict[str, Callable[[dict[str, Any]], Any]] = {}

        # Create tool callers only for filtered tools
        for tool_def in self._filtered_definitions:
            name = tool_def["name"]
            self._tools[name] = create_tool_caller(handler, name)

    @property
    def tool_definitions(self) -> list[dict[str, Any]]:
        """Get MCP tool definitions filtered by handler type."""
        return list(self._filtered_definitions)

    async def call_tool(self, name: str, params: dict[str, Any]) -> Any:
        """Call a tool by name.

        Args:
            name: Tool name
            params: Tool parameters

        Returns:
            Tool result

        Raises:
            KeyError: If tool not found
        """
        if name not in self._tools:
            raise KeyError(f"Unknown tool: {name}")
        return await self._tools[name](params)

    def get_tool_names(self) -> list[str]:
        """Get list of available tool names."""
        return list(self._tools.keys())

Collection of MCP tools from an ADCP handler.

Provides tool definitions and handlers for registering with an MCP server.

Create tool set from handler.

Args

handler

ADCP handler instance

Instance variables

prop tool_definitions : list[dict[str, Any]]

Expand source code

@property
def tool_definitions(self) -> list[dict[str, Any]]:
    """Get MCP tool definitions filtered by handler type."""
    return list(self._filtered_definitions)

Get MCP tool definitions filtered by handler type.

Methods

async def call_tool(self, name: str, params: dict[str, Any]) ‑> Any

Expand source code

async def call_tool(self, name: str, params: dict[str, Any]) -> Any:
    """Call a tool by name.

    Args:
        name: Tool name
        params: Tool parameters

    Returns:
        Tool result

    Raises:
        KeyError: If tool not found
    """
    if name not in self._tools:
        raise KeyError(f"Unknown tool: {name}")
    return await self._tools[name](params)

Call a tool by name.

Args

name

Tool name

params

Tool parameters

Returns

Tool result

Raises

KeyError

If tool not found

def get_tool_names(self) ‑> list[str]

Expand source code

def get_tool_names(self) -> list[str]:
    """Get list of available tool names."""
    return list(self._tools.keys())

Get list of available tool names.

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 : Error | None

var model_config

var reason : str

var supported : bool

class ProposalBuilder (name: str, proposal_id: str | None = None)

Expand source code

class ProposalBuilder:
    """Builder for ADCP Proposals.

    Helps construct valid proposals for get_products responses. Proposals
    represent recommended media plans with budget allocations.

    Example:
        proposal = (
            ProposalBuilder("Q1 Brand Campaign")
            .with_description("Balanced awareness campaign")
            .add_allocation("product-1", 60)
                .with_rationale("High-impact display")
            .add_allocation("product-2", 40)
                .with_rationale("Contextual targeting")
            .with_budget_guidance(min=10000, recommended=25000, max=50000)
            .build()
        )
    """

    def __init__(self, name: str, proposal_id: str | None = None):
        """Create a new proposal builder.

        Args:
            name: Human-readable name for the proposal
            proposal_id: Unique ID (auto-generated if not provided)
        """
        self._name = name
        self._proposal_id = proposal_id or f"proposal-{uuid4().hex[:8]}"
        self._description: str | None = None
        self._brief_alignment: str | None = None
        self._expires_at: datetime | None = None
        self._allocations: list[dict[str, Any]] = []
        self._budget_guidance: dict[str, Any] | None = None
        self._current_allocation: AllocationBuilder | None = None
        self._ext: dict[str, Any] | None = None

    def with_description(self, description: str) -> ProposalBuilder:
        """Add description explaining the proposal strategy.

        Args:
            description: What the proposal achieves
        """
        self._finalize_allocation()
        self._description = description
        return self

    def with_brief_alignment(self, alignment: str) -> ProposalBuilder:
        """Explain how proposal aligns with campaign brief.

        Args:
            alignment: Alignment explanation
        """
        self._finalize_allocation()
        self._brief_alignment = alignment
        return self

    def expires_in(self, days: int = 7) -> ProposalBuilder:
        """Set expiration relative to now.

        Args:
            days: Number of days until expiration
        """
        self._finalize_allocation()
        self._expires_at = datetime.now(timezone.utc) + timedelta(days=days)
        return self

    def expires_at(self, expires: datetime) -> ProposalBuilder:
        """Set absolute expiration time.

        Args:
            expires: When the proposal expires
        """
        self._finalize_allocation()
        self._expires_at = expires
        return self

    def add_allocation(
        self,
        product_id: str,
        allocation_percentage: float,
    ) -> ProposalBuilder:
        """Add a product allocation.

        After calling this, chain allocation methods (with_rationale, etc.)
        before adding another allocation or calling build().

        Args:
            product_id: ID of the product
            allocation_percentage: Percentage of budget (0-100)

        Returns:
            Self for method chaining
        """
        self._finalize_allocation()
        self._current_allocation = AllocationBuilder(product_id, allocation_percentage)
        return self

    def with_pricing_option(self, pricing_option_id: str) -> ProposalBuilder:
        """Set pricing option for current allocation."""
        if self._current_allocation:
            self._current_allocation.with_pricing_option(pricing_option_id)
        return self

    def with_rationale(self, rationale: str) -> ProposalBuilder:
        """Add rationale for current allocation."""
        if self._current_allocation:
            self._current_allocation.with_rationale(rationale)
        return self

    def with_sequence(self, sequence: int) -> ProposalBuilder:
        """Set sequence for current allocation."""
        if self._current_allocation:
            self._current_allocation.with_sequence(sequence)
        return self

    def with_tags(self, tags: list[str]) -> ProposalBuilder:
        """Add tags for current allocation."""
        if self._current_allocation:
            self._current_allocation.with_tags(tags)
        return self

    def with_budget_guidance(
        self,
        *,
        min: float | None = None,
        recommended: float | None = None,
        max: float | None = None,
        currency: str = "USD",
    ) -> ProposalBuilder:
        """Add budget guidance for the proposal.

        Args:
            min: Minimum recommended budget
            recommended: Optimal budget
            max: Maximum before diminishing returns
            currency: ISO 4217 currency code
        """
        self._finalize_allocation()
        self._budget_guidance = {
            "currency": currency,
        }
        if min is not None:
            self._budget_guidance["min"] = min
        if recommended is not None:
            self._budget_guidance["recommended"] = recommended
        if max is not None:
            self._budget_guidance["max"] = max
        return self

    def with_extension(self, ext: dict[str, Any]) -> ProposalBuilder:
        """Add extension data.

        Args:
            ext: Extension object
        """
        self._finalize_allocation()
        self._ext = ext
        return self

    def _finalize_allocation(self) -> None:
        """Finalize current allocation and add to list."""
        if self._current_allocation:
            self._allocations.append(self._current_allocation.build())
            self._current_allocation = None

    def build(self) -> dict[str, Any]:
        """Build the proposal dict.

        Returns:
            Proposal as a dict ready for use in get_products response

        Raises:
            ValueError: If allocations don't sum to 100
        """
        self._finalize_allocation()

        if not self._allocations:
            raise ValueError("Proposal must have at least one allocation")

        total = sum(a["allocation_percentage"] for a in self._allocations)
        if abs(total - 100.0) > 0.01:
            raise ValueError(f"Allocation percentages must sum to 100, got {total}")

        proposal: dict[str, Any] = {
            "proposal_id": self._proposal_id,
            "name": self._name,
            "allocations": self._allocations,
        }

        if self._description:
            proposal["description"] = self._description
        if self._brief_alignment:
            proposal["brief_alignment"] = self._brief_alignment
        if self._expires_at:
            proposal["expires_at"] = self._expires_at.isoformat()
        if self._budget_guidance:
            proposal["total_budget_guidance"] = self._budget_guidance
        if self._ext:
            proposal["ext"] = self._ext

        return proposal

    def validate(self) -> list[str]:
        """Validate the proposal without building.

        Returns:
            List of validation errors (empty if valid)
        """
        errors: list[str] = []

        if self._current_allocation:
            allocations = self._allocations + [self._current_allocation.build()]
        else:
            allocations = self._allocations

        if not allocations:
            errors.append("Proposal must have at least one allocation")
        else:
            total = sum(a["allocation_percentage"] for a in allocations)
            if abs(total - 100.0) > 0.01:
                errors.append(f"Allocation percentages must sum to 100, got {total}")

        return errors

Builder for ADCP Proposals.

Helps construct valid proposals for get_products responses. Proposals represent recommended media plans with budget allocations.

Example

proposal = ( ProposalBuilder("Q1 Brand Campaign") .with_description("Balanced awareness campaign") .add_allocation("product-1", 60) .with_rationale("High-impact display") .add_allocation("product-2", 40) .with_rationale("Contextual targeting") .with_budget_guidance(min=10000, recommended=25000, max=50000) .build() )

Create a new proposal builder.

Args

name

Human-readable name for the proposal

proposal_id

Unique ID (auto-generated if not provided)

Methods

def add_allocation(self, product_id: str, allocation_percentage: float) ‑> ProposalBuilder

Expand source code

def add_allocation(
    self,
    product_id: str,
    allocation_percentage: float,
) -> ProposalBuilder:
    """Add a product allocation.

    After calling this, chain allocation methods (with_rationale, etc.)
    before adding another allocation or calling build().

    Args:
        product_id: ID of the product
        allocation_percentage: Percentage of budget (0-100)

    Returns:
        Self for method chaining
    """
    self._finalize_allocation()
    self._current_allocation = AllocationBuilder(product_id, allocation_percentage)
    return self

Add a product allocation.

After calling this, chain allocation methods (with_rationale, etc.) before adding another allocation or calling build().

Args

product_id

ID of the product

allocation_percentage

Percentage of budget (0-100)

Returns

Self for method chaining

def build(self) ‑> dict[str, typing.Any]

Expand source code

def build(self) -> dict[str, Any]:
    """Build the proposal dict.

    Returns:
        Proposal as a dict ready for use in get_products response

    Raises:
        ValueError: If allocations don't sum to 100
    """
    self._finalize_allocation()

    if not self._allocations:
        raise ValueError("Proposal must have at least one allocation")

    total = sum(a["allocation_percentage"] for a in self._allocations)
    if abs(total - 100.0) > 0.01:
        raise ValueError(f"Allocation percentages must sum to 100, got {total}")

    proposal: dict[str, Any] = {
        "proposal_id": self._proposal_id,
        "name": self._name,
        "allocations": self._allocations,
    }

    if self._description:
        proposal["description"] = self._description
    if self._brief_alignment:
        proposal["brief_alignment"] = self._brief_alignment
    if self._expires_at:
        proposal["expires_at"] = self._expires_at.isoformat()
    if self._budget_guidance:
        proposal["total_budget_guidance"] = self._budget_guidance
    if self._ext:
        proposal["ext"] = self._ext

    return proposal

Build the proposal dict.

Returns

Proposal as a dict ready for use in get_products response

Raises

ValueError

If allocations don't sum to 100

def expires_at(self, expires: datetime) ‑> ProposalBuilder

Expand source code

def expires_at(self, expires: datetime) -> ProposalBuilder:
    """Set absolute expiration time.

    Args:
        expires: When the proposal expires
    """
    self._finalize_allocation()
    self._expires_at = expires
    return self

Set absolute expiration time.

Args

expires

When the proposal expires

def expires_in(self, days: int = 7) ‑> ProposalBuilder

Expand source code

def expires_in(self, days: int = 7) -> ProposalBuilder:
    """Set expiration relative to now.

    Args:
        days: Number of days until expiration
    """
    self._finalize_allocation()
    self._expires_at = datetime.now(timezone.utc) + timedelta(days=days)
    return self

Set expiration relative to now.

Args

days

Number of days until expiration

def validate(self) ‑> list[str]

Expand source code

def validate(self) -> list[str]:
    """Validate the proposal without building.

    Returns:
        List of validation errors (empty if valid)
    """
    errors: list[str] = []

    if self._current_allocation:
        allocations = self._allocations + [self._current_allocation.build()]
    else:
        allocations = self._allocations

    if not allocations:
        errors.append("Proposal must have at least one allocation")
    else:
        total = sum(a["allocation_percentage"] for a in allocations)
        if abs(total - 100.0) > 0.01:
            errors.append(f"Allocation percentages must sum to 100, got {total}")

    return errors

Validate the proposal without building.

Returns

List of validation errors (empty if valid)

def with_brief_alignment(self, alignment: str) ‑> ProposalBuilder

Expand source code

def with_brief_alignment(self, alignment: str) -> ProposalBuilder:
    """Explain how proposal aligns with campaign brief.

    Args:
        alignment: Alignment explanation
    """
    self._finalize_allocation()
    self._brief_alignment = alignment
    return self

Explain how proposal aligns with campaign brief.

Args

alignment

Alignment explanation

def with_budget_guidance(self,
*,
min: float | None = None,
recommended: float | None = None,
max: float | None = None,
currency: str = 'USD') ‑> ProposalBuilder

Expand source code

def with_budget_guidance(
    self,
    *,
    min: float | None = None,
    recommended: float | None = None,
    max: float | None = None,
    currency: str = "USD",
) -> ProposalBuilder:
    """Add budget guidance for the proposal.

    Args:
        min: Minimum recommended budget
        recommended: Optimal budget
        max: Maximum before diminishing returns
        currency: ISO 4217 currency code
    """
    self._finalize_allocation()
    self._budget_guidance = {
        "currency": currency,
    }
    if min is not None:
        self._budget_guidance["min"] = min
    if recommended is not None:
        self._budget_guidance["recommended"] = recommended
    if max is not None:
        self._budget_guidance["max"] = max
    return self

Add budget guidance for the proposal.

Args

min

Minimum recommended budget

recommended

Optimal budget

max

Maximum before diminishing returns

currency

ISO 4217 currency code

def with_description(self, description: str) ‑> ProposalBuilder

Expand source code

def with_description(self, description: str) -> ProposalBuilder:
    """Add description explaining the proposal strategy.

    Args:
        description: What the proposal achieves
    """
    self._finalize_allocation()
    self._description = description
    return self

Add description explaining the proposal strategy.

Args

description

What the proposal achieves

def with_extension(self, ext: dict[str, Any]) ‑> ProposalBuilder

Expand source code

def with_extension(self, ext: dict[str, Any]) -> ProposalBuilder:
    """Add extension data.

    Args:
        ext: Extension object
    """
    self._finalize_allocation()
    self._ext = ext
    return self

Add extension data.

Args

ext

Extension object

def with_pricing_option(self, pricing_option_id: str) ‑> ProposalBuilder

Expand source code

def with_pricing_option(self, pricing_option_id: str) -> ProposalBuilder:
    """Set pricing option for current allocation."""
    if self._current_allocation:
        self._current_allocation.with_pricing_option(pricing_option_id)
    return self

Set pricing option for current allocation.

def with_rationale(self, rationale: str) ‑> ProposalBuilder

Expand source code

def with_rationale(self, rationale: str) -> ProposalBuilder:
    """Add rationale for current allocation."""
    if self._current_allocation:
        self._current_allocation.with_rationale(rationale)
    return self

Add rationale for current allocation.

def with_sequence(self, sequence: int) ‑> ProposalBuilder

Expand source code

def with_sequence(self, sequence: int) -> ProposalBuilder:
    """Set sequence for current allocation."""
    if self._current_allocation:
        self._current_allocation.with_sequence(sequence)
    return self

Set sequence for current allocation.

def with_tags(self, tags: list[str]) ‑> ProposalBuilder

Expand source code

def with_tags(self, tags: list[str]) -> ProposalBuilder:
    """Add tags for current allocation."""
    if self._current_allocation:
        self._current_allocation.with_tags(tags)
    return self

Add tags for current allocation.

class ProposalNotSupported (**data: Any)

Expand source code

class ProposalNotSupported(BaseModel):
    """Response indicating proposal generation is not supported.

    Use this when your agent supports get_products but not proposal generation.
    """

    proposals_supported: bool = False
    reason: str = "This agent does not generate proposals"
    error: Error | None = None

Response indicating proposal generation is not supported.

Use this when your agent supports get_products but not proposal generation.

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 : Error | None

var model_config

var proposals_supported : bool

var reason : str

class SponsoredIntelligenceHandler

Expand source code

class SponsoredIntelligenceHandler(ADCPHandler):
    """Handler for Sponsored Intelligence protocol.

    Subclass this to implement a Sponsored Intelligence agent. All SI
    operations must be implemented via the handle_* methods.
    The public methods (si_get_offering, etc.) handle validation and
    error handling automatically.

    Non-SI operations (get_products, create_media_buy, content standards, etc.)
    return 'not supported' via the base class.

    Example:
        class MySIHandler(SponsoredIntelligenceHandler):
            async def handle_si_get_offering(
                self,
                request: SiGetOfferingRequest,
                context: ToolContext | None = None
            ) -> SiGetOfferingResponse:
                # Your implementation
                return SiGetOfferingResponse(...)
    """

    _agent_type: str = "Sponsored Intelligence agents"

    # ========================================================================
    # Sponsored Intelligence Operations - Override base class with validation
    # ========================================================================

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

        Validates params and delegates to handle_si_get_offering.
        """
        try:
            request = SiGetOfferingRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_si_get_offering(request, context)

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

        Validates params and delegates to handle_si_initiate_session.
        """
        try:
            request = SiInitiateSessionRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_si_initiate_session(request, context)

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

        Validates params and delegates to handle_si_send_message.
        """
        try:
            request = SiSendMessageRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_si_send_message(request, context)

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

        Validates params and delegates to handle_si_terminate_session.
        """
        try:
            request = SiTerminateSessionRequest.model_validate(params)
        except ValidationError as e:
            return NotImplementedResponse(
                supported=False,
                reason=f"Invalid request: {e}",
                error=Error(code="VALIDATION_ERROR", message=str(e)),
            )
        return await self.handle_si_terminate_session(request, context)

    # ========================================================================
    # Abstract handlers - Implement these in subclasses
    # ========================================================================

    @abstractmethod
    async def handle_si_get_offering(
        self,
        request: SiGetOfferingRequest,
        context: ToolContext | None = None,
    ) -> SiGetOfferingResponse:
        """Handle get offering request."""
        ...

    @abstractmethod
    async def handle_si_initiate_session(
        self,
        request: SiInitiateSessionRequest,
        context: ToolContext | None = None,
    ) -> SiInitiateSessionResponse:
        """Handle initiate session request."""
        ...

    @abstractmethod
    async def handle_si_send_message(
        self,
        request: SiSendMessageRequest,
        context: ToolContext | None = None,
    ) -> SiSendMessageResponse:
        """Handle send message request."""
        ...

    @abstractmethod
    async def handle_si_terminate_session(
        self,
        request: SiTerminateSessionRequest,
        context: ToolContext | None = None,
    ) -> SiTerminateSessionResponse:
        """Handle terminate session request."""
        ...

Handler for Sponsored Intelligence protocol.

Subclass this to implement a Sponsored Intelligence agent. All SI operations must be implemented via the handle_* methods. The public methods (si_get_offering, etc.) handle validation and error handling automatically.

Non-SI operations (get_products, create_media_buy, content standards, etc.) return 'not supported' via the base class.

Example

class MySIHandler(SponsoredIntelligenceHandler): async def handle_si_get_offering( self, request: SiGetOfferingRequest, context: ToolContext | None = None ) -> SiGetOfferingResponse: # Your implementation return SiGetOfferingResponse(…)

Ancestors

• ADCPHandler
• abc.ABC

Methods

async def handle_si_get_offering(self,
request: SiGetOfferingRequest,
context: ToolContext | None = None) ‑> SiGetOfferingResponse

Expand source code

@abstractmethod
async def handle_si_get_offering(
    self,
    request: SiGetOfferingRequest,
    context: ToolContext | None = None,
) -> SiGetOfferingResponse:
    """Handle get offering request."""
    ...

Handle get offering request.

async def handle_si_initiate_session(self,
request: SiInitiateSessionRequest,
context: ToolContext | None = None) ‑> SiInitiateSessionResponse

Expand source code

@abstractmethod
async def handle_si_initiate_session(
    self,
    request: SiInitiateSessionRequest,
    context: ToolContext | None = None,
) -> SiInitiateSessionResponse:
    """Handle initiate session request."""
    ...

Handle initiate session request.

async def handle_si_send_message(self,
request: SiSendMessageRequest,
context: ToolContext | None = None) ‑> SiSendMessageResponse

Expand source code

@abstractmethod
async def handle_si_send_message(
    self,
    request: SiSendMessageRequest,
    context: ToolContext | None = None,
) -> SiSendMessageResponse:
    """Handle send message request."""
    ...

Handle send message request.

async def handle_si_terminate_session(self,
request: SiTerminateSessionRequest,
context: ToolContext | None = None) ‑> SiTerminateSessionResponse

Expand source code

@abstractmethod
async def handle_si_terminate_session(
    self,
    request: SiTerminateSessionRequest,
    context: ToolContext | None = None,
) -> SiTerminateSessionResponse:
    """Handle terminate session request."""
    ...

Handle terminate session request.

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

Expand source code

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

    Validates params and delegates to handle_si_get_offering.
    """
    try:
        request = SiGetOfferingRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_si_get_offering(request, context)

Get sponsored intelligence offering.

Validates params and delegates to handle_si_get_offering.

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

Expand source code

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

    Validates params and delegates to handle_si_initiate_session.
    """
    try:
        request = SiInitiateSessionRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_si_initiate_session(request, context)

Initiate sponsored intelligence session.

Validates params and delegates to handle_si_initiate_session.

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

Expand source code

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

    Validates params and delegates to handle_si_send_message.
    """
    try:
        request = SiSendMessageRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_si_send_message(request, context)

Send message in sponsored intelligence session.

Validates params and delegates to handle_si_send_message.

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

Expand source code

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

    Validates params and delegates to handle_si_terminate_session.
    """
    try:
        request = SiTerminateSessionRequest.model_validate(params)
    except ValidationError as e:
        return NotImplementedResponse(
            supported=False,
            reason=f"Invalid request: {e}",
            error=Error(code="VALIDATION_ERROR", message=str(e)),
        )
    return await self.handle_si_terminate_session(request, context)

Terminate sponsored intelligence session.

Validates params and delegates to handle_si_terminate_session.

Inherited members

• ADCPHandler:
  • activate_signal
  • build_creative
  • calibrate_content
  • check_governance
  • create_content_standards
  • create_media_buy
  • create_property_list
  • delete_property_list
  • get_account_financials
  • get_adcp_capabilities
  • get_content_standards
  • get_creative_delivery
  • get_creative_features
  • get_media_buy_artifacts
  • get_media_buy_delivery
  • get_media_buys
  • get_plan_audit_logs
  • get_products
  • get_property_list
  • get_signals
  • list_accounts
  • list_content_standards
  • list_creative_formats
  • list_creatives
  • list_property_lists
  • log_event
  • preview_creative
  • provide_performance_feedback
  • report_plan_outcome
  • report_usage
  • sync_accounts
  • sync_audiences
  • sync_catalogs
  • sync_creatives
  • sync_event_sources
  • sync_plans
  • update_content_standards
  • update_media_buy
  • update_property_list
  • validate_content_delivery

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

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

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

Context passed to tool handlers.

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

Instance variables

var caller_identity : str | None

var metadata : dict[str, typing.Any]

var request_id : str | None