{"slug":"cold-sms-engine","title":"Cold SMS Engine — Python Sequencer with Slot Windows and GHL Threading","tags":["cold-sms","sms-sequencer","signal-house","ghl","tcpa","python","supabase"],"agent_summary":"Python-native cold SMS sequencer replacing n8n at scale. Cron-driven, slot-aware, TCPA-compliant. Supabase stores leads + sms_events + opt_outs. Dispatcher runs every 5 minutes, picks due leads by timezone-aware slot windows (morning/midday/evening), sends via Signal House API, mirrors to GHL sub-accounts as conversations + contacts + opportunities. Inbound webhook classifies replies and updates lead state.","trigger_phrases":["cold SMS engine","SMS sequencer Python","Signal House sender","GHL SMS threading","TCPA compliant SMS","bulk SMS at scale","n8n alternative SMS"],"runnable":true,"markdown":"\n# Cold SMS Engine\n\nPython-native cold SMS sequencer that replaces n8n at scale. Cron-driven, slot-aware, TCPA-compliant, and threads conversations into GHL sub-accounts.\n\n## When to use\n\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- Need TCPA-defensible audit trail\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## Supabase schema\n\n```sql\nCREATE TABLE leads (\n id UUID PRIMARY KEY DEFAULT gen_random_uuid(),\n phone TEXT NOT NULL UNIQUE,\n timezone TEXT DEFAULT 'America/New_York',\n sequence_id TEXT NOT NULL,\n sequence_step INT DEFAULT 0,\n next_send_at TIMESTAMPTZ,\n last_sent_at TIMESTAMPTZ,\n state TEXT DEFAULT 'active', -- active, opted_out, completed, bounced\n metadata JSONB,\n created_at TIMESTAMPTZ DEFAULT now()\n);\n\nCREATE INDEX idx_due_leads ON leads(next_send_at) WHERE state = 'active';\n\nCREATE TABLE sms_events (\n id UUID PRIMARY KEY DEFAULT gen_random_uuid(),\n lead_id UUID REFERENCES leads(id),\n direction TEXT NOT NULL, -- outbound, inbound\n from_number TEXT,\n to_number TEXT,\n body TEXT,\n status TEXT,\n provider_message_id TEXT,\n classification TEXT, -- positive, negative, optout, neutral\n occurred_at TIMESTAMPTZ DEFAULT now()\n);\n\nCREATE INDEX idx_lead_events ON sms_events(lead_id, occurred_at);\n\nCREATE TABLE opt_outs (\n phone TEXT PRIMARY KEY,\n reason TEXT,\n opted_out_at TIMESTAMPTZ DEFAULT now()\n);\n\nCREATE TABLE sequences (\n id TEXT PRIMARY KEY,\n name TEXT,\n steps JSONB -- array of {body_template, delay_hours, slot}\n);\n```\n\n## Slot windows\n\nThree slots per day, in the lead's local timezone:\n\n| Slot | Window | Use case |\n|---|---|---|\n| morning | 9:00 - 11:30 AM | Reach commuters and early decision-makers |\n| midday | 12:30 - 3:00 PM | Reach office workers on lunch break |\n| evening | 5:00 - 7:30 PM | Reach after-work decision-makers |\n\nFederal TCPA quiet hours: 8 AM - 9 PM local. All three slots safely inside. Some states (CA, FL) have stricter windows — narrow to 10 AM - 8 PM for those.\n\n## Dispatcher (Python)\n\n```python\nimport os\nfrom datetime import datetime, timezone\nfrom zoneinfo import ZoneInfo\nimport requests\nfrom supabase import create_client\n\nsb = create_client(os.environ[\"SUPABASE_URL\"], os.environ[\"SUPABASE_SERVICE_ROLE_KEY\"])\nSIGNAL_HOUSE_API = \"https://api.signalhouse.io/v1/sms\"\nSIGNAL_HOUSE_TOKEN = os.environ[\"SIGNAL_HOUSE_TOKEN\"]\nSIGNAL_HOUSE_FROM = os.environ[\"SIGNAL_HOUSE_FROM\"]\n\nSLOTS = {\n \"morning\": ((9, 0), (11, 30)),\n \"midday\": ((12, 30), (15, 0)),\n \"evening\": ((17, 0), (19, 30)),\n}\n\ndef in_slot(now_local, slot_name):\n start, end = SLOTS[slot_name]\n h, m = now_local.hour, now_local.minute\n minutes = h * 60 + m\n start_min = start[0] * 60 + start[1]\n end_min = end[0] * 60 + end[1]\n return start_min <= minutes < end_min\n\ndef get_due_leads(now_utc):\n return sb.table(\"leads\") \\\n .select(\"*\") \\\n .eq(\"state\", \"active\") \\\n .lte(\"next_send_at\", now_utc.isoformat()) \\\n .limit(500) \\\n .execute().data\n\ndef is_opted_out(phone):\n result = sb.table(\"opt_outs\").select(\"phone\").eq(\"phone\", phone).limit(1).execute()\n return len(result.data) > 0\n\ndef send_sms(to_phone, body):\n response = requests.post(\n SIGNAL_HOUSE_API,\n headers={\"Authorization\": f\"Bearer {SIGNAL_HOUSE_TOKEN}\"},\n json={\n \"from\": SIGNAL_HOUSE_FROM,\n \"to\": to_phone,\n \"body\": body,\n },\n timeout=30,\n )\n response.raise_for_status()\n return response.json()\n\ndef dispatch():\n now_utc = datetime.now(timezone.utc)\n leads = get_due_leads(now_utc)\n \n for lead in leads:\n # TCPA safety: check opt-out one more time\n if is_opted_out(lead[\"phone\"]):\n sb.table(\"leads\").update({\"state\": \"opted_out\"}).eq(\"id\", lead[\"id\"]).execute()\n continue\n \n # Get the sequence step\n sequence = sb.table(\"sequences\").select(\"steps\").eq(\"id\", lead[\"sequence_id\"]).single().execute().data\n step_index = lead[\"sequence_step\"]\n if step_index >= len(sequence[\"steps\"]):\n sb.table(\"leads\").update({\"state\": \"completed\"}).eq(\"id\", lead[\"id\"]).execute()\n continue\n \n step = sequence[\"steps\"][step_index]\n \n # Slot check — is now in the slot for this step (lead's local time)?\n tz = ZoneInfo(lead[\"timezone\"])\n now_local = now_utc.astimezone(tz)\n if not in_slot(now_local, step[\"slot\"]):\n continue # skip, will be picked up in a future cron run\n \n # Render template\n body = step[\"body_template\"].format(**(lead.get(\"metadata\") or {}))\n body = ensure_optout_footer(body)\n \n # Send\n try:\n result = send_sms(lead[\"phone\"], body)\n sb.table(\"sms_events\").insert({\n \"lead_id\": lead[\"id\"],\n \"direction\": \"outbound\",\n \"from_number\": SIGNAL_HOUSE_FROM,\n \"to_number\": lead[\"phone\"],\n \"body\": body,\n \"status\": \"sent\",\n \"provider_message_id\": result.get(\"id\"),\n }).execute()\n \n # Advance to next step\n next_step_index = step_index + 1\n if next_step_index >= len(sequence[\"steps\"]):\n sb.table(\"leads\").update({\n \"state\": \"completed\",\n \"last_sent_at\": now_utc.isoformat(),\n }).eq(\"id\", lead[\"id\"]).execute()\n else:\n next_step = sequence[\"steps\"][next_step_index]\n # Compute next send time = now + delay_hours, rounded to next slot\n next_at = compute_next_slot_time(now_utc, tz, next_step[\"slot\"], next_step[\"delay_hours\"])\n sb.table(\"leads\").update({\n \"sequence_step\": next_step_index,\n \"next_send_at\": next_at.isoformat(),\n \"last_sent_at\": now_utc.isoformat(),\n }).eq(\"id\", lead[\"id\"]).execute()\n \n mirror_to_ghl(lead, body)\n \n except Exception as e:\n sb.table(\"sms_events\").insert({\n \"lead_id\": lead[\"id\"],\n \"direction\": \"outbound\",\n \"to_number\": lead[\"phone\"],\n \"body\": body,\n \"status\": \"error\",\n \"metadata\": {\"error\": str(e)},\n }).execute()\n\ndef ensure_optout_footer(body):\n if \"STOP\" in body.upper():\n return body\n return f\"{body} Reply STOP to opt out.\"\n\ndef compute_next_slot_time(now_utc, tz, slot_name, delay_hours):\n # Add delay, then snap to the next occurrence of the slot window\n target = now_utc + timedelta(hours=delay_hours)\n target_local = target.astimezone(tz)\n # If target falls outside the slot, advance to next day's slot start\n if not in_slot(target_local, slot_name):\n # Move to slot start of the next day\n slot_start_h, slot_start_m = SLOTS[slot_name][0]\n target_local = target_local.replace(hour=slot_start_h, minute=slot_start_m, second=0, microsecond=0)\n if target_local < now_utc.astimezone(tz):\n target_local = target_local + timedelta(days=1)\n return target_local.astimezone(timezone.utc)\n\nif __name__ == \"__main__\":\n dispatch()\n```\n\nRun via cron every 5 minutes:\n\n```cron\n*/5 * * * * cd /opt/cold-sms-engine && /usr/bin/python3 dispatcher.py >> /var/log/cold-sms.log 2>&1\n```\n\n## Inbound classifier\n\nWhen Signal House delivers an inbound SMS webhook, classify and act:\n\n```python\n@app.route(\"/sms-inbound\", methods=[\"POST\"])\ndef sms_inbound():\n payload = request.json\n from_phone = payload[\"from\"]\n body = payload[\"body\"].strip()\n body_upper = body.upper()\n \n # Always log\n lead = sb.table(\"leads\").select(\"*\").eq(\"phone\", from_phone).single().execute().data\n \n sb.table(\"sms_events\").insert({\n \"lead_id\": lead[\"id\"] if lead else None,\n \"direction\": \"inbound\",\n \"from_number\": from_phone,\n \"body\": body,\n \"classification\": classify(body_upper),\n }).execute()\n \n # Opt-out keywords\n if body_upper in (\"STOP\", \"STOPALL\", \"UNSUBSCRIBE\", \"CANCEL\", \"END\", \"QUIT\", \"OPTOUT\"):\n sb.table(\"opt_outs\").upsert({\"phone\": from_phone, \"reason\": \"user_request\"}).execute()\n if lead:\n sb.table(\"leads\").update({\"state\": \"opted_out\"}).eq(\"id\", lead[\"id\"]).execute()\n # Confirm opt-out (CTIA requires it)\n send_sms(from_phone, \"You're unsubscribed. No more messages.\")\n return \"OK\"\n \n # HELP keyword\n if body_upper in (\"HELP\", \"INFO\"):\n send_sms(from_phone, f\"Reply STOP to unsubscribe. Contact: {SUPPORT_EMAIL}.\")\n return \"OK\"\n \n # Positive reply → pause sequence, route to GHL\n if classify(body_upper) == \"positive\":\n sb.table(\"leads\").update({\"state\": \"engaged\", \"next_send_at\": None}).eq(\"id\", lead[\"id\"]).execute()\n notify_ghl_sales_team(lead, body)\n \n return \"OK\"\n\ndef classify(body_upper):\n if body_upper in (\"STOP\", \"STOPALL\", \"UNSUBSCRIBE\", \"CANCEL\", \"END\", \"QUIT\"):\n return \"optout\"\n if any(word in body_upper for word in (\"YES\", \"INTERESTED\", \"TELL ME MORE\", \"CALL ME\", \"SURE\")):\n return \"positive\"\n if any(word in body_upper for word in (\"NO\", \"NOT INTERESTED\", \"FUCK OFF\")):\n return \"negative\"\n return \"neutral\"\n```\n\n## GHL mirror\n\nEach outbound and inbound message mirrors to GHL so the sales team sees the conversation:\n\n```python\ndef mirror_to_ghl(lead, body):\n if not lead.get(\"ghl_location_id\"):\n return\n \n # Upsert contact\n contact = requests.post(\n \"https://services.leadconnectorhq.com/contacts/upsert\",\n headers={\n \"Authorization\": f\"Bearer {GHL_TOKEN}\",\n \"Version\": \"2021-07-28\",\n },\n json={\n \"locationId\": lead[\"ghl_location_id\"],\n \"phone\": lead[\"phone\"],\n \"firstName\": (lead.get(\"metadata\") or {}).get(\"first_name\"),\n \"tags\": [lead[\"sequence_id\"], \"cold-sms-engine\"],\n }\n ).json()\n \n contact_id = contact[\"contact\"][\"id\"]\n \n # Log conversation message\n requests.post(\n \"https://services.leadconnectorhq.com/conversations/messages\",\n headers={\n \"Authorization\": f\"Bearer {GHL_TOKEN}\",\n \"Version\": \"2021-04-15\",\n },\n json={\n \"type\": \"SMS\",\n \"contactId\": contact_id,\n \"message\": body,\n \"direction\": \"outbound\",\n }\n )\n```\n\n## TCPA compliance posture\n\nThis is the audit-defensible posture:\n\n1. **Express consent on every lead** — opt-in source recorded in `leads.metadata.consent_source` and `leads.metadata.consent_timestamp`.\n2. **Opt-out honored within seconds** — STOP keyword checked synchronously in the inbound handler.\n3. **STOP confirmation** — CTIA-required, sent immediately.\n4. **Quiet hours respected** — slot windows are well inside 8 AM - 9 PM local.\n5. **State-specific windows** — leads in CA/FL get 10 AM - 8 PM windows (narrower).\n6. **Cool-down between sends** — 24 hours minimum between steps to avoid harassment patterns.\n7. **Opt-out is permanent** — opted-out phones can never re-enter sequences unless explicitly re-opted-in with a fresh consent timestamp.\n8. **Full audit trail** — every event in `sms_events` with timestamps, provider message IDs, classifications.\n\n## Throughput math\n\nSignal House offers per-second rate limits per registered campaign. Single 10DLC Standard - Medium campaign: 5 MPS to T-Mobile. Practical sustained: 4 MPS to stay under the cap.\n\n| Sustained MPS | Daily ceiling |\n|---|---|\n| 4 MPS | ~345K messages/day if running 24/7 |\n| 4 MPS, but 6 active hours | ~86K messages/day |\n| 75 MPS (Vetted tier) | ~6.5M messages/day if running 24/7 |\n\nMost cold-SMS operations run only during slot windows (8 hours/day total), so use the 6-active-hours math.\n\n## Operational signals to monitor\n\nAdd alerts on:\n\n- `leads.state = 'active' AND next_send_at < now() - interval '30 minutes'` — dispatcher dead\n- `sms_events.status = 'error'` rate > 5% in last hour — provider issues\n- `sms_events.classification = 'optout'` rate spike — message content burning lists\n- Daily opt-out rate > 3% — CTIA red flag, content too aggressive\n\n## Migration from n8n\n\nIf migrating from n8n cold-SMS workflows:\n\n1. Export leads to Supabase first (deduplicate against opt-outs).\n2. Run dispatcher in dry-run mode (logs but doesn't send) for 48 hours to verify slot logic.\n3. Compare expected sends against actual n8n history for the same period.\n4. Cut over by disabling n8n workflow and enabling cron.\n5. Monitor first week for delivery and reply patterns.\n\n## Common pitfalls\n\n- **Timezone naive datetimes** — leads in non-server timezones get sent at wrong local time. Always use `ZoneInfo(lead.timezone)`.\n- **Slot snap going backwards in time** — `compute_next_slot_time` must always return a time in the future, never the past.\n- **No opt-out check in inbound handler** — if classifier doesn't catch the keyword, future sends still go out. Hard-code STOP check first.\n- **GHL token expiration** — if not auto-refreshing, mirror silently fails. Add token refresh hook.\n- **Signal House rate-limit errors not retried** — 429 responses should backoff + retry, not be dropped.\n\n## Related patterns\n\n- [SMS best practices](/topic/sms-best-practices) — opt-in, opt-out, segment math\n- [A2P 10DLC campaign registry](/topic/a2p-10dlc-campaign-registry)\n- [A2P 10DLC vetting and trust scoring](/topic/a2p-10dlc-vetting-scoring)\n- [MMS handling](/topic/mms-handling)\n\n## References\n\n- Signal House API documentation\n- GHL Marketplace API — contacts and conversations\n- TCPA 47 USC §227 — autodialer and time-of-day rules\n- CTIA Messaging Principles v1.10\n","html":"

Cold SMS Engine

\n

Python-native cold SMS sequencer that replaces n8n at scale. Cron-driven, slot-aware, TCPA-compliant, 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

Supabase schema

\n
CREATE TABLE leads (\n    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),\n    phone TEXT NOT NULL UNIQUE,\n    timezone TEXT DEFAULT 'America/New_York',\n    sequence_id TEXT NOT NULL,\n    sequence_step INT DEFAULT 0,\n    next_send_at TIMESTAMPTZ,\n    last_sent_at TIMESTAMPTZ,\n    state TEXT DEFAULT 'active',  -- active, opted_out, completed, bounced\n    metadata JSONB,\n    created_at TIMESTAMPTZ DEFAULT now()\n);\n\nCREATE INDEX idx_due_leads ON leads(next_send_at) WHERE state = 'active';\n\nCREATE TABLE sms_events (\n    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),\n    lead_id UUID REFERENCES leads(id),\n    direction TEXT NOT NULL,  -- outbound, inbound\n    from_number TEXT,\n    to_number TEXT,\n    body TEXT,\n    status TEXT,\n    provider_message_id TEXT,\n    classification TEXT,  -- positive, negative, optout, neutral\n    occurred_at TIMESTAMPTZ DEFAULT now()\n);\n\nCREATE INDEX idx_lead_events ON sms_events(lead_id, occurred_at);\n\nCREATE TABLE opt_outs (\n    phone TEXT PRIMARY KEY,\n    reason TEXT,\n    opted_out_at TIMESTAMPTZ DEFAULT now()\n);\n\nCREATE TABLE sequences (\n    id TEXT PRIMARY KEY,\n    name TEXT,\n    steps JSONB  -- array of {body_template, delay_hours, slot}\n);\n
\n

Slot windows

\n

Three slots per day, in the lead's local timezone:

\n

| Slot | Window | Use case |\n|---|---|---|\n| morning | 9:00 - 11:30 AM | Reach commuters and early decision-makers |\n| midday | 12:30 - 3:00 PM | Reach office workers on lunch break |\n| evening | 5:00 - 7:30 PM | Reach after-work decision-makers |

\n

Federal TCPA quiet hours: 8 AM - 9 PM local. All three slots safely inside. Some states (CA, FL) have stricter windows — narrow to 10 AM - 8 PM for those.

\n

Dispatcher (Python)

\n
import os\nfrom datetime import datetime, timezone\nfrom zoneinfo import ZoneInfo\nimport requests\nfrom supabase import create_client\n\nsb = create_client(os.environ[\"SUPABASE_URL\"], os.environ[\"SUPABASE_SERVICE_ROLE_KEY\"])\nSIGNAL_HOUSE_API = \"https://api.signalhouse.io/v1/sms\"\nSIGNAL_HOUSE_TOKEN = os.environ[\"SIGNAL_HOUSE_TOKEN\"]\nSIGNAL_HOUSE_FROM = os.environ[\"SIGNAL_HOUSE_FROM\"]\n\nSLOTS = {\n    \"morning\": ((9, 0), (11, 30)),\n    \"midday\": ((12, 30), (15, 0)),\n    \"evening\": ((17, 0), (19, 30)),\n}\n\ndef in_slot(now_local, slot_name):\n    start, end = SLOTS[slot_name]\n    h, m = now_local.hour, now_local.minute\n    minutes = h * 60 + m\n    start_min = start[0] * 60 + start[1]\n    end_min = end[0] * 60 + end[1]\n    return start_min <= minutes < end_min\n\ndef get_due_leads(now_utc):\n    return sb.table(\"leads\") \\\n        .select(\"*\") \\\n        .eq(\"state\", \"active\") \\\n        .lte(\"next_send_at\", now_utc.isoformat()) \\\n        .limit(500) \\\n        .execute().data\n\ndef is_opted_out(phone):\n    result = sb.table(\"opt_outs\").select(\"phone\").eq(\"phone\", phone).limit(1).execute()\n    return len(result.data) > 0\n\ndef send_sms(to_phone, body):\n    response = requests.post(\n        SIGNAL_HOUSE_API,\n        headers={\"Authorization\": f\"Bearer {SIGNAL_HOUSE_TOKEN}\"},\n        json={\n            \"from\": SIGNAL_HOUSE_FROM,\n            \"to\": to_phone,\n            \"body\": body,\n        },\n        timeout=30,\n    )\n    response.raise_for_status()\n    return response.json()\n\ndef dispatch():\n    now_utc = datetime.now(timezone.utc)\n    leads = get_due_leads(now_utc)\n    \n    for lead in leads:\n        # TCPA safety: check opt-out one more time\n        if is_opted_out(lead[\"phone\"]):\n            sb.table(\"leads\").update({\"state\": \"opted_out\"}).eq(\"id\", lead[\"id\"]).execute()\n            continue\n        \n        # Get the sequence step\n        sequence = sb.table(\"sequences\").select(\"steps\").eq(\"id\", lead[\"sequence_id\"]).single().execute().data\n        step_index = lead[\"sequence_step\"]\n        if step_index >= len(sequence[\"steps\"]):\n            sb.table(\"leads\").update({\"state\": \"completed\"}).eq(\"id\", lead[\"id\"]).execute()\n            continue\n        \n        step = sequence[\"steps\"][step_index]\n        \n        # Slot check — is now in the slot for this step (lead's local time)?\n        tz = ZoneInfo(lead[\"timezone\"])\n        now_local = now_utc.astimezone(tz)\n        if not in_slot(now_local, step[\"slot\"]):\n            continue  # skip, will be picked up in a future cron run\n        \n        # Render template\n        body = step[\"body_template\"].format(**(lead.get(\"metadata\") or {}))\n        body = ensure_optout_footer(body)\n        \n        # Send\n        try:\n            result = send_sms(lead[\"phone\"], body)\n            sb.table(\"sms_events\").insert({\n                \"lead_id\": lead[\"id\"],\n                \"direction\": \"outbound\",\n                \"from_number\": SIGNAL_HOUSE_FROM,\n                \"to_number\": lead[\"phone\"],\n                \"body\": body,\n                \"status\": \"sent\",\n                \"provider_message_id\": result.get(\"id\"),\n            }).execute()\n            \n            # Advance to next step\n            next_step_index = step_index + 1\n            if next_step_index >= len(sequence[\"steps\"]):\n                sb.table(\"leads\").update({\n                    \"state\": \"completed\",\n                    \"last_sent_at\": now_utc.isoformat(),\n                }).eq(\"id\", lead[\"id\"]).execute()\n            else:\n                next_step = sequence[\"steps\"][next_step_index]\n                # Compute next send time = now + delay_hours, rounded to next slot\n                next_at = compute_next_slot_time(now_utc, tz, next_step[\"slot\"], next_step[\"delay_hours\"])\n                sb.table(\"leads\").update({\n                    \"sequence_step\": next_step_index,\n                    \"next_send_at\": next_at.isoformat(),\n                    \"last_sent_at\": now_utc.isoformat(),\n                }).eq(\"id\", lead[\"id\"]).execute()\n            \n            mirror_to_ghl(lead, body)\n        \n        except Exception as e:\n            sb.table(\"sms_events\").insert({\n                \"lead_id\": lead[\"id\"],\n                \"direction\": \"outbound\",\n                \"to_number\": lead[\"phone\"],\n                \"body\": body,\n                \"status\": \"error\",\n                \"metadata\": {\"error\": str(e)},\n            }).execute()\n\ndef ensure_optout_footer(body):\n    if \"STOP\" in body.upper():\n        return body\n    return f\"{body} Reply STOP to opt out.\"\n\ndef compute_next_slot_time(now_utc, tz, slot_name, delay_hours):\n    # Add delay, then snap to the next occurrence of the slot window\n    target = now_utc + timedelta(hours=delay_hours)\n    target_local = target.astimezone(tz)\n    # If target falls outside the slot, advance to next day's slot start\n    if not in_slot(target_local, slot_name):\n        # Move to slot start of the next day\n        slot_start_h, slot_start_m = SLOTS[slot_name][0]\n        target_local = target_local.replace(hour=slot_start_h, minute=slot_start_m, second=0, microsecond=0)\n        if target_local < now_utc.astimezone(tz):\n            target_local = target_local + timedelta(days=1)\n    return target_local.astimezone(timezone.utc)\n\nif __name__ == \"__main__\":\n    dispatch()\n
\n

Run via cron every 5 minutes:

\n
*/5 * * * * cd /opt/cold-sms-engine && /usr/bin/python3 dispatcher.py >> /var/log/cold-sms.log 2>&1\n
\n

Inbound classifier

\n

When Signal House delivers an inbound SMS webhook, classify and act:

\n
@app.route(\"/sms-inbound\", methods=[\"POST\"])\ndef sms_inbound():\n    payload = request.json\n    from_phone = payload[\"from\"]\n    body = payload[\"body\"].strip()\n    body_upper = body.upper()\n    \n    # Always log\n    lead = sb.table(\"leads\").select(\"*\").eq(\"phone\", from_phone).single().execute().data\n    \n    sb.table(\"sms_events\").insert({\n        \"lead_id\": lead[\"id\"] if lead else None,\n        \"direction\": \"inbound\",\n        \"from_number\": from_phone,\n        \"body\": body,\n        \"classification\": classify(body_upper),\n    }).execute()\n    \n    # Opt-out keywords\n    if body_upper in (\"STOP\", \"STOPALL\", \"UNSUBSCRIBE\", \"CANCEL\", \"END\", \"QUIT\", \"OPTOUT\"):\n        sb.table(\"opt_outs\").upsert({\"phone\": from_phone, \"reason\": \"user_request\"}).execute()\n        if lead:\n            sb.table(\"leads\").update({\"state\": \"opted_out\"}).eq(\"id\", lead[\"id\"]).execute()\n        # Confirm opt-out (CTIA requires it)\n        send_sms(from_phone, \"You're unsubscribed. No more messages.\")\n        return \"OK\"\n    \n    # HELP keyword\n    if body_upper in (\"HELP\", \"INFO\"):\n        send_sms(from_phone, f\"Reply STOP to unsubscribe. Contact: {SUPPORT_EMAIL}.\")\n        return \"OK\"\n    \n    # Positive reply → pause sequence, route to GHL\n    if classify(body_upper) == \"positive\":\n        sb.table(\"leads\").update({\"state\": \"engaged\", \"next_send_at\": None}).eq(\"id\", lead[\"id\"]).execute()\n        notify_ghl_sales_team(lead, body)\n    \n    return \"OK\"\n\ndef classify(body_upper):\n    if body_upper in (\"STOP\", \"STOPALL\", \"UNSUBSCRIBE\", \"CANCEL\", \"END\", \"QUIT\"):\n        return \"optout\"\n    if any(word in body_upper for word in (\"YES\", \"INTERESTED\", \"TELL ME MORE\", \"CALL ME\", \"SURE\")):\n        return \"positive\"\n    if any(word in body_upper for word in (\"NO\", \"NOT INTERESTED\", \"FUCK OFF\")):\n        return \"negative\"\n    return \"neutral\"\n
\n

GHL mirror

\n

Each outbound and inbound message mirrors to GHL so the sales team sees the conversation:

\n
def mirror_to_ghl(lead, body):\n    if not lead.get(\"ghl_location_id\"):\n        return\n    \n    # Upsert contact\n    contact = requests.post(\n        \"https://services.leadconnectorhq.com/contacts/upsert\",\n        headers={\n            \"Authorization\": f\"Bearer {GHL_TOKEN}\",\n            \"Version\": \"2021-07-28\",\n        },\n        json={\n            \"locationId\": lead[\"ghl_location_id\"],\n            \"phone\": lead[\"phone\"],\n            \"firstName\": (lead.get(\"metadata\") or {}).get(\"first_name\"),\n            \"tags\": [lead[\"sequence_id\"], \"cold-sms-engine\"],\n        }\n    ).json()\n    \n    contact_id = contact[\"contact\"][\"id\"]\n    \n    # Log conversation message\n    requests.post(\n        \"https://services.leadconnectorhq.com/conversations/messages\",\n        headers={\n            \"Authorization\": f\"Bearer {GHL_TOKEN}\",\n            \"Version\": \"2021-04-15\",\n        },\n        json={\n            \"type\": \"SMS\",\n            \"contactId\": contact_id,\n            \"message\": body,\n            \"direction\": \"outbound\",\n        }\n    )\n
\n

TCPA compliance posture

\n

This is the audit-defensible posture:

\n
    \n
  1. Express consent on every lead — opt-in source recorded in leads.metadata.consent_source and leads.metadata.consent_timestamp.
    2. \n
  2. Opt-out honored within seconds — STOP keyword checked synchronously in the inbound handler.
    3. \n
  3. STOP confirmation — CTIA-required, sent immediately.
    4. \n
  4. Quiet hours respected — slot windows are well inside 8 AM - 9 PM local.
    5. \n
  5. State-specific windows — leads in CA/FL get 10 AM - 8 PM windows (narrower).
    6. \n
  6. Cool-down between sends — 24 hours minimum between steps to avoid harassment patterns.
    7. \n
  7. Opt-out is permanent — opted-out phones can never re-enter sequences unless explicitly re-opted-in with a fresh consent timestamp.
    8. \n
  8. Full audit trail — every event in sms_events with timestamps, provider message IDs, classifications.
    9. \n
\n

Throughput math

\n

Signal House offers per-second rate limits per registered campaign. Single 10DLC Standard - Medium campaign: 5 MPS to T-Mobile. Practical sustained: 4 MPS to stay under the cap.

\n

| Sustained MPS | Daily ceiling |\n|---|---|\n| 4 MPS | ~345K messages/day if running 24/7 |\n| 4 MPS, but 6 active hours | ~86K messages/day |\n| 75 MPS (Vetted tier) | ~6.5M messages/day if running 24/7 |

\n

Most cold-SMS operations run only during slot windows (8 hours/day total), so use the 6-active-hours math.

\n

Operational signals to monitor

\n

Add alerts on:

\n\n

Migration from n8n

\n

If migrating from n8n cold-SMS workflows:

\n
    \n
  1. Export leads to Supabase first (deduplicate against opt-outs).
    2. \n
  2. Run dispatcher in dry-run mode (logs but doesn't send) for 48 hours to verify slot logic.
    3. \n
  3. Compare expected sends against actual n8n history for the same period.
    4. \n
  4. Cut over by disabling n8n workflow and enabling cron.
    5. \n
  5. Monitor first week for delivery and reply patterns.
    6. \n
\n

Common pitfalls

\n\n

Related patterns

\n\n

References

\n\n"}