import os, requests, time
MP_SA = os.environ["MIXPANEL_SERVICE_ACCOUNT"]
MP_SECRET = os.environ["MIXPANEL_SECRET"]
MP_PROJECT = os.environ["MIXPANEL_PROJECT_ID"]
MV = os.environ["MAVERA_API_KEY"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
mp_auth = (MP_SA, MP_SECRET)
FUNNEL_ID = os.environ.get("MIXPANEL_FUNNEL_ID", "12345")
r = requests.get(
"https://mixpanel.com/api/query/funnels",
auth=mp_auth,
params={
"project_id": MP_PROJECT,
"funnel_id": FUNNEL_ID,
"from_date": "2026-02-15",
"to_date": "2026-03-17",
"unit": "month",
},
)
if r.status_code == 429:
time.sleep(60)
r.raise_for_status()
funnel = r.json()
steps = []
meta = funnel.get("meta", {}).get("dates", {})
date_key = list(meta.keys())[0] if meta else None
funnel_data = funnel.get("data", {})
if date_key and date_key in funnel_data:
step_data = funnel_data[date_key].get("steps", [])
for step in step_data:
steps.append({
"name": step.get("event", "Unknown"),
"count": step.get("count", 0),
"avg_time": step.get("avg_time", None),
"goal": step.get("goal", False),
})
drop_offs = []
for i in range(1, len(steps)):
prev = steps[i - 1]["count"]
curr = steps[i]["count"]
rate = (prev - curr) / max(prev, 1)
drop_offs.append({
"from_step": steps[i - 1]["name"],
"to_step": steps[i]["name"],
"lost": prev - curr,
"drop_rate": rate,
"avg_time": steps[i].get("avg_time"),
})
drop_offs.sort(key=lambda d: d["drop_rate"], reverse=True)
funnel_summary = "\n".join(
f" {i+1}. {s['name']}: {s['count']} users" + (f" (avg {s['avg_time']:.0f}s to reach)" if s['avg_time'] else "")
for i, s in enumerate(steps)
)
drop_summary = "\n".join(
f" - {d['from_step']} → {d['to_step']}: {d['lost']} users lost ({d['drop_rate']:.0%})"
for d in drop_offs
)
PERSONA_IDS = os.environ.get("FUNNEL_PERSONA_IDS", "").split(",")
if not PERSONA_IDS[0]:
for name, desc in [
("New Free User", "Just signed up on a free plan. Exploring the product for the first time. Limited patience."),
("Team Admin", "Setting up the tool for their team. Needs onboarding to be fast and clear. Accountable for adoption."),
("Technical User", "Developer or analyst. Expects API-first workflows. Impatient with GUIs."),
]:
p = requests.post(f"{MB}/personas", headers=MH, json={"name": name, "description": desc}).json()
PERSONA_IDS.append(p["id"])
time.sleep(0.3)
worst = drop_offs[0] if drop_offs else {"from_step": "Unknown", "to_step": "Unknown", "drop_rate": 0}
fg = requests.post(f"{MB}/focus-groups", headers=MH, json={
"name": "Mixpanel Funnel Drop-off Investigation",
"persona_ids": [pid for pid in PERSONA_IDS if pid],
"questions": [
f"You just completed '{worst['from_step']}' in our product. What would prevent you from continuing to '{worst['to_step']}'? Be specific about what you'd be thinking or feeling.",
"Rank these barriers from most to least likely to make you quit: (1) Confusing UI, (2) Too many steps, (3) Need to invite teammates first, (4) Can't see value yet, (5) Technical error. Explain your #1.",
"What's the ONE thing you'd need to see or experience to keep going through the onboarding?",
"If you abandoned this flow, would you come back later or never return? What would trigger a return?",
],
"context": f"""Mixpanel funnel analysis for our product onboarding flow:
FUNNEL STEPS:
{funnel_summary}
DROP-OFF ANALYSIS:
{drop_summary}
Worst drop-off: {worst['from_step']} → {worst['to_step']} at {worst['drop_rate']:.0%}.""",
"responses_per_persona": 2,
}).json()
for _ in range(24):
time.sleep(5)
data = requests.get(f"{MB}/focus-groups/{fg['id']}", headers=MH).json()
if data.get("status") == "completed":
break
print(f"Focus Group: {data.get('id')} — {data.get('status')}\n")
for resp in data.get("responses", []):
print(f"[{resp.get('persona_id','?')}] {resp.get('question','')[:70]}")
print(f" → {resp.get('answer','')[:250]}\n")
