Webhook Event → Mave Research Trigger

Listen for Customer.io reporting webhooks on churn signals and trigger Mave Agent retention research tailored to the customer’s persona type

Scenario

Customer.io’s reporting webhooks fire on critical events — email bounces, unsubscribes, conversions, subscription changes. When a high-value customer unsubscribes, you don’t want to find out in a weekly report. This job sets up a webhook receiver that listens for specific events, maps the customer to a persona type using their attributes, then triggers Mave Agent to research retention strategies tailored to that persona. The result is an immediate, actionable research brief every time a churn signal fires.

Flow: Customer.io Reporting Webhook → Filter high-value events → Look up customer attributes → Map to persona type → Mavera POST /api/v1/mave/chat: “Research retention strategies for {persona type}” → Research brief

Architecture

Code

Python

import os, json, requests, hmac, hashlib
from http.server import HTTPServer, BaseHTTPRequestHandler

CIO_APP = os.environ["CIO_APP_KEY"]
CIO_WEBHOOK_SECRET = os.environ.get("CIO_WEBHOOK_SECRET", "")
MV = os.environ["MAVERA_API_KEY"]
APP_BASE = "https://api.customer.io/v1"
MB = "https://app.mavera.io/api/v1"
APP_H = {"Authorization": f"Bearer {CIO_APP}"}
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}

HIGH_VALUE_PLANS = {"pro", "enterprise", "business"}
ALERT_EVENTS = {"unsubscribed", "bounced", "complained"}

def verify_signature(payload: bytes, signature: str) -> bool:
    if not CIO_WEBHOOK_SECRET:
        return True
    expected = hmac.new(
        CIO_WEBHOOK_SECRET.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

def classify_persona(attrs: dict) -> str:
    plan = (attrs.get("plan_type") or "free").lower()
    industry = (attrs.get("industry") or "general").lower()
    usage = (attrs.get("usage_tier") or "low").lower()
    mrr = float(attrs.get("mrr", 0) or 0)

    if mrr > 500:
        return f"High-MRR {industry.title()} ({plan.title()} plan)"
    if plan in ("enterprise", "business"):
        return f"Enterprise {industry.title()} ({usage} usage)"
    return f"{plan.title()} {industry.title()} User"

def handle_churn_signal(event: dict):
    customer_id = event.get("customer_id", event.get("identifiers", {}).get("id"))
    event_type = event.get("event_type", event.get("metric", "unknown"))
    timestamp = event.get("timestamp", "")

    if not customer_id:
        print(f"No customer_id in event: {event_type}")
        return

    # 1. Enrich with customer attributes
    r = requests.get(f"{APP_BASE}/customers/{customer_id}/attributes",
        headers=APP_H)
    if not r.ok:
        print(f"Could not fetch attributes for {customer_id}: {r.status_code}")
        attrs = {}
    else:
        attrs = r.json().get("customer", {})

    plan = (attrs.get("plan_type") or "free").lower()
    if plan not in HIGH_VALUE_PLANS:
        print(f"Skipping free-tier {event_type} for {customer_id}")
        return

    persona_type = classify_persona(attrs)
    email = attrs.get("email", "unknown")
    mrr = attrs.get("mrr", "N/A")
    tenure_days = attrs.get("tenure_days", "N/A")
    last_active = attrs.get("last_active_at", "N/A")

    # 2. Research retention strategies
    research = requests.post(f"{MB}/mave/chat", headers=MV_H, json={
        "message": f"""A high-value customer just triggered a churn signal.

EVENT: {event_type}
PERSONA TYPE: {persona_type}
MRR: ${mrr}
TENURE: {tenure_days} days
LAST ACTIVE: {last_active}
PLAN: {plan}
INDUSTRY: {attrs.get('industry', 'N/A')}

Research retention strategies for this persona type:
1) Common reasons this persona type churns in {attrs.get('industry', 'SaaS')}
2) Proven retention tactics (with examples from similar companies)
3) Win-back email sequence outline (3 emails)
4) Offer structure that works for {plan} tier
5) Timing recommendations for outreach
6) Signals to watch for re-engagement potential"""
    }).json()

    brief = research.get("content", "")
    sources = research.get("sources", [])

    print(f"\n{'='*60}")
    print(f"CHURN ALERT: {event_type} | {email} | {persona_type}")
    print(f"MRR: ${mrr} | Tenure: {tenure_days}d | Plan: {plan}")
    print(f"{'='*60}")
    print(brief[:2000])
    print(f"\nSources: {len(sources)}")

    return {
        "customer_id": customer_id,
        "event_type": event_type,
        "persona_type": persona_type,
        "research_brief": brief,
        "sources": sources,
    }


class WebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length)
        sig = self.headers.get("X-CIO-Signature", "")

        if not verify_signature(body, sig):
            self.send_response(401)
            self.end_headers()
            return

        event = json.loads(body)
        event_type = event.get("event_type", event.get("metric", ""))

        if event_type in ALERT_EVENTS:
            result = handle_churn_signal(event)
            self.send_response(200)
            self.end_headers()
            self.wfile.write(json.dumps(result or {}).encode())
        else:
            self.send_response(200)
            self.end_headers()

    def log_message(self, format, *args):
        pass


if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", 8080), WebhookHandler)
    print("Webhook listener on :8080")
    server.serve_forever()

Example Output

============================================================
CHURN ALERT: unsubscribed | [e-mail] | High-MRR Fintech (Pro plan)
MRR: $850 | Tenure: 342d | Plan: pro
============================================================

## Retention Analysis: High-MRR Fintech Pro User

### Common Churn Reasons
1. Feature ceiling — Pro plan lacks API access they've outgrown
2. Compliance concerns — Fintech customers need SOC 2 + audit logs
3. Champion departure — Primary user left; no secondary adopter
4. Competitor poaching — 60% of fintech churn involves direct outreach

### Win-Back Sequence
**Email 1 (Day 0):** Personal note from CS lead. "We noticed you
unsubscribed — was something off?" Low pressure, genuine curiosity.
**Email 2 (Day 3):** Share roadmap preview relevant to fintech.
"We're shipping audit logs in 3 weeks — you asked for this."
**Email 3 (Day 7):** Offer: 30-day Enterprise trial at Pro price.
Include case study from similar fintech customer.

### Offer Structure
- 30-day Enterprise upgrade at current rate (test if features fix it)
- Quarterly billing option (reduce perceived commitment)
- Dedicated onboarding session for team beyond primary user

### Timing
- First outreach within 4 hours of unsubscribe
- Avoid end-of-month (budget stress in fintech)
- Best response rates: Tuesday 10am–12pm local time

Sources: 3

Error Handling

Webhook signature verification

Customer.io signs webhooks with HMAC SHA-256. Set CIO_WEBHOOK_SECRET from your workspace settings. The code gracefully skips verification if no secret is set (dev mode only — always verify in production).

Customer attribute lookup fails

If the customer was deleted or the App API is down, attributes will be empty. The code continues with defaults but the persona classification will be generic. Consider caching attributes locally.

Webhook event schema variations

Customer.io uses event_type in some webhook versions and metric in others. The code checks both fields. Test with Customer.io’s webhook tester before deploying.

High webhook volume

During a campaign blast, hundreds of events may fire simultaneously. Use a queue (Redis, SQS) between the webhook handler and Mave calls to avoid overwhelming Mavera’s rate limits.

---

What’s Next

Customer.io Integration

Back to Customer.io integration overview

Customer Attribute Personas

Build attribute-clustered personas from Customer.io segments

Campaign Messaging Strategy

Analyze campaign metrics for winning patterns

Mave Agent

Full reference for POST /api/v1/mave/chat