Prebid Sales Agent - Building a Custom Adapter

from src.adapters.base import AdServerAdapter

class MyAdapter(AdServerAdapter):

    async def create_media_buy(self, media_buy, products, creatives=None):
        """Push a new media buy to the ad server.

        Creates the corresponding campaign/order/flight in the ad server
        and returns a result with the external IDs.

        Args:
            media_buy: The MediaBuy object to create.
            products: List of Product objects referenced by the media buy.
            creatives: Optional list of Creative objects to attach.

        Returns:
            MediaBuyResult with ad server IDs and status.
        """
        ...

    async def update_media_buy(self, media_buy, products, creatives=None):
        """Modify an existing media buy in the ad server.

        Updates the campaign/order/flight with changed fields
        (budget, dates, targeting, etc.).

        Args:
            media_buy: The updated MediaBuy object.
            products: List of Product objects referenced by the media buy.
            creatives: Optional list of Creative objects to attach.

        Returns:
            MediaBuyResult with updated status.
        """
        ...

    async def get_media_buy_delivery(self, media_buy):
        """Fetch delivery metrics for a media buy.

        Pulls impressions, clicks, spend, and other delivery data
        from the ad server.

        Args:
            media_buy: The MediaBuy to get delivery for.

        Returns:
            DeliveryResult with current metrics.
        """
        ...

    async def check_media_buy_status(self, media_buy):
        """Check the current status of a media buy in the ad server.

        Returns whether the campaign is active, paused, completed, etc.

        Args:
            media_buy: The MediaBuy to check.

        Returns:
            StatusResult with current ad server status.
        """
        ...

    async def upload_creatives(self, media_buy, creatives):
        """Upload creative assets to the ad server.

        Uploads files, creates creative entities, and associates
        them with the campaign.

        Args:
            media_buy: The MediaBuy to associate creatives with.
            creatives: List of Creative objects to upload.

        Returns:
            CreativeUploadResult with ad server creative IDs.
        """
        ...

from typing import ClassVar
from src.adapters.base import (
    AdServerAdapter,
    AdapterCapabilities,
    TargetingCapabilities,
)


class MyAdServer(AdServerAdapter):
    adapter_name: ClassVar[str] = "my_ad_server"
    default_channels: list[str] = ["display"]
    default_delivery_measurement: dict = {
        "impressions": True,
        "clicks": True,
        "spend": True,
    }

    async def create_media_buy(self, media_buy, products, creatives=None):
        # Your implementation here
        ...

    async def update_media_buy(self, media_buy, products, creatives=None):
        ...

    async def get_media_buy_delivery(self, media_buy):
        ...

    async def check_media_buy_status(self, media_buy):
        ...

    async def upload_creatives(self, media_buy, creatives):
        ...

from pydantic import BaseModel, Field
from src.adapters.base import BaseConnectionConfig


class MyAdServerConnectionConfig(BaseConnectionConfig):
    api_url: str = Field(..., description="Base URL for the ad server API")
    api_key: str = Field(..., description="API authentication key")
    network_id: int = Field(..., description="Ad server network identifier")
    timeout_seconds: int = Field(default=30, description="API request timeout")


class MyAdServer(AdServerAdapter):
    adapter_name: ClassVar[str] = "my_ad_server"
    connection_config_class = MyAdServerConnectionConfig
    ...

class MyAdServer(AdServerAdapter):
    adapter_name: ClassVar[str] = "my_ad_server"

    capabilities = AdapterCapabilities(
        supports_inventory_sync=False,
        supports_inventory_profiles=False,
        inventory_entity_label="placement",
        supports_custom_targeting=True,
        supports_geo_targeting=True,
        supports_dynamic_products=False,
        supported_pricing_models=["cpm", "cpc"],
        supports_webhooks=False,
        supports_realtime_reporting=False,
    )

class MyAdServer(AdServerAdapter):
    ...

    def get_targeting_capabilities(self) -> TargetingCapabilities:
        return TargetingCapabilities(
            geo_countries=True,
            geo_regions=True,
            geo_nielsen_dma=False,
            geo_eurostat_nuts2=False,
            geo_uk_itl1=False,
            geo_uk_itl2=False,
            postal_us_zip=True,
            postal_us_zip4=False,
            postal_canadian=False,
            postal_uk=False,
            postal_german=False,
            postal_french=False,
            postal_australian=False,
        )

    def validate_media_buy_request(self, media_buy, products):
        """Validate targeting rules against adapter capabilities."""
        capabilities = self.get_targeting_capabilities()
        # Check that requested targeting dimensions are supported
        # Raise AdCPAdapterError if validation fails
        ...

from src.adapters.my_ad_server import MyAdServer

AVAILABLE_ADAPTERS = {
    "google_ad_manager": GoogleAdManager,
    "kevel": Kevel,
    "triton_digital": TritonDigital,
    "broadstreet": Broadstreet,
    "mock": MockAdServer,
    "my_ad_server": MyAdServer,  # Add your adapter here
}

from src.adapters.base import AdServerAdapter
from src.adapters.base_inventory import BaseInventoryManager


class MyAdServer(AdServerAdapter, BaseInventoryManager):
    adapter_name: ClassVar[str] = "my_ad_server"

    async def discover_inventory(self):
        # Fetch ad units from your ad server API
        return await self._client.list_ad_units()

    async def validate_inventory_ids(self, ids: list[str]):
        # Verify each ID exists in the ad server
        ...

    async def build_inventory_response(self):
        # Format for the Sales Agent API
        ...

    async def suggest_products(self):
        # Generate product suggestions from inventory
        ...

from src.adapters.base import AdServerAdapter
from src.adapters.base_workflow import BaseWorkflowManager


class MyAdServer(AdServerAdapter, BaseWorkflowManager):
    adapter_name: ClassVar[str] = "my_ad_server"

    async def create_media_buy(self, media_buy, products, creatives=None):
        # Require approval before pushing to ad server
        step = await self.create_workflow_step(
            step_type="media_buy_approval",
            owner="ops_team",
            request_data={"media_buy_id": media_buy.id, "budget": media_buy.budget},
        )
        return {"status": "pending_approval", "workflow_step_id": step.id}

from src.adapters.base import AdCPAdapterError

class MyAdServer(AdServerAdapter):
    async def create_media_buy(self, media_buy, products, creatives=None):
        try:
            response = await self._call_api(...)
        except TimeoutError:
            raise AdCPAdapterError(
                "Ad server API timed out",
                recovery="retry",  # Transient error, safe to retry
            )
        except AuthError:
            raise AdCPAdapterError(
                "Invalid API credentials",
                recovery="permanent",  # Not retryable
            )

import pytest
from unittest.mock import AsyncMock, patch
from src.adapters.my_ad_server import MyAdServer


@pytest.fixture
def adapter():
    config = {
        "api_url": "https://api.example.com",
        "api_key": "test-key",
        "network_id": 12345,
    }
    return MyAdServer(config=config, dry_run=True)


async def test_create_media_buy(adapter, sample_media_buy, sample_products):
    with patch.object(adapter, "_call_api", new_callable=AsyncMock) as mock_api:
        mock_api.return_value = {"order_id": "ext-001", "status": "created"}
        result = await adapter.create_media_buy(sample_media_buy, sample_products)
        assert result.external_id == "ext-001"
        mock_api.assert_called_once()

@pytest.mark.integration
async def test_end_to_end_flow(live_adapter, sample_media_buy, sample_products):
    # Create
    create_result = await live_adapter.create_media_buy(
        sample_media_buy, sample_products
    )
    assert create_result.external_id is not None

    # Check status
    status = await live_adapter.check_media_buy_status(sample_media_buy)
    assert status.state in ("draft", "pending")

    # Get delivery (may be zero for new campaigns)
    delivery = await live_adapter.get_media_buy_delivery(sample_media_buy)
    assert delivery.impressions >= 0

from typing import ClassVar
from src.adapters.base import (
    AdServerAdapter,
    AdapterCapabilities,
    AdCPAdapterError,
)


class MinimalAdapter(AdServerAdapter):
    """A minimal adapter implementation for reference."""

    adapter_name: ClassVar[str] = "minimal"
    default_channels: list[str] = ["display"]
    default_delivery_measurement: dict = {
        "impressions": True,
        "clicks": True,
        "spend": True,
    }

    capabilities = AdapterCapabilities(
        supports_inventory_sync=False,
        supports_inventory_profiles=False,
        inventory_entity_label="placement",
        supports_custom_targeting=False,
        supports_geo_targeting=False,
        supports_dynamic_products=False,
        supported_pricing_models=["cpm"],
        supports_webhooks=False,
        supports_realtime_reporting=False,
    )

    async def create_media_buy(self, media_buy, products, creatives=None):
        self.log(f"Creating media buy {media_buy.id}")
        try:
            # Call your ad server API to create the campaign
            external_id = await self._create_campaign(media_buy, products)
            return {"external_id": external_id, "status": "created"}
        except Exception as e:
            raise AdCPAdapterError(
                f"Failed to create campaign: {e}",
                recovery="retry",
            )

    async def update_media_buy(self, media_buy, products, creatives=None):
        self.log(f"Updating media buy {media_buy.id}")
        # Update the campaign in your ad server
        return {"status": "updated"}

    async def get_media_buy_delivery(self, media_buy):
        self.log(f"Fetching delivery for {media_buy.id}")
        # Pull delivery metrics from your ad server
        return {
            "impressions": 0,
            "clicks": 0,
            "spend": 0.0,
        }

    async def check_media_buy_status(self, media_buy):
        self.log(f"Checking status for {media_buy.id}")
        # Check campaign status in your ad server
        return {"state": "active"}

    async def upload_creatives(self, media_buy, creatives):
        self.log(f"Uploading {len(creatives)} creatives for {media_buy.id}")
        # Upload creative assets to your ad server
        return {"uploaded": len(creatives)}

    async def _create_campaign(self, media_buy, products):
        """Private helper to call the ad server API."""
        # Your API integration logic here
        raise NotImplementedError("Replace with real API call")