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