You build an AI calling agent and auto dialer as a reliable pipeline, not a single LLM prompt: a lead queue, a preflight check, a dialer worker, an explicit call-state machine, a speech path, CRM writeback, and operator controls. This article walks through a Python and FastAPI architecture for that stack—the same family of ideas behind tools like Routelix—with code outlines you can adapt. It assumes permission-based, compliant outreach only—no tricks for spam dialing.
Freight and sales desks do not need “an AI that talks.” They need:
Hosted dialer SaaS can cover some of that. Teams that want local control (Windows operator machine, their own data, their own voice/telephony path) often ask for a self-hosted console instead.
Think in six boxes:
[Lead / Carrier CRM]
│
▼
[FastAPI Operator API + Dashboard]
│
├──► [Dialer Worker] ──► telephony or browser-based calling surface
│
├──► [Call-State Machine]
│
├──► [STT → LLM → TTS pipeline] (only while connected / permitted)
│
└──► [Artifacts: recordings, transcripts, outcomes]
FastAPI gives you typed request/response models, OpenAPI docs for your own operators, and easy WebSocket or SSE hooks for live run status.
from enum import Enum
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
app = FastAPI(title="Operator Dialer API")
class CallState(str, Enum):
idle = "idle"
preflight = "preflight"
dialing = "dialing"
ringing = "ringing"
connected = "connected"
voicemail = "voicemail"
completed = "completed"
failed = "failed"
class StartRunRequest(BaseModel):
campaign_id: str
dry_run: bool = False
max_calls: int = Field(default=50, ge=1, le=500)
class StartRunResponse(BaseModel):
run_id: str
state: CallState
@app.post("/runs", response_model=StartRunResponse)
def start_run(body: StartRunRequest) -> StartRunResponse:
if worker_is_busy():
raise HTTPException(409, "A run is already active on this workstation")
run_id = enqueue_run(body.campaign_id, body.dry_run, body.max_calls)
return StartRunResponse(run_id=run_id, state=CallState.preflight)
Keep one active run per workstation (or per soft user) unless you have a real multi-line design. Heartbeats help enforce that.
UI bugs and double-dials come from implicit state. Make transitions explicit.
ALLOWED = {
CallState.idle: {CallState.preflight},
CallState.preflight: {CallState.dialing, CallState.failed, CallState.idle},
CallState.dialing: {CallState.ringing, CallState.failed},
CallState.ringing: {CallState.connected, CallState.voicemail, CallState.failed},
CallState.connected: {CallState.completed, CallState.failed},
CallState.voicemail: {CallState.completed, CallState.failed},
CallState.completed: {CallState.idle, CallState.dialing},
CallState.failed: {CallState.idle, CallState.dialing},
}
def transition(current: CallState, new: CallState) -> CallState:
if new not in ALLOWED[current]:
raise ValueError(f"illegal transition {current} → {new}")
persist_state(new)
broadcast_state(new) # websocket to dashboard
return new
Preflight checks belong before dialing: audio devices present, dial surface reachable, campaign not empty, compliance flags acknowledged.
There are two common patterns:
Pattern 2 (for example, driving a web calling UI with Selenium or Playwright) can be valid for authorized operator workflows, but it is not the same thing as an official calling API. Do not market it as one. Session cookies, UI selectors, and ToS all matter. Build kill switches and dry-run modes.
class DialerPort(Protocol):
def dial(self, e164_number: str) -> None: ...
def hangup(self) -> None: ...
def detect_connected(self) -> bool: ...
class DryRunDialer:
def dial(self, e164_number: str) -> None:
log.info("dry-run dial %s", e164_number)
def hangup(self) -> None:
return
def detect_connected(self) -> bool:
return False
Code against a port/interface so you can swap dry-run, PBX, or browser automation without rewriting the CRM.
A typical connected-call loop:
def on_connected(call_id: str, lead: dict) -> None:
transition(CallState.ringing, CallState.connected)
transcript_parts: list[str] = []
while is_connected(call_id):
partial = stt.read_partial()
if partial:
transcript_parts.append(partial)
reply = llm.next_utterance(
lead=lead,
transcript="\n".join(transcript_parts),
playbook=lead.get("playbook", "default"),
)
if reply.should_speak:
tts.speak(reply.text)
if reply.should_hangup:
dialer.hangup()
break
save_transcript(call_id, "\n".join(transcript_parts))
transition(CallState.connected, CallState.completed)
Practical tips:
The CRM does not need to be Salesforce on day one. A solid v1 is Postgres tables: accounts, contacts, calls, outcomes, tasks.
def writeback_call(session, call: dict) -> None:
session.execute(
"""
INSERT INTO calls (id, contact_id, started_at, ended_at, state, outcome, transcript_path)
VALUES (:id, :contact_id, :started_at, :ended_at, :state, :outcome, :transcript_path)
""",
call,
)
if call["outcome"] == "callback":
session.execute(
"""
INSERT INTO tasks (contact_id, due_at, kind, notes)
VALUES (:contact_id, :due_at, 'callback', :notes)
""",
{
"contact_id": call["contact_id"],
"due_at": call["callback_at"],
"notes": call.get("notes", ""),
},
)
session.commit()
Show the operator a carrier/lead profile with last outcome, AI summary of prior notes, and the next task. Sample/demo data in screenshots should stay labeled as sample.
Outbound calling is regulated differently by country and use case (consent, DNC lists, calling hours, caller ID rules). Engineering checklist:
Q: Do I need an LLM to ship an auto dialer?
A: No. Queue dialing + outcome HUD + CRM writeback is already valuable. Add AI speech only where it improves a permitted workflow.
Q: Should the dial loop live inside FastAPI request handlers?
A: No. Use a worker process or supervised task. HTTP handlers should start/stop/inspect runs, not block on multi-minute calls.
Q: How do I test without calling real people?
A: Dry-run dialers, recorded audio fixtures for STT, and a fake dial surface that emits state transitions. Keep a golden campaign of internal test numbers only where lawful.
Q: What is the first CRM screen worth building?
A: Contact/carrier profile + call history + next task. Fancy pipelines can wait until writeback is trustworthy.
Want a self-hosted AI dialer, voice-agent workflow, or carrier CRM? INDUS Web Agency builds Python/FastAPI operator tools for logistics and outbound teams.