Module adcp.server.sponsored_intelligence

Sponsored Intelligence protocol handler.

Provides a base class for implementing Sponsored Intelligence agents. Non-SI operations return 'not supported' via the base class.

Classes

class SponsoredIntelligenceHandler

Expand source code

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

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

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

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

    _agent_type: str = "Sponsored Intelligence agents"

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Handler for Sponsored Intelligence protocol.

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

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

Example

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

Ancestors

• ADCPHandler
• abc.ABC

Methods

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

Expand source code

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

Handle get offering request.

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

Expand source code

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

Handle initiate session request.

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

Expand source code

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

Handle send message request.

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

Expand source code

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

Handle terminate session request.

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

Expand source code

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

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

Get sponsored intelligence offering.

Validates params and delegates to handle_si_get_offering.

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

Expand source code

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

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

Initiate sponsored intelligence session.

Validates params and delegates to handle_si_initiate_session.

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

Expand source code

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

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

Send message in sponsored intelligence session.

Validates params and delegates to handle_si_send_message.

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

Expand source code

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

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

Terminate sponsored intelligence session.

Validates params and delegates to handle_si_terminate_session.

Inherited members

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