How to create an AI cold-calling agent with AssemblyAI

CRM / lead list
      │
      ▼
compliance gate ──► blocked (DNC, calling window, policy) ──► CRM
      │ passes
      ▼
Twilio Calls API ──► answering machine? ──► hang up, log no_answer
      │ human
      ▼
Media Stream ──► your server ──► Universal-3.6 Pro Realtime  (8 kHz mu-law in, turns out)
                             ──► LLM Gateway + tools          (script, tool calls)  ──► CRM
                             ──► AssemblyAI Text-to-Speech    (8 kHz mu-law out) ──► prospect
      │
      ▼
Twilio status callbacks ──► disposition reconciliation

pip install -U "assemblyai>=1.8.2" openai fastapi uvicorn "websockets>=14" twilio python-dotenv python-multipart

ASSEMBLYAI_API_KEY=your_assemblyai_key
TWILIO_ACCOUNT_SID=ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWILIO_AUTH_TOKEN=your_twilio_auth_token
TWILIO_FROM_NUMBER=+15551234567
PUBLIC_URL=https://your-subdomain.ngrok-free.app

# server.py
import asyncio
import base64
import json
import os
import re
import string
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
from urllib.parse import urlencode
from zoneinfo import ZoneInfo

import assemblyai as aai
import websockets
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request, Response, WebSocket
from openai import AsyncOpenAI
from twilio.request_validator import RequestValidator
from twilio.rest import Client as TwilioClient

load_dotenv()

# Calling rules your counsel owns. Keep them in config, not in code.
POLICY = json.load(open("calling_policy.json"))
# e.g. {"window_start_hour": 8, "window_end_hour": 21, "excluded_states": [...]}

@dataclass
class Lead:
    id: str
    phone: str            # E.164
    first_name: str
    company: str
    state: str            # two-letter code from your CRM
    timezone: Optional[str]
    record: bool = False

internal_suppression = set()  # load from your CRM; honor_dnc() adds to it live

def is_on_dnc_registry(phone):
    """Check your DNC provider. Raise if it can't answer, so the gate fails closed."""
    raise NotImplementedError("wire this to your DNC scrubbing provider")

def gate(lead):
    """Return (allowed, reason). Fail closed on every unknown."""
    if lead.phone in internal_suppression:
        return False, "internal_dnc"
    try:
        if is_on_dnc_registry(lead.phone):
            return False, "registry_dnc"
    except Exception:
        return False, "dnc_check_unavailable"  # fail closed

    if not lead.timezone:
        return False, "unknown_timezone"  # fail closed
    local = datetime.now(ZoneInfo(lead.timezone))
    if not POLICY["window_start_hour"] <= local.hour < POLICY["window_end_hour"]:
        return False, "outside_calling_window"

    if lead.state in POLICY["excluded_states"]:
        return False, "excluded_state"
    return True, "ok"

{
  "window_start_hour": 8,
  "window_end_hour": 21,
  "excluded_states": []
}

API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
PUBLIC_URL = os.environ["PUBLIC_URL"]  # e.g. https://your-app.ngrok.app
TWILIO_AUTH_TOKEN = os.environ["TWILIO_AUTH_TOKEN"]
twilio = TwilioClient(os.environ["TWILIO_ACCOUNT_SID"], TWILIO_AUTH_TOKEN)

app = FastAPI()
LEADS = {}  # lead id -> Lead; swap for your CRM
CRM = []    # stand-in for CRM writes; one dict per write

def crm_write(kind, **fields):
    CRM.append({"kind": kind, **fields})
    print(json.dumps({"crm": kind, **fields}, default=str))

@app.post("/dial/{lead_id}")
async def dial(lead_id: str):
    lead = LEADS.get(lead_id)
    if not lead:
        raise HTTPException(404, "unknown lead")
    allowed, reason = gate(lead)  # the only path to the dialer
    if not allowed:
        crm_write("blocked", lead_id=lead.id, reason=reason)
        return {"dialed": False, "reason": reason}

    call = await asyncio.to_thread(
        twilio.calls.create,
        to=lead.phone,
        from_=os.environ["TWILIO_FROM_NUMBER"],
        url=f"{PUBLIC_URL}/twilio/answer?{urlencode({'lead_id': lead.id})}",
        machine_detection="Enable",   # Twilio reports human or machine before /answer
        status_callback=f"{PUBLIC_URL}/twilio/status",
        status_callback_event=["initiated", "answered", "completed"],
        record=lead.record,
        time_limit=300,               # a ceiling on any one call, in seconds
    )
    crm_write("dialed", lead_id=lead.id, call_sid=call.sid)
    return {"dialed": True, "call_sid": call.sid}

@app.post("/twilio/answer")
async def answer(request: Request, lead_id: str):
    form = await request.form()
    if form.get("AnsweredBy", "human") not in ("human", "unknown"):
        crm_write("disposition", lead_id=lead_id, disposition="no_answer")
        return Response('<Response><Hangup/></Response>', media_type="application/xml")

    host = PUBLIC_URL.split("://", 1)[1]
    twiml = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
  <Connect>
    <Stream url="wss://{host}/media-stream">
      <Parameter name="lead_id" value="{lead_id}" />
    </Stream>
  </Connect>
</Response>"""
    return Response(twiml, media_type="application/xml")

STT_URL = "wss://streaming.assemblyai.com/v3/ws?" + urlencode({
    "speech_model": "universal-3-6-pro",
    "encoding": "pcm_mulaw",  # Twilio Media Streams: 8 kHz G.711 mu-law
    "sample_rate": 8000,
    "voice_focus": "near-field",
})

TTS_CONFIG = aai.TTSConfig(
    voice_id="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
    language="english",
    encoding="pcm_mulaw",
    sample_rate=8000,  # resampled on the server from the native 24 kHz
)

def free_slots():
    """Two open times from your calendar, ISO 8601 with offset."""
    return ["2026-11-03T15:00-05:00", "2026-11-04T11:00-05:00"]

class Call:
    """One outbound call: one STT socket, one TTS session, one history."""

    def __init__(self, twilio_ws, stt_ws, tts):
        self.twilio_ws, self.stt_ws, self.tts = twilio_ws, stt_ws, tts
        self.stream_sid = self.call_sid = self.lead = None
        self.messages, self.prospect_turns, self.offered_slots = [], [], []
        self.reply_task = None
        self.hang_up_after_reply = False

    def log(self, kind, **fields):
        print(json.dumps({"call": self.call_sid, "kind": kind, **fields}, default=str))

    def begin(self, start):
        self.stream_sid, self.call_sid = start["streamSid"], start["callSid"]
        self.lead = LEADS[start["customParameters"]["lead_id"]]
        self.offered_slots = free_slots()
        recorded = " and that this call is recorded" if self.lead.record else ""
        prompt = SCRIPT.format(company=self.lead.company, first_name=self.lead.first_name,
                               recorded=recorded, slot_a=self.offered_slots[0],
                               slot_b=self.offered_slots[1], **CAMPAIGN)
        opener = (f"Hi, is this {self.lead.first_name}? Quick heads up, I'm an AI assistant "
                  f"calling from {self.lead.company}{recorded.replace('that ', '')}.")
        self.messages = [{"role": "system", "content": prompt},
                         {"role": "assistant", "content": opener}]
        self.reply_task = asyncio.create_task(self.say(opener))

    def agent_speaking(self):
        return self.reply_task is not None and not self.reply_task.done()

    async def say(self, text):
        # Tell the speech model what the agent just asked, so a spelled email or a
        # short "the second one" is transcribed as the answer to that question.
        await self.stt_ws.send(json.dumps({"type": "UpdateConfiguration",
                                           "agent_context": text[-1750:]}))
        # Synthesis runs faster than playback: pace mu-law at 8,000 bytes a second
        # with a small lead, so this task lives exactly as long as the prospect hears it.
        loop = asyncio.get_running_loop()
        start, queued = loop.time(), 0.0
        async for chunk in self.tts.speak(text):  # raw 8 kHz mu-law, no header
            await self.twilio_ws.send_text(json.dumps({
                "event": "media", "streamSid": self.stream_sid,
                "media": {"payload": base64.b64encode(chunk).decode()},
            }))
            queued += len(chunk) / 8000
            lead = queued - (loop.time() - start)
            if lead > 0.2:
                await asyncio.sleep(lead - 0.2)
        await asyncio.sleep(max(0.0, queued - (loop.time() - start)))

    async def respond(self, prospect_text):
        reply = await run_turn(self, prospect_text)
        await self.say(reply)
        if self.hang_up_after_reply:
            await asyncio.to_thread(twilio.calls(self.call_sid).update, status="completed")

    async def barge_in(self):
        """The prospect started talking: stop the reply and empty Twilio's buffer."""
        if self.agent_speaking():
            self.reply_task.cancel()  # stops the LLM call, the tools or the playback
            await self.tts.cancel()   # drops synthesis of the request in flight
            await self.twilio_ws.send_text(json.dumps(
                {"event": "clear", "streamSid": self.stream_sid}))

# "mhm" while the agent talks is listening, not interrupting. "yes" and "no"
# are left out on purpose: on a sales call they are real answers.
BACKCHANNELS = frozenset({"mhm", "mm", "mmhm", "uh", "uhhuh", "huh", "um", "umm",
                          "hmm", "hm", "ah", "oh", "yeah", "yep", "okay", "ok",
                          "right", "alright", "gotcha", "sure"})
MIN_WORDS = 2  # shorter utterances while the agent talks count as filler

def is_backchannel(call, text):
    """True if this turn should not interrupt the agent."""
    if not call.agent_speaking():
        return False  # once the agent is quiet, every word is an answer
    tokens = text.lower().translate(str.maketrans("", "", string.punctuation)).split()
    return len(tokens) < MIN_WORDS or all(t in BACKCHANNELS for t in tokens)

@app.websocket("/media-stream")
async def media_stream(twilio_ws: WebSocket):
    await twilio_ws.accept()
    async with websockets.connect(
        STT_URL, additional_headers={"Authorization": API_KEY}
    ) as stt_ws, aai.AsyncTTSSession(TTS_CONFIG, api_key=API_KEY) as tts:
        call = Call(twilio_ws, stt_ws, tts)

        async def from_twilio():
            async for raw in twilio_ws.iter_text():
                msg = json.loads(raw)
                if msg["event"] == "start":
                    call.begin(msg["start"])
                    call.log("start", lead=call.lead.id, tts_session=tts.begin.id)
                elif msg["event"] == "media":
                    await stt_ws.send(base64.b64decode(msg["media"]["payload"]))
                elif msg["event"] == "stop":
                    await stt_ws.send(json.dumps({"type": "Terminate"}))
                    break

        async def from_assemblyai():
            prospect_speaking = False
            async for raw in stt_ws:
                msg = json.loads(raw)
                if msg.get("type") != "Turn":
                    continue
                transcript = msg.get("transcript", "").strip()
                end_of_turn = msg.get("end_of_turn", False)
                if transcript and is_backchannel(call, transcript):
                    prospect_speaking = prospect_speaking and not end_of_turn
                    continue
                if transcript and not prospect_speaking:
                    prospect_speaking = True  # first words of a real turn
                    await call.barge_in()
                if end_of_turn:
                    prospect_speaking = False
                    if transcript:
                        call.prospect_turns.append(transcript)
                        call.log("prospect", text=transcript)
                        call.reply_task = asyncio.create_task(call.respond(transcript))

        async def keep_tts_alive():
            while True:
                await asyncio.sleep(60)
                await tts.keep_alive()  # the session closes after 2 idle minutes

        keepalive = asyncio.create_task(keep_tts_alive())
        listener = asyncio.create_task(from_assemblyai())
        try:
            await from_twilio()
        finally:
            for task in (keepalive, listener, call.reply_task):
                if task:
                    task.cancel()

SCRIPT = """You are a scheduling assistant calling on behalf of {company}.
You have already said you are an AI assistant{recorded}.

OPENER (under 15 seconds):
Confirm you're speaking to {first_name}. State why you're calling in one
sentence. Ask for permission to continue.

DISCOVERY (maximum 2 questions, then stop):
1. Does {team_function} sit with you or someone else?
2. How are you handling {problem_area} today?

PITCH (two sentences, no more):
{company} provides {one_line_description}. Teams use it to {primary_outcome}.

CALL TO ACTION:
Offer exactly these two times: {slot_a} or {slot_b}. Never offer an open-ended
"sometime next week." Read the email address back and wait for a yes before
you call book_meeting.

OBJECTIONS: map, don't argue:
"Not interested"       -> thank them, call log_disposition, say goodbye.
"Send me an email"     -> confirm the address, call log_disposition, say goodbye.
"Not the right person" -> ask who is, call log_disposition, say goodbye.
"Bad time"             -> offer a callback, call mark_callback, say goodbye.
After book_meeting succeeds, confirm the time, call log_disposition with
"booked", and say goodbye.

STOP IMMEDIATELY:
If the person asks not to be called again, in any wording, call honor_dnc
before saying anything else. Then confirm and say goodbye.

NEVER state a price, a customer name, a statistic, or a commitment that did
not come from a tool result in this conversation. If you do not have the
value, say you will follow up with it.

This is a phone call: one or two short sentences per reply, no lists, no markdown."""

CAMPAIGN = {  # one campaign's fill-ins
    "team_function": "revenue operations",
    "problem_area": "call QA",
    "one_line_description": "automated QA scoring for sales calls",
    "primary_outcome": "review every call instead of a sample",
}

DISPOSITIONS = ["booked", "not_now", "not_interested", "wrong_person", "send_email", "dnc"]

def fn(name, description, properties, required):
    return {"type": "function", "function": {
        "name": name, "description": description,
        "parameters": {"type": "object", "properties": properties, "required": required},
    }}

TOOLS = [
    fn("book_meeting",
       "Book a meeting. Call this only after the prospect has agreed to one of the "
       "two offered times and confirmed their email. Do not call it to check availability.",
       {"email": {"type": "string", "format": "email",
                  "description": "Prospect's work email, as they spelled it out."},
        "starts_at": {"type": "string", "pattern": r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}",
                      "description": "One of the two offered times, exactly as given."}},
       ["email", "starts_at"]),
    fn("log_disposition", "Record how the call ended. Call it once, at the end.",
       {"disposition": {"type": "string", "enum": DISPOSITIONS},
        "note": {"type": "string", "description": "One sentence, facts the prospect said."}},
       ["disposition"]),
    fn("honor_dnc", "The prospect asked not to be called again, in any wording.",
       {}, []),
    fn("mark_callback", "The prospect asked to be called back at a better time.",
       {"when": {"type": "string", "description": "When they asked for, in their words."}},
       ["when"]),
]

EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[a-z]{2,}$", re.I)

def heard(call, value):
    """True if the prospect said this value: 'j dot smith at acme dot co' or 'j.smith@acme.co'."""
    spoken = re.sub(r"\b(at|dot)\b", "", " ".join(call.prospect_turns).lower())
    norm = lambda s: re.sub(r"[^a-z0-9]", "", s.lower())
    return bool(norm(value)) and norm(value) in norm(spoken)

def run_tool(call, name, args):
    """Run one tool call. Every argument is checked before anything is written."""
    lead = call.lead
    if name == "book_meeting":
        email, starts_at = args.get("email", ""), args.get("starts_at", "")
        if not EMAIL_RE.match(email) or not heard(call, email):
            return {"error": "email_not_confirmed",
                    "message": "Ask the prospect to spell the email again."}
        if starts_at not in call.offered_slots:
            return {"error": "slot_not_offered",
                    "message": f"Only these times are open: {', '.join(call.offered_slots)}."}
        crm_write("meeting", lead_id=lead.id, email=email, starts_at=starts_at)
        return {"status": "booked", "starts_at": starts_at}

    if name == "log_disposition":
        if args.get("disposition") not in DISPOSITIONS:
            return {"error": "invalid_disposition", "allowed": DISPOSITIONS}
        crm_write("disposition", lead_id=lead.id, call_sid=call.call_sid, **args)
        call.hang_up_after_reply = True
        return {"status": "logged"}

    if name == "honor_dnc":
        internal_suppression.add(lead.phone)  # takes effect before the next dial
        crm_write("disposition", lead_id=lead.id, call_sid=call.call_sid, disposition="dnc")
        call.hang_up_after_reply = True
        return {"status": "suppressed"}

    if name == "mark_callback":
        crm_write("callback", lead_id=lead.id, when=args.get("when", ""))
        crm_write("disposition", lead_id=lead.id, call_sid=call.call_sid, disposition="not_now")
        call.hang_up_after_reply = True
        return {"status": "callback_scheduled"}

    return {"error": "unknown_tool"}

llm = AsyncOpenAI(base_url="https://llm-gateway.assemblyai.com/v1", api_key=API_KEY)
LLM_MODEL = "claude-haiku-4-5-20251001"  # any model on Available models that supports tools
MAX_TOOL_ROUNDS = 4

async def run_turn(call, prospect_text):
    """One prospect turn: call the LLM with tools until it answers in plain text."""
    call.messages.append({"role": "user", "content": prospect_text})
    for _ in range(MAX_TOOL_ROUNDS):
        response = await llm.chat.completions.create(
            model=LLM_MODEL, messages=call.messages, tools=TOOLS, max_tokens=300,
        )
        # Some models return text and tool calls as separate choices: read them all.
        text = " ".join(c.message.content for c in response.choices if c.message.content)
        tool_calls = [tc for c in response.choices for tc in (c.message.tool_calls or [])]
        if not tool_calls:
            call.messages.append({"role": "assistant", "content": text})
            return text

        call.messages.append({"role": "assistant", "content": text or None,
                              "tool_calls": [tc.model_dump() for tc in tool_calls]})
        filler = asyncio.create_task(call.say(text)) if text else None  # keep the line warm
        for tc in tool_calls:
            args = json.loads(tc.function.arguments or "{}")
            result = run_tool(call, tc.function.name, args)
            call.log("tool", name=tc.function.name, args=args, result=result)
            call.messages.append({"role": "tool", "tool_call_id": tc.id,
                                  "content": json.dumps(result)})
        if filler:
            await filler
    return "Sorry, I'll have someone follow up by email. Thanks for your time."

validator = RequestValidator(TWILIO_AUTH_TOKEN)
seen = set()  # use your database in production

@app.post("/twilio/status")
async def status(request: Request):
    form = dict(await request.form())
    signature = request.headers.get("X-Twilio-Signature", "")
    if not validator.validate(f"{PUBLIC_URL}/twilio/status", form, signature):
        raise HTTPException(403, "bad signature")

    key = (form.get("CallSid"), form.get("CallStatus"))
    if key in seen:  # Twilio retries: make the write idempotent
        return Response(status_code=204)
    seen.add(key)

    crm_write("call_status", call_sid=form.get("CallSid"), status=form.get("CallStatus"),
              duration_s=form.get("CallDuration"), answered_by=form.get("AnsweredBy"))
    return Response(status_code=204)

uvicorn server:app --port 8000
ngrok http 8000