"""Ad group management tools."""
from typing import Any
from mcp.server import Server
from ..google_ads_client import ads_client
def register_ad_group_tools(server: Server) -> None:
"""Register ad group management tools with the MCP server.
Args:
server: The MCP server instance.
"""
@server.tool()
async def list_ad_groups(
campaign_id: str | None = None,
status_filter: str | None = None,
) -> dict[str, Any]:
"""List ad groups, optionally filtered by campaign.
Args:
campaign_id: Optional campaign ID to filter by.
status_filter: Optional filter by status (ENABLED, PAUSED, REMOVED).
Returns:
Dictionary containing list of ad groups.
"""
query = """
SELECT
ad_group.id,
ad_group.name,
ad_group.status,
ad_group.type,
campaign.id,
campaign.name
FROM ad_group
"""
conditions = []
if campaign_id:
conditions.append(f"campaign.id = {campaign_id}")
if status_filter:
conditions.append(f"ad_group.status = '{status_filter}'")
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += " ORDER BY campaign.name, ad_group.name"
try:
results = ads_client.search(query)
ad_groups = []
for row in results:
ad_groups.append({
"id": str(row.ad_group.id),
"name": row.ad_group.name,
"status": row.ad_group.status.name,
"type": row.ad_group.type_.name,
"campaign_id": str(row.campaign.id),
"campaign_name": row.campaign.name,
})
return {"ad_groups": ad_groups, "count": len(ad_groups)}
except Exception as e:
return {"error": str(e), "ad_groups": []}
@server.tool()
async def create_ad_group(
campaign_id: str,
name: str,
cpc_bid_micros: int | None = None,
) -> dict[str, Any]:
"""Create a new ad group in a campaign.
Args:
campaign_id: The campaign ID to create the ad group in.
name: Ad group name.
cpc_bid_micros: Default CPC bid in micros (optional).
Returns:
Dictionary with created ad group details or error.
"""
# TODO: Implement ad group creation using mutate operations
return {
"status": "not_implemented",
"message": "Ad group creation will be implemented with Google Ads mutate operations.",
"planned_config": {
"campaign_id": campaign_id,
"name": name,
"cpc_bid_micros": cpc_bid_micros,
},
}
@server.tool()
async def update_ad_group(
ad_group_id: str,
name: str | None = None,
cpc_bid_micros: int | None = None,
status: str | None = None,
) -> dict[str, Any]:
"""Update an existing ad group.
Args:
ad_group_id: The ad group ID to update.
name: New ad group name (optional).
cpc_bid_micros: New default CPC bid in micros (optional).
status: New status (ENABLED, PAUSED, REMOVED) (optional).
Returns:
Dictionary with update result.
"""
# TODO: Implement ad group update using mutate operations
return {
"status": "not_implemented",
"message": "Ad group update will be implemented with Google Ads mutate operations.",
"ad_group_id": ad_group_id,
"planned_updates": {
"name": name,
"cpc_bid_micros": cpc_bid_micros,
"status": status,
},
}
