Documentation Index
Fetch the complete documentation index at: https://docs.mavera.io/llms.txt
Use this file to discover all available pages before exploring further.
Scenario
Your CX survey measures satisfaction by touchpoint — onboarding, product usage, support, billing, renewal. When a touchpoint scores below 7/10, you need to understand why — fast. This job identifies low-scoring touchpoints from your CX data, creates personas representing dissatisfied customers at each touchpoint, then runs Focus Groups asking: “Describe your experience. What would improve it?” The result is a touchpoint-specific improvement plan grounded in the voice of the customer. Flow: Qualtrics CX export → Identify low-scoring touchpoints →POST /api/v1/personas per touchpoint → POST /api/v1/focus-groups: “Describe your experience. What would improve it?” → Touchpoint improvement plan
Architecture
Code
import os, csv, io, zipfile, requests, time
from collections import defaultdict
QT = os.environ["QUALTRICS_TOKEN"]
DC = os.environ["QUALTRICS_DC"]
MV = os.environ["MAVERA_API_KEY"]
Q_BASE = f"https://{DC}.qualtrics.com/API/v3"
MB = "https://app.mavera.io/api/v1"
Q_H = {"X-API-TOKEN": QT, "Content-Type": "application/json"}
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
SURVEY_ID = os.environ.get("CX_SURVEY_ID", "SV_xxxxx")
TOUCHPOINTS = {
"Onboarding": os.environ.get("COL_ONBOARDING", "CX_Onboarding"),
"Product Usage": os.environ.get("COL_PRODUCT", "CX_Product"),
"Support": os.environ.get("COL_SUPPORT", "CX_Support"),
"Billing": os.environ.get("COL_BILLING", "CX_Billing"),
"Renewal": os.environ.get("COL_RENEWAL", "CX_Renewal"),
}
THRESHOLD = 7.0
# 1. Export + parse
export = requests.post(f"{Q_BASE}/surveys/{SURVEY_ID}/export-responses",
headers=Q_H, json={"format": "csv"}).json()
pid = export["result"]["progressId"]
file_id = None
for _ in range(60):
time.sleep(5)
s = requests.get(f"{Q_BASE}/surveys/{SURVEY_ID}/export-responses/{pid}",
headers=Q_H).json()
if s["result"].get("percentComplete") == 100:
file_id = s["result"]["fileId"]; break
zip_data = requests.get(
f"{Q_BASE}/surveys/{SURVEY_ID}/export-responses/{file_id}/file",
headers=Q_H).content
with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
csv_name = [n for n in zf.namelist() if n.endswith(".csv")][0]
with zf.open(csv_name) as f:
rows = list(csv.DictReader(io.TextIOWrapper(f, encoding="utf-8-sig")))
responses = [r for r in rows if r.get("Finished") == "1"]
print(f"CX responses: {len(responses)}")
# 2. Score touchpoints
tp_scores = {}
tp_comments = defaultdict(list)
for tp_name, col in TOUCHPOINTS.items():
scores = []
comment_col = f"{col}_Comment"
for r in responses:
try:
val = float(r.get(col, 0) or 0)
scores.append(val)
if r.get(comment_col, "").strip():
tp_comments[tp_name].append(r[comment_col].strip())
except ValueError:
continue
if scores:
tp_scores[tp_name] = {
"avg": sum(scores) / len(scores),
"n": len(scores),
"low_count": sum(1 for s in scores if s < THRESHOLD),
"low_pct": sum(1 for s in scores if s < THRESHOLD) / len(scores),
}
print("\nTouchpoint Scores:")
for tp, data in sorted(tp_scores.items(), key=lambda x: x[1]["avg"]):
flag = " ⚠️ BELOW THRESHOLD" if data["avg"] < THRESHOLD else ""
print(f" {tp}: {data['avg']:.1f}/10 (n={data['n']}, "
f"{data['low_pct']:.0%} below {THRESHOLD}){flag}")
# 3. Create personas for low-scoring touchpoints
low_tps = {tp: data for tp, data in tp_scores.items() if data["avg"] < THRESHOLD}
if not low_tps:
print("\nAll touchpoints above threshold. No focus groups needed.")
exit()
persona_ids = {}
for tp_name, data in low_tps.items():
comments = tp_comments.get(tp_name, [])[:5]
comment_str = "; ".join(comments[:3]) if comments else "No comments"
p = requests.post(f"{MB}/personas", headers=MV_H, json={
"name": f"QX CX: Dissatisfied at {tp_name}",
"description": (
f"Customer who scored {tp_name} below {THRESHOLD}/10. "
f"Avg score: {data['avg']:.1f}. {data['low_count']}/{data['n']} "
f"respondents dissatisfied. Feedback: {comment_str[:200]}"
),
"psychographic": {
"satisfaction": "low",
"touchpoint": tp_name,
"avg_score": data["avg"],
},
})
p.raise_for_status()
persona_ids[tp_name] = p.json()["id"]
time.sleep(0.3)
# 4. Focus Group per touchpoint
for tp_name, pid in persona_ids.items():
data = low_tps[tp_name]
comments = tp_comments.get(tp_name, [])
fg = requests.post(f"{MB}/focus-groups", headers=MV_H, json={
"name": f"CX Deep-Dive: {tp_name}",
"persona_ids": [pid],
"context": (
f"You recently experienced the {tp_name.lower()} process and "
f"gave it {data['avg']:.1f}/10. Other customers in your "
f"situation said: {'; '.join(comments[:3])}"
),
"questions": [
f"Describe your {tp_name.lower()} experience in detail. What happened?",
"What was the single most frustrating moment?",
"What would have made this experience a 9 or 10?",
"How does this experience affect your likelihood to renew or recommend us?",
"If you were redesigning this process, what would you change first?",
f"Compare your {tp_name.lower()} experience to the best you've had with any vendor.",
],
"responses_per_persona": 3,
}).json()
for _ in range(20):
time.sleep(5)
result = requests.get(f"{MB}/focus-groups/{fg['id']}",
headers=MV_H).json()
if result.get("status") == "completed":
break
print(f"\n{'='*50}")
print(f"CX DEEP-DIVE: {tp_name} (avg: {data['avg']:.1f}/10)")
print(f"{'='*50}")
for resp in result.get("responses", []):
print(f"\nQ: {resp.get('question', '')[:60]}...")
print(f"A: {resp.get('answer', '')[:300]}")
# 5. Generate improvement plan
all_touchpoint_data = "\n".join(
f"- {tp}: {d['avg']:.1f}/10, {d['low_pct']:.0%} dissatisfied"
for tp, d in sorted(tp_scores.items(), key=lambda x: x[1]["avg"]))
all_comments = "\n".join(
f"[{tp}] {c}"
for tp, comments in tp_comments.items()
for c in comments[:3])
plan = requests.post(f"{MB}/mave/chat", headers=MV_H, json={
"message": f"""Create a prioritized CX improvement plan based on this data.
TOUCHPOINT SCORES:
{all_touchpoint_data}
SAMPLE CUSTOMER FEEDBACK:
{all_comments[:2000]}
Produce:
1) Priority ranking (which touchpoint to fix first and why)
2) Root cause analysis per low-scoring touchpoint
3) Quick wins (implementable in 2 weeks)
4) Medium-term improvements (1-3 months)
5) Metrics to track improvement
6) Estimated impact on overall NPS"""
}).json()
print(f"\n{'='*50}")
print("CX IMPROVEMENT PLAN")
print(f"{'='*50}")
print(plan.get("content", "")[:2000])
Example Output
Touchpoint Scores:
Onboarding: 5.8/10 (n=892, 64% below 7) ⚠️ BELOW THRESHOLD
Billing: 6.3/10 (n=743, 48% below 7) ⚠️ BELOW THRESHOLD
Support: 7.4/10 (n=1204, 28% below 7)
Product Usage: 8.1/10 (n=1350, 12% below 7)
Renewal: 7.8/10 (n=456, 18% below 7)
==================================================
CX DEEP-DIVE: Onboarding (avg: 5.8/10)
==================================================
Q: Describe your onboarding experience in detail.
A: I was handed login credentials and a link to a help center with 200
articles. No guided setup, no quick-start, no "here are the 3 things
to do first." I spent 4 hours trying to figure out the basic workflow
that should have taken 20 minutes with guidance.
Q: What would have made this a 9 or 10?
A: A structured 5-step onboarding checklist that shows progress. A 30-
minute live session in week 1 (not a webinar — a real person looking
at my account). And templates — don't make me start from scratch.
Q: Compare to the best onboarding you've had.
A: Notion. Day 1: pre-built workspace with sample content. Day 3:
personalized email based on what I'd actually set up. Day 7: check-in
from a human. Total time to value: 2 hours. Yours: still figuring
it out after 2 weeks.
==================================================
CX IMPROVEMENT PLAN
==================================================