Google Ads MCP
by cohnen
Verified
from typing import Any, Dict, List, Optional, Union
from pydantic import Field
import os
import json
import requests
from datetime import datetime, timedelta
from google_auth_oauthlib.flow import InstalledAppFlow
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
# MCP
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
"google-ads-server",
dependencies=[
"google-auth-oauthlib",
"google-auth",
"requests"
]
)
# Constants and configuration
SCOPES = ['https://www.googleapis.com/auth/adwords']
API_VERSION = "v19" # Google Ads API version
# Get credentials from environment variables
GOOGLE_ADS_CREDENTIALS_PATH = os.environ.get("GOOGLE_ADS_CREDENTIALS_PATH")
GOOGLE_ADS_DEVELOPER_TOKEN = os.environ.get("GOOGLE_ADS_DEVELOPER_TOKEN")
GOOGLE_ADS_LOGIN_CUSTOMER_ID = os.environ.get("GOOGLE_ADS_LOGIN_CUSTOMER_ID", "")
def format_customer_id(customer_id: str) -> str:
"""Format customer ID to ensure it's 10 digits without dashes."""
# Convert to string if passed as integer or another type
customer_id = str(customer_id)
# Remove any quotes surrounding the customer_id (both escaped and unescaped)
customer_id = customer_id.replace('\"', '').replace('"', '')
# Remove any non-digit characters (including dashes, braces, etc.)
customer_id = ''.join(char for char in customer_id if char.isdigit())
# Ensure it's 10 digits with leading zeros if needed
return customer_id.zfill(10)
def get_credentials():
"""Get and refresh OAuth credentials."""
creds = None
client_config = None # Initialize this variable
if not GOOGLE_ADS_CREDENTIALS_PATH:
raise ValueError("GOOGLE_ADS_CREDENTIALS_PATH environment variable not set")
# Check if token file exists and load credentials
if os.path.exists(GOOGLE_ADS_CREDENTIALS_PATH):
with open(GOOGLE_ADS_CREDENTIALS_PATH, 'r') as f:
creds_data = json.load(f)
# Check if this is a client config or saved credentials
if "installed" in creds_data:
client_config = creds_data
else:
creds = Credentials.from_authorized_user_info(creds_data, SCOPES)
# If credentials don't exist or are invalid, get new ones
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
# If no client_config is defined yet, create one from environment variables
if not client_config:
client_id = os.environ.get("GOOGLE_ADS_CLIENT_ID")
client_secret = os.environ.get("GOOGLE_ADS_CLIENT_SECRET")
if not client_id or not client_secret:
raise ValueError("GOOGLE_ADS_CLIENT_ID and GOOGLE_ADS_CLIENT_SECRET must be set if no client config file exists")
client_config = {
"installed": {
"client_id": client_id,
"client_secret": client_secret,
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"token_uri": "https://oauth2.googleapis.com/token",
"redirect_uris": ["urn:ietf:wg:oauth:2.0:oob", "http://localhost"]
}
}
# Run the OAuth flow
flow = InstalledAppFlow.from_client_config(client_config, SCOPES)
creds = flow.run_local_server(port=0)
# Save the credentials
with open(GOOGLE_ADS_CREDENTIALS_PATH, 'w') as f:
f.write(creds.to_json())
return creds
def get_headers(creds):
"""Get headers for Google Ads API requests."""
if not GOOGLE_ADS_DEVELOPER_TOKEN:
raise ValueError("GOOGLE_ADS_DEVELOPER_TOKEN environment variable not set")
headers = {
'Authorization': f'Bearer {creds.token}',
'developer-token': GOOGLE_ADS_DEVELOPER_TOKEN,
'content-type': 'application/json'
}
if GOOGLE_ADS_LOGIN_CUSTOMER_ID:
headers['login-customer-id'] = format_customer_id(GOOGLE_ADS_LOGIN_CUSTOMER_ID)
return headers
@mcp.tool()
async def list_accounts() -> str:
"""
Lists all accessible Google Ads accounts.
This is typically the first command you should run to identify which accounts
you have access to. The returned account IDs can be used in subsequent commands.
Returns:
A formatted list of all Google Ads accounts accessible with your credentials
"""
try:
creds = get_credentials()
headers = get_headers(creds)
url = f"https://googleads.googleapis.com/{API_VERSION}/customers:listAccessibleCustomers"
response = requests.get(url, headers=headers)
if response.status_code != 200:
return f"Error accessing accounts: {response.text}"
customers = response.json()
if not customers.get('resourceNames'):
return "No accessible accounts found."
# Format the results
result_lines = ["Accessible Google Ads Accounts:"]
result_lines.append("-" * 50)
for resource_name in customers['resourceNames']:
customer_id = resource_name.split('/')[-1]
formatted_id = format_customer_id(customer_id)
result_lines.append(f"Account ID: {formatted_id}")
return "\n".join(result_lines)
except Exception as e:
return f"Error listing accounts: {str(e)}"
@mcp.tool()
async def execute_gaql_query(
customer_id: Union[str, int] = Field(description="Google Ads customer ID (10 digits, no dashes)"),
query: str = Field(description="Valid GAQL query string following Google Ads Query Language syntax")
) -> str:
"""
Execute a custom GAQL (Google Ads Query Language) query.
This tool allows you to run any valid GAQL query against the Google Ads API.
Always specify the customer_id as a string (even if it looks like a number).
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
query: The GAQL query to execute (must follow GAQL syntax)
Returns:
Formatted query results or error message
Example:
customer_id: "1234567890"
query: "SELECT campaign.id, campaign.name FROM campaign LIMIT 10"
"""
try:
creds = get_credentials()
headers = get_headers(creds)
formatted_customer_id = format_customer_id(customer_id)
url = f"https://googleads.googleapis.com/{API_VERSION}/customers/{formatted_customer_id}/googleAds:search"
payload = {"query": query}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
return f"Error executing query: {response.text}"
results = response.json()
if not results.get('results'):
return "No results found for the query."
# Format the results as a table
result_lines = [f"Query Results for Account {formatted_customer_id}:"]
result_lines.append("-" * 80)
# Get field names from the first result
fields = []
first_result = results['results'][0]
for key in first_result:
if isinstance(first_result[key], dict):
for subkey in first_result[key]:
fields.append(f"{key}.{subkey}")
else:
fields.append(key)
# Add header
result_lines.append(" | ".join(fields))
result_lines.append("-" * 80)
# Add data rows
for result in results['results']:
row_data = []
for field in fields:
if "." in field:
parent, child = field.split(".")
value = str(result.get(parent, {}).get(child, ""))
else:
value = str(result.get(field, ""))
row_data.append(value)
result_lines.append(" | ".join(row_data))
return "\n".join(result_lines)
except Exception as e:
return f"Error executing GAQL query: {str(e)}"
@mcp.tool()
async def get_campaign_performance(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes) as a string"),
days: int = Field(default=30, description="Number of days to look back (7, 30, 90, etc.)")
) -> str:
"""
Get campaign performance metrics for the specified time period.
RECOMMENDED WORKFLOW:
1. First run list_accounts() to get available account IDs
2. Then run get_account_currency() to see what currency the account uses
3. Finally run this command to get campaign performance
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
days: Number of days to look back (default: 30)
Returns:
Formatted table of campaign performance data
Note:
Cost values are in micros (millionths) of the account currency
(e.g., 1000000 = 1 USD in a USD account)
Example:
customer_id: "1234567890"
days: 14
"""
query = f"""
SELECT
campaign.id,
campaign.name,
campaign.status,
metrics.impressions,
metrics.clicks,
metrics.cost_micros,
metrics.conversions,
metrics.average_cpc
FROM campaign
WHERE segments.date DURING LAST_{days}DAYS
ORDER BY metrics.cost_micros DESC
LIMIT 50
"""
return await execute_gaql_query(customer_id, query)
@mcp.tool()
async def get_ad_performance(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes) as a string"),
days: int = Field(default=30, description="Number of days to look back (7, 30, 90, etc.)")
) -> str:
"""
Get ad performance metrics for the specified time period.
RECOMMENDED WORKFLOW:
1. First run list_accounts() to get available account IDs
2. Then run get_account_currency() to see what currency the account uses
3. Finally run this command to get ad performance
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
days: Number of days to look back (default: 30)
Returns:
Formatted table of ad performance data
Note:
Cost values are in micros (millionths) of the account currency
(e.g., 1000000 = 1 USD in a USD account)
Example:
customer_id: "1234567890"
days: 14
"""
query = f"""
SELECT
ad_group_ad.ad.id,
ad_group_ad.ad.name,
ad_group_ad.status,
campaign.name,
ad_group.name,
metrics.impressions,
metrics.clicks,
metrics.cost_micros,
metrics.conversions
FROM ad_group_ad
WHERE segments.date DURING LAST_{days}DAYS
ORDER BY metrics.impressions DESC
LIMIT 50
"""
return await execute_gaql_query(customer_id, query)
@mcp.tool()
async def run_gaql(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes)"),
query: str = Field(description="Valid GAQL query string following Google Ads Query Language syntax"),
format: str = Field(default="table", description="Output format: 'table', 'json', or 'csv'")
) -> str:
"""
Execute any arbitrary GAQL (Google Ads Query Language) query with custom formatting options.
This is the most powerful tool for custom Google Ads data queries. Always format your
customer_id as a string, even though it looks like a number.
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
query: The GAQL query to execute (any valid GAQL query)
format: Output format ("table", "json", or "csv")
Returns:
Query results in the requested format
EXAMPLE QUERIES:
1. Basic campaign metrics:
SELECT
campaign.name,
metrics.clicks,
metrics.impressions,
metrics.cost_micros
FROM campaign
WHERE segments.date DURING LAST_7DAYS
2. Ad group performance:
SELECT
ad_group.name,
metrics.conversions,
metrics.cost_micros,
campaign.name
FROM ad_group
WHERE metrics.clicks > 100
3. Keyword analysis:
SELECT
keyword.text,
metrics.average_position,
metrics.ctr
FROM keyword_view
ORDER BY metrics.impressions DESC
4. Get conversion data:
SELECT
campaign.name,
metrics.conversions,
metrics.conversions_value,
metrics.cost_micros
FROM campaign
WHERE segments.date DURING LAST_30DAYS
Note:
Cost values are in micros (millionths) of the account currency
(e.g., 1000000 = 1 USD in a USD account)
"""
try:
creds = get_credentials()
headers = get_headers(creds)
formatted_customer_id = format_customer_id(customer_id)
url = f"https://googleads.googleapis.com/{API_VERSION}/customers/{formatted_customer_id}/googleAds:search"
payload = {"query": query}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
return f"Error executing query: {response.text}"
results = response.json()
if not results.get('results'):
return "No results found for the query."
if format.lower() == "json":
return json.dumps(results, indent=2)
elif format.lower() == "csv":
# Get field names from the first result
fields = []
first_result = results['results'][0]
for key, value in first_result.items():
if isinstance(value, dict):
for subkey in value:
fields.append(f"{key}.{subkey}")
else:
fields.append(key)
# Create CSV string
csv_lines = [",".join(fields)]
for result in results['results']:
row_data = []
for field in fields:
if "." in field:
parent, child = field.split(".")
value = str(result.get(parent, {}).get(child, "")).replace(",", ";")
else:
value = str(result.get(field, "")).replace(",", ";")
row_data.append(value)
csv_lines.append(",".join(row_data))
return "\n".join(csv_lines)
else: # default table format
result_lines = [f"Query Results for Account {formatted_customer_id}:"]
result_lines.append("-" * 100)
# Get field names and maximum widths
fields = []
field_widths = {}
first_result = results['results'][0]
for key, value in first_result.items():
if isinstance(value, dict):
for subkey in value:
field = f"{key}.{subkey}"
fields.append(field)
field_widths[field] = len(field)
else:
fields.append(key)
field_widths[key] = len(key)
# Calculate maximum field widths
for result in results['results']:
for field in fields:
if "." in field:
parent, child = field.split(".")
value = str(result.get(parent, {}).get(child, ""))
else:
value = str(result.get(field, ""))
field_widths[field] = max(field_widths[field], len(value))
# Create formatted header
header = " | ".join(f"{field:{field_widths[field]}}" for field in fields)
result_lines.append(header)
result_lines.append("-" * len(header))
# Add data rows
for result in results['results']:
row_data = []
for field in fields:
if "." in field:
parent, child = field.split(".")
value = str(result.get(parent, {}).get(child, ""))
else:
value = str(result.get(field, ""))
row_data.append(f"{value:{field_widths[field]}}")
result_lines.append(" | ".join(row_data))
return "\n".join(result_lines)
except Exception as e:
return f"Error executing GAQL query: {str(e)}"
@mcp.tool()
async def get_ad_creatives(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes) as a string")
) -> str:
"""
Get ad creative details including headlines, descriptions, and URLs.
This tool retrieves the actual ad content (headlines, descriptions)
for review and analysis. Great for creative audits.
RECOMMENDED WORKFLOW:
1. First run list_accounts() to get available account IDs
2. Then run this command with the desired account ID
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
Returns:
Formatted list of ad creative details
Example:
customer_id: "1234567890"
"""
query = """
SELECT
ad_group_ad.ad.id,
ad_group_ad.ad.name,
ad_group_ad.ad.type,
ad_group_ad.ad.final_urls,
ad_group_ad.status,
ad_group_ad.ad.responsive_search_ad.headlines,
ad_group_ad.ad.responsive_search_ad.descriptions,
ad_group.name,
campaign.name
FROM ad_group_ad
WHERE ad_group_ad.status != 'REMOVED'
ORDER BY campaign.name, ad_group.name
LIMIT 50
"""
try:
creds = get_credentials()
headers = get_headers(creds)
formatted_customer_id = format_customer_id(customer_id)
url = f"https://googleads.googleapis.com/{API_VERSION}/customers/{formatted_customer_id}/googleAds:search"
payload = {"query": query}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
return f"Error retrieving ad creatives: {response.text}"
results = response.json()
if not results.get('results'):
return "No ad creatives found for this customer ID."
# Format the results in a readable way
output_lines = [f"Ad Creatives for Customer ID {formatted_customer_id}:"]
output_lines.append("=" * 80)
for i, result in enumerate(results['results'], 1):
ad = result.get('adGroupAd', {}).get('ad', {})
ad_group = result.get('adGroup', {})
campaign = result.get('campaign', {})
output_lines.append(f"\n{i}. Campaign: {campaign.get('name', 'N/A')}")
output_lines.append(f" Ad Group: {ad_group.get('name', 'N/A')}")
output_lines.append(f" Ad ID: {ad.get('id', 'N/A')}")
output_lines.append(f" Ad Name: {ad.get('name', 'N/A')}")
output_lines.append(f" Status: {result.get('adGroupAd', {}).get('status', 'N/A')}")
output_lines.append(f" Type: {ad.get('type', 'N/A')}")
# Handle Responsive Search Ads
rsa = ad.get('responsiveSearchAd', {})
if rsa:
if 'headlines' in rsa:
output_lines.append(" Headlines:")
for headline in rsa['headlines']:
output_lines.append(f" - {headline.get('text', 'N/A')}")
if 'descriptions' in rsa:
output_lines.append(" Descriptions:")
for desc in rsa['descriptions']:
output_lines.append(f" - {desc.get('text', 'N/A')}")
# Handle Final URLs
final_urls = ad.get('finalUrls', [])
if final_urls:
output_lines.append(f" Final URLs: {', '.join(final_urls)}")
output_lines.append("-" * 80)
return "\n".join(output_lines)
except Exception as e:
return f"Error retrieving ad creatives: {str(e)}"
@mcp.tool()
async def get_account_currency(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes)")
) -> str:
"""
Retrieve the default currency code used by the Google Ads account.
IMPORTANT: Run this first before analyzing cost data to understand which currency
the account uses. Cost values are always displayed in the account's currency.
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
Returns:
The account's default currency code (e.g., 'USD', 'EUR', 'GBP')
Example:
customer_id: "1234567890"
"""
query = """
SELECT
customer.id,
customer.currency_code
FROM customer
LIMIT 1
"""
try:
creds = get_credentials()
headers = get_headers(creds)
formatted_customer_id = format_customer_id(customer_id)
url = f"https://googleads.googleapis.com/{API_VERSION}/customers/{formatted_customer_id}/googleAds:search"
payload = {"query": query}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
return f"Error retrieving account currency: {response.text}"
results = response.json()
if not results.get('results'):
return "No account information found for this customer ID."
# Extract the currency code from the results
customer = results['results'][0].get('customer', {})
currency_code = customer.get('currencyCode', 'Not specified')
return f"Account {formatted_customer_id} uses currency: {currency_code}"
except Exception as e:
return f"Error retrieving account currency: {str(e)}"
@mcp.resource("gaql://reference")
def gaql_reference() -> str:
"""Google Ads Query Language (GAQL) reference documentation."""
return """
# Google Ads Query Language (GAQL) Reference
GAQL is similar to SQL but with specific syntax for Google Ads. Here's a quick reference:
## Basic Query Structure
```
SELECT field1, field2, ...
FROM resource_type
WHERE condition
ORDER BY field [ASC|DESC]
LIMIT n
```
## Common Field Types
### Resource Fields
- campaign.id, campaign.name, campaign.status
- ad_group.id, ad_group.name, ad_group.status
- ad_group_ad.ad.id, ad_group_ad.ad.final_urls
- keyword.text, keyword.match_type
### Metric Fields
- metrics.impressions
- metrics.clicks
- metrics.cost_micros
- metrics.conversions
- metrics.ctr
- metrics.average_cpc
### Segment Fields
- segments.date
- segments.device
- segments.day_of_week
## Common WHERE Clauses
### Date Ranges
- WHERE segments.date DURING LAST_7DAYS
- WHERE segments.date DURING LAST_30DAYS
- WHERE segments.date BETWEEN '2023-01-01' AND '2023-01-31'
### Filtering
- WHERE campaign.status = 'ENABLED'
- WHERE metrics.clicks > 100
- WHERE campaign.name LIKE '%Brand%'
## Tips
- Always check account currency before analyzing cost data
- Cost values are in micros (millionths): 1000000 = 1 unit of currency
- Use LIMIT to avoid large result sets
"""
@mcp.prompt("google_ads_workflow")
def google_ads_workflow() -> str:
"""Provides guidance on the recommended workflow for using Google Ads tools."""
return """
I'll help you analyze your Google Ads account data. Here's the recommended workflow:
1. First, let's list all the accounts you have access to:
- Run the `list_accounts()` tool to get available account IDs
2. Before analyzing cost data, let's check which currency the account uses:
- Run `get_account_currency(customer_id="ACCOUNT_ID")` with your selected account
3. Now we can explore the account data:
- For campaign performance: `get_campaign_performance(customer_id="ACCOUNT_ID", days=30)`
- For ad performance: `get_ad_performance(customer_id="ACCOUNT_ID", days=30)`
- For ad creative review: `get_ad_creatives(customer_id="ACCOUNT_ID")`
4. For custom queries, use the GAQL query tool:
- `run_gaql(customer_id="ACCOUNT_ID", query="YOUR_QUERY", format="table")`
5. Let me know if you have specific questions about:
- Campaign performance
- Ad performance
- Keywords
- Budgets
- Conversions
Important: Always provide the customer_id as a string, even though it looks like a number.
For example: customer_id="1234567890" (not customer_id=1234567890)
"""
@mcp.prompt("gaql_help")
def gaql_help() -> str:
"""Provides assistance for writing GAQL queries."""
return """
I'll help you write a Google Ads Query Language (GAQL) query. Here are some examples to get you started:
## Get campaign performance last 30 days
```
SELECT
campaign.id,
campaign.name,
campaign.status,
metrics.impressions,
metrics.clicks,
metrics.cost_micros,
metrics.conversions
FROM campaign
WHERE segments.date DURING LAST_30DAYS
ORDER BY metrics.cost_micros DESC
```
## Get keyword performance
```
SELECT
keyword.text,
keyword.match_type,
metrics.impressions,
metrics.clicks,
metrics.cost_micros,
metrics.conversions
FROM keyword_view
WHERE segments.date DURING LAST_30DAYS
ORDER BY metrics.clicks DESC
```
## Get ads with poor performance
```
SELECT
ad_group_ad.ad.id,
ad_group_ad.ad.name,
campaign.name,
ad_group.name,
metrics.impressions,
metrics.clicks,
metrics.conversions
FROM ad_group_ad
WHERE
segments.date DURING LAST_30DAYS
AND metrics.impressions > 1000
AND metrics.ctr < 0.01
ORDER BY metrics.impressions DESC
```
Once you've chosen a query, use it with:
```
run_gaql(customer_id="YOUR_ACCOUNT_ID", query="YOUR_QUERY_HERE")
```
Remember:
- Always provide the customer_id as a string
- Cost values are in micros (1,000,000 = 1 unit of currency)
- Use LIMIT to avoid large result sets
- Check the account currency before analyzing cost data
"""
@mcp.tool()
async def get_image_assets(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes) as a string"),
limit: int = Field(default=50, description="Maximum number of image assets to return")
) -> str:
"""
Retrieve all image assets in the account including their full-size URLs.
This tool allows you to get details about image assets used in your Google Ads account,
including the URLs to download the full-size images for further processing or analysis.
RECOMMENDED WORKFLOW:
1. First run list_accounts() to get available account IDs
2. Then run this command with the desired account ID
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
limit: Maximum number of image assets to return (default: 50)
Returns:
Formatted list of image assets with their download URLs
Example:
customer_id: "1234567890"
limit: 100
"""
query = f"""
SELECT
asset.id,
asset.name,
asset.type,
asset.image_asset.full_size.url,
asset.image_asset.full_size.height_pixels,
asset.image_asset.full_size.width_pixels,
asset.image_asset.file_size
FROM
asset
WHERE
asset.type = 'IMAGE'
LIMIT {limit}
"""
try:
creds = get_credentials()
headers = get_headers(creds)
formatted_customer_id = format_customer_id(customer_id)
url = f"https://googleads.googleapis.com/{API_VERSION}/customers/{formatted_customer_id}/googleAds:search"
payload = {"query": query}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
return f"Error retrieving image assets: {response.text}"
results = response.json()
if not results.get('results'):
return "No image assets found for this customer ID."
# Format the results in a readable way
output_lines = [f"Image Assets for Customer ID {formatted_customer_id}:"]
output_lines.append("=" * 80)
for i, result in enumerate(results['results'], 1):
asset = result.get('asset', {})
image_asset = asset.get('imageAsset', {})
full_size = image_asset.get('fullSize', {})
output_lines.append(f"\n{i}. Asset ID: {asset.get('id', 'N/A')}")
output_lines.append(f" Name: {asset.get('name', 'N/A')}")
if full_size:
output_lines.append(f" Image URL: {full_size.get('url', 'N/A')}")
output_lines.append(f" Dimensions: {full_size.get('widthPixels', 'N/A')} x {full_size.get('heightPixels', 'N/A')} px")
file_size = image_asset.get('fileSize', 'N/A')
if file_size != 'N/A':
# Convert to KB for readability
file_size_kb = int(file_size) / 1024
output_lines.append(f" File Size: {file_size_kb:.2f} KB")
output_lines.append("-" * 80)
return "\n".join(output_lines)
except Exception as e:
return f"Error retrieving image assets: {str(e)}"
@mcp.tool()
async def download_image_asset(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes) as a string"),
asset_id: str = Field(description="The ID of the image asset to download"),
output_dir: str = Field(default="./ad_images", description="Directory to save the downloaded image")
) -> str:
"""
Download a specific image asset from a Google Ads account.
This tool allows you to download the full-size version of an image asset
for further processing, analysis, or backup.
RECOMMENDED WORKFLOW:
1. First run list_accounts() to get available account IDs
2. Then run get_image_assets() to get available image asset IDs
3. Finally use this command to download specific images
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
asset_id: The ID of the image asset to download
output_dir: Directory where the image should be saved (default: ./ad_images)
Returns:
Status message indicating success or failure of the download
Example:
customer_id: "1234567890"
asset_id: "12345"
output_dir: "./my_ad_images"
"""
query = f"""
SELECT
asset.id,
asset.name,
asset.image_asset.full_size.url
FROM
asset
WHERE
asset.type = 'IMAGE'
AND asset.id = {asset_id}
LIMIT 1
"""
try:
creds = get_credentials()
headers = get_headers(creds)
formatted_customer_id = format_customer_id(customer_id)
url = f"https://googleads.googleapis.com/{API_VERSION}/customers/{formatted_customer_id}/googleAds:search"
payload = {"query": query}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
return f"Error retrieving image asset: {response.text}"
results = response.json()
if not results.get('results'):
return f"No image asset found with ID {asset_id}"
# Extract the image URL
asset = results['results'][0].get('asset', {})
image_url = asset.get('imageAsset', {}).get('fullSize', {}).get('url')
asset_name = asset.get('name', f"image_{asset_id}")
if not image_url:
return f"No download URL found for image asset ID {asset_id}"
# Create output directory if it doesn't exist
os.makedirs(output_dir, exist_ok=True)
# Download the image
image_response = requests.get(image_url)
if image_response.status_code != 200:
return f"Failed to download image: HTTP {image_response.status_code}"
# Clean the filename to be safe for filesystem
safe_name = ''.join(c for c in asset_name if c.isalnum() or c in ' ._-')
filename = f"{asset_id}_{safe_name}.jpg"
file_path = os.path.join(output_dir, filename)
# Save the image
with open(file_path, 'wb') as f:
f.write(image_response.content)
return f"Successfully downloaded image asset {asset_id} to {file_path}"
except Exception as e:
return f"Error downloading image asset: {str(e)}"
@mcp.tool()
async def get_asset_usage(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes) as a string"),
asset_id: str = Field(default=None, description="Optional: specific asset ID to look up (leave empty to get all image assets)"),
asset_type: str = Field(default="IMAGE", description="Asset type to search for ('IMAGE', 'TEXT', 'VIDEO', etc.)")
) -> str:
"""
Find where specific assets are being used in campaigns, ad groups, and ads.
This tool helps you analyze how assets are linked to campaigns and ads across your account,
which is useful for creative analysis and optimization.
RECOMMENDED WORKFLOW:
1. First run list_accounts() to get available account IDs
2. Run get_image_assets() to see available assets
3. Use this command to see where specific assets are used
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
asset_id: Optional specific asset ID to look up (leave empty to get all assets of the specified type)
asset_type: Type of asset to search for (default: 'IMAGE')
Returns:
Formatted report showing where assets are used in the account
Example:
customer_id: "1234567890"
asset_id: "12345"
asset_type: "IMAGE"
"""
# Build the query based on whether a specific asset ID was provided
where_clause = f"asset.type = '{asset_type}'"
if asset_id:
where_clause += f" AND asset.id = {asset_id}"
# First get the assets themselves
assets_query = f"""
SELECT
asset.id,
asset.name,
asset.type
FROM
asset
WHERE
{where_clause}
LIMIT 100
"""
# Then get the associations between assets and campaigns/ad groups
# Try using campaign_asset instead of asset_link
associations_query = f"""
SELECT
campaign.id,
campaign.name,
asset.id,
asset.name,
asset.type
FROM
campaign_asset
WHERE
{where_clause}
LIMIT 500
"""
# Also try ad_group_asset for ad group level information
ad_group_query = f"""
SELECT
ad_group.id,
ad_group.name,
asset.id,
asset.name,
asset.type
FROM
ad_group_asset
WHERE
{where_clause}
LIMIT 500
"""
try:
creds = get_credentials()
headers = get_headers(creds)
formatted_customer_id = format_customer_id(customer_id)
# First get the assets
url = f"https://googleads.googleapis.com/{API_VERSION}/customers/{formatted_customer_id}/googleAds:search"
payload = {"query": assets_query}
assets_response = requests.post(url, headers=headers, json=payload)
if assets_response.status_code != 200:
return f"Error retrieving assets: {assets_response.text}"
assets_results = assets_response.json()
if not assets_results.get('results'):
return f"No {asset_type} assets found for this customer ID."
# Now get the associations
payload = {"query": associations_query}
assoc_response = requests.post(url, headers=headers, json=payload)
if assoc_response.status_code != 200:
return f"Error retrieving asset associations: {assoc_response.text}"
assoc_results = assoc_response.json()
# Format the results in a readable way
output_lines = [f"Asset Usage for Customer ID {formatted_customer_id}:"]
output_lines.append("=" * 80)
# Create a dictionary to organize asset usage by asset ID
asset_usage = {}
# Initialize the asset usage dictionary with basic asset info
for result in assets_results.get('results', []):
asset = result.get('asset', {})
asset_id = asset.get('id')
if asset_id:
asset_usage[asset_id] = {
'name': asset.get('name', 'Unnamed asset'),
'type': asset.get('type', 'Unknown'),
'usage': []
}
# Add usage information from the associations
for result in assoc_results.get('results', []):
asset = result.get('asset', {})
asset_id = asset.get('id')
if asset_id and asset_id in asset_usage:
campaign = result.get('campaign', {})
ad_group = result.get('adGroup', {})
ad = result.get('adGroupAd', {}).get('ad', {}) if 'adGroupAd' in result else {}
asset_link = result.get('assetLink', {})
usage_info = {
'campaign_id': campaign.get('id', 'N/A'),
'campaign_name': campaign.get('name', 'N/A'),
'ad_group_id': ad_group.get('id', 'N/A'),
'ad_group_name': ad_group.get('name', 'N/A'),
'ad_id': ad.get('id', 'N/A') if ad else 'N/A',
'ad_name': ad.get('name', 'N/A') if ad else 'N/A'
}
asset_usage[asset_id]['usage'].append(usage_info)
# Format the output
for asset_id, info in asset_usage.items():
output_lines.append(f"\nAsset ID: {asset_id}")
output_lines.append(f"Name: {info['name']}")
output_lines.append(f"Type: {info['type']}")
if info['usage']:
output_lines.append("\nUsed in:")
output_lines.append("-" * 60)
output_lines.append(f"{'Campaign':<30} | {'Ad Group':<30}")
output_lines.append("-" * 60)
for usage in info['usage']:
campaign_str = f"{usage['campaign_name']} ({usage['campaign_id']})"
ad_group_str = f"{usage['ad_group_name']} ({usage['ad_group_id']})"
output_lines.append(f"{campaign_str[:30]:<30} | {ad_group_str[:30]:<30}")
output_lines.append("=" * 80)
return "\n".join(output_lines)
except Exception as e:
return f"Error retrieving asset usage: {str(e)}"
@mcp.tool()
async def analyze_image_assets(
customer_id: str = Field(description="Google Ads customer ID (10 digits, no dashes) as a string"),
days: int = Field(default=30, description="Number of days to look back (7, 30, 90, etc.)")
) -> str:
"""
Analyze image assets with their performance metrics across campaigns.
This comprehensive tool helps you understand which image assets are performing well
by showing metrics like impressions, clicks, and conversions for each image.
RECOMMENDED WORKFLOW:
1. First run list_accounts() to get available account IDs
2. Then run get_account_currency() to see what currency the account uses
3. Finally run this command to analyze image asset performance
Args:
customer_id: The Google Ads customer ID as a string (10 digits, no dashes)
days: Number of days to look back (default: 30)
Returns:
Detailed report of image assets and their performance metrics
Example:
customer_id: "1234567890"
days: 14
"""
# Make sure to use a valid date range format
# Valid formats are: LAST_7_DAYS, LAST_14_DAYS, LAST_30_DAYS, etc. (with underscores)
if days == 7:
date_range = "LAST_7_DAYS"
elif days == 14:
date_range = "LAST_14_DAYS"
elif days == 30:
date_range = "LAST_30_DAYS"
else:
# Default to 30 days if not a standard range
date_range = "LAST_30_DAYS"
query = f"""
SELECT
asset.id,
asset.name,
asset.image_asset.full_size.url,
asset.image_asset.full_size.width_pixels,
asset.image_asset.full_size.height_pixels,
campaign.name,
metrics.impressions,
metrics.clicks,
metrics.conversions,
metrics.cost_micros
FROM
campaign_asset
WHERE
asset.type = 'IMAGE'
AND segments.date DURING LAST_30_DAYS
ORDER BY
metrics.impressions DESC
LIMIT 200
"""
try:
creds = get_credentials()
headers = get_headers(creds)
formatted_customer_id = format_customer_id(customer_id)
url = f"https://googleads.googleapis.com/{API_VERSION}/customers/{formatted_customer_id}/googleAds:search"
payload = {"query": query}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
return f"Error analyzing image assets: {response.text}"
results = response.json()
if not results.get('results'):
return "No image asset performance data found for this customer ID and time period."
# Group results by asset ID
assets_data = {}
for result in results.get('results', []):
asset = result.get('asset', {})
asset_id = asset.get('id')
if asset_id not in assets_data:
assets_data[asset_id] = {
'name': asset.get('name', f"Asset {asset_id}"),
'url': asset.get('imageAsset', {}).get('fullSize', {}).get('url', 'N/A'),
'dimensions': f"{asset.get('imageAsset', {}).get('fullSize', {}).get('widthPixels', 'N/A')} x {asset.get('imageAsset', {}).get('fullSize', {}).get('heightPixels', 'N/A')}",
'impressions': 0,
'clicks': 0,
'conversions': 0,
'cost_micros': 0,
'campaigns': set(),
'ad_groups': set()
}
# Aggregate metrics
metrics = result.get('metrics', {})
assets_data[asset_id]['impressions'] += int(metrics.get('impressions', 0))
assets_data[asset_id]['clicks'] += int(metrics.get('clicks', 0))
assets_data[asset_id]['conversions'] += float(metrics.get('conversions', 0))
assets_data[asset_id]['cost_micros'] += int(metrics.get('costMicros', 0))
# Add campaign and ad group info
campaign = result.get('campaign', {})
ad_group = result.get('adGroup', {})
if campaign.get('name'):
assets_data[asset_id]['campaigns'].add(campaign.get('name'))
if ad_group.get('name'):
assets_data[asset_id]['ad_groups'].add(ad_group.get('name'))
# Format the results
output_lines = [f"Image Asset Performance Analysis for Customer ID {formatted_customer_id} (Last {days} days):"]
output_lines.append("=" * 100)
# Sort assets by impressions (highest first)
sorted_assets = sorted(assets_data.items(), key=lambda x: x[1]['impressions'], reverse=True)
for asset_id, data in sorted_assets:
output_lines.append(f"\nAsset ID: {asset_id}")
output_lines.append(f"Name: {data['name']}")
output_lines.append(f"Dimensions: {data['dimensions']}")
# Calculate CTR if there are impressions
ctr = (data['clicks'] / data['impressions'] * 100) if data['impressions'] > 0 else 0
# Format metrics
output_lines.append(f"\nPerformance Metrics:")
output_lines.append(f" Impressions: {data['impressions']:,}")
output_lines.append(f" Clicks: {data['clicks']:,}")
output_lines.append(f" CTR: {ctr:.2f}%")
output_lines.append(f" Conversions: {data['conversions']:.2f}")
output_lines.append(f" Cost (micros): {data['cost_micros']:,}")
# Show where it's used
output_lines.append(f"\nUsed in {len(data['campaigns'])} campaigns:")
for campaign in list(data['campaigns'])[:5]: # Show first 5 campaigns
output_lines.append(f" - {campaign}")
if len(data['campaigns']) > 5:
output_lines.append(f" - ... and {len(data['campaigns']) - 5} more")
# Add URL
if data['url'] != 'N/A':
output_lines.append(f"\nImage URL: {data['url']}")
output_lines.append("-" * 100)
return "\n".join(output_lines)
except Exception as e:
return f"Error analyzing image assets: {str(e)}"
@mcp.tool()
async def list_resources(customer_id: str) -> str:
"""
List valid resources that can be used in GAQL FROM clauses.
Args:
customer_id: The Google Ads customer ID as a string
Returns:
Formatted list of valid resources
"""
# Example query that lists some common resources
# This might need to be adjusted based on what's available in your API version
query = """
SELECT
google_ads_field.name,
google_ads_field.category,
google_ads_field.data_type
FROM
google_ads_field
WHERE
google_ads_field.category = 'RESOURCE'
ORDER BY
google_ads_field.name
"""
# Use your existing run_gaql function to execute this query
return await run_gaql(customer_id, query)
if __name__ == "__main__":
# Start the MCP server on stdio transport
mcp.run(transport="stdio")