Google Ads MCP Server

import os import json import logging from typing import Any from datetime import datetime, timedelta from mcp.server import Server from mcp.types import Tool import mcp.types as types from google.ads.googleads.client import GoogleAdsClient import csv from io import StringIO # Setup logging logging.basicConfig(level=logging.ERROR) logger = logging.getLogger(__name__) # Configuration from environment variables DEVELOPER_TOKEN = os.getenv("GOOGLE_ADS_DEVELOPER_TOKEN", "") LOGIN_CUSTOMER_ID = os.getenv("GOOGLE_ADS_LOGIN_CUSTOMER_ID", "") CLIENT_ID = os.getenv("GOOGLE_ADS_CLIENT_ID", "") CLIENT_SECRET = os.getenv("GOOGLE_ADS_CLIENT_SECRET", "") REFRESH_TOKEN = os.getenv("GOOGLE_ADS_REFRESH_TOKEN", "") # Create MCP server server = Server("google-ads-mcp") # GAQL Help documentation GAQL_HELP = { "overview": """Google Ads Query Language (GAQL) is used to query Google Ads data. Basic syntax: SELECT [fields] FROM [resource] WHERE [conditions] Common resources: campaign, ad_group, ad_group_ad, ad_group_criterion, customer, etc. Example: SELECT campaign.id, campaign.name, metrics.clicks FROM campaign WHERE campaign.status = 'ENABLED'""", "resources": """Key GAQL resources: - campaign: Campaign data - ad_group: Ad group data - ad_group_ad: Ads within ad groups - ad_group_criterion: Keywords, placements, audiences - customer: Account information - feed: Feeds for extensions - asset: Ad assets - conversion_action: Conversion tracking - accessible_bidding_strategy: Bidding strategies - campaign_criterion: Campaign-level targeting""", "metrics": """Common metrics: - metrics.impressions: Number of impressions - metrics.clicks: Number of clicks - metrics.conversions: Conversion count - metrics.cost_micros: Cost in micros (divide by 1,000,000 for currency) - metrics.ctr: Click-through rate - metrics.conversion_rate: Conversion rate - metrics.average_cpc: Average cost per click - metrics.average_cpm: Average cost per mille - metrics.quality_score: Quality score (1-10) - metrics.absolute_top_impression_percentage: % at top position""", "segments": """Segmentation dimensions: - segments.date: By date - segments.device: Mobile, Tablet, Desktop, Unknown - segments.geo_target_city: By city - segments.geo_target_country: By country - segments.language: By language - segments.age_range: Age ranges (18-24, 25-34, etc) - segments.gender: By gender - segments.ad_type: Type of ad - segments.day_of_week: Day of week - segments.hour_of_day: Hour of day""", "filters": """WHERE clause conditions: - Equality: field = 'value' - Inequality: field != 'value' - Comparison: field > 100, field < 50 - IN operator: field IN ('value1', 'value2') - LIKE: field LIKE '%pattern%' - DURING: segments.date DURING LAST_7_DAYS Date options: LAST_7_DAYS, LAST_30_DAYS, THIS_MONTH, LAST_MONTH, LAST_QUARTER, LAST_YEAR - STARTS_WITH: field STARTS_WITH 'prefix'""", "best_practices": """GAQL best practices: - Always filter by date when querying metrics - Use pagination for large result sets - Specify only needed fields to reduce query time - Use segments for dimensional analysis - Test queries with LIMIT before fetching all data - Use STARTS_WITH for partial name matching - Combine multiple conditions with AND/OR""" } # Google Ads resource help RESOURCES_HELP = { "campaigns": "Campaign settings, budgets, targeting, status, bidding strategy", "ad_groups": "Ad group settings, bids, status, targeting within campaigns", "ads": "Ad copy, creative, headlines, descriptions, display URLs, landing pages", "keywords": "Keywords, match types (Broad, Phrase, Exact), bids, quality scores", "extensions": "Sitelinks, callouts, structured snippets, price extensions, promotion extensions", "audiences": "Audience lists, remarketing lists, custom intent, affinity audiences", "display_placements": "Website placements for Display Network campaigns", "shopping": "Product groups, merchants, shopping feeds for Shopping campaigns", "video": "Video ads, YouTube placements, video targeting", "labels": "Custom labels for organizing campaigns, ad groups, ads, keywords" } def get_google_ads_client() -> GoogleAdsClient: """Initialize and return a Google Ads API client.""" if not all([DEVELOPER_TOKEN, LOGIN_CUSTOMER_ID, CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN]): raise ValueError("Missing required Google Ads credentials in environment variables") config = { "developer_token": DEVELOPER_TOKEN, "login_customer_id": LOGIN_CUSTOMER_ID, "use_proto_plus": True, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "refresh_token": REFRESH_TOKEN, } return GoogleAdsClient.load_from_dict(config) @server.list_tools() async def handle_list_tools() -> list[Tool]: """ List available tools. Per Anthropic's recommendations, expose a unified 'call_tool' interface that lets Claude write code to interact with Google Ads API. This reduces token usage by ~98.7% vs listing all tool definitions upfront. See: https://www.anthropic.com/engineering/code-execution-with-mcp """ return [ Tool( name="call_tool", description="""Call Google Ads API operations. Claude should write code to call this efficiently. AVAILABLE OPERATIONS: 📊 LIST_RESOURCES - List various Google Ads entities list_accounts() - List accessible customer accounts list_campaigns(customer_id) - List all campaigns list_ad_groups(customer_id, campaign_id_optional) - List ad groups list_ads(customer_id, ad_group_id) - List ads list_keywords(customer_id, ad_group_id) - List keywords list_extensions(customer_id) - List ad extensions list_audiences(customer_id) - List audiences list_labels(customer_id) - List custom labels list_bidding_strategies(customer_id) - List bidding strategies 🔍 QUERY_DATA - Execute GAQL queries execute_gaql(query, customer_id, output_format, auto_paginate) Formats: 'json', 'table', 'csv' Query: "SELECT campaign.id, campaign.name, metrics.clicks FROM campaign WHERE campaign.status = 'ENABLED'" 📈 GET_PERFORMANCE - Performance metrics & analytics get_performance(level, customer_id, date_range, metrics, segments, filters, output_format) Levels: 'account', 'campaign', 'ad_group', 'ad', 'keyword' Date ranges: 'LAST_7_DAYS', 'LAST_30_DAYS', 'THIS_MONTH', 'LAST_MONTH', etc. Example: get_performance('campaign', '1234567890', 'LAST_30_DAYS', ['impressions', 'clicks', 'conversions', 'cost_micros']) ❓ GAQL_HELP - Get help with GAQL syntax & examples gaql_help(topic) Topics: 'overview', 'resources', 'metrics', 'segments', 'filters', 'best_practices' 📋 SEARCH_TOOLS - Search available operations search_tools(query) Returns matching tool names and descriptions 💡 EXAMPLES: # Filter campaigns by performance campaigns = call_tool('execute_gaql', { 'query': 'SELECT campaign.id, campaign.name, metrics.clicks FROM campaign WHERE metrics.clicks > 1000', 'customer_id': '1234567890' }) # Get detailed performance report perf = call_tool('get_performance', { 'level': 'keyword', 'customer_id': '1234567890', 'date_range': 'LAST_30_DAYS', 'metrics': ['clicks', 'conversions', 'cost_micros'], 'output_format': 'json' }) filtered = [k for k in perf if k.get('metrics', {}).get('clicks', 0) > 10] # Find high quality score keywords keywords = call_tool('execute_gaql', { 'query': 'SELECT ad_group_criterion.keyword.text, metrics.quality_score FROM ad_group_criterion WHERE metrics.quality_score >= 8' }) """, inputSchema={ "type": "object", "properties": { "method": { "type": "string", "description": "The method to call" }, "arguments": { "type": "object", "description": "Arguments to pass to the method" } }, "required": ["method", "arguments"] } ) ] @server.call_tool() async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent]: """Handle unified tool calls.""" try: if name != "call_tool": return [types.TextContent(type="text", text=f"Unknown tool: {name}")] method = arguments.get("method") args = arguments.get("arguments", {}) # Route to appropriate method result = None # LIST operations if method == "list_accounts": result = list_accounts() elif method == "list_campaigns": result = list_campaigns(args.get("customer_id")) elif method == "list_ad_groups": result = list_ad_groups(args.get("customer_id"), args.get("campaign_id")) elif method == "list_ads": result = list_ads(args.get("customer_id"), args.get("ad_group_id")) elif method == "list_keywords": result = list_keywords(args.get("customer_id"), args.get("ad_group_id")) elif method == "list_extensions": result = list_extensions(args.get("customer_id")) elif method == "list_audiences": result = list_audiences(args.get("customer_id")) elif method == "list_labels": result = list_labels(args.get("customer_id")) elif method == "list_bidding_strategies": result = list_bidding_strategies(args.get("customer_id")) # QUERY operations elif method == "execute_gaql": result = execute_gaql( args.get("query"), args.get("customer_id"), args.get("output_format", "json"), args.get("page_size", 1000), args.get("page_token"), args.get("auto_paginate", False), args.get("max_pages") ) # PERFORMANCE operations elif method == "get_performance": result = get_performance( args.get("level"), args.get("customer_id"), args.get("date_range", "LAST_30_DAYS"), args.get("days"), args.get("metrics"), args.get("segments"), args.get("filters"), args.get("output_format", "json"), args.get("page_size", 1000), args.get("auto_paginate", False) ) # HELP operations elif method == "gaql_help": result = gaql_help(args.get("topic"), args.get("search")) elif method == "search_tools": result = search_tools(args.get("query", "")) else: return [types.TextContent(type="text", text=f"Unknown method: {method}")] # Format result as JSON return [types.TextContent(type="text", text=json.dumps(result, indent=2))] except Exception as e: logger.exception(f"Error in call_tool") return [types.TextContent(type="text", text=f"Error: {str(e)}")] def get_google_ads_client() -> GoogleAdsClient: """Initialize and return a Google Ads API client.""" if not all([DEVELOPER_TOKEN, LOGIN_CUSTOMER_ID, CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN]): raise ValueError("Missing required Google Ads credentials in environment variables") config = { "developer_token": DEVELOPER_TOKEN, "login_customer_id": LOGIN_CUSTOMER_ID, "use_proto_plus": True, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "refresh_token": REFRESH_TOKEN, } return GoogleAdsClient.load_from_dict(config) def execute_gaql(query: str, customer_id: str = None, output_format: str = "json", page_size: int = 1000, page_token: str = None, auto_paginate: bool = False, max_pages: int = None) -> Any: """Execute a GAQL query with pagination support.""" try: if not customer_id: customer_id = LOGIN_CUSTOMER_ID client = get_google_ads_client() ga_service = client.get_service("GoogleAdsService") all_results = [] page_count = 0 while True: search_request = client.get_type("SearchGoogleAdsRequest") search_request.customer_id = customer_id search_request.query = query if page_token: search_request.page_token = page_token response = ga_service.search(request=search_request) for row in response: all_results.append(json.loads(type(row).to_json(row))) page_count += 1 page_token = response.next_page_token if not auto_paginate or not page_token or (max_pages and page_count >= max_pages): break return _format_output(all_results, output_format) except Exception as e: logger.exception("GAQL query failed") raise def run_gaql(customer_id: str, query: str) -> list[dict]: """Run a GAQL query and return structured results.""" return execute_gaql(query, customer_id) def get_performance(level: str, customer_id: str = None, date_range: str = "LAST_30_DAYS", days: int = None, metrics: list = None, segments: list = None, filters: dict = None, output_format: str = "json", page_size: int = 1000, auto_paginate: bool = False) -> Any: """Get performance metrics for a specific level (account, campaign, ad_group, ad, keyword).""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID # Build field list based on level resource_map = { "account": "customer", "campaign": "campaign", "ad_group": "ad_group", "ad": "ad_group_ad", "keyword": "ad_group_criterion" } resource = resource_map.get(level, level) # Default metrics if not metrics: metrics = ["impressions", "clicks", "conversions", "cost_micros", "ctr"] # Build SELECT clause select_parts = [f"{resource}.id", f"{resource}.name"] for metric in metrics: if "." not in metric: metric = f"metrics.{metric}" select_parts.append(metric) # Add segments if provided if segments: for segment in segments: if "." not in segment: segment = f"segments.{segment}" select_parts.append(segment) select_clause = ", ".join(select_parts) # Build WHERE clause where_parts = [] if date_range or days: if days: where_parts.append(f"segments.date DURING LAST_{days}_DAYS") else: where_parts.append(f"segments.date DURING {date_range}") if filters: for field, value in filters.items(): if isinstance(value, str): where_parts.append(f"{field} = '{value}'") else: where_parts.append(f"{field} = {value}") where_clause = " AND ".join(where_parts) if where_parts else "" # Build final query query = f"SELECT {select_clause} FROM {resource}" if where_clause: query += f" WHERE {where_clause}" return execute_gaql(query, customer_id, output_format, page_size, auto_paginate=auto_paginate) def list_accounts() -> dict: """List all accessible customer accounts.""" try: client = get_google_ads_client() customer_service = client.get_service("CustomerService") accessible_customers = customer_service.list_accessible_customers() accounts = [] for resource_name in accessible_customers.resource_names: accounts.append({ "resource_name": resource_name, "customer_id": resource_name.split("/")[-1] }) return {"accounts": accounts} except Exception as e: logger.exception("Failed to list accounts") raise def list_campaigns(customer_id: str = None) -> list: """List all campaigns for a customer.""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID query = """ SELECT campaign.id, campaign.name, campaign.status, campaign.advertising_channel_type, campaign_budget.amount_micros FROM campaign ORDER BY campaign.id """ return execute_gaql(query, customer_id) def list_ad_groups(customer_id: str = None, campaign_id: str = None) -> list: """List ad groups, optionally filtered by campaign.""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID if campaign_id: query = f""" SELECT ad_group.id, ad_group.name, ad_group.status, campaign.id, campaign.name FROM ad_group WHERE campaign.id = {campaign_id} ORDER BY ad_group.id """ else: query = """ SELECT ad_group.id, ad_group.name, ad_group.status, campaign.id, campaign.name FROM ad_group ORDER BY ad_group.id """ return execute_gaql(query, customer_id) def list_ads(customer_id: str = None, ad_group_id: str = None) -> list: """List ads, optionally filtered by ad group.""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID if ad_group_id: query = f""" SELECT ad_group_ad.ad.id, ad_group_ad.ad.type, ad_group_ad.status, ad_group.id, ad_group.name FROM ad_group_ad WHERE ad_group.id = {ad_group_id} """ else: query = """ SELECT ad_group_ad.ad.id, ad_group_ad.ad.type, ad_group_ad.status, ad_group.id, ad_group.name, campaign.id, campaign.name FROM ad_group_ad """ return execute_gaql(query, customer_id) def list_keywords(customer_id: str = None, ad_group_id: str = None) -> list: """List keywords, optionally filtered by ad group.""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID if ad_group_id: query = f""" SELECT ad_group_criterion.keyword.text, ad_group_criterion.keyword.match_type, ad_group_criterion.status, metrics.quality_score FROM ad_group_criterion WHERE ad_group.id = {ad_group_id} AND ad_group_criterion.type = 'KEYWORD' """ else: query = """ SELECT ad_group_criterion.keyword.text, ad_group_criterion.keyword.match_type, ad_group_criterion.status, ad_group.id, ad_group.name, campaign.id, campaign.name, metrics.quality_score FROM ad_group_criterion WHERE ad_group_criterion.type = 'KEYWORD' """ return execute_gaql(query, customer_id) def list_extensions(customer_id: str = None) -> list: """List ad extensions.""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID query = """ SELECT asset.id, asset.type, asset.name, asset.status FROM asset WHERE asset.type IN ('SITELINK', 'CALLOUT', 'STRUCTURED_SNIPPET', 'PRICE', 'PROMOTION') """ return execute_gaql(query, customer_id) def list_audiences(customer_id: str = None) -> list: """List audiences.""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID query = """ SELECT audience.id, audience.name, audience.type, audience.status FROM audience """ return execute_gaql(query, customer_id) def list_labels(customer_id: str = None) -> list: """List custom labels.""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID query = """ SELECT label.id, label.name, label.status, label.text_label.description FROM label """ return execute_gaql(query, customer_id) def list_bidding_strategies(customer_id: str = None) -> list: """List bidding strategies.""" if not customer_id: customer_id = LOGIN_CUSTOMER_ID query = """ SELECT accessible_bidding_strategy.id, accessible_bidding_strategy.name, accessible_bidding_strategy.type, accessible_bidding_strategy.owner_customer_id FROM accessible_bidding_strategy """ return execute_gaql(query, customer_id) def gaql_help(topic: str = None, search: str = None) -> dict: """Provide GAQL help and documentation.""" if search: search_lower = search.lower() results = {} for key, value in GAQL_HELP.items(): if search_lower in key.lower() or search_lower in value.lower(): results[key] = value return results if results else {"error": f"No help topics match '{search}'"} if topic and topic in GAQL_HELP: return {topic: GAQL_HELP[topic]} if topic: return {"error": f"Unknown topic: {topic}. Available topics: {', '.join(GAQL_HELP.keys())}"} return GAQL_HELP def search_tools(query: str) -> dict: """Search for available tools and resources.""" if not query: return {"resources": RESOURCES_HELP} query_lower = query.lower() results = {} # Search in resources for resource, desc in RESOURCES_HELP.items(): if query_lower in resource.lower() or query_lower in desc.lower(): results[resource] = desc return results if results else {"error": f"No resources match '{query}'"} def _format_output(data: list, format_type: str = "json") -> Any: """Format output in different formats.""" if format_type == "table": if not data: return {"error": "No data to format"} # Create ASCII table headers = list(data[0].keys()) if isinstance(data[0], dict) else [] lines = [] lines.append(" | ".join(str(h) for h in headers)) lines.append("-" * (sum(len(str(h)) for h in headers) + 3 * len(headers))) for row in data: if isinstance(row, dict): lines.append(" | ".join(str(row.get(h, "")) for h in headers)) return "\n".join(lines) elif format_type == "csv": if not data: return "" output = StringIO() if isinstance(data[0], dict): writer = csv.DictWriter(output, fieldnames=data[0].keys()) writer.writeheader() writer.writerows(data) return output.getvalue() else: # json return data if __name__ == "__main__": server.run()