"""Campaign management tools."""
from typing import Any
from mcp.server import Server
from ..google_ads_client import ads_client
def register_campaign_tools(server: Server) -> None:
"""Register campaign management tools with the MCP server.
Args:
server: The MCP server instance.
"""
@server.tool()
async def list_campaigns(
status_filter: str | None = None,
) -> dict[str, Any]:
"""List all campaigns in the Google Ads account.
Args:
status_filter: Optional filter by status (ENABLED, PAUSED, REMOVED).
Returns:
Dictionary containing list of campaigns with their details.
"""
query = """
SELECT
campaign.id,
campaign.name,
campaign.status,
campaign.advertising_channel_type,
campaign.bidding_strategy_type,
campaign_budget.amount_micros
FROM campaign
"""
if status_filter:
query += f" WHERE campaign.status = '{status_filter}'"
query += " ORDER BY campaign.name"
try:
results = ads_client.search(query)
campaigns = []
for row in results:
campaigns.append({
"id": str(row.campaign.id),
"name": row.campaign.name,
"status": row.campaign.status.name,
"channel_type": row.campaign.advertising_channel_type.name,
"bidding_strategy": row.campaign.bidding_strategy_type.name,
"budget_micros": row.campaign_budget.amount_micros,
})
return {"campaigns": campaigns, "count": len(campaigns)}
except Exception as e:
return {"error": str(e), "campaigns": []}
@server.tool()
async def get_campaign(campaign_id: str) -> dict[str, Any]:
"""Get detailed information about a specific campaign.
Args:
campaign_id: The campaign ID to retrieve.
Returns:
Dictionary containing campaign details.
"""
query = f"""
SELECT
campaign.id,
campaign.name,
campaign.status,
campaign.advertising_channel_type,
campaign.bidding_strategy_type,
campaign.start_date,
campaign.end_date,
campaign_budget.amount_micros,
campaign_budget.delivery_method
FROM campaign
WHERE campaign.id = {campaign_id}
"""
try:
results = ads_client.search(query)
for row in results:
return {
"id": str(row.campaign.id),
"name": row.campaign.name,
"status": row.campaign.status.name,
"channel_type": row.campaign.advertising_channel_type.name,
"bidding_strategy": row.campaign.bidding_strategy_type.name,
"start_date": row.campaign.start_date,
"end_date": row.campaign.end_date,
"budget_micros": row.campaign_budget.amount_micros,
"delivery_method": row.campaign_budget.delivery_method.name,
}
return {"error": f"Campaign {campaign_id} not found"}
except Exception as e:
return {"error": str(e)}
@server.tool()
async def create_campaign(
name: str,
budget_amount: float,
channel_type: str = "SEARCH",
bidding_strategy: str = "MAXIMIZE_CLICKS",
) -> dict[str, Any]:
"""Create a new campaign.
Args:
name: Campaign name.
budget_amount: Daily budget in account currency.
channel_type: Advertising channel (SEARCH, DISPLAY, SHOPPING, VIDEO).
bidding_strategy: Bidding strategy type.
Returns:
Dictionary with created campaign details or error.
"""
# TODO: Implement campaign creation using mutate operations
return {
"status": "not_implemented",
"message": "Campaign creation will be implemented with Google Ads mutate operations.",
"planned_config": {
"name": name,
"budget_amount": budget_amount,
"channel_type": channel_type,
"bidding_strategy": bidding_strategy,
},
}
@server.tool()
async def update_campaign(
campaign_id: str,
name: str | None = None,
budget_amount: float | None = None,
) -> dict[str, Any]:
"""Update an existing campaign.
Args:
campaign_id: The campaign ID to update.
name: New campaign name (optional).
budget_amount: New daily budget (optional).
Returns:
Dictionary with update result.
"""
# TODO: Implement campaign update using mutate operations
return {
"status": "not_implemented",
"message": "Campaign update will be implemented with Google Ads mutate operations.",
"campaign_id": campaign_id,
"planned_updates": {
"name": name,
"budget_amount": budget_amount,
},
}
@server.tool()
async def set_campaign_status(
campaign_id: str,
status: str,
) -> dict[str, Any]:
"""Set the status of a campaign (pause, enable, or remove).
Args:
campaign_id: The campaign ID to update.
status: New status (ENABLED, PAUSED, REMOVED).
Returns:
Dictionary with status change result.
"""
# TODO: Implement status change using mutate operations
return {
"status": "not_implemented",
"message": "Campaign status change will be implemented with Google Ads mutate operations.",
"campaign_id": campaign_id,
"planned_status": status,
}
