Voice AI agents: Vapi, Retell, Bland.ai, real phone calls

Customer speaks
  → STT (speech recognition): hundreds of milliseconds
  → LLM (generating the answer): hundreds of milliseconds
  → TTS (voice synthesis): hundreds of milliseconds
  → Customer hears the answer
  
Total: you need to fit into roughly one second; the faster, the better

VAPI_KEY=your-vapi-key-here
VAPI_PHONE_NUMBER_ID=your-phone-number-id
ANTHROPIC_API_KEY=sk-ant-your-key-here

# voice_agent.py
import requests
import os
from dotenv import load_dotenv

load_dotenv()

VAPI_KEY = os.getenv("VAPI_KEY")
HEADERS = {"Authorization": f"Bearer {VAPI_KEY}"}

def create_lead_qualifier() -> str:
    """
    Creates a voice agent for qualifying leads.
    Returns the ID of the created assistant.
    """
    assistant = requests.post(
        "https://api.vapi.ai/assistant",
        headers=HEADERS,
        json={
            "name": "Lead Qualifier EN",
            "model": {
                "provider": "anthropic",
                "model": "claude-sonnet-5-5",   # check that the model is on Vapi's list; current models: the What's current page
                "messages": [{"role": "system", "content": """You are a professional sales rep.
Your job is to qualify an inbound lead in 3-5 minutes.

Ask these four questions during the conversation:
1. What budget is the customer considering?
2. What's the timeline: when do they plan to decide?
3. What exactly are they looking for: specific requirements?
4. Who makes the final decision?

Conversation rules:
- Keep it short and friendly, don't sound scripted
- Don't ask all the questions in a row: weave them into the conversation
- If the customer already answered a question, don't ask it again
- At the end, offer to book a meeting with a manager
- Keep the call under 5 minutes
- If asked whether you're a person, honestly say you're an AI assistant"""}],
                "temperature": 0.7,
                "maxTokens": 150
            },
            "voice": {
                "provider": "11labs",
                "voiceId": "pNInz6obpgDQGcFmaJgB"
            },
            "firstMessage": "Hi! This is the company's AI assistant. You submitted a request on our website. Do you have a couple of minutes to talk?",
            "endCallMessage": "Great, I've got everything I need. One of our managers will get in touch within the hour. Have a great day!",
            "maxDurationSeconds": 300
        }
    ).json()

    assistant_id = assistant["id"]
    print(f"Assistant created: {assistant_id}")
    return assistant_id


def make_call(assistant_id: str, phone_number: str) -> dict:
    """
    Starts a call to a customer.
    phone_number in E.164 format: +15551234567
    Only call people who have given consent (see the section on laws below).
    """
    call = requests.post(
        "https://api.vapi.ai/call",
        headers=HEADERS,
        json={
            "assistantId": assistant_id,
            "phoneNumberId": os.getenv("VAPI_PHONE_NUMBER_ID"),   # the number we call from
            "customer": {
                "number": phone_number,
                "name": "Customer"  # optional, for the logs
            }
        }
    ).json()

    print(f"Call started: {call['id']}")
    print(f"Status: {call['status']}")
    return call


def get_call_transcript(call_id: str) -> str:
    """
    Gets the transcript of a finished call.
    """
    call = requests.get(
        f"https://api.vapi.ai/call/{call_id}",
        headers=HEADERS
    ).json()

    if call.get("artifact", {}).get("transcript"):
        return call["artifact"]["transcript"]
    return ""


# Run it
if __name__ == "__main__":
    assistant_id = create_lead_qualifier()

    # For testing, call your own number
    call = make_call(assistant_id, "+15551234567")
    print(f"\nCall ID for getting the transcript: {call['id']}")

pip install requests python-dotenv

# knowledge_loader.py
from pathlib import Path

def load_knowledge_base(filename: str) -> str:
    """Reads a knowledge base file"""
    path = Path("knowledge") / filename
    if path.exists():
        return path.read_text(encoding="utf-8")
    return ""


def build_system_prompt() -> str:
    """Builds the system prompt with the knowledge base"""

    services = load_knowledge_base("services.md")
    faq = load_knowledge_base("faq.md")
    objections = load_knowledge_base("objections.md")

    return f"""You are a consultant for a real estate agency in Ecuador.

OUR SERVICES AND PRICES:
{services}

FREQUENTLY ASKED QUESTIONS:
{faq}

HOW TO HANDLE OBJECTIONS:
{objections}

RULES:
- Never quote prices higher than the ones listed; always say "starting at X"
- If the question isn't in the FAQ, say "I'll check the details and we'll call you back"
- Always offer to book a meeting or a video consultation
- Speak English unless the customer switches to another language (for example, Spanish)
- Don't read out lists: weave the information into the conversation"""


# Example: create an assistant with the knowledge base
def create_realty_agent() -> str:
    import requests
    import os

    VAPI_KEY = os.getenv("VAPI_KEY")

    assistant = requests.post(
        "https://api.vapi.ai/assistant",
        headers={"Authorization": f"Bearer {VAPI_KEY}"},
        json={
            "name": "Realty Consultant",
            "model": {
                "provider": "anthropic",
                "model": "claude-sonnet-5-5",
                "messages": [{"role": "system", "content": build_system_prompt()}]
            },
            "voice": {
                "provider": "11labs",
                "voiceId": "pNInz6obpgDQGcFmaJgB"
            },
            "firstMessage": "Hi! This is the agency's AI consultant. How can I help?"
        }
    ).json()

    return assistant["id"]

knowledge/
  services.md    — list of services with prices
  faq.md         — 10-15 frequent questions with answers
  objections.md  — typical objections and how to respond

# webhook_handler.py
from flask import Flask, request, jsonify
import anthropic
import json
import os
from dotenv import load_dotenv

load_dotenv()

app = Flask(__name__)
claude = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))


def analyze_call(transcript: str) -> dict:
    """
    Claude analyzes the transcript and extracts structured data.
    """
    response = "".join(b.text for b in claude.messages.create(
        model="claude-sonnet-5-5",   # current models: the What's current page
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Analyze this conversation with a customer and return JSON.

TRANSCRIPT:
{transcript}

Return ONLY valid JSON with no explanation:
{{
  "qualified": true or false,
  "budget": "amount or range, or 'not specified'",
  "timeline": "when they plan to decide",
  "requirements": "what exactly they're looking for (2-3 sentences)",
  "decision_maker": true or false,
  "next_action": "what to do next (for example: have a manager call, send a presentation, remove from the funnel)",
  "sentiment": "positive, neutral or negative",
  "summary": "2-3 sentences about the conversation"
}}"""
        }]
    ).content if b.type == "text")

    # Strip markdown if the model added it
    if "```json" in response:
        response = response.split("```json")[1].split("```")[0]
    elif "```" in response:
        response = response.split("```")[1].split("```")[0]

    try:
        return json.loads(response.strip())
    except json.JSONDecodeError:
        return {
            "qualified": False,
            "summary": "Couldn't analyze the transcript",
            "raw_response": response
        }


def save_to_crm(analysis: dict, customer: dict, call_id: str):
    """
    Saves the result to your system.
    Replace this with your real CRM: Notion, Airtable, Google Sheets, HubSpot.
    """
    record = {
        "call_id": call_id,
        "customer_phone": customer.get("number", "unknown"),
        "customer_name": customer.get("name", "unknown"),
        **analysis
    }

    # The simple version: save to a JSON file
    import datetime
    filename = f"calls/call-{call_id}-{datetime.date.today()}.json"
    os.makedirs("calls", exist_ok=True)

    with open(filename, "w", encoding="utf-8") as f:
        json.dump(record, f, ensure_ascii=False, indent=2)

    print(f"Saved: {filename}")

    # If qualified, notify a manager
    if analysis.get("qualified"):
        notify_manager(record)


def notify_manager(record: dict):
    """
    Notifies a manager about a qualified lead.
    Replace with a real channel: Slack, SMS, email.
    """
    # Example: print to the console (in production, a Slack webhook or email)
    print(f"""
=== QUALIFIED LEAD ===
Phone: {record['customer_phone']}
Budget: {record.get('budget', 'not specified')}
Timeline: {record.get('timeline', 'not specified')}
Next step: {record.get('next_action', '')}
Summary: {record.get('summary', '')}
==============================
    """)


@app.post("/vapi-webhook")
def handle_vapi_event():
    """The main handler for events from Vapi"""
    # Vapi puts the event inside the "message" key
    message = (request.json or {}).get("message", {})
    event_type = message.get("type")

    if event_type == "end-of-call-report":
        call = message.get("call", {})
        call_id = call.get("id", "unknown")
        customer = call.get("customer", {})
        transcript = message.get("artifact", {}).get("transcript", "")

        print(f"Call ended: {call_id}")
        print(f"Reason it ended: {message.get('endedReason', 'unknown')}")

        if transcript:
            print("Analyzing the transcript...")
            analysis = analyze_call(transcript)
            save_to_crm(analysis, customer, call_id)
        else:
            print("Empty transcript: the call didn't connect or the customer didn't answer")

    elif event_type == "status-update":
        call_id = message.get("call", {}).get("id", "unknown")
        print(f"Call {call_id} status: {message.get('status', '?')}")

    return jsonify({"status": "ok"})


if __name__ == "__main__":
    # In production, protect the webhook with a secret in the header and don't enable debug
    app.run(port=5000, debug=False)

pip install flask anthropic requests python-dotenv

# Start the webhook server
python webhook_handler.py

# In another terminal, forward the port through ngrok for testing
ngrok http 5000
# ngrok will give you a public URL like https://abc123.ngrok.io

# Update the assistant so it sends events to your webhook
import requests

VAPI_KEY = "your-key"
ASSISTANT_ID = "your-assistant-id"
WEBHOOK_URL = "https://abc123.ngrok.io/vapi-webhook"

requests.patch(
    f"https://api.vapi.ai/assistant/{ASSISTANT_ID}",
    headers={"Authorization": f"Bearer {VAPI_KEY}"},
    json={"server": {"url": WEBHOOK_URL}}   # older examples use a serverUrl field
)

# auto_caller.py
"""
A script for automated calling.
Connect it to a webhook from your CRM or contact form.
"""
import requests
import os
from dotenv import load_dotenv

load_dotenv()

VAPI_KEY = os.getenv("VAPI_KEY")
ASSISTANT_ID = os.getenv("VAPI_ASSISTANT_ID")
PHONE_NUMBER_ID = os.getenv("VAPI_PHONE_NUMBER_ID")


def call_new_lead(phone: str, name: str = "", source: str = "") -> str:
    """
    Calls a new lead right after they submit a request.
    Returns a call_id for tracking.
    """
    call = requests.post(
        "https://api.vapi.ai/call",
        headers={"Authorization": f"Bearer {VAPI_KEY}"},
        json={
            "assistantId": ASSISTANT_ID,
            "phoneNumberId": PHONE_NUMBER_ID,
            "customer": {
                "number": phone,
                "name": name
            },
            # Pass context to the agent through overrides
            "assistantOverrides": {
                "variableValues": {
                    "lead_source": source,
                    "lead_name": name
                }
            }
        }
    ).json()

    return call.get("id", "")


def batch_call(leads: list) -> list:
    """
    Calls a list of leads.
    leads = [{"phone": "+1...", "name": "...", "source": "..."}, ...]
    """
    results = []

    for lead in leads:
        print(f"Calling: {lead['name']} ({lead['phone']})")
        call_id = call_new_lead(
            phone=lead["phone"],
            name=lead.get("name", ""),
            source=lead.get("source", "")
        )
        results.append({
            "lead": lead,
            "call_id": call_id
        })

        # Pause between calls so we don't overload anything
        import time
        time.sleep(2)

    return results


# Example: calling leads after exporting them from the CRM
if __name__ == "__main__":
    today_leads = [
        {"phone": "+15551111111", "name": "John Smith", "source": "Website"},
        {"phone": "+15552222222", "name": "Maria Garcia", "source": "Ads"},
    ]

    results = batch_call(today_leads)
    print(f"\nCalls started: {len(results)}")
    for r in results:
        print(f"  {r['lead']['name']}: call_id = {r['call_id']}")