Module adcp.server.governance

Governance protocol handler.

Provides a base class for implementing Governance agents that manage property lists for brand safety, compliance, and quality filtering.

Classes

class GovernanceHandler

Expand source code

class GovernanceHandler(ADCPHandler[TContext], Generic[TContext]):
    """Handler for Governance protocol (Property Lists).

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

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

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

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

    _agent_type: str = "Governance agents"

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Handler for Governance protocol (Property Lists).

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

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

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

Example

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

Ancestors

• ADCPHandler
• abc.ABC
• typing.Generic

Methods

async def check_governance(self,
params: CheckGovernanceRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.governance.check_governance_response.CheckGovernanceResponse | NotImplementedResponse

Expand source code

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

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

async def create_property_list(self,
params: CreatePropertyListRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.property.create_property_list_response.CreatePropertyListResponse | NotImplementedResponse

Expand source code

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

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

Create a property list for governance filtering.

Validates params and delegates to handle_create_property_list.

async def delete_property_list(self,
params: DeletePropertyListRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.property.delete_property_list_response.DeletePropertyListResponse | NotImplementedResponse

Expand source code

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

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

Delete a property list.

Validates params and delegates to handle_delete_property_list.

async def get_creative_features(self,
params: GetCreativeFeaturesRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.creative.get_creative_features_response.GetCreativeFeaturesResponse1 | adcp.types.generated_poc.creative.get_creative_features_response.GetCreativeFeaturesResponse2 | NotImplementedResponse

Expand source code

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

Evaluate governance features for a creative manifest.

async def get_plan_audit_logs(self,
params: GetPlanAuditLogsRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.governance.get_plan_audit_logs_response.GetPlanAuditLogsResponse | NotImplementedResponse

Expand source code

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

Retrieve governance audit logs for one or more plans.

async def get_property_list(self,
params: GetPropertyListRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.property.get_property_list_response.GetPropertyListResponse | NotImplementedResponse

Expand source code

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

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

Get a property list with optional resolution.

Validates params and delegates to handle_get_property_list.

async def handle_check_governance(self, request: CheckGovernanceRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.governance.check_governance_response.CheckGovernanceResponse

Expand source code

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

Handle a governance check request.

async def handle_create_property_list(self, request: CreatePropertyListRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.property.create_property_list_response.CreatePropertyListResponse

Expand source code

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

Handle create property list request.

async def handle_delete_property_list(self, request: DeletePropertyListRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.property.delete_property_list_response.DeletePropertyListResponse

Expand source code

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

Handle delete property list request.

async def handle_get_creative_features(self, request: GetCreativeFeaturesRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.creative.get_creative_features_response.GetCreativeFeaturesResponse1 | adcp.types.generated_poc.creative.get_creative_features_response.GetCreativeFeaturesResponse2

Expand source code

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

Handle creative feature evaluation.

async def handle_get_plan_audit_logs(self, request: GetPlanAuditLogsRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.governance.get_plan_audit_logs_response.GetPlanAuditLogsResponse

Expand source code

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

Handle retrieval of governance audit logs.

async def handle_get_property_list(self, request: GetPropertyListRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.property.get_property_list_response.GetPropertyListResponse

Expand source code

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

Handle get property list request.

async def handle_list_property_lists(self, request: ListPropertyListsRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.property.list_property_lists_response.ListPropertyListsResponse

Expand source code

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

Handle list property lists request.

async def handle_report_plan_outcome(self, request: ReportPlanOutcomeRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.governance.report_plan_outcome_response.ReportPlanOutcomeResponse

Expand source code

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

Handle reporting of a governed action outcome.

async def handle_sync_plans(self, request: SyncPlansRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.governance.sync_plans_response.SyncPlansResponse

Expand source code

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

Handle campaign governance plan sync.

async def handle_update_property_list(self, request: UpdatePropertyListRequest, context: TContext | None = None) ‑> adcp.types.generated_poc.property.update_property_list_response.UpdatePropertyListResponse

Expand source code

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

Handle update property list request.

async def list_property_lists(self,
params: ListPropertyListsRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.property.list_property_lists_response.ListPropertyListsResponse | NotImplementedResponse

Expand source code

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

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

List property lists.

Validates params and delegates to handle_list_property_lists.

async def report_plan_outcome(self,
params: ReportPlanOutcomeRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.governance.report_plan_outcome_response.ReportPlanOutcomeResponse | NotImplementedResponse

Expand source code

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

Report the outcome of a previously governed action.

async def sync_plans(self,
params: SyncPlansRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.governance.sync_plans_response.SyncPlansResponse | NotImplementedResponse

Expand source code

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

Sync campaign governance plans to the agent.

async def update_property_list(self,
params: UpdatePropertyListRequest | dict[str, Any],
context: TContext | None = None) ‑> adcp.types.generated_poc.property.update_property_list_response.UpdatePropertyListResponse | NotImplementedResponse

Expand source code

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

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

Update a property list.

Validates params and delegates to handle_update_property_list.

Inherited members

• ADCPHandler:
  • acquire_rights
  • activate_signal
  • build_creative
  • calibrate_content
  • comply_test_controller
  • context_match
  • create_collection_list
  • create_content_standards
  • create_media_buy
  • delete_collection_list
  • get_account_financials
  • get_adcp_capabilities
  • get_brand_identity
  • get_collection_list
  • get_content_standards
  • get_creative_delivery
  • get_media_buy_artifacts
  • get_media_buy_delivery
  • get_media_buys
  • get_products
  • get_rights
  • get_signals
  • get_task_status
  • identity_match
  • list_accounts
  • list_collection_lists
  • list_content_standards
  • list_creative_formats
  • list_creatives
  • list_tasks
  • list_transformers
  • log_event
  • preview_creative
  • provide_performance_feedback
  • report_usage
  • si_get_offering
  • si_initiate_session
  • si_send_message
  • si_terminate_session
  • sync_accounts
  • sync_audiences
  • sync_catalogs
  • sync_creatives
  • sync_event_sources
  • sync_governance
  • update_collection_list
  • update_content_standards
  • update_media_buy
  • update_rights
  • validate_content_delivery
  • validate_input
  • verify_brand_claim
  • verify_brand_claims