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 not_supported(reason: str = 'This operation is not supported by this agent') ‑> NotImplementedResponse

Expand source code

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

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

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

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

Create a standard 'not supported' response.

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

Args

reason

Human-readable explanation of why the operation is not supported

Returns

NotImplementedResponse with supported=False

Classes

class ADCPHandler

Expand source code

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

    # ========================================================================
    # 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 not_supported("get_products is not implemented by this agent")

    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 not_supported("list_creative_formats is not implemented by this agent")

    # ========================================================================
    # 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 not_supported("sync_creatives is not implemented by this agent")

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

        Override this to list synced creatives.
        """
        return not_supported("list_creatives is not implemented by this agent")

    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 not_supported("build_creative is not implemented by this agent")

    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 not_supported("preview_creative is not implemented by this agent")

    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 not_supported("get_creative_delivery is not implemented by this agent")

    # ========================================================================
    # 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 not_supported("create_media_buy is not implemented by this agent")

    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 not_supported("update_media_buy is not implemented by this agent")

    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 not_supported("get_media_buy_delivery is not implemented by this agent")

    # ========================================================================
    # 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 not_supported("get_signals is not implemented by this agent")

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

        Override this to handle signal activation.
        """
        return not_supported("activate_signal is not implemented by this agent")

    # ========================================================================
    # 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 not_supported("provide_performance_feedback is not implemented by this agent")

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

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

        Override this to provide functionality.
        """
        return not_supported("list_accounts is not implemented by this agent")

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

        Override this to provide functionality.
        """
        return not_supported("sync_accounts is not implemented by this agent")

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

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

        Override this to provide functionality.
        """
        return not_supported("log_event is not implemented by this agent")

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

        Override this to provide functionality.
        """
        return not_supported("sync_event_sources is not implemented by this agent")

    # ========================================================================
    # 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 not_supported("get_adcp_capabilities is not implemented by this agent")

    # ========================================================================
    # 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 not_supported("create_content_standards is not implemented by this agent")

    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 not_supported("get_content_standards is not implemented by this agent")

    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 not_supported("list_content_standards is not implemented by this agent")

    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 not_supported("update_content_standards is not implemented by this agent")

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

        Override this in ContentStandardsHandler subclasses.
        """
        return not_supported("calibrate_content is not implemented by this agent")

    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 not_supported("validate_content_delivery is not implemented by this agent")

    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 not_supported("get_media_buy_artifacts is not implemented by this agent")

    # ========================================================================
    # 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 not_supported("si_get_offering is not implemented by this agent")

    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 not_supported("si_initiate_session is not implemented by this agent")

    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 not_supported("si_send_message is not implemented by this agent")

    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 not_supported("si_terminate_session is not implemented by this agent")

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

    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 not_supported("create_property_list is not implemented by this agent")

    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 not_supported("get_property_list is not implemented by this agent")

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

        Override this in GovernanceHandler subclasses.
        """
        return not_supported("list_property_lists is not implemented by this agent")

    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 not_supported("update_property_list is not implemented by this agent")

    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 not_supported("delete_property_list is not implemented by this agent")

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

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 not_supported("activate_signal is not implemented by this agent")

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 not_supported("build_creative is not implemented by this agent")

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 not_supported("calibrate_content is not implemented by this agent")

Calibrate content against standards.

Override this in ContentStandardsHandler 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 not_supported("create_content_standards is not implemented by this agent")

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 not_supported("create_media_buy is not implemented by this agent")

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 not_supported("create_property_list is not implemented by this agent")

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 not_supported("delete_property_list is not implemented by this agent")

Delete a property list.

Override this in GovernanceHandler subclasses.

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 not_supported("get_adcp_capabilities is not implemented by this agent")

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 not_supported("get_content_standards is not implemented by this agent")

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 not_supported("get_creative_delivery is not implemented by this agent")

Get creative delivery metrics.

Override this to provide functionality.

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 not_supported("get_media_buy_artifacts is not implemented by this agent")

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 not_supported("get_media_buy_delivery is not implemented by this agent")

Get media buy delivery metrics.

Override this to provide delivery reporting.

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 not_supported("get_products is not implemented by this agent")

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 not_supported("get_property_list is not implemented by this agent")

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 not_supported("get_signals is not implemented by this agent")

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 not_supported("list_accounts is not implemented by this agent")

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 not_supported("list_content_standards is not implemented by this agent")

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 not_supported("list_creative_formats is not implemented by this agent")

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 not_supported("list_creatives is not implemented by this agent")

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 not_supported("list_property_lists is not implemented by this agent")

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 not_supported("log_event is not implemented by this agent")

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 not_supported("preview_creative is not implemented by this agent")

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 not_supported("provide_performance_feedback is not implemented by this agent")

Provide performance feedback.

Override this to handle performance feedback ingestion.

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 not_supported("si_get_offering is not implemented by this agent")

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 not_supported("si_initiate_session is not implemented by this agent")

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 not_supported("si_send_message is not implemented by this agent")

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 not_supported("si_terminate_session is not implemented by this agent")

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 not_supported("sync_accounts is not implemented by this agent")

Sync accounts.

Override this to provide functionality.

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 not_supported("sync_creatives is not implemented by this agent")

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 not_supported("sync_event_sources is not implemented by this agent")

Sync event sources.

Override this to provide functionality.

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 not_supported("update_content_standards is not implemented by this agent")

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 not_supported("update_media_buy is not implemented by this agent")

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 not_supported("update_property_list is not implemented by this agent")

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 not_supported("validate_content_delivery is not implemented by this agent")

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

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

    # ========================================================================
    # 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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Content standards creation response, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Content standards response, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            List of content standards, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Updated content standards response, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Calibration response with scores and feedback, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Validation response, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Media buy artifacts response, or error response
        """
        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.

        Must be implemented by Content Standards agents.

        Args:
            request: Validated content standards creation request
            context: Optional tool context

        Returns:
            Content standards creation response
        """
        ...

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

        Must be implemented by Content Standards agents.

        Args:
            request: Validated content standards retrieval request
            context: Optional tool context

        Returns:
            Content standards response
        """
        ...

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

        Must be implemented by Content Standards agents.

        Args:
            request: Validated list content standards request
            context: Optional tool context

        Returns:
            List of content standards
        """
        ...

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

        Must be implemented by Content Standards agents.

        Args:
            request: Validated content standards update request
            context: Optional tool context

        Returns:
            Updated content standards response
        """
        ...

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

        Must be implemented by Content Standards agents.

        Args:
            request: Validated calibration request with content to evaluate
            context: Optional tool context

        Returns:
            Calibration response with scores and feedback
        """
        ...

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

        Must be implemented by Content Standards agents.

        Args:
            request: Validated request with delivery data
            context: Optional tool context

        Returns:
            Validation response
        """
        ...

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

        Must be implemented by Content Standards agents.

        Args:
            request: Validated artifacts retrieval request
            context: Optional tool context

        Returns:
            Media buy artifacts response
        """
        ...

    # ========================================================================
    # Non-Content-Standards Operations - Return 'not supported'
    # ========================================================================

    async def get_products(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported(
            "get_products is not supported by Content Standards agents. "
            "This agent handles content calibration and validation, not product catalog operations."
        )

    async def list_creative_formats(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("list_creative_formats is not supported by Content Standards agents.")

    async def sync_creatives(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("sync_creatives is not supported by Content Standards agents.")

    async def list_creatives(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("list_creatives is not supported by Content Standards agents.")

    async def build_creative(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("build_creative is not supported by Content Standards agents.")

    async def preview_creative(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("preview_creative is not supported by Content Standards agents.")

    async def get_creative_delivery(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("get_creative_delivery is not supported by Content Standards agents.")

    async def create_media_buy(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported(
            "create_media_buy is not supported by Content Standards agents. "
            "This agent handles content calibration and validation, not media buying."
        )

    async def update_media_buy(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("update_media_buy is not supported by Content Standards agents.")

    async def get_media_buy_delivery(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("get_media_buy_delivery is not supported by Content Standards agents.")

    async def get_signals(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("get_signals is not supported by Content Standards agents.")

    async def activate_signal(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("activate_signal is not supported by Content Standards agents.")

    async def provide_performance_feedback(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported(
            "provide_performance_feedback is not supported by Content Standards agents."
        )

    async def list_accounts(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("list_accounts is not supported by Content Standards agents.")

    async def sync_accounts(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("sync_accounts is not supported by Content Standards agents.")

    async def log_event(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("log_event is not supported by Content Standards agents.")

    async def sync_event_sources(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("sync_event_sources is not supported by Content Standards agents.")

    # ========================================================================
    # V3 Sponsored Intelligence - Not supported
    # ========================================================================

    async def si_get_offering(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported(
            "si_get_offering is not supported by Content Standards agents. "
            "Use a Sponsored Intelligence agent for SI operations."
        )

    async def si_initiate_session(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("si_initiate_session is not supported by Content Standards agents.")

    async def si_send_message(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("si_send_message is not supported by Content Standards agents.")

    async def si_terminate_session(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("si_terminate_session is not supported by Content Standards agents.")

    # ========================================================================
    # V3 Governance (Property Lists) - Not supported
    # ========================================================================

    async def create_property_list(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported(
            "create_property_list is not supported by Content Standards agents. "
            "Use a Governance agent for property list operations."
        )

    async def get_property_list(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("get_property_list is not supported by Content Standards agents.")

    async def list_property_lists(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("list_property_lists is not supported by Content Standards agents.")

    async def update_property_list(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("update_property_list is not supported by Content Standards agents.")

    async def delete_property_list(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Content Standards agents."""
        return not_supported("delete_property_list is not supported by Content Standards agents.")

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

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

Expand source code

async def activate_signal(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("activate_signal is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def build_creative(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("build_creative is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Calibration response with scores and feedback, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Calibration response with scores and feedback, or error response

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

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Content standards creation response, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Content standards creation response, or error response

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

Expand source code

async def create_media_buy(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported(
        "create_media_buy is not supported by Content Standards agents. "
        "This agent handles content calibration and validation, not media buying."
    )

Not supported by Content Standards agents.

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

Expand source code

async def create_property_list(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported(
        "create_property_list is not supported by Content Standards agents. "
        "Use a Governance agent for property list operations."
    )

Not supported by Content Standards agents.

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

Expand source code

async def delete_property_list(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("delete_property_list is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Content standards response, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Content standards response, or error response

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

Expand source code

async def get_creative_delivery(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("get_creative_delivery is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Media buy artifacts response, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Media buy artifacts response, or error response

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

Expand source code

async def get_media_buy_delivery(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("get_media_buy_delivery is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def get_products(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported(
        "get_products is not supported by Content Standards agents. "
        "This agent handles content calibration and validation, not product catalog operations."
    )

Not supported by Content Standards agents.

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

Expand source code

async def get_property_list(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("get_property_list is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def get_signals(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("get_signals is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

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

    Must be implemented by Content Standards agents.

    Args:
        request: Validated calibration request with content to evaluate
        context: Optional tool context

    Returns:
        Calibration response with scores and feedback
    """
    ...

Handle calibrate content request.

Must be implemented by Content Standards agents.

Args

request

Validated calibration request with content to evaluate

context

Optional tool context

Returns

Calibration response with scores and feedback

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

Expand source code

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

    Must be implemented by Content Standards agents.

    Args:
        request: Validated content standards creation request
        context: Optional tool context

    Returns:
        Content standards creation response
    """
    ...

Handle create content standards request.

Must be implemented by Content Standards agents.

Args

request

Validated content standards creation request

context

Optional tool context

Returns

Content standards creation response

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

Expand source code

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

    Must be implemented by Content Standards agents.

    Args:
        request: Validated content standards retrieval request
        context: Optional tool context

    Returns:
        Content standards response
    """
    ...

Handle get content standards request.

Must be implemented by Content Standards agents.

Args

request

Validated content standards retrieval request

context

Optional tool context

Returns

Content standards response

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

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.

    Must be implemented by Content Standards agents.

    Args:
        request: Validated artifacts retrieval request
        context: Optional tool context

    Returns:
        Media buy artifacts response
    """
    ...

Handle get media buy artifacts request.

Must be implemented by Content Standards agents.

Args

request

Validated artifacts retrieval request

context

Optional tool context

Returns

Media buy artifacts response

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

Expand source code

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

    Must be implemented by Content Standards agents.

    Args:
        request: Validated list content standards request
        context: Optional tool context

    Returns:
        List of content standards
    """
    ...

Handle list content standards request.

Must be implemented by Content Standards agents.

Args

request

Validated list content standards request

context

Optional tool context

Returns

List of content standards

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

Expand source code

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

    Must be implemented by Content Standards agents.

    Args:
        request: Validated content standards update request
        context: Optional tool context

    Returns:
        Updated content standards response
    """
    ...

Handle update content standards request.

Must be implemented by Content Standards agents.

Args

request

Validated content standards update request

context

Optional tool context

Returns

Updated content standards response

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

Expand source code

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

    Must be implemented by Content Standards agents.

    Args:
        request: Validated request with delivery data
        context: Optional tool context

    Returns:
        Validation response
    """
    ...

Handle validate content delivery request.

Must be implemented by Content Standards agents.

Args

request

Validated request with delivery data

context

Optional tool context

Returns

Validation response

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

Expand source code

async def list_accounts(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("list_accounts is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        List of content standards, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

List of content standards, or error response

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

Expand source code

async def list_creative_formats(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("list_creative_formats is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def list_creatives(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("list_creatives is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def list_property_lists(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("list_property_lists is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def log_event(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("log_event is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def preview_creative(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("preview_creative is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def provide_performance_feedback(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported(
        "provide_performance_feedback is not supported by Content Standards agents."
    )

Not supported by Content Standards agents.

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

Expand source code

async def si_get_offering(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported(
        "si_get_offering is not supported by Content Standards agents. "
        "Use a Sponsored Intelligence agent for SI operations."
    )

Not supported by Content Standards agents.

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

Expand source code

async def si_initiate_session(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("si_initiate_session is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def si_send_message(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("si_send_message is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def si_terminate_session(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("si_terminate_session is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def sync_accounts(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("sync_accounts is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def sync_creatives(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("sync_creatives is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def sync_event_sources(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("sync_event_sources is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Updated content standards response, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Updated content standards response, or error response

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

Expand source code

async def update_media_buy(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("update_media_buy is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

Expand source code

async def update_property_list(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Content Standards agents."""
    return not_supported("update_property_list is not supported by Content Standards agents.")

Not supported by Content Standards agents.

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

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Validation response, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Validation response, or error response

Inherited members

• ADCPHandler:
  • get_adcp_capabilities

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

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

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

    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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Response with created property list metadata, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Response with list metadata and optionally resolved identifiers
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Response with array of property list metadata
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Response with updated 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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Response confirming deletion
        """
        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_create_property_list(
        self,
        request: CreatePropertyListRequest,
        context: ToolContext | None = None,
    ) -> CreatePropertyListResponse:
        """Handle create property list request.

        Must be implemented by Governance agents.

        Args:
            request: Validated property list creation request
            context: Optional tool context

        Returns:
            Response with created property list metadata
        """
        ...

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

        Must be implemented by Governance agents.

        When resolve=true, evaluates filters and returns matching property
        identifiers. Otherwise returns only metadata.

        Args:
            request: Validated request with list_id and optional resolve flag
            context: Optional tool context

        Returns:
            Response with list metadata and optionally resolved identifiers
        """
        ...

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

        Must be implemented by Governance agents.

        Args:
            request: Validated request with optional filtering and pagination
            context: Optional tool context

        Returns:
            Response with array of property list metadata
        """
        ...

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

        Must be implemented by Governance agents.

        Args:
            request: Validated request with list_id and updates
            context: Optional tool context

        Returns:
            Response with updated property list
        """
        ...

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

        Must be implemented by Governance agents.

        Args:
            request: Validated request with list_id
            context: Optional tool context

        Returns:
            Response confirming deletion
        """
        ...

    # ========================================================================
    # Non-Governance Operations - Return 'not supported'
    # ========================================================================

    async def get_products(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported(
            "get_products is not supported by Governance agents. "
            "This agent manages property lists for filtering, not product catalogs."
        )

    async def list_creative_formats(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("list_creative_formats is not supported by Governance agents.")

    async def sync_creatives(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("sync_creatives is not supported by Governance agents.")

    async def list_creatives(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("list_creatives is not supported by Governance agents.")

    async def build_creative(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("build_creative is not supported by Governance agents.")

    async def preview_creative(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("preview_creative is not supported by Governance agents.")

    async def get_creative_delivery(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("get_creative_delivery is not supported by Governance agents.")

    async def create_media_buy(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported(
            "create_media_buy is not supported by Governance agents. "
            "This agent manages property lists, not media buying."
        )

    async def update_media_buy(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("update_media_buy is not supported by Governance agents.")

    async def get_media_buy_delivery(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("get_media_buy_delivery is not supported by Governance agents.")

    async def get_signals(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("get_signals is not supported by Governance agents.")

    async def activate_signal(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("activate_signal is not supported by Governance agents.")

    async def provide_performance_feedback(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("provide_performance_feedback is not supported by Governance agents.")

    async def list_accounts(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("list_accounts is not supported by Governance agents.")

    async def sync_accounts(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("sync_accounts is not supported by Governance agents.")

    async def log_event(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("log_event is not supported by Governance agents.")

    async def sync_event_sources(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("sync_event_sources is not supported by Governance agents.")

    # ========================================================================
    # V3 Content Standards - Not supported
    # ========================================================================

    async def create_content_standards(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported(
            "create_content_standards is not supported by Governance agents. "
            "Use a Content Standards agent for content calibration."
        )

    async def get_content_standards(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("get_content_standards is not supported by Governance agents.")

    async def list_content_standards(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("list_content_standards is not supported by Governance agents.")

    async def update_content_standards(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("update_content_standards is not supported by Governance agents.")

    async def calibrate_content(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("calibrate_content is not supported by Governance agents.")

    async def validate_content_delivery(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("validate_content_delivery is not supported by Governance agents.")

    async def get_media_buy_artifacts(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("get_media_buy_artifacts is not supported by Governance agents.")

    # ========================================================================
    # V3 Sponsored Intelligence - Not supported
    # ========================================================================

    async def si_get_offering(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported(
            "si_get_offering is not supported by Governance agents. "
            "Use a Sponsored Intelligence agent for SI operations."
        )

    async def si_initiate_session(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("si_initiate_session is not supported by Governance agents.")

    async def si_send_message(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("si_send_message is not supported by Governance agents.")

    async def si_terminate_session(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Governance agents."""
        return not_supported("si_terminate_session is not supported by Governance agents.")

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

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

Expand source code

async def activate_signal(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("activate_signal is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def build_creative(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("build_creative is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def calibrate_content(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("calibrate_content is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def create_content_standards(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported(
        "create_content_standards is not supported by Governance agents. "
        "Use a Content Standards agent for content calibration."
    )

Not supported by Governance agents.

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

Expand source code

async def create_media_buy(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported(
        "create_media_buy is not supported by Governance agents. "
        "This agent manages property lists, not media buying."
    )

Not supported by Governance agents.

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Response with created property list metadata, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Response with created property list metadata, or error response

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Response confirming deletion
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Response confirming deletion

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

Expand source code

async def get_content_standards(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("get_content_standards is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def get_creative_delivery(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("get_creative_delivery is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def get_media_buy_artifacts(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("get_media_buy_artifacts is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def get_media_buy_delivery(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("get_media_buy_delivery is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def get_products(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported(
        "get_products is not supported by Governance agents. "
        "This agent manages property lists for filtering, not product catalogs."
    )

Not supported by Governance agents.

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Response with list metadata and optionally resolved identifiers
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Response with list metadata and optionally resolved identifiers

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

Expand source code

async def get_signals(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("get_signals is not supported by Governance agents.")

Not supported by Governance agents.

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.

    Must be implemented by Governance agents.

    Args:
        request: Validated property list creation request
        context: Optional tool context

    Returns:
        Response with created property list metadata
    """
    ...

Handle create property list request.

Must be implemented by Governance agents.

Args

request

Validated property list creation request

context

Optional tool context

Returns

Response with created property list metadata

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.

    Must be implemented by Governance agents.

    Args:
        request: Validated request with list_id
        context: Optional tool context

    Returns:
        Response confirming deletion
    """
    ...

Handle delete property list request.

Must be implemented by Governance agents.

Args

request

Validated request with list_id

context

Optional tool context

Returns

Response confirming deletion

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.

    Must be implemented by Governance agents.

    When resolve=true, evaluates filters and returns matching property
    identifiers. Otherwise returns only metadata.

    Args:
        request: Validated request with list_id and optional resolve flag
        context: Optional tool context

    Returns:
        Response with list metadata and optionally resolved identifiers
    """
    ...

Handle get property list request.

Must be implemented by Governance agents.

When resolve=true, evaluates filters and returns matching property identifiers. Otherwise returns only metadata.

Args

request

Validated request with list_id and optional resolve flag

context

Optional tool context

Returns

Response with list metadata and optionally resolved identifiers

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.

    Must be implemented by Governance agents.

    Args:
        request: Validated request with optional filtering and pagination
        context: Optional tool context

    Returns:
        Response with array of property list metadata
    """
    ...

Handle list property lists request.

Must be implemented by Governance agents.

Args

request

Validated request with optional filtering and pagination

context

Optional tool context

Returns

Response with array of property list metadata

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.

    Must be implemented by Governance agents.

    Args:
        request: Validated request with list_id and updates
        context: Optional tool context

    Returns:
        Response with updated property list
    """
    ...

Handle update property list request.

Must be implemented by Governance agents.

Args

request

Validated request with list_id and updates

context

Optional tool context

Returns

Response with updated property list

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

Expand source code

async def list_accounts(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("list_accounts is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def list_content_standards(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("list_content_standards is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def list_creative_formats(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("list_creative_formats is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def list_creatives(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("list_creatives is not supported by Governance agents.")

Not supported by Governance agents.

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Response with array of property list metadata
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Response with array of property list metadata

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

Expand source code

async def log_event(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("log_event is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def preview_creative(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("preview_creative is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def provide_performance_feedback(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("provide_performance_feedback is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def si_get_offering(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported(
        "si_get_offering is not supported by Governance agents. "
        "Use a Sponsored Intelligence agent for SI operations."
    )

Not supported by Governance agents.

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

Expand source code

async def si_initiate_session(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("si_initiate_session is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def si_send_message(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("si_send_message is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def si_terminate_session(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("si_terminate_session is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def sync_accounts(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("sync_accounts is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def sync_creatives(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("sync_creatives is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def sync_event_sources(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("sync_event_sources is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def update_content_standards(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("update_content_standards is not supported by Governance agents.")

Not supported by Governance agents.

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

Expand source code

async def update_media_buy(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("update_media_buy is not supported by Governance agents.")

Not supported by Governance agents.

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Response with updated 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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Response with updated property list

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

Expand source code

async def validate_content_delivery(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Governance agents."""
    return not_supported("validate_content_delivery is not supported by Governance agents.")

Not supported by Governance agents.

Inherited members

• ADCPHandler:
  • get_adcp_capabilities

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._tools: dict[str, Callable[[dict[str, Any]], Any]] = {}

        # Create tool callers for all methods
        for tool_def in ADCP_TOOL_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."""
        return ADCP_TOOL_DEFINITIONS.copy()

    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."""
    return ADCP_TOOL_DEFINITIONS.copy()

Get MCP tool definitions.

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

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

    # ========================================================================
    # 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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            SI offering response with capabilities and pricing, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Session initiation response with session ID, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Message response with AI-generated content, or error response
        """
        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.

        Args:
            params: Request parameters as dict
            context: Optional tool context

        Returns:
            Termination response with session summary, or error response
        """
        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.

        Must be implemented by Sponsored Intelligence agents.

        Args:
            request: Validated SI offering request
            context: Optional tool context

        Returns:
            SI offering response with capabilities and pricing
        """
        ...

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

        Must be implemented by Sponsored Intelligence agents.

        Args:
            request: Validated session initiation request
            context: Optional tool context

        Returns:
            Session initiation response with session ID
        """
        ...

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

        Must be implemented by Sponsored Intelligence agents.

        Args:
            request: Validated message request with session ID and content
            context: Optional tool context

        Returns:
            Message response with AI-generated content
        """
        ...

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

        Must be implemented by Sponsored Intelligence agents.

        Args:
            request: Validated session termination request
            context: Optional tool context

        Returns:
            Termination response with session summary
        """
        ...

    # ========================================================================
    # Non-SI Operations - Return 'not supported'
    # ========================================================================

    async def get_products(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "get_products is not supported by Sponsored Intelligence agents. "
            "This agent handles conversational AI sponsorship, not product catalog operations."
        )

    async def list_creative_formats(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "list_creative_formats is not supported by Sponsored Intelligence agents."
        )

    async def sync_creatives(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("sync_creatives is not supported by Sponsored Intelligence agents.")

    async def list_creatives(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("list_creatives is not supported by Sponsored Intelligence agents.")

    async def build_creative(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("build_creative is not supported by Sponsored Intelligence agents.")

    async def preview_creative(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("preview_creative is not supported by Sponsored Intelligence agents.")

    async def get_creative_delivery(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "get_creative_delivery is not supported by Sponsored Intelligence agents."
        )

    async def create_media_buy(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "create_media_buy is not supported by Sponsored Intelligence agents. "
            "SI sessions are initiated via si_initiate_session, not media buys."
        )

    async def update_media_buy(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("update_media_buy is not supported by Sponsored Intelligence agents.")

    async def get_media_buy_delivery(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "get_media_buy_delivery is not supported by Sponsored Intelligence agents."
        )

    async def get_signals(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("get_signals is not supported by Sponsored Intelligence agents.")

    async def activate_signal(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("activate_signal is not supported by Sponsored Intelligence agents.")

    async def provide_performance_feedback(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "provide_performance_feedback is not supported by Sponsored Intelligence agents."
        )

    async def list_accounts(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "list_accounts is not supported by Sponsored Intelligence agents."
        )

    async def sync_accounts(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "sync_accounts is not supported by Sponsored Intelligence agents."
        )

    async def log_event(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("log_event is not supported by Sponsored Intelligence agents.")

    async def sync_event_sources(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "sync_event_sources is not supported by Sponsored Intelligence agents."
        )

    # ========================================================================
    # V3 Content Standards - Not supported
    # ========================================================================

    async def create_content_standards(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "create_content_standards is not supported by Sponsored Intelligence agents. "
            "Use a Content Standards agent for content calibration."
        )

    async def get_content_standards(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "get_content_standards is not supported by Sponsored Intelligence agents."
        )

    async def list_content_standards(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "list_content_standards is not supported by Sponsored Intelligence agents."
        )

    async def update_content_standards(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "update_content_standards is not supported by Sponsored Intelligence agents."
        )

    async def calibrate_content(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("calibrate_content is not supported by Sponsored Intelligence agents.")

    async def validate_content_delivery(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "validate_content_delivery is not supported by Sponsored Intelligence agents."
        )

    async def get_media_buy_artifacts(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "get_media_buy_artifacts is not supported by Sponsored Intelligence agents."
        )

    # ========================================================================
    # V3 Governance (Property Lists) - Not supported
    # ========================================================================

    async def create_property_list(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "create_property_list is not supported by Sponsored Intelligence agents. "
            "Use a Governance agent for property list operations."
        )

    async def get_property_list(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported("get_property_list is not supported by Sponsored Intelligence agents.")

    async def list_property_lists(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "list_property_lists is not supported by Sponsored Intelligence agents."
        )

    async def update_property_list(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "update_property_list is not supported by Sponsored Intelligence agents."
        )

    async def delete_property_list(
        self, params: dict[str, Any], context: ToolContext | None = None
    ) -> NotImplementedResponse:
        """Not supported by Sponsored Intelligence agents."""
        return not_supported(
            "delete_property_list is not supported by Sponsored Intelligence agents."
        )

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

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

Expand source code

async def activate_signal(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("activate_signal is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def build_creative(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("build_creative is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def calibrate_content(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("calibrate_content is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def create_content_standards(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "create_content_standards is not supported by Sponsored Intelligence agents. "
        "Use a Content Standards agent for content calibration."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def create_media_buy(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "create_media_buy is not supported by Sponsored Intelligence agents. "
        "SI sessions are initiated via si_initiate_session, not media buys."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def create_property_list(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "create_property_list is not supported by Sponsored Intelligence agents. "
        "Use a Governance agent for property list operations."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def delete_property_list(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "delete_property_list is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def get_content_standards(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "get_content_standards is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def get_creative_delivery(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "get_creative_delivery is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def get_media_buy_artifacts(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "get_media_buy_artifacts is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def get_media_buy_delivery(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "get_media_buy_delivery is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def get_products(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "get_products is not supported by Sponsored Intelligence agents. "
        "This agent handles conversational AI sponsorship, not product catalog operations."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def get_property_list(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("get_property_list is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def get_signals(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("get_signals is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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.

    Must be implemented by Sponsored Intelligence agents.

    Args:
        request: Validated SI offering request
        context: Optional tool context

    Returns:
        SI offering response with capabilities and pricing
    """
    ...

Handle get offering request.

Must be implemented by Sponsored Intelligence agents.

Args

request

Validated SI offering request

context

Optional tool context

Returns

SI offering response with capabilities and pricing

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.

    Must be implemented by Sponsored Intelligence agents.

    Args:
        request: Validated session initiation request
        context: Optional tool context

    Returns:
        Session initiation response with session ID
    """
    ...

Handle initiate session request.

Must be implemented by Sponsored Intelligence agents.

Args

request

Validated session initiation request

context

Optional tool context

Returns

Session initiation response with session ID

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.

    Must be implemented by Sponsored Intelligence agents.

    Args:
        request: Validated message request with session ID and content
        context: Optional tool context

    Returns:
        Message response with AI-generated content
    """
    ...

Handle send message request.

Must be implemented by Sponsored Intelligence agents.

Args

request

Validated message request with session ID and content

context

Optional tool context

Returns

Message response with AI-generated content

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.

    Must be implemented by Sponsored Intelligence agents.

    Args:
        request: Validated session termination request
        context: Optional tool context

    Returns:
        Termination response with session summary
    """
    ...

Handle terminate session request.

Must be implemented by Sponsored Intelligence agents.

Args

request

Validated session termination request

context

Optional tool context

Returns

Termination response with session summary

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

Expand source code

async def list_accounts(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "list_accounts is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def list_content_standards(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "list_content_standards is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def list_creative_formats(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "list_creative_formats is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def list_creatives(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("list_creatives is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def list_property_lists(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "list_property_lists is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def log_event(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("log_event is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def preview_creative(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("preview_creative is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def provide_performance_feedback(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "provide_performance_feedback is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        SI offering response with capabilities and pricing, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

SI offering response with capabilities and pricing, or error response

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Session initiation response with session ID, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Session initiation response with session ID, or error response

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Message response with AI-generated content, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Message response with AI-generated content, or error response

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.

    Args:
        params: Request parameters as dict
        context: Optional tool context

    Returns:
        Termination response with session summary, or error response
    """
    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.

Args

params

Request parameters as dict

context

Optional tool context

Returns

Termination response with session summary, or error response

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

Expand source code

async def sync_accounts(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "sync_accounts is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def sync_creatives(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("sync_creatives is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def sync_event_sources(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "sync_event_sources is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def update_content_standards(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "update_content_standards is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def update_media_buy(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported("update_media_buy is not supported by Sponsored Intelligence agents.")

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def update_property_list(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "update_property_list is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

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

Expand source code

async def validate_content_delivery(
    self, params: dict[str, Any], context: ToolContext | None = None
) -> NotImplementedResponse:
    """Not supported by Sponsored Intelligence agents."""
    return not_supported(
        "validate_content_delivery is not supported by Sponsored Intelligence agents."
    )

Not supported by Sponsored Intelligence agents.

Inherited members

• ADCPHandler:
  • get_adcp_capabilities

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