#!/usr/bin/env python3
"""
competitor_monitor.py — Analyze your competitive positioning
using account structure, campaign performance, and creative analysis
via GoMarble MCP tools + Claude API.
Outputs: competitive_brief.txt, your_creatives.csv
Usage:
python competitor_monitor.py --meta act_123456 --google 1234567890 --competitors "Brand A, Brand B"
"""
import argparse, csv, json, re, sys, time, subprocess
try:
import requests
except ImportError:
subprocess.check_call([sys.executable, "-m", "pip", "install", "requests"])
import requests
# ┌──────────────────────────────────────────────────────────┐
# │ EDIT THESE VALUES BEFORE RUNNING │
# └──────────────────────────────────────────────────────────┘
ANTHROPIC_API_KEY = "" # Your Anthropic API key
GOMARBLE_API_KEY = "" # From GoMarble dashboard
API_URL = "https://api.anthropic.com/v1/messages"
MODEL = "claude-sonnet-4-20250514"
def try_parse_json(text):
text = re.sub(r"```(?:json)?\s*", "", 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 = try_parse_json(text)
if data is None: return []
if isinstance(data, list): return data
if isinstance(data, dict):
for key in ("results", "rows", "data"):
if key in data and isinstance(data[key], list): return data[key]
return [data]
return []
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)
return flat
def banner(step, total, title):
print(f"\n{'─'*60}\n Step {step}/{total} │ {title}\n{'─'*60}")
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 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=120)
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_text(response):
parts = []
for block in response.get("content", []):
if block.get("type") == "text": parts.append(block["text"])
elif 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"): parts.append(c["text"])
elif isinstance(content, str): parts.append(content)
return "\n".join(parts)
# ── Step 1: Account Structure ─────────────────────────────────
def step1_account_structure(google_cid, meta_id):
banner(1, 5, "Account Structure")
google_text, meta_text = None, None
if google_cid:
msg = (f"Run google_ads_run_gaql with customer_id='{google_cid}' and query:\n"
"SELECT campaign.name, campaign.status, campaign.advertising_channel_type, "
"campaign.bidding_strategy_type "
"FROM campaign WHERE campaign.status = 'ENABLED'\n\n"
"Return the raw JSON.")
resp = mcp_request([{"role": "user", "content": msg}])
google_text = get_text(resp)
print(f" Google campaigns: {len(to_rows(google_text))}")
if meta_id:
msg = (f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='campaign', date_preset='last_30d', "
"fields=['campaign_name','objective','status']. "
"Return the raw JSON.")
resp = mcp_request([{"role": "user", "content": msg}])
meta_text = get_text(resp)
print(f" Meta campaigns: {len(to_rows(meta_text))}")
return google_text, meta_text
# ── Step 2: Your Performance ──────────────────────────────────
def step2_performance(google_cid, meta_id):
banner(2, 5, "Your Performance (30d)")
google_text, meta_text = None, None
if meta_id:
msg = (f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='campaign', date_preset='last_30d', "
"fields=['campaign_name','spend','impressions','clicks','ctr','cpc',"
"'purchase_roas','actions']. "
"Return the raw JSON.")
resp = mcp_request([{"role": "user", "content": msg}])
meta_text = get_text(resp)
print(f" Meta performance rows: {len(to_rows(meta_text))}")
if google_cid:
msg = (f"Run google_ads_run_gaql with customer_id='{google_cid}' and query:\n"
"SELECT campaign.name, metrics.impressions, metrics.clicks, "
"metrics.cost_micros, metrics.conversions, metrics.conversions_value, "
"metrics.ctr, metrics.average_cpc "
"FROM campaign WHERE segments.date DURING LAST_30_DAYS "
"AND campaign.status = 'ENABLED' ORDER BY metrics.cost_micros DESC LIMIT 30\n\n"
"Return the raw JSON.")
resp = mcp_request([{"role": "user", "content": msg}])
google_text = get_text(resp)
print(f" Google performance rows: {len(to_rows(google_text))}")
return google_text, meta_text
# ── Step 3: Your Creatives (Meta) ─────────────────────────────
def step3_creatives(meta_id):
banner(3, 5, "Your Creatives (Meta)")
if not meta_id: print(" Skipped (no Meta account)"); return None
msg = (f"Run facebook_get_ad_creative_details with ad_account_id='{meta_id}', "
"limit=20. Return the raw JSON including headlines, body text, and CTAs.")
resp = mcp_request([{"role": "user", "content": msg}])
text = get_text(resp)
print(f" Creatives fetched: {len(to_rows(text))}")
return text
# ── Step 4: AI Analysis ──────────────────────────────────────
def step4_analysis(google_struct, meta_struct, google_perf, meta_perf, creatives_text, competitors):
banner(4, 5, "AI Competitive Positioning Analysis")
competitor_context = f"Known competitors: {', '.join(competitors)}" if competitors else "No specific competitors provided."
prompt = f"""You are a competitive positioning analyst for paid advertising. Analyze the data below and produce a competitive positioning brief.
{competitor_context}
Sections:
1. ACCOUNT STRUCTURE OVERVIEW — Summarize campaign types, objectives, and bidding strategies across platforms.
2. PERFORMANCE SNAPSHOT — Key metrics: spend, impressions, clicks, CTR, CPC, ROAS, conversions.
3. CREATIVE STRATEGY ANALYSIS — Based on the ad creatives:
- What messaging themes are being used?
- Strengths and weaknesses in creative approach
- Headline and CTA patterns
4. COMPETITIVE MESSAGING GAPS — What messaging angles are competitors likely using that you're not?
5. SUGGESTED TEST ANGLES — 5-10 new creative/messaging tests to run based on gaps identified.
6. COMPETITIVE POSITIONING BRIEF — How you're positioned vs competitors, and strategic recommendations.
Use specific data points, percentages, and dollar amounts where available.
### Google Account Structure
{google_struct[:3000] if google_struct else 'N/A'}
### Meta Account Structure
{meta_struct[:3000] if meta_struct else 'N/A'}
### Google Performance (30d)
{google_perf[:4000] if google_perf else 'N/A'}
### Meta Performance (30d)
{meta_perf[:4000] if meta_perf else 'N/A'}
### Your Meta Creatives (Top 20)
{creatives_text[:5000] if creatives_text else 'N/A'}"""
resp = mcp_request([{"role": "user", "content": prompt}])
brief = get_text(resp)
print(f" Brief: {len(brief)} chars")
return brief
# ── Step 5: Output ────────────────────────────────────────────
def step5_output(creatives_text, brief):
banner(5, 5, "Write Output Files")
if creatives_text:
rows = [flatten_row(r) for r in to_rows(creatives_text)]
if rows:
fields = list(rows[0].keys())[:15]
write_csv("your_creatives.csv", rows, fields)
with open("competitive_brief.txt", "w", encoding="utf-8") as f:
f.write("COMPETITIVE POSITIONING BRIEF\n")
f.write(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write("=" * 60 + "\n\n")
f.write(brief)
print(f" -> competitive_brief.txt — {len(brief)} chars")
print(f"\n{'='*60}\n COMPETITIVE POSITIONING ANALYSIS COMPLETE\n{'='*60}")
def main():
parser = argparse.ArgumentParser(description="Competitive Positioning Monitor via GoMarble MCP")
parser.add_argument("--meta", default="", help="Meta ad account ID, e.g. act_123456")
parser.add_argument("--google", default="", help="Google Ads customer ID, e.g. 1234567890")
parser.add_argument("--competitors", default="", help="Comma-separated competitor names")
args = parser.parse_args()
google_cid = args.google.strip().replace("-", "") or None
meta_id = args.meta.strip() or None
competitors = [c.strip() for c in args.competitors.split(",") if c.strip()] if args.competitors else []
if not google_cid and not meta_id:
sys.exit("Provide at least one of --google or --meta.")
if not ANTHROPIC_API_KEY or not GOMARBLE_API_KEY:
sys.exit("Set ANTHROPIC_API_KEY and GOMARBLE_API_KEY at the top of the file.")
print(f"\nCompetitive Positioning Monitor")
if google_cid: print(f" Google: {google_cid}")
if meta_id: print(f" Meta: {meta_id}")
if competitors: print(f" Competitors: {competitors}")
google_struct, meta_struct = step1_account_structure(google_cid, meta_id)
google_perf, meta_perf = step2_performance(google_cid, meta_id)
creatives_text = step3_creatives(meta_id)
brief = step4_analysis(google_struct, meta_struct, google_perf, meta_perf, creatives_text, competitors)
step5_output(creatives_text, brief)
if __name__ == "__main__":
main()