Product analytics with AI: PostHog, Mixpanel and smart insights without a data analyst

claude mcp add --transport http posthog https://mcp.posthog.com/mcp

import requests

def get_mixpanel_funnel(project_id, funnel_id, api_secret):
    url = f"https://mixpanel.com/api/2.0/funnels"
    params = {
        "funnel_id": funnel_id,
        "from_date": "2026-09-01",   # example: put in your own dates
        "to_date": "2026-09-30",
    }
    response = requests.get(
        url,
        params=params,
        auth=(api_secret, "")
    )
    return response.json()

import anthropic
import json

def predict_churn(users_activity_data: list[dict]) -> dict:
    """
    users_activity_data is a list of dicts with each user's activity:
    [{"user_id": "u123", "last_login_days_ago": 8, "sessions_last_30d": 2,
      "feature_usage": {"core": 5, "advanced": 0}, "plan": "free"}]
    """
    client = anthropic.Anthropic()

    prompt = f"""
You are a product analyst. Here is activity data for {len(users_activity_data)} users over the last 30 days:

{json.dumps(users_activity_data, ensure_ascii=False, indent=2)}

Analyze the patterns and:
1. Single out HIGH RISK users (likely to leave in the next 2 weeks)
2. Explain the signals you used to determine the risk
3. Suggest a specific action for each risk group (email, call, in-app notification)

Reply in JSON format:
{{
  "high_risk": [list of user_id],
  "medium_risk": [list of user_id],
  "risk_signals": "explanation",
  "actions": {{"high_risk": "action", "medium_risk": "action"}}
}}
"""

    message = client.messages.create(
        model="claude-sonnet-5-5",   # current models: the "What's current" page
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )

    return json.loads("".join(b.text for b in message.content if b.type == "text"))

def interpret_ab_test(test_data: dict) -> str:
    client = anthropic.Anthropic()

    message = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"""
Analyze the results of an A/B test:
{json.dumps(test_data, ensure_ascii=False)}

Answer:
1. Which version won, and how confidently?
2. Is the result statistically significant (sample of {test_data.get('total_users', 'N/A')} users)?
3. Recommend: roll out the winner or wait for more data?
Answer in English, briefly, in 3-4 sentences.
"""
        }]
    )

    return "".join(b.text for b in message.content if b.type == "text")

def analyze_session_patterns(sessions_metadata: list[dict]) -> str:
    rage_clicks = [s for s in sessions_metadata if s.get("rage_clicks", 0) > 2]
    high_exit_pages = {}

    for session in sessions_metadata:
        exit_page = session.get("exit_page", "")
        high_exit_pages[exit_page] = high_exit_pages.get(exit_page, 0) + 1

    analysis_data = {
        "total_sessions": len(sessions_metadata),
        "frustration_sessions": len(rage_clicks),
        "top_exit_pages": sorted(
            high_exit_pages.items(), key=lambda x: x[1], reverse=True
        )[:5],
        "avg_session_duration_minutes": sum(
            s.get("duration_seconds", 0) for s in sessions_metadata
        ) / len(sessions_metadata) / 60
    }

    client = anthropic.Anthropic()
    message = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""
User session data:
{json.dumps(analysis_data, ensure_ascii=False, indent=2)}

Find the main user pain points. What needs fixing first?
Answer: three specific problems with suggested fixes.
"""
        }]
    )

    return "".join(b.text for b in message.content if b.type == "text")

import os
import json
import anthropic
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv

load_dotenv()

# Keys come from .env, they're not written into the code
POSTHOG_PERSONAL_API_KEY = os.getenv("POSTHOG_PERSONAL_API_KEY")  # personal key (phx_...) for reading data
POSTHOG_PROJECT_ID = os.getenv("POSTHOG_PROJECT_ID")
POSTHOG_HOST = os.getenv("POSTHOG_HOST", "https://us.posthog.com")  # for Europe: https://eu.posthog.com
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")

def run_hogql(query: str) -> list:
    """Runs a HogQL query (PostHog's query language) through the API."""
    response = requests.post(
        f"{POSTHOG_HOST}/api/projects/{POSTHOG_PROJECT_ID}/query/",
        headers={"Authorization": f"Bearer {POSTHOG_PERSONAL_API_KEY}"},
        json={"query": {"kind": "HogQLQuery", "query": query}, "name": "weekly report"},
    )
    return response.json().get("results", []) if response.ok else []

def fetch_posthog_insights() -> dict:
    """Pull the key metrics for last week"""
    end_date = datetime.now()
    start_date = end_date - timedelta(days=7)

    # Pageviews by day
    pageviews_by_day = run_hogql(
        "select toDate(timestamp) as day, count() from events "
        "where event = '$pageview' and timestamp >= now() - interval 7 day "
        "group by day order by day"
    )

    # Active users for the week
    active_users = run_hogql(
        "select count(distinct person_id) from events "
        "where timestamp >= now() - interval 7 day"
    )

    # New sign-ups
    signups = run_hogql(
        "select count() from events "
        "where event = 'signed_up' and timestamp >= now() - interval 7 day"
    )

    return {
        "period": f"{start_date.strftime('%m/%d')} to {end_date.strftime('%m/%d/%Y')}",
        "pageviews_by_day": pageviews_by_day,
        "active_users": active_users,
        "signups": signups,
    }

def generate_weekly_report(insights: dict) -> str:
    """Claude writes a human-readable report"""
    client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)

    message = client.messages.create(
        model="claude-sonnet-5-5",   # current models: the "What's current" page
        max_tokens=1500,
        messages=[{
            "role": "user",
            "content": f"""
You are a product analyst. Here is the data for the week {insights['period']}:

{json.dumps(insights, ensure_ascii=False, indent=2, default=str)}

Write a weekly report for a startup founder. Structure:
📊 WEEK {insights['period']}

🟢 What's going well (2-3 points)
🔴 What needs attention (1-2 points)
🎯 This week's priority (1 specific action)

Tone: friendly, specific, no fluff. Write in English.
If there isn't much data, say so, don't invent reasons.
"""
        }]
    )

    return "".join(b.text for b in message.content if b.type == "text")

def send_to_telegram(text: str):
    """Send the report to Telegram"""
    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
        json={
            "chat_id": TELEGRAM_CHAT_ID,
            "text": text,
        }
    )

def run_weekly_report():
    print("Collecting data from PostHog...")
    insights = fetch_posthog_insights()

    print("Claude is analyzing...")
    report = generate_weekly_report(insights)

    print("Sending to Telegram...")
    send_to_telegram(report)
    print("Done!")

if __name__ == "__main__":
    run_weekly_report()

<script>
  !function(t,e){var o,n,p,r;e.__SV||(window.posthog=e,e._i=[],e.init=function(i,s,a){...}
  posthog.init('phc_your_key', {api_host: 'https://us.i.posthog.com'})   // for Europe: https://eu.i.posthog.com
</script>

pip install posthog

from posthog import Posthog

posthog = Posthog('phc_your_key', host='https://us.i.posthog.com')
# Keyword arguments work across different versions of the library
posthog.capture(event='signed_up', distinct_id='user_123', properties={'plan': 'free', 'source': 'google'})

mkdir analytics_bot
cd analytics_bot
pip install anthropic requests python-dotenv

POSTHOG_PERSONAL_API_KEY=phx_your_personal_key
POSTHOG_PROJECT_ID=your_project_id
POSTHOG_HOST=https://us.posthog.com
ANTHROPIC_API_KEY=sk-ant-your_key
TELEGRAM_BOT_TOKEN=your_token
TELEGRAM_CHAT_ID=your_chat_id

python weekly_report.py

from dotenv import load_dotenv
import os
import anthropic
import requests
import json

load_dotenv()

POSTHOG_HOST = os.getenv("POSTHOG_HOST", "https://us.posthog.com")

def get_at_risk_users():
    """Get users who haven't logged in for 7+ days (simplified: check the filter and fields against the PostHog docs)"""
    headers = {"Authorization": f"Bearer {os.getenv('POSTHOG_PERSONAL_API_KEY')}"}

    # Simplified query: users with low activity
    response = requests.get(
        f"{POSTHOG_HOST}/api/projects/{os.getenv('POSTHOG_PROJECT_ID')}/persons/",
        headers=headers,
        params={"properties": '[{"key":"$last_seen","value":"-7d","operator":"is_before_date"}]'}
    )

    if not response.ok:
        return []

    persons = response.json().get("results", [])
    # Send only internal IDs and counters to the model, no names or emails
    return [
        {
            "user_id": p.get("distinct_ids", ["unknown"])[0],
            "last_seen": p.get("properties", {}).get("$last_seen"),
            "total_sessions": p.get("properties", {}).get("$session_count", 0),
        }
        for p in persons[:50]  # Take the first 50
    ]

def run_churn_check():
    users = get_at_risk_users()
    if not users:
        print("No users at risk")
        return

    client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
    message = client.messages.create(
        model="claude-haiku-4-5",  # Haiku is enough for this task (check that the model is still available in the API)
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""
Here are users who haven't logged in for a while ({len(users)} people):
{json.dumps(users[:10], ensure_ascii=False)}

What should we write to bring them back? One short email (3-4 sentences) for re-engagement.
"""
        }]
    )

    print("Suggested re-engagement text:")
    print("".join(b.text for b in message.content if b.type == "text"))
    print(f"\nUsers at risk: {len(users)}")

if __name__ == "__main__":
    run_churn_check()

# Open crontab
crontab -e

# Add these lines:
# Monday 9:00: weekly report
0 9 * * 1 cd /path/to/analytics_bot && python weekly_report.py

# Friday 17:00: churn check
0 17 * * 5 cd /path/to/analytics_bot && python churn_detector.py >> churn_log.txt