---
title: "Build a CallSphere-Style Outbound Voice Campaign Tool"
description: "Replace expensive outbound SDR tooling with a self-hosted dialer that runs OpenAI Realtime agents at 100 concurrent calls. Full architecture and code."
canonical: https://callsphere.ai/blog/vw3h-build-callsphere-style-outbound-voice-campaign-tool
category: "AI Voice Agents"
tags: ["Outbound", "Campaign", "Dialer", "Multi-agent", "Tutorial"]
author: "CallSphere Team"
published: 2026-05-03T00:00:00.000Z
updated: 2026-05-07T09:59:37.315Z
---

# Build a CallSphere-Style Outbound Voice Campaign Tool

> Replace expensive outbound SDR tooling with a self-hosted dialer that runs OpenAI Realtime agents at 100 concurrent calls. Full architecture and code.

> **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

```mermaid
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}](https://you/cb?lead=%7Blead.id%7D)",
                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](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

CallSphere's outbound stack runs the same shape — campaigns, workers, agents, results — at production scale. OneRoof Property dials at 30+ concurrent across 10 specialists over WebRTC + Pion + NATS. Healthcare uses outbound for appointment reminders (FastAPI :8084, 14 HIPAA tools). Salon's recall outbound emits `GB-YYYYMMDD-###` refs from 4 ElevenLabs agents. 37 agents · 90+ tools · 115+ DB tables · 6 verticals. Pricing $149/$499/$1499 with [14-day trial](/trial). [/affiliate](/affiliate) is 22%.

## 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](https://www.twilio.com/docs/voice)
- [TCPA + AMD](https://www.twilio.com/docs/voice/answering-machine-detection)
- [OpenAI Realtime](https://platform.openai.com/docs/guides/realtime)
- [/pricing](https://callsphere.ai/pricing)

---

Source: https://callsphere.ai/blog/vw3h-build-callsphere-style-outbound-voice-campaign-tool