Call Support AI: Vapi + Bland.ai, voice calls in customer support

Customer calls → Vapi picks up → STT (speech to text) → LLM (thinks) → TTS (text to speech) → customer hears the answer

import requests
import json

VAPI_API_KEY = "your_vapi_key"

def create_support_agent(
    company_name: str,
    knowledge_base: str,
    transfer_phone: str  # The live agent's number
) -> dict:
    """Creates a voice support agent through Vapi"""

    headers = {
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    }

    # Field and model names in Vapi change: before launching, check the configuration
    # against the docs at docs.vapi.ai (model, voice, transcription, call transfer)
    agent_config = {
        "name": f"{company_name} Support Agent",
        "model": {
            "provider": "anthropic",
            "model": "claude-sonnet-5-5",
            "messages": [{"role": "system", "content": f"""You are a customer support agent for {company_name}.

KNOWLEDGE BASE:
{knowledge_base}

RULES:
1. Be friendly and professional
2. Answer only from the knowledge base; don't make things up
3. If you don't know the answer, say so honestly and offer to connect them with an agent
4. Always check that the customer understood
5. If the customer is angry, show empathy and don't argue

TRANSFER TO AN AGENT (say this word for word):
"Let me connect you with one of our specialists who can help with this. 
Please hold."

Speak naturally. Pause. Don't rush."""}],
            "temperature": 0.3,  # Less variation for support
            # Transferring the call to a live agent: the transferCall tool
            "tools": [{
                "type": "transferCall",
                "destinations": [{
                    "type": "number",
                    "number": transfer_phone,
                    "description": "Transfer the call to an agent if the AI doesn't know the answer or the customer asks for a person",
                    "message": "Connecting you with a specialist"
                }]
            }]
        },
        "voice": {
            "provider": "11labs",  # ElevenLabs
            "voiceId": "voice_ID_from_ElevenLabs"  # pick a voice that sounds natural in your customers' language
        },
        "transcriber": {
            "provider": "deepgram",
            "language": "en"
        },
        "firstMessage": f"Hi! You've reached {company_name} support. How can I help?",
        "endCallMessage": "Thanks for calling! If you have any more questions, give us a call. Goodbye!",
        "maxDurationSeconds": 600  # 10 minutes max
    }

    response = requests.post(
        "https://api.vapi.ai/assistant",
        headers=headers,
        json=agent_config
    )

    return response.json()


def assign_phone_number(assistant_id: str, phone_number: str) -> dict:
    """Links a phone number to the agent"""

    headers = {
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    }

    response = requests.post(
        "https://api.vapi.ai/phone-number",
        headers=headers,
        json={
            "assistantId": assistant_id,
            "number": phone_number,
            "provider": "twilio"  # or vonage, telnyx
        }
    )

    return response.json()


# Example: an agent for an online store
knowledge_base = """
ORDER QUESTIONS:
- Order status: ask for the order number, go to track.example.com/[number]
- Delivery time: local 1-2 days, nationwide 3-5 days, international 7-14 days
- Changing the address: possible until the order ships, not after

PAYMENT QUESTIONS:
- Methods: Visa/Mastercard, PayPal, Apple Pay, Google Pay
- Returns: within 14 days under our return policy
- Receipt: sent to your email automatically

TECHNICAL PROBLEMS:
- The site isn't working: try clearing your browser cache
- The email didn't arrive: check your Spam folder
- Can't log in: reset your password at example.com/reset

NEEDS AN AGENT:
- Complaints about product quality
- Legal questions
- Situations that require compensation
"""

agent = create_support_agent(
    company_name="My Store",
    knowledge_base=knowledge_base,
    transfer_phone="+15555550123"
)

print(f"Agent created: {agent.get('id')}")
print(f"Name: {agent.get('name')}")

import requests

BLAND_API_KEY = "your_bland_key"

def make_batch_calls(
    contacts: list[dict],  # [{"phone": "+1...", "name": "John", ...}]
    call_purpose: str,
    script_template: str
) -> dict:
    """Launches a mass calling campaign through Bland.ai"""

    # Bland's request schema changes: check the fields and endpoint against the docs at docs.bland.ai
    headers = {
        "Authorization": BLAND_API_KEY,
        "Content-Type": "application/json"
    }

    # Build a task for each contact
    call_objects = []
    for contact in contacts:
        # Personalize the script for each person
        personalized_script = script_template.format(**contact)

        call_objects.append({
            "phone_number": contact["phone"],
            "task": personalized_script
        })

    # Launch the batch: shared settings go in global, calls go in call_objects
    response = requests.post(
        "https://api.bland.ai/v2/batches/create",
        headers=headers,
        json={
            "global": {
                "task": script_template,           # fallback script
                "voice": "voice_ID_from_dashboard",  # pick a voice in the Bland dashboard
                "language": "en",                  # see the docs for language codes
                "max_duration": 2,                 # 2 minutes max
                "record": True                     # record the call (see the warning below)
            },
            "call_objects": call_objects,
            "description": call_purpose[:60]
        }
    )

    return response.json()


# Example: appointment reminders
contacts_for_reminder = [
    {"phone": "+15555550111", "name": "John", "date": "tomorrow", "time": "2:00 PM", "service": "a haircut"},
    {"phone": "+15555550122", "name": "Maria", "date": "on Wednesday", "time": "11:30 AM", "service": "a manicure"},
]

reminder_script = """
Hi, {name}! This is Beauty Studio calling.
Just a reminder that you have an appointment for {service} {date} at {time}.
If your plans have changed, please give us a call or send us a text.
See you soon! Goodbye.
"""

result = make_batch_calls(
    contacts=contacts_for_reminder,
    call_purpose="Appointment reminders — 2026-10-15",
    script_template=reminder_script
)

print(f"Batch created: {result.get('batch_id')}")
print(f"Calls scheduled: {len(contacts_for_reminder)}")

import anthropic
from typing import Optional

# A simple RAG system for a voice agent
def create_rag_support_handler(
    documents: list[str],  # List of knowledge base documents
) -> callable:
    """Creates a question handler that searches the knowledge base"""

    client = anthropic.Anthropic()

    # In a real project this would be a vector database (Pinecone, Chroma)
    # To keep it simple, we pass all documents into the context
    knowledge_context = "\n\n---\n\n".join(documents)

    def handle_voice_query(
        user_query: str,
        conversation_history: list[dict],
        customer_info: Optional[dict] = None
    ) -> dict:
        """Handles a request from the voice agent"""

        system_prompt = f"""You are a support agent. Answer ONLY based on the knowledge base.
If the information isn't there, say honestly that you don't know and offer to connect them with an agent.

IMPORTANT FOR VOICE ANSWERS:
- The answer should sound natural when read out loud
- Don't use list markers (•, -, *); they don't sound good
- Write numbers as words: not "14:00" but "at two in the afternoon"
- 3-4 sentences max per answer

KNOWLEDGE BASE:
{knowledge_context}"""

        messages = conversation_history + [{
            "role": "user",
            "content": user_query
        }]

        response = client.messages.create(
            model="claude-sonnet-5-5",
            max_tokens=200,  # Voice answers are short
            system=system_prompt,
            messages=messages
        )

        answer = "".join(b.text for b in response.content if b.type == "text")

        # Detect whether a transfer to an agent is needed
        transfer_signals = [
            "don't know", "can't help", "need a specialist",
            "connect you with an agent", "pass your situation"
        ]
        needs_transfer = any(signal in answer.lower() for signal in transfer_signals)

        return {
            "response_text": answer,
            "needs_human_transfer": needs_transfer,
            "transfer_reason": "The AI couldn't answer" if needs_transfer else None
        }

    return handle_voice_query


# Example usage
faq_documents = [
    """RETURNS
    You can return an item within fourteen days of purchase.
    The item must be in its original packaging with no signs of use.
    We refund the money within three business days to the card you paid with.""",

    """SHIPPING
    Local delivery takes one to two days.
    Nationwide delivery takes three to seven days.
    Shipping is free on orders over fifty dollars.
    You can track your order on the website under My Orders.""",

    """PAYMENT
    We accept Visa, Mastercard and American Express.
    You can also pay with PayPal or Apple Pay.
    Installment plans are available through a partner for three, six or twelve months."""
]

handler = create_rag_support_handler(faq_documents)

# Test conversation
conversation = []
queries = [
    "I want to return something I bought a week ago",
    "what if I don't have the packaging anymore?"
]

for query in queries:
    print(f"\nCustomer: {query}")
    result = handler(query, conversation)
    print(f"AI: {result['response_text']}")
    if result['needs_human_transfer']:
        print("⚠️ TRANSFER TO AN AGENT")

    # Update the conversation history
    conversation.append({"role": "user", "content": query})
    conversation.append({"role": "assistant", "content": result['response_text']})

def analyze_call_transcripts(transcripts: list[str]) -> dict:
    """Analyzes a batch of call transcripts and finds patterns"""

    client = anthropic.Anthropic()

    # Combine the transcripts with separators
    combined = "\n\n=== CALL ===\n\n".join(transcripts[:20])  # First 20

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": f"""Analyze these support call transcripts.

TRANSCRIPTS:
{combined}

I need:
1. **Top 10 questions** asked most often (with percentages)
2. **Top 5 problems** the AI couldn't solve (needed a person)
3. **Average call length** by category
4. **Emotional tone**: how many angry/satisfied/neutral customers
5. **Recommendations**: what to add to the knowledge base so the AI can handle more
6. **Red flags**: what urgently needs fixing in the product/service

Format: a structured report with numbers."""
        }]
    )

    return "".join(b.text for b in response.content if b.type == "text")


# Cost vs ROI
def calculate_roi(
    current_calls_per_month: int,
    avg_call_duration_minutes: float,
    operator_hourly_cost: float,
    ai_cost_per_minute: float,  # the sum of the five per-minute cost items; calculate it for your stack
    ai_deflection_rate: float = 0.75  # an assumption for the example: measure it on your own calls
) -> dict:
    """Calculates the ROI of rolling out voice AI (all numbers in the example are made up)"""

    # Current costs
    total_minutes = current_calls_per_month * avg_call_duration_minutes
    current_cost = (total_minutes / 60) * operator_hourly_cost

    # After AI
    ai_handled = current_calls_per_month * ai_deflection_rate
    human_handled = current_calls_per_month * (1 - ai_deflection_rate)

    ai_cost = ai_handled * avg_call_duration_minutes * ai_cost_per_minute
    human_cost = (human_handled * avg_call_duration_minutes / 60) * operator_hourly_cost

    new_total_cost = ai_cost + human_cost
    savings = current_cost - new_total_cost

    return {
        "current_monthly_cost": round(current_cost, 2),
        "new_monthly_cost": round(new_total_cost, 2),
        "monthly_savings": round(savings, 2),
        "annual_savings": round(savings * 12, 2),
        "ai_handled_calls": int(ai_handled),
        "human_calls_freed": int(ai_handled),
        "roi_percent": round((savings / new_total_cost) * 100, 1)
    }

# Example calculation
roi = calculate_roi(
    current_calls_per_month=2000,
    avg_call_duration_minutes=4.5,
    operator_hourly_cost=15,   # an assumption for the example: plug in the real rate
    ai_cost_per_minute=0.20    # an assumption for the example: plug in the total for your stack
)

print("=== ROI ANALYSIS ===")
for k, v in roi.items():
    print(f"{k}: {v}")

# Go to vapi.ai
# Create an account
# Get an API key in Dashboard → API Keys
# Save it to .env:

echo "VAPI_API_KEY=your_key_here" >> .env

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

load_dotenv()
VAPI_API_KEY = os.environ["VAPI_API_KEY"]

# Create the agent
agent = requests.post(
    "https://api.vapi.ai/assistant",
    headers={"Authorization": f"Bearer {VAPI_API_KEY}"},
    json={
        "name": "Test support agent",
        "model": {
            "provider": "anthropic",
            "model": "claude-sonnet-5-5",
            "messages": [{"role": "system", "content": """You are a support agent for a test store.

You know:
- Shipping: 2-5 days nationwide, free on orders over $50
- Returns: 14 days, item in its packaging
- Payment: cards, PayPal, Apple Pay
- Hours: Mon-Fri 9-6, Sat 10-3

If a question is off-topic, say you're connecting them with an agent.
Keep it short and to the point. Be friendly."""}]
        },
        "voice": {
            "provider": "11labs",
            "voiceId": "voice_ID_from_ElevenLabs"
        },
        "firstMessage": "Hi there! Test support line. How can I help?"
    }
).json()

print(f"Agent ID: {agent['id']}")
print(f"To make a test call in the browser:")
print(f"Use Dashboard → Test Phone Call")

# get_call_analytics.py
import requests
import os

VAPI_API_KEY = os.environ["VAPI_API_KEY"]

# Get the list of recent calls
calls = requests.get(
    "https://api.vapi.ai/call",
    headers={"Authorization": f"Bearer {VAPI_API_KEY}"},
    params={"limit": 10}
).json()

for call in calls.get("results", []):
    print(f"\nCall: {call.get('id')}")
    print(f"Duration: {call.get('duration', 0)} sec")
    print(f"Status: {call.get('status')}")
    if call.get('transcript'):
        print(f"Transcript:\n{call['transcript'][:300]}...")