google_ads_custom_query
Execute custom Google Ads Query Language (GAQL) queries to extract specific performance data. Write your own query to retrieve metrics, keywords, or campaign details.
Instructions
Execute a custom Google Ads Query Language (GAQL) query.
For advanced users who want to write their own GAQL queries. Use the Google Ads Query Builder to construct queries: https://developers.google.com/google-ads/api/fields/latest/overview_query_builder
Args: customer_id: Customer ID (without hyphens) query: GAQL query string response_format: Output format ('json' or 'markdown')
Returns: Query results in specified format
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| customer_id | Yes | ||
| query | Yes | ||
| response_format | No | json |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- google_ads_mcp.py:166-166 (registration)The tool is registered using the @mcp.tool() decorator on FastMCP instance.
@mcp.tool() - google_ads_mcp.py:167-219 (handler)The handler function that executes a custom GAQL query against the Google Ads API via GoogleAdsService.search(), supporting JSON and markdown output formats.
def google_ads_custom_query( customer_id: str, query: str, response_format: str = "json", ) -> str: """ Execute a custom Google Ads Query Language (GAQL) query. For advanced users who want to write their own GAQL queries. Use the Google Ads Query Builder to construct queries: https://developers.google.com/google-ads/api/fields/latest/overview_query_builder Args: customer_id: Customer ID (without hyphens) query: GAQL query string response_format: Output format ('json' or 'markdown') Returns: Query results in specified format """ try: client = get_auth_manager().get_client() ga_service = client.get_service("GoogleAdsService") clean_id = customer_id.replace("-", "") response = ga_service.search(customer_id=clean_id, query=query) results = [] for row in response: row_dict = {} for field_name in type(row).meta.fields.keys(): value = getattr(row, field_name, None) if value: row_dict[field_name] = str(value) results.append(row_dict) header = f"# Custom Query Results\n\n**Query**: {query}\n\n**Result Count**: {len(results)}\n\n" if response_format == "json": return header + json.dumps(results, indent=2, default=str) if not results: return header + "No results found." keys = list(results[0].keys()) table = "| " + " | ".join(keys) + " |\n" table += "| " + " | ".join("---" for _ in keys) + " |\n" for r in results: table += "| " + " | ".join(str(r.get(k, ""))[:60] for k in keys) + " |\n" return header + table except Exception as exc: return f"❌ Custom query failed: {exc}" - google_ads_mcp.py:167-171 (schema)Function signature defines parameters: customer_id (str), query (str), response_format (str, default='json') with defaults and docstring acting as schema.
def google_ads_custom_query( customer_id: str, query: str, response_format: str = "json", ) -> str: