"""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,
}
