Google Ads MCP Server

"""Ad creation and management tools."""

from typing import Any
from mcp.server import Server

from ..google_ads_client import ads_client


def register_ad_tools(server: Server) -> None:
    """Register ad management tools with the MCP server.

    Args:
        server: The MCP server instance.
    """

    @server.tool()
    async def list_ads(
        ad_group_id: str | None = None,
        campaign_id: str | None = None,
    ) -> dict[str, Any]:
        """List ads, optionally filtered by ad group or campaign.

        Args:
            ad_group_id: Optional ad group ID to filter by.
            campaign_id: Optional campaign ID to filter by.

        Returns:
            Dictionary containing list of ads.
        """
        query = """
            SELECT
                ad_group_ad.ad.id,
                ad_group_ad.ad.type,
                ad_group_ad.ad.responsive_search_ad.headlines,
                ad_group_ad.ad.responsive_search_ad.descriptions,
                ad_group_ad.status,
                ad_group.id,
                ad_group.name,
                campaign.id,
                campaign.name
            FROM ad_group_ad
        """

        conditions = []
        if ad_group_id:
            conditions.append(f"ad_group.id = {ad_group_id}")
        if campaign_id:
            conditions.append(f"campaign.id = {campaign_id}")

        if conditions:
            query += " WHERE " + " AND ".join(conditions)

        query += " ORDER BY campaign.name, ad_group.name"

        try:
            results = ads_client.search(query)
            ads = []
            for row in results:
                ad_data = {
                    "id": str(row.ad_group_ad.ad.id),
                    "type": row.ad_group_ad.ad.type_.name,
                    "status": row.ad_group_ad.status.name,
                    "ad_group_id": str(row.ad_group.id),
                    "ad_group_name": row.ad_group.name,
                    "campaign_id": str(row.campaign.id),
                    "campaign_name": row.campaign.name,
                }

                # Extract RSA headlines and descriptions if available
                rsa = row.ad_group_ad.ad.responsive_search_ad
                if rsa:
                    ad_data["headlines"] = [h.text for h in rsa.headlines]
                    ad_data["descriptions"] = [d.text for d in rsa.descriptions]

                ads.append(ad_data)
            return {"ads": ads, "count": len(ads)}
        except Exception as e:
            return {"error": str(e), "ads": []}

    @server.tool()
    async def create_responsive_search_ad(
        ad_group_id: str,
        headlines: list[str],
        descriptions: list[str],
        final_url: str,
        path1: str | None = None,
        path2: str | None = None,
    ) -> dict[str, Any]:
        """Create a new Responsive Search Ad (RSA).

        Args:
            ad_group_id: The ad group ID to create the ad in.
            headlines: List of headline texts (3-15 headlines, max 30 chars each).
            descriptions: List of description texts (2-4 descriptions, max 90 chars each).
            final_url: The landing page URL.
            path1: Optional display URL path 1 (max 15 chars).
            path2: Optional display URL path 2 (max 15 chars).

        Returns:
            Dictionary with created ad details or error.
        """
        # Validate inputs
        if len(headlines) < 3 or len(headlines) > 15:
            return {"error": "RSA requires 3-15 headlines"}
        if len(descriptions) < 2 or len(descriptions) > 4:
            return {"error": "RSA requires 2-4 descriptions"}

        for i, h in enumerate(headlines):
            if len(h) > 30:
                return {"error": f"Headline {i+1} exceeds 30 character limit"}

        for i, d in enumerate(descriptions):
            if len(d) > 90:
                return {"error": f"Description {i+1} exceeds 90 character limit"}

        # TODO: Implement ad creation using mutate operations
        return {
            "status": "not_implemented",
            "message": "RSA creation will be implemented with Google Ads mutate operations.",
            "planned_config": {
                "ad_group_id": ad_group_id,
                "headlines": headlines,
                "descriptions": descriptions,
                "final_url": final_url,
                "path1": path1,
                "path2": path2,
            },
        }

    @server.tool()
    async def update_ad_status(
        ad_group_id: str,
        ad_id: str,
        status: str,
    ) -> dict[str, Any]:
        """Update the status of an ad.

        Args:
            ad_group_id: The ad group ID containing the ad.
            ad_id: The ad ID to update.
            status: New status (ENABLED, PAUSED, REMOVED).

        Returns:
            Dictionary with update result.
        """
        # TODO: Implement ad status update using mutate operations
        return {
            "status": "not_implemented",
            "message": "Ad status update will be implemented with Google Ads mutate operations.",
            "ad_group_id": ad_group_id,
            "ad_id": ad_id,
            "planned_status": status,
        }