Module adcp.server.content_standards

Content Standards protocol handler.

Provides a base class for implementing Content Standards agents. Non-Content-Standards operations return 'not supported' via the base class.

Classes

class ContentStandardsHandler

Expand source code

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

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

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

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

    _agent_type: str = "Content Standards agents"

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Handler for Content Standards protocol.

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

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

Example

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

Ancestors

• ADCPHandler
• abc.ABC

Methods

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

Expand source code

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

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

Calibrate content against standards.

Validates params and delegates to handle_calibrate_content.

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

Expand source code

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

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

Create content standards configuration.

Validates params and delegates to handle_create_content_standards.

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

Expand source code

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

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

Get content standards configuration.

Validates params and delegates to handle_get_content_standards.

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

Expand source code

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

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

Get artifacts associated with a media buy.

Validates params and delegates to handle_get_media_buy_artifacts.

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

Expand source code

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

Handle calibrate content request.

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

Expand source code

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

Handle create content standards request.

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

Expand source code

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

Handle get content standards request.

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

Expand source code

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

Handle get media buy artifacts request.

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

Expand source code

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

Handle list content standards request.

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

Expand source code

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

Handle update content standards request.

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

Expand source code

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

Handle validate content delivery request.

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

Expand source code

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

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

List content standards configurations.

Validates params and delegates to handle_list_content_standards.

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

Expand source code

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

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

Update content standards configuration.

Validates params and delegates to handle_update_content_standards.

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

Expand source code

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

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

Validate content delivery against standards.

Validates params and delegates to handle_validate_content_delivery.

Inherited members

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