Cold SMS Engine

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

When To Use

Volume > 5K sends/day (n8n chokes on batch logic here)
Need deterministic slot windows, not webhook-driven chaos
Need git-diffable, testable send logic

Architecture

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

Tables (Supabase)

leads (
  id uuid pk,
  phone_e164 text unique,
  first_name text,
  business_name text,
  timezone text,       -- derived from area code
  status text,         -- new | sequencing | replied | booked | opted_out | dead
  touchpoint int default 0,
  next_send_at timestamptz,
  sub_account_id text  -- GHL location
)

sms_events (
  id uuid pk, lead_id uuid, direction text,
  body text, segments int, status text,
  provider_id text, sent_at timestamptz
)

opt_outs ( phone_e164 text pk, reason text, created_at timestamptz )

Slot Logic

SLOTS = {
    "morning": (9, 11),
    "midday":  (12, 14),
    "evening": (16, 19),
}

def in_slot(tz: str, now_utc) -> bool:
    local = now_utc.astimezone(ZoneInfo(tz))
    h = local.hour
    return any(lo <= h < hi for lo, hi in SLOTS.values())

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.

Touchpoint Sequence

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

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

Sending (Signal House)

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

GHL Mirroring

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

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

Inbound Classification

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

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

Routing:

stop → opt_outs + status=opted_out
positive → status=replied, create GHL opportunity, notify rep (Slack)
negative → status=dead, retire
question → route to Conversation AI agent in GHL
neutral → keep sequencing

Compliance Non-Negotiables

No sends before 9am / after 8pm local (enforced by slot windows)
STOP honored instantly (opt_outs table checked on every send)
3-touch cap after any negative signal
Sample messages match the A2P-registered campaign

Metrics to Track

Delivery rate (sent → delivered)
Reply rate (delivered → inbound)
Positive rate (inbound → intent=positive)
Opt-out rate (target: < 2%)
Per-number health (rotate underperformers)

Lineage

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

Cold SMS Engine — Python Sequencer Full Implementation