How to Connect Google Ads & Google Keyword Planner to Claude via MCP

from google_auth_oauthlib.flow import InstalledAppFlow

CLIENT_CONFIG = {
    "installed": {
        "client_id": "YOUR_CLIENT_ID.apps.googleusercontent.com",
        "client_secret": "YOUR_CLIENT_SECRET",
        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
        "token_uri": "https://oauth2.googleapis.com/token",
    }
}

SCOPES = ["https://www.googleapis.com/auth/adwords"]

def get_refresh_token():
    flow = InstalledAppFlow.from_client_config(CLIENT_CONFIG, scopes=SCOPES)
    credentials = flow.run_local_server(port=8080, prompt="consent", access_type="offline")
    print(f"\nRefresh Token: {credentials.refresh_token}")

if __name__ == "__main__":
    get_refresh_token()

import os, json
from mcp.server.fastmcp import FastMCP
from google.ads.googleads.client import GoogleAdsClient

mcp = FastMCP("Google Ads MCP Server")

def get_google_ads_client():
    credentials = {
        "developer_token": os.environ["GOOGLE_ADS_DEVELOPER_TOKEN"],
        "client_id": os.environ["GOOGLE_ADS_CLIENT_ID"],
        "client_secret": os.environ["GOOGLE_ADS_CLIENT_SECRET"],
        "refresh_token": os.environ["GOOGLE_ADS_REFRESH_TOKEN"],
        "use_proto_plus": True
    }
    if os.environ.get("GOOGLE_ADS_LOGIN_CUSTOMER_ID"):
        credentials["login_customer_id"] = os.environ["GOOGLE_ADS_LOGIN_CUSTOMER_ID"]
    return GoogleAdsClient.load_from_dict(credentials)

@mcp.tool()
def execute_gaql_query(customer_id: str, query: str) -> str:
    """Executes a read-only GAQL SELECT query against the specified Google Ads customer_id."""
    clean = query.strip()
    if not clean.upper().startswith("SELECT"):
        return json.dumps({"error": "Safety violation: Only SELECT queries are permitted."})

    client = get_google_ads_client()
    ga_service = client.get_service("GoogleAdsService")
    try:
        response = ga_service.search_stream(customer_id=customer_id.replace("-", ""), query=clean)
        results = []
        for batch in response:
            for row in batch.results:
                results.append(row._pb.to_dict() if hasattr(row, "_pb") else str(row))
                if len(results) >= 100: break
            if len(results) >= 100: break
        return json.dumps({"count": len(results), "data": results}, default=str)
    except Exception as e:
        return json.dumps({"error": str(e)})

if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "google-ads": {
      "command": "/usr/local/bin/python3",
      "args": ["/path/to/google_ads_mcp.py"],
      "env": {
        "GOOGLE_ADS_DEVELOPER_TOKEN": "YOUR_DEV_TOKEN",
        "GOOGLE_ADS_CLIENT_ID": "YOUR_CLIENT_ID.apps.googleusercontent.com",
        "GOOGLE_ADS_CLIENT_SECRET": "YOUR_CLIENT_SECRET",
        "GOOGLE_ADS_REFRESH_TOKEN": "1//YOUR_REFRESH_TOKEN",
        "GOOGLE_ADS_LOGIN_CUSTOMER_ID": "1234567890"
      }
    }
  }
}

SELECT 
  search_term_view.search_term,
  search_term_view.status,
  campaign.name,
  ad_group.name,
  metrics.impressions,
  metrics.clicks,
  metrics.cost_micros,
  metrics.conversions
FROM search_term_view
WHERE segments.date DURING LAST_30_DAYS
  AND metrics.clicks > 20
  AND metrics.conversions = 0
ORDER BY metrics.cost_micros DESC
LIMIT 100

@mcp.tool()
def get_keyword_ideas(customer_id: str, keywords: list[str], location_id: str = "2840", language_id: str = "1000") -> str:
    """Fetches keyword search volume, CPC estimates, and related terms from Google Keyword Planner."""
    client = get_google_ads_client()
    kp_service = client.get_service("KeywordPlanIdeaService")
    
    request = client.get_type("GenerateKeywordIdeasRequest")
    request.customer_id = customer_id.replace("-", "")
    request.language = f"languageConstants/{language_id}"
    request.geo_target_constants.append(f"geoTargetConstants/{location_id}")
    request.keyword_plan_network = client.enums.KeywordPlanNetworkEnum.GOOGLE_SEARCH
    request.keyword_seed.keywords.extend(keywords)

    response = kp_service.generate_keyword_ideas(request=request)
    ideas = []
    for idea in response:
        m = idea.keyword_idea_metrics
        ideas.append({
            "keyword": idea.text,
            "monthly_searches": m.avg_monthly_searches,
            "competition": m.competition.name,
            "low_bid": (m.low_top_of_page_bid_micros or 0) / 1_000_000,
            "high_bid": (m.high_top_of_page_bid_micros or 0) / 1_000_000
        })
        if len(ideas) >= 50: break
    return json.dumps({"ideas": ideas})