{"slug":"cold-sms-engine-python-skill","title":"Cold SMS Engine — Python Sequencer Full Implementation","tags":["sms","python","signal-house","supabase","ghl","tcpa","automation"],"agent_summary":"Complete Python cold SMS sequencer replacing n8n at scale: Supabase schema, TCPA slot windows, Signal House sending, GHL mirroring, inbound classification, and compliance rules.","trigger_phrases":["python sms sequencer","signal house python","cold sms python code","sms slot windows","ghl sms threading","tcpa sms compliance","n8n replacement sms","cold sms schema"],"runnable":true,"markdown":"# Cold SMS Engine\n\nPython-native cold SMS sequencer that replaces n8n at scale. Cron-driven, slot-aware, TCPA-safe, and threads conversations into GHL sub-accounts.\n\n## When To Use\n- Volume > 5K sends/day (n8n chokes on batch logic here)\n- Need deterministic slot windows, not webhook-driven chaos\n- Need git-diffable, testable send logic\n\n## Architecture\n\n```\nSupabase (leads, sms_events, opt_outs)\n ↑↓\nPython Dispatcher (cron every 5 min)\n → Slot check (morning/midday/evening, per lead timezone)\n → Pull due leads\n → Signal House API (send)\n → Mirror to GHL via public API (contact + conversation + opportunity)\n ← Inbound webhook → classify → update lead state\n```\n\n## Tables (Supabase)\n```sql\nleads (\n id uuid pk,\n phone_e164 text unique,\n first_name text,\n business_name text,\n timezone text, -- derived from area code\n status text, -- new | sequencing | replied | booked | opted_out | dead\n touchpoint int default 0,\n next_send_at timestamptz,\n sub_account_id text -- GHL location\n)\n\nsms_events (\n id uuid pk, lead_id uuid, direction text,\n body text, segments int, status text,\n provider_id text, sent_at timestamptz\n)\n\nopt_outs ( phone_e164 text pk, reason text, created_at timestamptz )\n```\n\n## Slot Logic\n```python\nSLOTS = {\n \"morning\": (9, 11),\n \"midday\": (12, 14),\n \"evening\": (16, 19),\n}\n\ndef in_slot(tz: str, now_utc) -> bool:\n local = now_utc.astimezone(ZoneInfo(tz))\n h = local.hour\n return any(lo <= h < hi for lo, hi in SLOTS.values())\n```\n\nCron every 5 min runs the dispatcher. It only sends to leads whose `next_send_at <= now()` AND whose timezone is currently in a slot.\n\n## Touchpoint Sequence\n```python\nSEQUENCE = [\n (\"t0_permission\", timedelta(0)),\n (\"t0_followup\", timedelta(minutes=3)), # Eugene 3-min gem\n (\"t1_value\", timedelta(days=1)),\n (\"t3_soft_cta\", timedelta(days=3)),\n (\"t7_hard_cta\", timedelta(days=7)),\n (\"t14_breakup\", timedelta(days=14)),\n]\n```\n\nOn send, set `next_send_at = now + SEQUENCE[touchpoint+1].delta`.\n\n## Sending (Signal House)\n```python\ndef send(lead, body):\n if lead.phone_e164 in opt_outs: return\n resp = httpx.post(\n \"https://api.signalhouse.io/v1/messages\",\n headers={\"Authorization\": f\"Bearer {SH_TOKEN}\"},\n json={\"to\": lead.phone_e164, \"from\": FROM_NUMBER, \"body\": body},\n )\n resp.raise_for_status()\n log_event(lead, \"outbound\", body, resp.json()[\"id\"])\n mirror_to_ghl(lead, body, \"outbound\")\n```\n\n## GHL Mirroring\nUse GHL public API to keep reps' UI in sync:\n```python\ndef mirror_to_ghl(lead, body, direction):\n contact = upsert_ghl_contact(lead)\n create_ghl_conversation_message(contact[\"id\"], body, direction)\n if lead.status == \"booked\":\n create_ghl_opportunity(contact[\"id\"], lead)\n```\n\n## Inbound Classification\nWebhook from Signal House → FastAPI → classify via a small prompt (Gemma 4 local or Claude Haiku):\n```\nGiven this SMS reply: \"{body}\"\nReturn JSON: {\"intent\": \"positive|negative|stop|neutral|question\", \"confidence\": 0-1}\n```\n\nRouting:\n- `stop` → opt_outs + status=opted_out\n- `positive` → status=replied, create GHL opportunity, notify rep (Slack)\n- `negative` → status=dead, retire\n- `question` → route to Conversation AI agent in GHL\n- `neutral` → keep sequencing\n\n## Compliance Non-Negotiables\n- No sends before 9am / after 8pm local (enforced by slot windows)\n- STOP honored instantly (opt_outs table checked on every send)\n- 3-touch cap after any negative signal\n- Sample messages match the A2P-registered campaign\n\n## Metrics to Track\n- Delivery rate (sent → delivered)\n- Reply rate (delivered → inbound)\n- Positive rate (inbound → intent=positive)\n- Opt-out rate (target: < 2%)\n- Per-number health (rotate underperformers)\n\n## Lineage\nBuilt on Iron Automations livestream, replacing n8n after ~5K sends/day breakpoint.\n","html":"

Cold SMS Engine

\n

Python-native cold SMS sequencer that replaces n8n at scale. Cron-driven, slot-aware, TCPA-safe, and threads conversations into GHL sub-accounts.

\n

When To Use

\n\n

Architecture

\n
Supabase (leads, sms_events, opt_outs)\n  ↑↓\nPython Dispatcher (cron every 5 min)\n  → Slot check (morning/midday/evening, per lead timezone)\n  → Pull due leads\n  → Signal House API (send)\n  → Mirror to GHL via public API (contact + conversation + opportunity)\n  ← Inbound webhook → classify → update lead state\n
\n

Tables (Supabase)

\n
leads (\n  id uuid pk,\n  phone_e164 text unique,\n  first_name text,\n  business_name text,\n  timezone text,       -- derived from area code\n  status text,         -- new | sequencing | replied | booked | opted_out | dead\n  touchpoint int default 0,\n  next_send_at timestamptz,\n  sub_account_id text  -- GHL location\n)\n\nsms_events (\n  id uuid pk, lead_id uuid, direction text,\n  body text, segments int, status text,\n  provider_id text, sent_at timestamptz\n)\n\nopt_outs ( phone_e164 text pk, reason text, created_at timestamptz )\n
\n

Slot Logic

\n
SLOTS = {\n    \"morning\": (9, 11),\n    \"midday\":  (12, 14),\n    \"evening\": (16, 19),\n}\n\ndef in_slot(tz: str, now_utc) -> bool:\n    local = now_utc.astimezone(ZoneInfo(tz))\n    h = local.hour\n    return any(lo <= h < hi for lo, hi in SLOTS.values())\n
\n

Cron every 5 min runs the dispatcher. It only sends to leads whose next_send_at <= now() AND whose timezone is currently in a slot.

\n

Touchpoint Sequence

\n
SEQUENCE = [\n    (\"t0_permission\",  timedelta(0)),\n    (\"t0_followup\",    timedelta(minutes=3)),    # Eugene 3-min gem\n    (\"t1_value\",       timedelta(days=1)),\n    (\"t3_soft_cta\",    timedelta(days=3)),\n    (\"t7_hard_cta\",    timedelta(days=7)),\n    (\"t14_breakup\",    timedelta(days=14)),\n]\n
\n

On send, set next_send_at = now + SEQUENCE[touchpoint+1].delta.

\n

Sending (Signal House)

\n
def send(lead, body):\n    if lead.phone_e164 in opt_outs: return\n    resp = httpx.post(\n        \"https://api.signalhouse.io/v1/messages\",\n        headers={\"Authorization\": f\"Bearer {SH_TOKEN}\"},\n        json={\"to\": lead.phone_e164, \"from\": FROM_NUMBER, \"body\": body},\n    )\n    resp.raise_for_status()\n    log_event(lead, \"outbound\", body, resp.json()[\"id\"])\n    mirror_to_ghl(lead, body, \"outbound\")\n
\n

GHL Mirroring

\n

Use GHL public API to keep reps' UI in sync:

\n
def mirror_to_ghl(lead, body, direction):\n    contact = upsert_ghl_contact(lead)\n    create_ghl_conversation_message(contact[\"id\"], body, direction)\n    if lead.status == \"booked\":\n        create_ghl_opportunity(contact[\"id\"], lead)\n
\n

Inbound Classification

\n

Webhook from Signal House → FastAPI → classify via a small prompt (Gemma 4 local or Claude Haiku):

\n
Given this SMS reply: \"{body}\"\nReturn JSON: {\"intent\": \"positive|negative|stop|neutral|question\", \"confidence\": 0-1}\n
\n

Routing:

\n\n

Compliance Non-Negotiables

\n\n

Metrics to Track

\n\n

Lineage

\n

Built on Iron Automations livestream, replacing n8n after ~5K sends/day breakpoint.

\n"}