"""Keyword management tools."""
from typing import Any
from mcp.server import Server
from ..google_ads_client import ads_client
def register_keyword_tools(server: Server) -> None:
"""Register keyword management tools with the MCP server.
Args:
server: The MCP server instance.
"""
@server.tool()
async def list_keywords(
ad_group_id: str | None = None,
campaign_id: str | None = None,
status_filter: str | None = None,
) -> dict[str, Any]:
"""List keywords, 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.
status_filter: Optional filter by status (ENABLED, PAUSED, REMOVED).
Returns:
Dictionary containing list of keywords.
"""
query = """
SELECT
ad_group_criterion.criterion_id,
ad_group_criterion.keyword.text,
ad_group_criterion.keyword.match_type,
ad_group_criterion.status,
ad_group_criterion.cpc_bid_micros,
ad_group.id,
ad_group.name,
campaign.id,
campaign.name
FROM ad_group_criterion
WHERE ad_group_criterion.type = 'KEYWORD'
"""
if ad_group_id:
query += f" AND ad_group.id = {ad_group_id}"
if campaign_id:
query += f" AND campaign.id = {campaign_id}"
if status_filter:
query += f" AND ad_group_criterion.status = '{status_filter}'"
query += " ORDER BY campaign.name, ad_group.name, ad_group_criterion.keyword.text"
try:
results = ads_client.search(query)
keywords = []
for row in results:
keywords.append({
"id": str(row.ad_group_criterion.criterion_id),
"text": row.ad_group_criterion.keyword.text,
"match_type": row.ad_group_criterion.keyword.match_type.name,
"status": row.ad_group_criterion.status.name,
"cpc_bid_micros": row.ad_group_criterion.cpc_bid_micros,
"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,
})
return {"keywords": keywords, "count": len(keywords)}
except Exception as e:
return {"error": str(e), "keywords": []}
@server.tool()
async def add_keywords(
ad_group_id: str,
keywords: list[dict[str, Any]],
) -> dict[str, Any]:
"""Add keywords to an ad group.
Args:
ad_group_id: The ad group ID to add keywords to.
keywords: List of keyword configs, each with:
- text: The keyword text
- match_type: BROAD, PHRASE, or EXACT
- cpc_bid_micros: Optional CPC bid in micros
Returns:
Dictionary with add result.
"""
# Validate inputs
valid_match_types = {"BROAD", "PHRASE", "EXACT"}
for kw in keywords:
if "text" not in kw:
return {"error": "Each keyword must have a 'text' field"}
if kw.get("match_type", "BROAD") not in valid_match_types:
return {"error": f"Invalid match_type. Must be one of: {valid_match_types}"}
# TODO: Implement keyword addition using mutate operations
return {
"status": "not_implemented",
"message": "Keyword addition will be implemented with Google Ads mutate operations.",
"ad_group_id": ad_group_id,
"planned_keywords": keywords,
}
@server.tool()
async def update_keyword_bid(
ad_group_id: str,
keyword_id: str,
cpc_bid_micros: int,
) -> dict[str, Any]:
"""Update the CPC bid for a keyword.
Args:
ad_group_id: The ad group ID containing the keyword.
keyword_id: The keyword criterion ID.
cpc_bid_micros: New CPC bid in micros.
Returns:
Dictionary with update result.
"""
# TODO: Implement bid update using mutate operations
return {
"status": "not_implemented",
"message": "Keyword bid update will be implemented with Google Ads mutate operations.",
"ad_group_id": ad_group_id,
"keyword_id": keyword_id,
"planned_bid_micros": cpc_bid_micros,
}
@server.tool()
async def pause_keyword(
ad_group_id: str,
keyword_id: str,
) -> dict[str, Any]:
"""Pause a keyword.
Args:
ad_group_id: The ad group ID containing the keyword.
keyword_id: The keyword criterion ID to pause.
Returns:
Dictionary with pause result.
"""
# TODO: Implement keyword pause using mutate operations
return {
"status": "not_implemented",
"message": "Keyword pause will be implemented with Google Ads mutate operations.",
"ad_group_id": ad_group_id,
"keyword_id": keyword_id,
"planned_status": "PAUSED",
}
