Build a CallSphere-Style Outbound Voice Campaign Tool

TL;DR — Outbound vendors charge $0.20–$0.40/min once you add their concurrency tier. Build the same thing self-hosted: a Python dialer worker pool, OpenAI Realtime agents, Twilio Programmable Voice with concurrency tokens, and per-list throttle/DNC enforcement.

What you'll build

A complete outbound voice campaign system: campaigns table, leads ingestion, concurrency-controlled dialer, AI agent per call, real-time results dashboard, SMS follow-up, and DNC/TCPA enforcement.

Prerequisites

1. Twilio account with the right concurrency tier (call Quota API to confirm).
2. OpenAI Realtime + Python 3.11.
3. Postgres for campaigns, leads, results.
4. Reviewed TCPA + state outbound rules with counsel.
5. Active DNC subscription (national + relevant state lists).

Architecture

flowchart TB
  CSV[Lead CSV] --> ING[Ingestion]
  ING --> PG[(Leads)]
  SCHED[Scheduler] --> WK[Worker Pool]
  WK --> TW[Twilio Calls]
  TW --> AG[Realtime Agent]
  AG --> RES[(Results)]
  RES --> DASH[Dashboard]
  DNC[DNC Lists] --> ING

Step 1 — Schema

```sql CREATE TABLE campaigns ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), name text, agent_prompt text, status text, max_concurrent int DEFAULT 25, start_local_hour int DEFAULT 9, end_local_hour int DEFAULT 19); CREATE TABLE leads ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), campaign_id uuid REFERENCES campaigns, phone text, name text, timezone text, state_code text, attempts int DEFAULT 0, next_attempt_at timestamptz, dnc bool DEFAULT false, outcome text); CREATE INDEX ON leads (campaign_id, next_attempt_at) WHERE outcome IS NULL AND NOT dnc; ```

Step 2 — DNC and TCPA gate

```python async def is_dialable(lead) -> bool: if lead.dnc: return False if await dnc.is_listed(lead.phone): return False if not within_local_hours(lead.timezone): return False if state_specific_holiday(lead.state_code): return False return True ```

Step 3 — Worker pool

```python import asyncio

async def dial_loop(campaign_id, max_concurrent=25): sem = asyncio.Semaphore(max_concurrent) while True: leads = await db.next_due_leads(campaign_id, limit=max_concurrent) if not leads: await asyncio.sleep(5); continue tasks = [asyncio.create_task(dial_lead(sem, l)) for l in leads] await asyncio.gather(*tasks)

async def dial_lead(sem, lead): async with sem: if not await is_dialable(lead): await db.mark_skipped(lead.id, "not_dialable"); return try: twilio.calls.create( to=lead.phone, from_=BUSINESS_NUMBER, twiml=f'', machine_detection="DetectMessageEnd", async_amd=True, status_callback=f"https://you/cb?lead={lead.id}", status_callback_event=["completed"]) await db.mark_attempted(lead.id) except TwilioRestException as e: await db.mark_error(lead.id, str(e)) ```

Step 4 — Agent prompt and tools

```python @function_tool async def mark_outcome(lead_id: str, outcome: str, notes: str) -> dict: await db.set_outcome(lead_id, outcome, notes); return {"ok": True}

@function_tool async def request_dnc(lead_id: str) -> dict: await db.set_dnc(lead_id); return {"ok": True}

@function_tool async def book_followup(lead_id: str, when_iso: str, channel: str) -> dict: await db.schedule_followup(lead_id, when_iso, channel); return {"ok": True} ```

```md You are calling on behalf of [Company]. Open with "This is an automated call from [Company]". If the caller asks not to be called, immediately call request_dnc and end politely. Never claim to be human. Always call mark_outcome before goodbye. ```

Step 5 — AMD (answering machine detection)

When Twilio's AMD reports AnsweredBy=machine_end_other, drop a 12-second voicemail and exit; don't waste agent minutes.

```python async def amd_callback(req): if req["AnsweredBy"].startswith("machine"): twilio.calls(req["CallSid"]).update( twiml='https://cdn/voicemail.mp3') ```

Step 6 — Dashboard

Streaming view of in-flight calls + outcomes histogram + per-list pacing. 80 lines of Next.js + a Postgres LISTEN/NOTIFY channel.

Step 7 — Throttle and ramp

Start at 5 concurrent. Watch carrier-block rate. Ramp up to your tier ceiling. Twilio enforces concurrency at the project level — pull your tier with /v1/Limits.

Common pitfalls

• TCPA without consent. Outbound to consumer cells without prior express written consent is a multi-million-dollar mistake.
• AMD false positives. Some carriers misclassify; tune MachineDetectionTimeout.
• Local presence rotation. Use Twilio Verify/Trust Hub to maintain caller-ID reputation.

How CallSphere does this in production

Explore a live demo and compare current plans to find the right fit for your business.

FAQ

TCPA risk? Substantial — get counsel and PEWC before B2C cell outbound.

Twilio concurrency? Tied to trust score + tier; can hit 250+.

AMD accuracy? ~85–92% depending on carrier.

Spanish? Realtime supports Spanish natively; campaign language field.

Cost at 50k attempts/mo? ~$0.07/min × ~30s avg = ~$1,750 + Twilio.

Sources

• Twilio Programmable Voice
• TCPA + AMD
• OpenAI Realtime
• /pricing