# SMS + Call Connector — Build Spec
## Context
Command Sovereignty currently has zero telephony. We need to add SMS messaging and outbound calling as **customer-provided connectors** — the customer connects their own Twilio/Plivo/Bandwidth account (like they connect HubSpot). Our system receives their webhooks and sends outbound via their credentials.
## Requirements
1. SMS sent / delivered / failed (with reason)
2. SMS replies (content + timestamp) + keyword flags (STOP, YES, CALL ME)
3. Opt-outs with auto-suppression (TCPA compliance)
4. Call triggered from SMS reply (CALL ME keyword) — timestamped for speed-to-lead
5. Call disposition (connected, voicemail, no answer, appointment set)
6. Time-to-first-contact from lead creation and from SMS sent
7. One unified, chronological touchpoint timeline per lead (SMS + calls + email)
8. Lead status synced across channels
## Architecture
### 1. New Models (in `app/models.py`)
#### `SmsMessage`
```python
class SmsMessage(db.Model):
__tablename__ = 'sms_messages'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
connector_id = db.Column(db.String(36), nullable=False, index=True) # FK to Connector
lead_id = db.Column(db.String(36), db.ForeignKey('angi_leads.id', ondelete='SET NULL'), nullable=True, index=True)
from_phone = db.Column(db.String(50), nullable=False) # Our business number
to_phone = db.Column(db.String(50), nullable=False) # Lead's phone
direction = db.Column(db.String(10), nullable=False) # 'outbound' | 'inbound'
body = db.Column(db.Text, nullable=False) # Message content
status = db.Column(db.String(20), default='queued') # queued, sent, delivered, failed, rejected
error_code = db.Column(db.String(50), default='') # Provider error code on failure
error_message = db.Column(db.Text, default='') # Human-readable error
provider_message_sid = db.Column(db.String(128), default='') # Twilio/Plivo message SID
keyword_flag = db.Column(db.String(20), default='') # STOP, YES, CALL_ME (for inbound)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint("direction IN ('outbound', 'inbound')", name='ck_sms_direction'),
db.CheckConstraint("status IN ('queued', 'sent', 'delivered', 'failed', 'rejected')", name='ck_sms_status'),
)
```
#### `Call`
```python
class Call(db.Model):
__tablename__ = 'calls'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
connector_id = db.Column(db.String(36), nullable=False, index=True)
lead_id = db.Column(db.String(36), db.ForeignKey('angi_leads.id', ondelete='SET NULL'), nullable=True, index=True)
triggered_sms_id = db.Column(db.String(36), db.ForeignKey('sms_messages.id', ondelete='SET NULL'), nullable=True) # If triggered by CALL ME
from_phone = db.Column(db.String(50), nullable=False)
to_phone = db.Column(db.String(50), nullable=False)
direction = db.Column(db.String(10), nullable=False) # 'outbound'
disposition = db.Column(db.String(20), default='') # connected, voicemail, no_answer, busy, appointment_set, unknown
duration_seconds = db.Column(db.Integer, nullable=True)
recording_url = db.Column(db.String(500), default='') # Provider recording URL
provider_call_sid = db.Column(db.String(128), default='')
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
started_at = db.Column(db.DateTime, nullable=True)
ended_at = db.Column(df.DateTime, nullable=True)
__table_args__ = (
db.CheckConstraint("direction IN ('outbound', 'inbound')", name='ck_call_direction'),
db.CheckConstraint("disposition IN ('', 'connected', 'voicemail', 'no_answer', 'busy', 'appointment_set', 'unknown')", name='ck_call_disposition'),
)
```
#### `LeadTouchpoint`
Unified timeline entry — every SMS, call, email, or status change creates one.
```python
class LeadTouchpoint(db.Model):
__tablename__ = 'lead_touchpoints'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
lead_id = db.Column(db.String(36), db.ForeignKey('angi_leads.id', ondelete='CASCADE'), nullable=False, index=True)
touchpoint_type = db.Column(db.String(20), nullable=False) # sms_sent, sms_received, sms_delivered, sms_failed, call_outbound, call_connected, email_sent, status_change, note_added
direction = db.Column(db.String(10), default='outbound') # outbound | inbound | system
content = db.Column(db.Text, default='') # Summary or message snippet
reference_id = db.Column(db.String(36), default='') # FK to SmsMessage.id, Call.id, etc.
reference_type = db.Column(db.String(20), default='') # sms_message, call, angi_lead_action
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.CheckConstraint("touchpoint_type IN ('sms_sent', 'sms_received', 'sms_delivered', 'sms_failed', 'call_outbound', 'call_connected', 'call_voicemail', 'call_no_answer', 'email_sent', 'status_change', 'note_added')", name='ck_touchpoint_type'),
db.CheckConstraint("direction IN ('outbound', 'inbound', 'system')", name='ck_touchpoint_direction'),
)
```
#### `OptOut`
```python
class OptOut(db.Model):
__tablename__ = 'opt_outs'
id = db.Column(db.String(36), primary_key=True, default=gen_uuid)
company_id = db.Column(db.String(36), db.ForeignKey('companies.id', ondelete='CASCADE'), nullable=False, index=True)
phone = db.Column(db.String(50), nullable=False, index=True)
reason = db.Column(db.String(50), default='stop') # stop, manual, bounced, complaint
source_message_id = db.Column(db.String(36), db.ForeignKey('sms_messages.id', ondelete='SET NULL'), nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
db.UniqueConstraint('company_id', 'phone', name='uq_optout_company_phone'),
db.CheckConstraint("reason IN ('stop', 'manual', 'bounced', 'complaint')", name='ck_optout_reason'),
)
```
### 2. New Connector: `app/connectors/sms.py`
Follows the existing `BaseConnector` pattern. Service name: `sms`.
Config fields (encrypted at rest via `Connector.config` property):
```python
{
"provider": "twilio", # twilio | plivo | bandwidth
"account_sid": "...",
"auth_token": "...", # For Twilio
"from_phone": "+1XXXXXXXXXX", # Business number
# Plivo: auth_id, auth_token, from_phone
# Bandwidth: app_id, app_secret, from_phone
# ...provider-specific fields
}
```
Methods:
- `connect()` — validate credentials (Twilio: list phone numbers, Plivo: get account info)
- `send_sms(to_phone, body, lead_id)` — send outbound SMS, returns message record
- `send_sms_bulk(phone_list, body)` — bulk send (future)
- `make_call(to_phone, from_phone)` — outbound call via Twilio Voice
- `is_opted_out(phone)` — check suppression list
- `sync()` — no-op for SMS (webhook-driven), returns success
- `status()` — check provider account health
Webhook routes (in `app/routes/sms_webhooks.py`):
```
POST /api/webhooks/sms/inbound — inbound SMS message
POST /api/webhooks/sms/delivery — delivery status (sent, delivered, failed)
POST /api/webhooks/calls/events — call status events (initiated, ringing, answered, completed, failed)
```
### 3. Keyword Processing
On inbound SMS (`/api/webhooks/sms/inbound`):
1. Uppercase the body, check for keywords
2. **STOP** → create `OptOut` record → respond with "You have been unsubscribed. Reply YES to re-subscribe."
3. **YES** → remove `OptOut` record → respond with "You have been re-subscribed."
4. **CALL ME** → flag on `SmsMessage` → trigger outbound call via connector → create `Call` record
5. Any other content → store as inbound message, create touchpoint, notify dashboard
### 4. Speed-to-Lead Metrics
On `AngiLead`:
- `response_time_minutes` — already exists (time from receipt to first response)
- Add computed property: `time_to_first_contact()` — min of all `LeadTouchpoint.created_at` where direction='outbound' minus `received_at`
- Add computed property: `time_to_first_call()` — min of `Call.started_at` minus `received_at`
### 5. Lead Status Machine
Extend `AngiLead.status` with SMS/call states. Current states: `new, accepted, responded, contacted, scheduled, won, lost, expired, rejected`
New transitions (triggered automatically):
- SMS sent → status = `sms_sent` (if was `new`/`accepted`)
- SMS reply received → status = `replied`
- CALL ME detected → status = `call_attempted`
- Call connected → status = `contacted`
- Appointment discussed → status = `scheduled` (manual confirmation)
- Opt-out → status = `opted_out` (no more outbound SMS)
### 6. Routes
New blueprint: `app/routes/sms.py` → `sms_bp`
```
GET /api/sms/messages — list SMS messages (filter by lead, date, status)
POST /api/sms/send — send outbound SMS (checks opt-out list first!)
GET /api/sms/messages/<id> — get single message
GET /api/sms/opt-outs — list opt-outs
POST /api/sms/opt-outs — manual opt-out
DELETE /api/sms/opt-outs/<phone> — remove opt-out (e.g., after YES)
GET /api/sms/timeline/<lead_id> — unified chronological timeline for a lead
GET /api/calls — list calls
POST /api/calls — make outbound call
GET /api/calls/<id> — get call details
GET /api/leads/<id>/metrics — speed-to-lead metrics
```
### 7. Frontend
For now, focus on **backend + API**. Frontend will come later. The API endpoints should be complete and testable with curl/Postman.
### 8. Testing
Test file: `tests/test_sms_call.py`
Test cases:
1. Send outbound SMS → creates SmsMessage, LeadTouchpoint, updates lead status
2. Inbound SMS with STOP → creates OptOut, blocks future sends
3. Inbound SMS with YES → removes OptOut
4. Inbound SMS with CALL ME → creates Call record, triggers outbound call
5. Opt-out suppression → attempting to send to opted-out phone returns error
6. Timeline for lead with mixed touchpoints → chronological order
7. Speed-to-lead metrics → correct time calculations
8. Connector registration → 'sms' appears in /api/connectors/available
### 9. Implementation Order
1. **Models** — add to `app/models.py`, run migration
2. **Connector** — `app/connectors/sms.py` (Twilio-first, Plivo-compatible)
3. **Webhook routes** — `app/routes/sms_webhooks.py`
4. **Management routes** — `app/routes/sms.py`
5. **Keyword processor** — `app/utils/sms_keywords.py`
6. **Timeline + metrics** — extend AngiLead routes
7. **Tests** — `tests/test_sms_call.py`
8. **Register connector** — add to `_import_submodules()` in `app/connectors/__init__.py`
### 10. Important Notes
- **SUPPRESSION IS MANDATORY** — every outbound SMS must check `OptOut` first. This is TCPA compliance, not optional.
- **Provider abstraction** — the connector should work with Twilio, Plivo, or Bandwidth. Start with Twilio as the implementation, but structure so others can be added.
- **Webhook security** — Twilio sends `X-Twilio-Signature`, Plivo sends `X-Plivo-Auth-Id`. Validate signatures on webhook endpoints.
- **No schema migration framework** — the project uses raw SQLAlchemy models. Add a migration script like the existing `scripts/migrate_crm_tables.py` pattern.
- **Follow existing patterns** — look at `app/connectors/angi.py` for a connector example, `app/routes/connectors.py` for route patterns, `tests/test_angi_leads.py` for test patterns.