keyword_opportunity.py — Find Google Ads keyword opportunities via GoMarble MCP + Claude.
Outputs: keyword_performance.csv, keyword_opportunities.csv, keyword_brief.txt
"""
# ┌──────────────────────────────────────────────────────────┐
# │ ✏️ EDIT THESE 4 VALUES BEFORE RUNNING │
# └──────────────────────────────────────────────────────────┘
ANTHROPIC_API_KEY = "PASTE_YOUR_ANTHROPIC_API_KEY_HERE" # e.g. "sk-ant-api03-..."
GOMARBLE_API_KEY = "PASTE_YOUR_GOMARBLE_API_KEY_HERE" # from GoMarble dashboard
GOOGLE_ADS_CID = "PASTE_YOUR_GOOGLE_ADS_CUSTOMER_ID" # 10-digit ID, e.g. "4254753542"
LANDING_PAGE_URL = "" # optional, e.g. "https://yoursite.com"
import csv, json, sys, time, subprocess, re
try:
import requests
except ImportError:
subprocess.check_call([sys.executable, "-m", "pip", "install", "requests"])
import requests
API_URL = "https://api.anthropic.com/v1/messages"
MODEL = "claude-sonnet-4-20250514"
def mcp_request(messages, system=None, max_tokens=16000):
headers = {"x-api-key": ANTHROPIC_API_KEY, "anthropic-version": "2023-06-01", "anthropic-beta": "mcp-client-2025-04-04", "content-type": "application/json"}
payload = {"model": MODEL, "max_tokens": max_tokens, "messages": messages, "mcp_servers": [{"type": "url", "url": "https://apps.gomarble.ai/mcp-api/sse", "name": "gomarble", "authorization_token": GOMARBLE_API_KEY}]}
if system: payload["system"] = system
for attempt in range(3):
try:
resp = requests.post(API_URL, headers=headers, json=payload, timeout=600)
if resp.status_code != 200: print(f" API error {resp.status_code}: {resp.text[:300]}"); resp.raise_for_status()
return resp.json()
except requests.exceptions.ReadTimeout:
print(f" Timeout (attempt {attempt+1}/3), retrying...")
if attempt == 2: raise
except requests.exceptions.ConnectionError:
print(f" Connection error (attempt {attempt+1}/3), retrying..."); time.sleep(5)
if attempt == 2: raise
def get_tool_data(response):
texts = []
for block in response.get("content", []):
if block.get("type") == "mcp_tool_result":
content = block.get("content", "")
if isinstance(content, list):
for c in content:
if isinstance(c, dict) and c.get("text"): texts.append(c["text"])
elif isinstance(content, str): texts.append(content)
return "\n".join(texts)
def get_text(response):
return "\n".join(b["text"] for b in response.get("content", []) if b.get("type") == "text")
def get_all_text(response):
tool_data = get_tool_data(response); text_data = get_text(response)
return tool_data + "\n" + text_data if tool_data else text_data
def strip_fences(text): return re.sub(r'```(?:json)?\s*', '', text)
def find_json(text):
text = strip_fences(text)
for open_ch, close_ch in [("{", "}"), ("[", "]")]:
pos = 0
while pos < len(text):
start = text.find(open_ch, pos)
if start == -1: break
depth = 0
for i in range(start, len(text)):
if text[i] == open_ch: depth += 1
elif text[i] == close_ch: depth -= 1
if depth == 0:
try: return json.loads(text[start : i + 1])
except json.JSONDecodeError: pass
break
pos = start + 1
return None
def to_rows(text):
data = find_json(text)
if data is None: return []
if isinstance(data, list): return data
if isinstance(data, dict):
for key in ("results", "rows", "data", "keywords", "keyword_ideas", "keyword_metrics"):
if key in data and isinstance(data[key], list): return data[key]
return [data]
return []
def pluck_keywords(text):
kws = []
for row in to_rows(text):
if isinstance(row, str): kws.append(row); continue
if not isinstance(row, dict): continue
for nested_key in ("searchTermView", "search_term_view"):
if nested_key in row and isinstance(row[nested_key], dict):
for sk in ("searchTerm", "search_term"):
if sk in row[nested_key] and row[nested_key][sk]: kws.append(str(row[nested_key][sk])); break
break
else:
for k in ("keyword", "keyword_text", "text", "searchTerm", "ad_group_criterion.keyword.text", "search_term", "search_term_view.search_term"):
if k in row and row[k]: kws.append(str(row[k])); break
return kws
def banner(step, title): print(f"\n{'─'*60}\n Step {step}/7 │ {title}\n{'─'*60}")
def flatten_row(row):
if not isinstance(row, dict): return {}
flat = {}
def _flatten(obj, prefix=""):
if isinstance(obj, dict):
for k, v in obj.items(): _flatten(v, f"{prefix}_{k}" if prefix else k)
elif not isinstance(obj, list): flat[prefix] = obj
_flatten(row)
aliases = {}
for k, v in flat.items():
aliases[k.replace(".", "_")] = v
if k.startswith("metrics_"): aliases[k[8:]] = v
for pfx in ("adGroupCriterion_keyword_", "ad_group_criterion_keyword_"):
if k.startswith(pfx):
short = k[len(pfx):]
if short in ("text", "keyword_text"): aliases["keyword_text"] = v
if short in ("matchType", "match_type"): aliases["match_type"] = v
for pfx in ("searchTermView_", "search_term_view_"):
if k.startswith(pfx):
short = k[len(pfx):]
if short in ("searchTerm", "search_term"): aliases["search_term"] = v
if short == "status": aliases["status"] = v
flat.update(aliases); return flat
def write_csv(path, rows, fields):
if not rows: print(f" (no data for {path})"); return
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore"); w.writeheader(); w.writerows(rows)
print(f" ✓ {path} — {len(rows)} rows")
def step1_account_info(cid):
banner(1, "Account Info")
msg = f"Run google_ads_run_gaql with customer_id='{cid}' and query:\nSELECT customer.id, customer.descriptive_name, customer.currency_code, customer.time_zone, metrics.impressions, metrics.clicks, metrics.cost, metrics.conversions, metrics.conversions_value FROM customer WHERE segments.date DURING LAST_30_DAYS LIMIT 1\n\nReturn the raw JSON."
resp = mcp_request([{"role": "user", "content": msg}]); text = get_all_text(resp); print(text[:500]); return text
def step2_current_keywords(cid):
banner(2, "Current Keywords (30d)")
msg = f"Run google_ads_run_gaql with customer_id='{cid}' and query:\nSELECT campaign.id, campaign.name, ad_group_criterion.keyword.text, ad_group_criterion.keyword.match_type, metrics.impressions, metrics.clicks, metrics.cost, metrics.conversions, metrics.conversions_value, metrics.ctr, metrics.average_cpc FROM keyword_view WHERE segments.date DURING LAST_30_DAYS AND campaign.status = 'ENABLED' ORDER BY metrics.impressions DESC LIMIT 200\n\nReturn the raw JSON."
resp = mcp_request([{"role": "user", "content": msg}]); text = get_all_text(resp); print(f" Got {len(to_rows(text))} keyword rows"); return text
def step3_search_term_gaps(cid):
banner(3, "Search Term Gaps")
msg = f"Run google_ads_run_gaql with customer_id='{cid}' and query:\nSELECT campaign.id, campaign.name, search_term_view.search_term, search_term_view.status, metrics.impressions, metrics.clicks, metrics.cost, metrics.conversions, metrics.conversions_value, metrics.ctr FROM search_term_view WHERE segments.date DURING LAST_30_DAYS AND metrics.conversions > 0 ORDER BY metrics.conversions DESC LIMIT 200\n\nReturn the raw JSON."
resp = mcp_request([{"role": "user", "content": msg}]); text = get_all_text(resp); print(f" Got {len(to_rows(text))} converting search terms"); return text
def step4_discover(cid, seeds, url=None):
banner(4, "Discover Keywords")
parts = ["Run google_ads_keyword_discover with:", f" customer_id: '{cid}'", f" keywords: {json.dumps(seeds[:5])}"]
if url: parts.append(f' url: "{url}"')
parts.append(" page_size: 100\n\nReturn the raw JSON.")
resp = mcp_request([{"role": "user", "content": "\n".join(parts)}]); text = get_all_text(resp); rows = to_rows(text)
print(f" Discovered {len(rows)} keyword ideas"); return text
def step5_metrics(cid, keywords):
banner(5, "Keyword Metrics")
batch = keywords[:200]
msg = f"Run google_ads_keyword_metrics with:\n customer_id: '{cid}'\n keywords: {json.dumps(batch)}\n\nReturn the raw JSON."
resp = mcp_request([{"role": "user", "content": msg}]); text = get_all_text(resp); print(f" Metrics: {len(to_rows(text))} rows"); return text
def step6_analysis(acct, perf, gaps, discovered, metrics):
banner(6, "AI Analysis")
prompt = f"You are a senior Google Ads keyword strategist. Analyze the following data and write a concise keyword brief with these sections:\n\n1. Account Snapshot -- 2-3 line health summary\n2. Expansion Plan -- Top 10 new keywords to add with match type and rationale\n3. Match Type Optimization -- Keywords to migrate broad->phrase->exact\n4. Gap Keywords -- Converting search terms to add as exact-match\n5. Low Performers to Pause -- High spend, zero/low conversion keywords\n6. Budget Reallocation -- Where to shift spend for max ROI\n\nBe specific: use keyword names, numbers, and concrete next steps.\n\n### Account Info\n{acct[:2000]}\n\n### Current Keywords (30d)\n{perf[:6000]}\n\n### Search Term Gaps\n{gaps[:5000]}\n\n### Discovered Keywords\n{discovered[:5000]}\n\n### Keyword Metrics\n{metrics[:5000]}"
resp = mcp_request([{"role": "user", "content": prompt}]); text = get_text(resp); print(f" Analysis: {len(text)} chars"); return text
def step7_output(perf_text, gaps_text, discovered_text, metrics_text, analysis_text):
banner(7, "Write Output Files")
perf_rows_raw = to_rows(perf_text); gaps_rows_raw = to_rows(gaps_text)
all_perf_rows = []; perf_fields = ["keyword_text", "match_type", "campaign_name", "impressions", "clicks", "cost", "conversions", "conversions_value", "ctr", "average_cpc"]
if perf_rows_raw:
for row in perf_rows_raw: all_perf_rows.append(flatten_row(row))
if gaps_rows_raw:
if not all_perf_rows: perf_fields = ["search_term", "campaign_name", "impressions", "clicks", "cost", "conversions", "conversions_value", "ctr"]
for row in gaps_rows_raw:
flat = flatten_row(row)
if "search_term" in flat and "keyword_text" not in flat: flat["keyword_text"] = flat["search_term"]; flat.setdefault("match_type", "SEARCH_TERM")
all_perf_rows.append(flat)
write_csv("keyword_performance.csv", all_perf_rows, perf_fields)
disc_rows = to_rows(discovered_text); opp_fields = ["keyword", "avg_monthly_searches", "competition", "competition_index", "low_top_of_page_bid", "high_top_of_page_bid"]; opp_rows = []
for row in disc_rows:
if not isinstance(row, dict): continue
flat = flatten_row(row); r = {}
r["keyword"] = flat.get("keyword", flat.get("keyword_text", flat.get("text", "")))
if not r["keyword"]: continue
r["avg_monthly_searches"] = flat.get("avg_monthly_searches", flat.get("avgMonthlySearches", ""))
r["competition"] = flat.get("competition", ""); r["competition_index"] = flat.get("competition_index", flat.get("competitionIndex", ""))
low_bid = flat.get("low_top_of_page_bid_micros", flat.get("lowTopOfPageBidMicros", "")); high_bid = flat.get("high_top_of_page_bid_micros", flat.get("highTopOfPageBidMicros", ""))
try: r["low_top_of_page_bid"] = f"${int(low_bid)/1_000_000:.2f}" if low_bid else ""
except (ValueError, TypeError): r["low_top_of_page_bid"] = str(low_bid)
try: r["high_top_of_page_bid"] = f"${int(high_bid)/1_000_000:.2f}" if high_bid else ""
except (ValueError, TypeError): r["high_top_of_page_bid"] = str(high_bid)
opp_rows.append(r)
write_csv("keyword_opportunities.csv", opp_rows, opp_fields)
with open("keyword_brief.txt", "w", encoding="utf-8") as f:
f.write("KEYWORD OPPORTUNITY BRIEF\n"); f.write(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"); f.write("=" * 60 + "\n\n"); f.write(analysis_text)
print(f" ✓ keyword_brief.txt — {len(analysis_text)} chars")
def main():
cid = GOOGLE_ADS_CID.replace("-", ""); url = LANDING_PAGE_URL.strip() or None
if "PASTE_YOUR" in ANTHROPIC_API_KEY: sys.exit("✗ Edit ANTHROPIC_API_KEY at the top of the file")
if "PASTE_YOUR" in GOMARBLE_API_KEY: sys.exit("✗ Edit GOMARBLE_API_KEY at the top of the file")
if "PASTE_YOUR" in GOOGLE_ADS_CID: sys.exit("✗ Edit GOOGLE_ADS_CID at the top of the file")
print(f"\nKeyword Opportunity Finder\n Account: {cid}")
if url: print(f" URL: {url}")
acct = step1_account_info(cid); perf = step2_current_keywords(cid); gaps = step3_search_term_gaps(cid)
seeds = pluck_keywords(perf) + pluck_keywords(gaps)
seen = set(); unique_seeds = []
for s in seeds:
sl = s.lower().strip()
if sl not in seen: seen.add(sl); unique_seeds.append(s)
seeds = unique_seeds if unique_seeds else ["brand keywords"]
print(f" Seeds ({len(seeds)} total): {seeds[:5]}")
discovered = step4_discover(cid, seeds[:5], url=url); disc_kws = pluck_keywords(discovered)
metrics = step5_metrics(cid, disc_kws) if disc_kws else discovered
analysis = step6_analysis(acct, perf, gaps, discovered, metrics)
step7_output(perf, gaps, discovered, metrics, analysis)
print(f"\n{'─'*60}\n Done! Files written:\n keyword_performance.csv\n keyword_opportunities.csv\n keyword_brief.txt\n{'─'*60}\n")
if __name__ == "__main__":
main()