What this is: A complete, production-ready blueprint for setting up an AI support agent on OpenClaw that handles Intercom support for Shopify apps. Based on a battle-tested setup running in production since early 2026.
How to use it: Give this file to your OpenClaw agent instance. It will walk you through a guided setup, ask the right questions about your specific app, and auto-configure itself. No webhooks needed — this uses a polling architecture that's simpler, more reliable, and doesn't require exposing endpoints.
- Architecture Overview
- Guided Self-Configuration
- Agent Identity (SOUL.md)
- Intercom Integration
- Scripts — The Polling Engine
- Cron Jobs — The Heartbeat
- Conversation State Management
- Playbooks — Repeatable Workflows
- Escalation Rules
- Lessons Architecture
- Health Monitoring
- Database Integration (Optional)
- Common Mistakes — Learn From Ours
- File Structure Reference
- Quick Start Checklist
Webhooks sound elegant but introduce real problems for a support agent:
| Problem | Webhooks | Polling |
|---|---|---|
| Setup complexity | Need public endpoint, SSL, auth, retry logic | Zero infrastructure — just a cron job |
| Reliability | Missed events if endpoint is down | Always catches up on next poll |
| Duplicate handling | Must deduplicate webhook retries | Stateful diffing built in |
| Rate limits | Burst of events can overwhelm | Controlled, predictable API usage |
| Debugging | Hard to replay missed events | Just run the script manually |
| Security | Exposed endpoint = attack surface | No public endpoints needed |
┌─────────────────────────────────────────────────────┐
│ OpenClaw Gateway │
│ │
│ ┌──────────┐ every 2 min ┌──────────────────┐ │
│ │ Cron Job │───────────────▶│ poll_inbox.py │ │
│ │(inbox-poll)│ │ (stateful diff) │ │
│ └──────────┘ └────────┬─────────┘ │
│ │ │
│ changes detected? │
│ │ │ │
│ yes no → exit │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Agent Session │ │
│ │ (isolated run) │ │
│ │ │ │
│ │ 1. Read context │ │
│ │ 2. Fetch conv │ │
│ │ 3. Investigate │ │
│ │ 4. Reply/escalate│ │
│ │ 5. Save state │ │
│ └─────────────────┘ │
│ │
│ ┌──────────┐ every 15 min ┌──────────────────┐ │
│ │ Cron Job │──────────────▶│ Health Monitor │ │
│ │(health) │ │ (self-diagnostic) │ │
│ └──────────┘ └──────────────────┘ │
│ │
│ ┌──────────┐ every 30 min ┌──────────────────┐ │
│ │ Cron Job │──────────────▶│ Heartbeat │ │
│ │(heartbeat)│ │ (cleanup + audit) │ │
│ └──────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────┘
- Investigate before responding. Never ask the merchant for information you can look up yourself.
- One conversation per poll cycle. Handle the highest-priority ticket fully, then stop. The next cycle catches the rest. This prevents timeouts and ensures quality.
- State is file-based. Conversation progress is tracked in JSON files, not in memory. The agent is stateless between runs.
- Duplicate checking is non-negotiable. Always re-fetch the conversation before sending a reply.
- Escalate when uncertain. Better to escalate than to send a wrong answer.
When this agent starts for the first time, it should run through this questionnaire and save the answers to its workspace. The agent should ask these questions conversationally, not dump them all at once.
QUESTIONS TO ASK THE FOUNDER:
1. What's the agent's name? (e.g., "Lisa", "Alex", "Sam")
2. What's the agent's email address? (e.g., name@company.com)
3. What's the company name? (e.g., "Arcspeed", "Stackify")
4. What's the founder's name? (The person the agent escalates to)
5. What's the founder's Telegram chat ID or preferred alert channel?
6. What tone should the agent use? (casual/professional/friendly-professional)
7. What language(s) should the agent support? (e.g., English only, match customer language)
QUESTIONS TO ASK:
1. What's your Intercom API token? (Settings → Developers → Access Token)
2. What's the agent's Intercom Admin ID?
→ Help them find it: "Go to Settings → Teammates, click on the agent teammate,
the ID is in the URL: app.intercom.com/a/apps/YOUR_APP/admins/ADMIN_ID"
3. What's YOUR (founder's) Intercom Admin ID? (So we skip tickets assigned to you)
4. Do you have any contacts that should be permanently skipped?
(We'll use a custom attribute called `bot_skip` for this)
QUESTIONS TO ASK:
1. How many Shopify apps do you support? List each with:
- App name (as shown in Shopify App Store)
- App listing URL
- Codebase path (if you want the agent to read code for investigations)
- The `app_name` value in Intercom custom attributes (if you use contact tagging)
- Any unique domain or URL pattern to identify the app
2. For each app, what are the top 5 most common support questions?
(The agent will build a response playbook from these)
3. Do you have a knowledge base / help center URL?
4. Do you have a production database the agent can query (read-only)?
If yes: connection string, key tables, and useful queries.
QUESTIONS TO ASK:
1. Should the agent auto-close stale conversations? If yes, after how many hours? (Default: 24h)
2. Should the agent snooze conversations after replying? (Default: yes — 1 day for chat, 7 days for email)
3. What should ALWAYS be escalated to you? (e.g., refund requests, Shopify compliance, bug reports)
4. Do you have any engagement playbooks? (e.g., asking for reviews, upselling features)
5. What's your preferred polling frequency? (Default: every 2 minutes)
After collecting answers, the agent should:
- Generate
AGENTS.mdwith all app-specific details, workflows, and common patterns - Generate
SOUL.mdwith the agent's identity and voice - Generate
IDENTITY.mdwith email/contact details - Create all scripts in
scripts/(see Scripts section) - Set up cron jobs (see Cron Jobs section)
- Create the
conversations/andconversations/archive/directories - Create an empty
lessons.mdfile - Create
HEARTBEAT.mdwith the heartbeat checklist - Test the Intercom API connection
- Do a dry run of
check_inbox.pyto verify everything works
The agent should generate a SOUL.md based on the founder's answers. Here's the structure:
# SOUL.md — {AGENT_NAME}, {COMPANY_NAME} Support
You are **{AGENT_NAME}**, the customer support agent for {COMPANY_NAME}'s Shopify apps.
You handle support via Intercom.
## Identity — NON-NEGOTIABLE
- **You are {AGENT_NAME}.** Always. Every reply. No exceptions.
- **Intercom Admin ID: `{AGENT_ADMIN_ID}`** — this is YOUR id. Use it for every API call.
- **Never reply as {FOUNDER_NAME}** (admin ID `{FOUNDER_ADMIN_ID}`). They are your boss, not you.
## Voice
- Sound human, not like AI. Write like a real support person.
- Keep responses short. 1-3 sentences when possible.
- {TONE_GUIDELINES — generated from founder's tone preference}
- Match the customer's energy. Casual if they're casual, professional if formal.
- Be direct, warm, slightly witty when it fits.
- Respond in the customer's language (match conversation language).
## High Agency Problem Solving
- **Fix, don't just report.** When you encounter issues, diagnose and solve them.
- **Root causes, not symptoms.** Think deeply about why something broke.
- **Simple is superior.** Prefer minimal solutions that solve problems at scale.
- **Learn from every mistake.** Each error makes you better.
## Boundaries
- **Never close conversations.** Leave them open for {FOUNDER_NAME} to review.
(Exception: auto-close via close_stale.py after {STALE_HOURS}h of no customer response)
- **Never send emails, tweets, or anything outside Intercom** (unless explicitly configured).
- **Never run destructive database queries** (INSERT/UPDATE/DELETE) unless explicitly granted.
- When unsure, escalate to {FOUNDER_NAME} via {ALERT_CHANNEL}.# IDENTITY.md
- **Name:** {AGENT_NAME}
- **Email:** {AGENT_EMAIL}
- **Role:** {COMPANY_NAME} AI Support Agent (powered by OpenClaw)
- **Vibe:** Warm, helpful, efficient support repAll Intercom API calls use the same pattern:
import urllib.request, json
TOKEN = "{INTERCOM_API_TOKEN}"
def api(method, path, data=None):
url = f"https://api.intercom.io{path}"
body = json.dumps(data).encode() if data else None
req = urllib.request.Request(url, data=body, method=method, headers={
"Authorization": f"Bearer {TOKEN}",
"Intercom-Version": "2.11",
"Content-Type": "application/json",
"Accept": "application/json"
})
with urllib.request.urlopen(req, timeout=15) as r:
return json.loads(r.read())Important: Use urllib.request (stdlib) — no external dependencies needed. This makes scripts portable and eliminates pip install steps.
| Action | Method | Path |
|---|---|---|
| Search conversations | POST | /conversations/search |
| Get conversation | GET | /conversations/{id} |
| Reply to conversation | POST | /conversations/{id}/reply |
| Add note | POST | /conversations/{id}/reply (with message_type: note) |
| Close conversation | POST | /conversations/{id}/parts (with message_type: close) |
| Snooze conversation | POST | /conversations/{id}/parts (with message_type: snoozed) |
| Get contact | GET | /contacts/{id} |
| Tag conversation | POST | /tags |
- In Intercom, go to Settings → Teammates
- Create a new teammate for the agent (or use an existing one)
- Note the Admin ID from the URL
- Generate an API token with conversation read/write permissions
- Test: run
check_inbox.pyand verify it returns results
Create a tag called bot_pause in Intercom. When you (the founder) are manually handling a conversation, apply this tag. The polling scripts will skip any conversation with this tag, preventing duplicate replies.
To create it via API:
api("POST", "/tags", {"name": "bot_pause"})
# Note the returned tag ID for referenceCreate these scripts in your workspace's scripts/ directory. They use zero external dependencies (stdlib only).
This is the main engine. It runs every 2 minutes, compares the current inbox state to the previous state, and only outputs when something changes. Silent exits mean "nothing to do."
#!/usr/bin/env python3
"""Stateful inbox poller. Only outputs when something changed.
Safety net: if a conversation has been waiting on us for >STALE_THRESHOLD_SEC
with no change in waiting_since, re-emit it as a 'stale_waiting' event so the
handler gets another chance to process it.
"""
import urllib.request, json, sys, os, time
# ──────────────────────────────────────────────
# CONFIGURE THESE VALUES DURING SETUP
# ──────────────────────────────────────────────
TOKEN = "{INTERCOM_API_TOKEN}"
FOUNDER_ADMIN_ID = "{FOUNDER_ADMIN_ID}" # Skip tickets assigned to founder
STATE_FILE = os.path.join(os.path.dirname(__file__), ".inbox_state.json")
STALE_THRESHOLD_SEC = 300 # 5 minutes — re-emit if waiting this long unchanged
# ──────────────────────────────────────────────
def api(method, path, data=None):
url = f"https://api.intercom.io{path}"
body = json.dumps(data).encode() if data else None
req = urllib.request.Request(url, data=body, method=method, headers={
"Authorization": f"Bearer {TOKEN}",
"Intercom-Version": "2.11",
"Content-Type": "application/json",
"Accept": "application/json"
})
with urllib.request.urlopen(req, timeout=15) as r:
return json.loads(r.read())
def get_contact(contact_id):
try:
return api("GET", f"/contacts/{contact_id}")
except:
return {}
def load_state():
try:
with open(STATE_FILE) as f:
return json.load(f)
except:
return {}
def save_state(state):
with open(STATE_FILE, "w") as f:
json.dump(state, f)
def main():
# Get open, non-snoozed conversations
try:
data = api("POST", "/conversations/search", {
"query": {
"operator": "AND",
"value": [
{"field": "state", "operator": "=", "value": "open"},
{"field": "snoozed_until", "operator": "=", "value": None}
]
},
"pagination": {"per_page": 50}
})
except Exception as e:
print(json.dumps({"error": f"Search failed: {e}"}))
sys.exit(1)
conversations = data.get("conversations", [])
prev_state = load_state()
prev_waiting = prev_state.get("waiting_on_us", {})
prev_open_ids = set(prev_state.get("open_ids", []))
current_waiting = {}
current_open_ids = set()
skipped = []
changes = []
for conv in conversations:
cid = conv["id"]
current_open_ids.add(cid)
# Skip tickets assigned to founder
assignee = conv.get("admin_assignee_id")
if str(assignee) == FOUNDER_ADMIN_ID:
skipped.append(cid)
continue
# Check bot_skip on contact
contacts = conv.get("contacts", {}).get("contacts", [])
should_skip = False
for ct in contacts:
contact = get_contact(ct["id"])
if contact.get("custom_attributes", {}).get("bot_skip"):
should_skip = True
break
if should_skip:
skipped.append(cid)
continue
ws = conv.get("waiting_since")
if ws:
current_waiting[cid] = ws
# Detect changes
now = int(time.time())
prev_first_seen = prev_state.get("first_seen_waiting", {})
current_first_seen = {}
changed_ids = set()
# 1. New tickets
new_tickets = current_open_ids - prev_open_ids - set(skipped)
for cid in new_tickets:
changes.append({"type": "new_ticket", "id": cid})
changed_ids.add(cid)
# 2. Customer replied (waiting_since changed)
for cid, ws in current_waiting.items():
if cid in skipped:
continue
prev_ws = prev_waiting.get(cid)
if prev_ws != ws:
changes.append({"type": "customer_replied", "id": cid, "waiting_since": ws})
changed_ids.add(cid)
current_first_seen[cid] = now
else:
current_first_seen[cid] = prev_first_seen.get(cid, now)
# 3. Safety net: re-emit stale waiting conversations
for cid, ws in current_waiting.items():
if cid in skipped or cid in changed_ids:
continue
first_seen = current_first_seen.get(cid, now)
if now - first_seen >= STALE_THRESHOLD_SEC:
changes.append({"type": "stale_waiting", "id": cid, "waiting_since": ws,
"stale_seconds": now - first_seen})
current_first_seen[cid] = now # Reset clock
# Save state
save_state({
"open_ids": list(current_open_ids),
"waiting_on_us": current_waiting,
"first_seen_waiting": current_first_seen,
"last_poll": now,
"skipped": skipped
})
# Only output if changes exist
if changes:
print(json.dumps({
"changes": changes,
"total_open": len(conversations),
"skipped": len(skipped)
}, indent=2))
else:
sys.exit(0) # Silent — nothing changed
if __name__ == "__main__":
main()How the state diffing works:
- On each run, it snapshots all open conversations and their
waiting_sincetimestamps - Compares against the previous snapshot (saved in
.inbox_state.json) - Detects: new tickets, customer replies (waiting_since changed), and stale-waiting safety net
- If nothing changed → silent exit (exit code 0, no output)
- If changes → outputs JSON for the agent to process
The stale_waiting safety net:
- If the agent fails to process a change (timeout, error, etc.), the state file already recorded it
- Without the safety net, that change would be permanently lost
- The safety net re-emits any conversation that's been waiting >5 minutes with no change
- This gives the agent another chance to handle it on the next cycle
Every reply goes through this script. No exceptions. It handles duplicate checking, correct admin ID, and auto-snoozing.
#!/usr/bin/env python3
"""Send Intercom reply as the support agent. No external deps."""
import urllib.request, json, argparse, sys, time
# ──────────────────────────────────────────────
# CONFIGURE THESE VALUES DURING SETUP
# ──────────────────────────────────────────────
TOKEN = "{INTERCOM_API_TOKEN}"
ADMIN_ID = "{AGENT_ADMIN_ID}"
SNOOZE_CHAT = 1 * 24 * 3600 # 1 day for chat
SNOOZE_EMAIL = 7 * 24 * 3600 # 7 days for email
# ──────────────────────────────────────────────
def api(method, path, data=None):
url = f"https://api.intercom.io{path}"
body = json.dumps(data).encode() if data else None
req = urllib.request.Request(url, data=body, method=method, headers={
"Authorization": f"Bearer {TOKEN}",
"Intercom-Version": "2.11",
"Content-Type": "application/json",
"Accept": "application/json"
})
with urllib.request.urlopen(req, timeout=15) as r:
return json.loads(r.read())
def check_duplicate(conv_id, message):
"""Check if last admin reply already covers this message."""
conv = api("GET", f"/conversations/{conv_id}")
parts = conv.get("conversation_parts", {}).get("conversation_parts", [])
for p in reversed(parts[-3:]):
if p.get("author", {}).get("type") == "admin" and p.get("part_type") == "comment":
existing = (p.get("body") or "").lower()
if message[:50].lower() in existing:
return True
return False
def main():
parser = argparse.ArgumentParser()
parser.add_argument("-c", "--conv-id", required=True)
parser.add_argument("-m", "--message", required=True)
parser.add_argument("--note", action="store_true", help="Send as internal note")
parser.add_argument("--no-snooze", action="store_true", help="Don't snooze after reply")
parser.add_argument("--channel", default="chat", choices=["chat", "email"],
help="Channel type for snooze duration")
args = parser.parse_args()
if check_duplicate(args.conv_id, args.message):
print(json.dumps({"ok": False, "reason": "duplicate_detected"}))
sys.exit(0)
msg_type = "note" if args.note else "comment"
result = api("POST", f"/conversations/{args.conv_id}/reply", {
"message_type": msg_type,
"type": "admin",
"admin_id": ADMIN_ID,
"body": args.message
})
# Snooze after reply (unless --no-snooze or it's a note)
snoozed = False
if not args.note and not args.no_snooze:
snooze_secs = SNOOZE_EMAIL if args.channel == "email" else SNOOZE_CHAT
snooze_until = int(time.time()) + snooze_secs
try:
api("POST", f"/conversations/{args.conv_id}/parts", {
"message_type": "snoozed",
"admin_id": ADMIN_ID,
"snoozed_until": snooze_until
})
snoozed = True
except Exception as e:
print(json.dumps({"ok": True, "conversation_id": args.conv_id,
"snoozed": False, "snooze_error": str(e)}))
sys.exit(0)
print(json.dumps({"ok": True, "action": msg_type,
"conversation_id": args.conv_id, "snoozed": snoozed}))
if __name__ == "__main__":
main()Usage:
# Regular reply (auto-snoozes for 1 day)
python3 scripts/reply.py -c CONV_ID -m "Your message here"
# Email reply (auto-snoozes for 7 days)
python3 scripts/reply.py -c CONV_ID -m "Your message here" --channel email
# Internal note (no snooze)
python3 scripts/reply.py -c CONV_ID -m "Internal note" --note
# Reply without snoozing
python3 scripts/reply.py -c CONV_ID -m "Quick response" --no-snoozeWhy snooze after reply?
- When the agent replies, the conversation moves to "waiting on customer"
- If the customer doesn't reply within the snooze period, the snooze expires
- The
close_stale.pyscript then picks it up and auto-closes it - This creates a natural lifecycle: reply → snooze → expire → close (if no response)
Unlike poll_inbox.py (which only reports changes), this gives a complete view of the inbox. Used for heartbeat checks and health monitoring.
#!/usr/bin/env python3
"""Check Intercom inbox for ALL open conversations needing support."""
import urllib.request, json, sys
# ──────────────────────────────────────────────
# CONFIGURE THESE VALUES DURING SETUP
# ──────────────────────────────────────────────
TOKEN = "{INTERCOM_API_TOKEN}"
FOUNDER_ADMIN_ID = "{FOUNDER_ADMIN_ID}"
BOT_PAUSE_TAG = "bot_pause"
# ──────────────────────────────────────────────
def api(method, path, data=None):
url = f"https://api.intercom.io{path}"
body = json.dumps(data).encode() if data else None
req = urllib.request.Request(url, data=body, method=method, headers={
"Authorization": f"Bearer {TOKEN}",
"Intercom-Version": "2.11",
"Content-Type": "application/json",
"Accept": "application/json"
})
with urllib.request.urlopen(req, timeout=15) as r:
return json.loads(r.read())
def get_contact(contact_id):
try:
return api("GET", f"/contacts/{contact_id}")
except:
return {}
def analyze_conversation(conv_id):
try:
full_conv = api("GET", f"/conversations/{conv_id}")
# Check bot_pause tag
tags = full_conv.get("tags", {}).get("tags", [])
for tag in tags:
if tag.get("name") == BOT_PAUSE_TAG:
return "skip"
# Check bot_skip on contact
contacts = full_conv.get("contacts", {}).get("contacts", [])
for ct in contacts:
contact = get_contact(ct["id"])
if contact.get("custom_attributes", {}).get("bot_skip"):
return "skip"
# Check last message author
parts = full_conv.get("conversation_parts", {}).get("conversation_parts", [])
comments = [p for p in parts if p.get("part_type") == "comment"]
if comments:
last = comments[-1]
if last.get("author", {}).get("type") == "user":
return "needs_action"
else:
return "waiting_on_customer"
else:
source_author = full_conv.get("source", {}).get("author", {})
if source_author.get("type") in ("user", "lead"):
return "needs_action"
else:
return "unknown"
except Exception as e:
return f"error: {e}"
def main():
try:
data = api("POST", "/conversations/search", {
"query": {
"operator": "AND",
"value": [
{"field": "state", "operator": "=", "value": "open"},
{"field": "snoozed_until", "operator": "=", "value": None}
]
},
"pagination": {"per_page": 50}
})
except Exception as e:
print(json.dumps({"error": f"Search failed: {e}"}))
return
conversations = data.get("conversations", [])
needs_action = []
waiting = []
skipped = []
errors = []
for conv in conversations:
cid = conv["id"]
title = conv.get("custom_attributes", {}).get("AI Title", "No title")
assignee = conv.get("admin_assignee_id") or conv.get("assignee", {}).get("id")
if str(assignee) == FOUNDER_ADMIN_ID:
skipped.append({"id": cid, "title": title, "reason": "assigned_to_founder"})
continue
status = analyze_conversation(cid)
if status == "needs_action":
needs_action.append({"id": cid, "title": title,
"waiting_since": conv.get("waiting_since")})
elif status == "waiting_on_customer":
waiting.append({"id": cid, "title": title})
elif status == "skip":
skipped.append({"id": cid, "title": title})
else:
errors.append({"id": cid, "title": title, "error": status})
print(json.dumps({
"total_open": len(conversations),
"needs_action": needs_action,
"waiting_on_customer": waiting,
"skipped": skipped,
"errors": errors,
"summary": {
"needs_action": len(needs_action),
"waiting": len(waiting),
"skipped": len(skipped),
"errors": len(errors)
}
}, indent=2))
if __name__ == "__main__":
main()#!/usr/bin/env python3
"""Close stale conversations where merchant hasn't responded in STALE_HOURS hours.
Criteria:
- Open and NOT snoozed (snooze already expired)
- Last message (comment/assignment/open) was from admin
- That message is older than STALE_HOURS
- NOT assigned to founder
- Contact doesn't have bot_skip=true
"""
import urllib.request, json, sys, time
# ──────────────────────────────────────────────
# CONFIGURE THESE VALUES DURING SETUP
# ──────────────────────────────────────────────
TOKEN = "{INTERCOM_API_TOKEN}"
ADMIN_ID = "{AGENT_ADMIN_ID}"
FOUNDER_ADMIN_ID = "{FOUNDER_ADMIN_ID}"
STALE_THRESHOLD_SECS = 24 * 3600 # 24 hours — adjust as needed
# ──────────────────────────────────────────────
def api(method, path, data=None):
url = f"https://api.intercom.io{path}"
body = json.dumps(data).encode() if data else None
req = urllib.request.Request(url, data=body, method=method, headers={
"Authorization": f"Bearer {TOKEN}",
"Intercom-Version": "2.11",
"Content-Type": "application/json",
"Accept": "application/json"
})
with urllib.request.urlopen(req, timeout=15) as r:
return json.loads(r.read())
def get_contact(contact_id):
try:
return api("GET", f"/contacts/{contact_id}")
except:
return {}
def should_close(conv):
cid = conv["id"]
assignee = conv.get("admin_assignee_id") or conv.get("assignee", {}).get("id")
if str(assignee) == FOUNDER_ADMIN_ID:
return False, "assigned_to_founder"
full_conv = api("GET", f"/conversations/{cid}")
contacts = full_conv.get("contacts", {}).get("contacts", [])
for ct in contacts:
contact = get_contact(ct["id"])
if contact.get("custom_attributes", {}).get("bot_skip"):
return False, "bot_skip"
parts = full_conv.get("conversation_parts", {}).get("conversation_parts", [])
# CRITICAL: Include all message-bearing part types
# "comment" = normal messages, "assignment" = admin replies via assignment,
# "open" = customer reopens with a message
MESSAGE_TYPES = ("comment", "assignment", "open")
all_messages = [p for p in parts if p.get("body")
and p.get("part_type") in MESSAGE_TYPES]
if not all_messages:
return False, "no_messages"
last = all_messages[-1]
if last.get("author", {}).get("type") != "admin":
return False, "last_message_from_user"
last_created = last.get("created_at", 0)
age_secs = time.time() - last_created
if age_secs < STALE_THRESHOLD_SECS:
hours_remaining = (STALE_THRESHOLD_SECS - age_secs) / 3600
return False, f"too_recent ({hours_remaining:.1f}h remaining)"
return True, "eligible"
def main():
dry_run = "--dry-run" in sys.argv
try:
data = api("POST", "/conversations/search", {
"query": {
"operator": "AND",
"value": [
{"field": "state", "operator": "=", "value": "open"},
{"field": "snoozed_until", "operator": "=", "value": None}
]
},
"pagination": {"per_page": 50}
})
except Exception as e:
print(json.dumps({"error": f"Search failed: {e}"}))
return
conversations = data.get("conversations", [])
closed = []
skipped = []
for conv in conversations:
cid = conv["id"]
title = conv.get("custom_attributes", {}).get("AI Title", "No title")
try:
eligible, reason = should_close(conv)
if eligible:
if dry_run:
closed.append({"id": cid, "title": title, "action": "would_close"})
else:
# Add internal note explaining closure
api("POST", f"/conversations/{cid}/reply", {
"message_type": "note",
"type": "admin",
"admin_id": ADMIN_ID,
"body": "Closing — merchant has not responded in over 24 hours. "
"They can reopen by replying anytime."
})
api("POST", f"/conversations/{cid}/parts", {
"message_type": "close",
"type": "admin",
"admin_id": ADMIN_ID,
"body": ""
})
closed.append({"id": cid, "title": title, "action": "closed"})
else:
skipped.append({"id": cid, "title": title, "reason": reason})
except Exception as e:
skipped.append({"id": cid, "title": title, "reason": f"error: {e}"})
print(json.dumps({
"dry_run": dry_run,
"total_checked": len(conversations),
"closed": closed,
"skipped": skipped,
"summary": {"closed": len(closed), "skipped": len(skipped)}
}, indent=2))
if __name__ == "__main__":
main()#!/usr/bin/env python3
"""
Archive conversation state files where the Intercom conversation is closed.
Keeps the workspace clean without losing history.
"""
import urllib.request, json, os, shutil
# ──────────────────────────────────────────────
# CONFIGURE THESE VALUES DURING SETUP
# ──────────────────────────────────────────────
TOKEN = "{INTERCOM_API_TOKEN}"
# ──────────────────────────────────────────────
CONV_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
'conversations')
ARCHIVE_DIR = os.path.join(CONV_DIR, 'archive')
def get_conversation_state(conv_id):
try:
req = urllib.request.Request(
f'https://api.intercom.io/conversations/{conv_id}',
headers={
'Authorization': f'Bearer {TOKEN}',
'Intercom-Version': '2.11',
'Accept': 'application/json'
})
with urllib.request.urlopen(req) as r:
data = json.loads(r.read())
return data.get('state', 'unknown')
except Exception as e:
return f'error:{e}'
def main():
os.makedirs(ARCHIVE_DIR, exist_ok=True)
files = [f for f in os.listdir(CONV_DIR) if f.endswith('.json')]
if not files:
print(json.dumps({"archived": 0, "remaining": 0, "errors": []}))
return
archived, remaining, errors = [], [], []
for f in sorted(files):
conv_id = f.replace('.json', '')
state = get_conversation_state(conv_id)
if state == 'closed':
shutil.move(os.path.join(CONV_DIR, f), os.path.join(ARCHIVE_DIR, f))
archived.append(conv_id)
elif state.startswith('error:'):
errors.append({"conv_id": conv_id, "error": state})
remaining.append(conv_id)
else:
remaining.append(conv_id)
print(json.dumps({
"archived": len(archived), "remaining": len(remaining),
"errors": errors, "archived_ids": archived, "remaining_ids": remaining
}, indent=2))
if __name__ == '__main__':
main()Set up these cron jobs via OpenClaw's cron system. These are the minimum required for a functioning support agent.
This is the core job. It runs poll_inbox.py, and when changes are detected, handles them.
{
"name": "inbox-poll",
"schedule": { "kind": "every", "everyMs": 120000 },
"sessionTarget": "isolated",
"payload": {
"kind": "agentTurn",
"message": "Run `python3 {WORKSPACE}/scripts/poll_inbox.py`.\n\nIf the script exits with no output (no changes), respond with NO_REPLY.\n\nIf the script produces JSON output (changes detected), you MUST handle each change:\n\n⚠️ MANDATORY PRE-FLIGHT:\n1. Read AGENTS.md — full workflow, app detection, common patterns.\n2. Read SOUL.md — voice, identity, boundaries.\n3. Read lessons.md — accumulated mistakes. Every lesson exists because it was violated.\n\n⚠️ CRITICAL RULES:\n- READ THE FULL CONVERSATION FIRST. Fetch the complete conversation and read EVERY part before responding.\n- DUPLICATE CHECK BEFORE EVERY SEND. Re-fetch right before sending. If an admin already responded similarly, DO NOT send.\n- You are {AGENT_NAME} (admin ID {AGENT_ADMIN_ID}). Never reply as anyone else.\n- ALWAYS use reply.py to send replies. Never make raw Intercom API calls.\n- HANDLE ONE CONVERSATION MAXIMUM PER RUN. Pick highest priority, handle fully, stop.\n\nWorkflow:\n1. Fetch the FULL conversation. Read ALL parts.\n2. Check conversations/{CONV_ID}.json for existing state.\n3. Identify the app and understand the full context.\n4. Investigate (check DB, URLs, codebase) before replying.\n5. If it's an escalation case: DO NOT reply. Escalate to {FOUNDER_NAME} via {ALERT_CHANNEL}.\n6. FINAL DUPLICATE CHECK before calling reply.py.\n7. Save/update conversation state.\n8. Run `python3 {WORKSPACE}/scripts/close_stale.py` to handle expired snoozes.\n9. Briefly summarize what you did.",
"timeoutSeconds": 300,
"model": "anthropic/claude-sonnet-4-6"
},
"delivery": { "mode": "none" }
}Why claude-sonnet-4-6 and not Opus? The poller runs every 2 minutes. Most cycles are silent (no changes). Even when there are changes, Sonnet handles support conversations well. Save Opus for complex investigations or manual interventions. This keeps costs manageable.
Watches the poller and catches tickets that slip through.
{
"name": "health-monitor",
"schedule": { "kind": "every", "everyMs": 900000 },
"sessionTarget": "isolated",
"payload": {
"kind": "agentTurn",
"message": "You are {AGENT_NAME}'s self-monitoring health check. Run silently. ONLY alert for CRITICAL issues.\n\n## Check 1: Inbox-poll cron health\nCheck if the inbox-poll cron job has 5+ consecutive errors or is disabled. If so, alert {FOUNDER_NAME}.\n\n## Check 2: Unhandled tickets >15 min\nRun: `python3 {WORKSPACE}/scripts/check_inbox.py`\nOnly flag tickets waiting >15 min AND <60 min. Ignore older ones (already tracked). If a flagged ticket has no state file in conversations/, handle it.\n\n## Output\n- If ALL checks pass: respond with NO_REPLY\n- Only alert on: inbox-poll broken, or genuinely new unhandled tickets with no state file.",
"timeoutSeconds": 300,
"model": "anthropic/claude-sonnet-4-6"
},
"delivery": { "mode": "none" }
}The heartbeat is configured in OpenClaw's agent settings and runs automatically. It handles:
- Self-diagnostic checks
- Active playbook state review
- Stale conversation cleanup
- Any other periodic maintenance
HEARTBEAT.md template:
# Heartbeat Checklist
**Frequency: Every 30 minutes**
⚠️ Ticket handling is done by the inbox-poll cron job, NOT the heartbeat.
Do NOT check inbox, reply to tickets, or run playbook steps during heartbeats.
1. **Self-diagnostic** — Check operational status (API connections, message delivery)
2. **Check active playbook states** — Read all conversations/*.json for ongoing processes.
Report any stuck for 24h+.
3. **Archive stale conversations** — Run `python3 {WORKSPACE}/scripts/cleanup_conversations.py`
4. If nothing needs attention, reply HEARTBEAT_OK.The agent is stateless between runs. Each cron cycle spawns a fresh isolated session. To maintain continuity across cycles, conversation progress is tracked in JSON files.
Save in conversations/{CONV_ID}.json:
{
"conv_id": "123456789",
"shop": "store-name.myshopify.com",
"app": "your-app-name",
"type": "general_support",
"channel": "chat",
"step": 1,
"started": "2026-03-27T09:00:00Z",
"last_action": "2026-03-27T09:00:00Z",
"notes": "Customer asked about feature X. Investigated, found Y."
}For playbook-driven conversations (like a review ask flow), add playbook-specific fields:
{
"type": "daily_sync",
"step": 3,
"review_asked": false,
"sentiment": "positive"
}New ticket → conversations/{ID}.json created (step 1)
→ Agent processes → updates step/notes
→ Ticket closed → cleanup_conversations.py moves to archive/
- Always check for existing state before handling a conversation
- Always update state after any action
- Never delete state files manually — let
cleanup_conversations.pyhandle it - Archive, don't delete — closed conversation states go to
conversations/archive/
Playbooks are step-by-step flows for common scenarios. The agent follows them using conversation state files to track progress.
This is the most common playbook for Shopify apps. The goal is to engage merchants, provide value, and (when appropriate) ask for an App Store review.
# Review Ask Playbook
## When to Use
When a merchant requests a feature upgrade, configuration change, or anything
where you're providing clear value.
## Detect Channel
- **Chat:** Source URL contains `embedded=1`, rapid responses
- **Email:** Slow response times (hours between messages)
## Chat Flow
### Step 1: Acknowledge + Engage
Don't just fulfill the request immediately. Ask a brief, relevant question first.
> "To make sure this works best for you — [relevant question about their use case]?"
### Step 2: Fulfill the Request
After they respond, do the thing they asked for.
> "Done! [Confirm what you did]. Could you verify on your end?"
### Step 3: Gauge Sentiment
Read their response carefully. You need GENUINELY positive sentiment to ask for a review.
**Positive signals (proceed to review ask):**
- Enthusiastic language ("love it!", "great app", "awesome")
- Detailed, engaged responses
- Unprompted praise
**Neutral/ambiguous (skip review ask):**
- Short replies ("ok", "thanks", "sure")
- Just answering without elaboration
- Polite but disengaged
**Rule of thumb:** If you have to convince yourself it's positive, it's not positive enough.
### Step 4: Ask for Review (conditional)
**Only if ALL conditions met:**
1. Merchant responded recently (chat: <5 min)
2. Paid Shopify plan (NOT staff/developer/trial/partner_test)
3. Clearly positive sentiment
> "Glad to hear you're having a good experience! If you have a moment,
> we'd really appreciate a quick review — it helps us a lot as a small team. 🙏
> [REVIEW_LINK]"
**Review link format:**
`https://apps.shopify.com/{APP_SLUG}#modal-show=WriteReviewModal`
## Email Flow
Combine steps to minimize emails. Don't chase if they don't reply.
## State File Fields
- step: 1-4
- channel: "chat" or "email"
- review_asked: boolean
- sentiment: "positive", "neutral", "negative"During setup, ask the founder:
- What scenarios repeat often enough to warrant a playbook?
- What's the desired outcome for each?
- Are there any conditional branches (e.g., different handling for paid vs. free plans)?
Save playbooks in playbooks/{name}.md.
Always escalate these to the founder — never attempt to handle them:
- Shopify App Store audits / compliance emails — anything from
@shopify.com - Partnership / business proposals — not support, needs founder judgment
- Legal / GDPR / privacy requests — requires legal review
- Bug reports that need code fixes — agent can diagnose but not fix code
- Refund requests — financial decisions need founder approval
- Anything you're not confident handling — better safe than sorry
🚨 **Escalation: {store} ({app})**
**Issue:** {one-line summary}
**Severity:** Low / Medium / High
**What I found:** {investigation details}
**What's needed:** {specific action required}
The agent should send escalations via the configured alert channel (Telegram, Slack, Discord, etc.) using OpenClaw's messaging tools. Never reply to the merchant when escalating — leave the conversation open for the founder.
The agent will make mistakes. Without a structured way to learn from them, it will repeat the same mistakes every session (because each session is stateless).
# Lessons Learned
## {Date} — {Short Title}
- **What happened:** {factual description}
- **Why it was wrong:** {root cause analysis}
- **Concrete change:** {what was actually changed — file, config, script, or rule}- Read
lessons.mdat every session startup. Non-negotiable. - A lesson without a concrete change is not a lesson. Every entry must describe what was actually changed (a file, a script, a rule in AGENTS.md, a config value). "I'll be more careful next time" is worthless — the agent is code.
- Append after every mistake. When the founder corrects something, log it immediately.
- Never let lessons live only in session transcripts. If it's worth learning, it goes in
lessons.md.
The cron job prompt includes "Read lessons.md" as a mandatory pre-flight step. Since each cron cycle is an isolated session, the lessons file is the only way institutional knowledge persists.
- Check your own operational status regularly (API connections, script health)
- Detect problems before they impact merchants (stale tickets, broken cron jobs)
- Alert proactively — don't wait to be asked
- Test your own systems — run dry checks even when things seem fine
| Check | Frequency | Alert If |
|---|---|---|
| Inbox-poll cron errors | Every 15 min | 5+ consecutive errors |
| Unhandled tickets | Every 15 min | Waiting >15 min with no state file |
| Intercom API connectivity | Every heartbeat | Connection fails |
| State file cleanup | Every heartbeat | Files for closed conversations accumulating |
If your app has a database, give the agent read-only access. This lets it investigate issues without asking the merchant for information they don't know.
- Create a read-only database user (never give write access unless absolutely necessary)
- Document key tables and useful queries in AGENTS.md
- Include the connection string in your config
-- Store settings for a specific merchant
SELECT * FROM store_settings WHERE shop = '{store}.myshopify.com';
-- Recent activity/sync/operations
SELECT * FROM operations WHERE shop = '{store}.myshopify.com'
ORDER BY created_at DESC LIMIT 5;
-- User identity (who installed the app)
SELECT * FROM users WHERE shop = '{store}.myshopify.com';Never ask a merchant for their store URL. You should already have it from:
- Intercom contact
external_id(typically{store}.myshopify.com) - Contact
custom_attributes.domain - Your app's database
These are real mistakes from a production support agent. Each one cost merchant trust or developer time. Don't repeat them.
What happened: Raw API call used the founder's admin ID instead of the agent's.
Fix: All replies go through reply.py which has the correct admin ID hardcoded. Never make raw API calls for sending messages.
What happened: Scripts only checked part_type == "comment". Admin replies sent via assignment had part_type == "assignment". Customer reopens had part_type == "open". These were invisible to filters.
Fix: Always check all message-bearing part types: ("comment", "assignment", "open"). When filtering Intercom parts, think about ALL ways a message can arrive.
What happened: Reply text contained \n which was rendered literally in the chat, not as newlines.
Fix: Write replies as natural flowing text. If you need paragraph breaks, send multiple separate messages via reply.py or use HTML <br> tags (Intercom renders HTML).
What happened: Founder manually replied to a ticket. Cron job fired, saw the same unanswered question, sent a second similar reply.
Fix: The bot_pause tag system. When manually handling a conversation, tag it first. Polling scripts skip tagged conversations.
What happened: Founder gave feedback about a past reply. Agent sent a "corrected" version, creating a confusing duplicate for the merchant. Fix: Feedback about a sent message = lesson learned, not an action. Log the lesson, don't re-reply.
What happened: Poller saved state (marking a change as "seen") before the handler finished processing. Handler timed out. Change was permanently lost.
Fix: The stale_waiting safety net in poll_inbox.py. Re-emits events that have been waiting >5 minutes unchanged.
What happened: Asked for a review after a merchant gave short, minimal responses. Not negative, but not enthusiastic either. Fix: Explicit sentiment signal lists. Short replies ≠ positive sentiment. If you have to convince yourself it's positive, skip the review ask.
What happened: Agent responded based on partial context, missing critical information in earlier messages. Fix: Always fetch and read the COMPLETE conversation, including all parts (not just comments), before responding.
What happened: Merchant opened two conversations about the same topic. Agent redirected one as a duplicate, but it contained additional context the merchant wanted help with. Fix: Before labeling anything as a duplicate, read the full content of both. If the "duplicate" has ANY new information, address it.
What happened: Merchant explicitly asked for a simple change. Agent ran them through a multi-step engagement flow when they just wanted the thing done. Fix: Playbooks are tools, not mandates. For simple, clear requests, just do the thing.
{workspace}/
├── AGENTS.md # Main agent instructions (apps, workflows, patterns)
├── SOUL.md # Agent identity and voice
├── IDENTITY.md # Email/contact details
├── HEARTBEAT.md # Heartbeat checklist
├── TOOLS.md # Environment-specific notes
├── USER.md # About the founder
├── lessons.md # Accumulated lessons (read at startup)
│
├── scripts/
│ ├── poll_inbox.py # Stateful inbox poller (core engine)
│ ├── reply.py # Send replies safely (duplicate check + snooze)
│ ├── check_inbox.py # Full inbox snapshot (for health checks)
│ ├── close_stale.py # Auto-close unresponsive tickets
│ ├── cleanup_conversations.py # Archive state files for closed tickets
│ └── .inbox_state.json # Poller state (auto-generated, don't edit)
│
├── conversations/ # Active conversation state files
│ ├── {CONV_ID}.json
│ └── archive/ # Archived (closed) conversation states
│
├── playbooks/ # Repeatable workflow guides
│ └── review-ask.md
│
└── memory/ # Daily logs and memory
└── YYYY-MM-DD.md
When the agent receives this file and starts for the first time, it should:
- Ask Phase 1 questions (identity)
- Ask Phase 2 questions (Intercom setup)
- Ask Phase 3 questions (app details)
- Ask Phase 4 questions (workflow preferences)
- Generate AGENTS.md with all collected details
- Generate SOUL.md with identity and voice
- Generate IDENTITY.md with contact details
- Create scripts/ with all 5 scripts (tokens filled in)
- Create conversations/ and conversations/archive/
- Create empty lessons.md
- Generate HEARTBEAT.md
- Test Intercom API — run
check_inbox.pyand verify it returns results - Set up cron jobs — inbox-poll (2 min), health-monitor (15 min)
- Do a dry run —
close_stale.py --dry-runto verify logic - Send a test message — reply to a test conversation to verify
reply.pyworks - Confirm with founder — "Everything is set up. I'm monitoring your inbox."
Q: What if I have more than 50 open conversations? A: The scripts paginate at 50. For higher volumes, add pagination handling to loop through all pages. Most Shopify app support inboxes are well under 50 concurrent open conversations.
Q: Can I use a different helpdesk (not Intercom)? A: The architecture (polling + state files + playbooks) works with any helpdesk that has an API. You'd need to rewrite the scripts for your helpdesk's API format.
Q: How much does this cost to run? A: With Sonnet for the poller (most cycles are silent/no-op) and 2-minute intervals, expect ~$5-15/day depending on ticket volume. The health monitor and heartbeat add minimal cost.
Q: What about attachments and images? A: Intercom API returns attachment URLs in conversation parts. The agent can view images and reference them. Sending attachments in replies requires additional Intercom API work.
Q: Can I run multiple agents for multiple apps? A: Yes. Either one agent with multi-app detection (recommended for <5 apps) or separate OpenClaw agent instances for each app.
This guide was created by an agent that's been running this exact setup in production. Every recommendation comes from real experience, and every "common mistake" was made at least once. Learn from our bruises.