Build a Voice Call Center Triage System with Vapi + Swarms

Caller (PSTN / SIP)
        |
        v
+----------------------+
|  Vapi                |   <- STT, TTS, barge-in, audio loop
|  (or Retell / 11Labs)|
+----------+-----------+
           |  POST /webhook/vapi  { transcript, call_id, caller_phone, context }
           v
+----------------------+
|  FastAPI endpoint    |
+----------+-----------+
           |
           v
+----------------------+
|  MultiAgentRouter    |   Booking | Billing | Emergency | FAQ
|  /v1/swarm/          |   (auto-routes by transcript)
|  completions         |
+----------+-----------+
           |
           v
   reply text -> Vapi -> TTS -> Caller


At end-of-call:

   Vapi  --POST /webhook/vapi/end-of-call-->  FastAPI
                                                  |
                                                  v
                                       Summary Agent (single)
                                       /v1/agent/completions
                                                  |
                                                  v
                                         Airtable / HubSpot

pip install fastapi uvicorn requests python-dotenv
export SWARMS_API_KEY="your-api-key-here"
export AIRTABLE_API_KEY="your-airtable-key"
export AIRTABLE_BASE_ID="appXXXXXXXXXXXXXX"

import os
import requests
from fastapi import FastAPI, Request
from dotenv import load_dotenv

load_dotenv()

SWARMS_API_KEY = os.getenv("SWARMS_API_KEY")
SWARMS_BASE_URL = "https://api.swarms.world"

if not SWARMS_API_KEY:
    raise ValueError("SWARMS_API_KEY environment variable is required")

swarms_headers = {
    "x-api-key": SWARMS_API_KEY,
    "Content-Type": "application/json",
}

app = FastAPI()

{
  "name": "Front Desk",
  "model": {
    "provider": "openai",
    "model": "gpt-4.1-mini",
    "messages": [
      { "role": "system", "content": "You are the receptionist. Always defer to the server webhook for replies." }
    ]
  },
  "voice": { "provider": "11labs", "voiceId": "your-voice-id" },
  "transcriber": { "provider": "deepgram", "model": "nova-2" },
  "serverUrl": "https://your-domain.com/webhook/vapi",
  "serverMessages": ["function-call", "end-of-call-report"]
}

BOOKING_AGENT = {
    "agent_name": "Booking Specialist",
    "description": (
        "Handles appointment scheduling, reschedules, cancellations, "
        "and availability questions. Use for anything about times, dates, or slots."
    ),
    "system_prompt": (
        "You are a voice booking specialist on a live phone call. "
        "Keep every reply under two sentences. "
        "Speak naturally — no lists, no markdown, no special characters. "
        "When you need information from the caller (name, callback number, preferred time), "
        "ask one question at a time. "
        "If you commit to a booking, restate the time and date once for confirmation."
    ),
    "model_name": "claude-haiku-4.5",
    "max_loops": 1,
    "temperature": 0.3,
}

BILLING_AGENT = {
    "agent_name": "Billing Inquiry Handler",
    "description": (
        "Handles invoice questions, payment status, refunds, and account balance inquiries. "
        "Use for anything about money, charges, or statements."
    ),
    "system_prompt": (
        "You are a voice billing specialist on a live phone call. "
        "Keep every reply under two sentences. "
        "Never read out card numbers or full account numbers. "
        "If the caller needs to confirm sensitive info, ask them to verify the last four digits only. "
        "If you cannot resolve the issue on the call, offer to email a statement to the email on file."
    ),
    "model_name": "gpt-4.1-mini",
    "max_loops": 1,
    "temperature": 0.2,
}

EMERGENCY_AGENT = {
    "agent_name": "Emergency Triage",
    "description": (
        "Handles urgent, safety-critical, or after-hours emergency calls. "
        "Use when the caller mentions injury, severe pain, flooding, fire, break-in, "
        "or anything time-sensitive."
    ),
    "system_prompt": (
        "You are an emergency triage agent on a live phone call. "
        "Stay calm and direct. Replies must be under two sentences. "
        "If the situation is life-threatening, immediately tell the caller to hang up and dial 911. "
        "Otherwise, capture the caller's name, callback number, and the nature of the emergency, "
        "then tell them an on-call specialist will call back within fifteen minutes."
    ),
    "model_name": "claude-sonnet-4.5",
    "max_loops": 1,
    "temperature": 0.1,
}

FAQ_AGENT = {
    "agent_name": "FAQ Bot",
    "description": (
        "Handles general questions about hours, location, services offered, pricing, "
        "and anything else informational. Default agent for non-urgent, non-transactional calls."
    ),
    "system_prompt": (
        "You are a voice FAQ agent on a live phone call. "
        "Keep every reply under two sentences. "
        "If you do not know the answer, say so and offer to take a message. "
        "Speak naturally — no lists, no markdown."
    ),
    "model_name": "grok-4",
    "max_loops": 1,
    "temperature": 0.3,
}

VOICE_AGENTS = [BOOKING_AGENT, BILLING_AGENT, EMERGENCY_AGENT, FAQ_AGENT]

def run_voice_swarm(transcript: str, caller_phone: str, call_id: str) -> str:
    """Route a caller turn to the right specialist and return the spoken reply."""
    swarm_config = {
        "name": "Voice Call Center Router",
        "description": "Routes inbound voice calls to the right specialist agent.",
        "swarm_type": "MultiAgentRouter",
        "task": (
            f"Caller phone: {caller_phone}\n"
            f"Call ID: {call_id}\n\n"
            f"Live transcript so far (caller's latest turn at the end):\n{transcript}\n\n"
            "Respond as if speaking to the caller right now. "
            "Keep the reply under two sentences."
        ),
        "agents": VOICE_AGENTS,
        "max_loops": 1,
    }

    response = requests.post(
        f"{SWARMS_BASE_URL}/v1/swarm/completions",
        headers=swarms_headers,
        json=swarm_config,
        timeout=20,
    )
    response.raise_for_status()
    data = response.json()

    output = data.get("output", [{}])
    if isinstance(output, list) and output:
        return output[0].get("content", "I'm sorry, could you repeat that?")
    return "I'm sorry, could you repeat that?"


@app.post("/webhook/vapi")
async def vapi_webhook(request: Request):
    """Vapi calls this on every user turn."""
    payload = await request.json()
    message = payload.get("message", {})

    # On user turn, Vapi sends the running transcript and call context
    transcript = message.get("transcript", "")
    call = message.get("call", {})
    call_id = call.get("id", "unknown")
    caller_phone = call.get("customer", {}).get("number", "unknown")

    # End-of-call event is handled by a separate route — see Step 5
    if message.get("type") == "end-of-call-report":
        return await handle_end_of_call(payload)

    reply = run_voice_swarm(
        transcript=transcript,
        caller_phone=caller_phone,
        call_id=call_id,
    )

    # Vapi expects { "result": "<text to speak>" } for function-call style responses
    return {"result": reply}

SUMMARY_SYSTEM_PROMPT = (
    "You are a call summary agent. Given a full call transcript, output a STRICT JSON object "
    "with exactly these keys and nothing else:\n\n"
    "{\n"
    '  "caller_name": "<string or null>",\n'
    '  "callback_number": "<string>",\n'
    '  "intent": "BOOKING" | "BILLING" | "EMERGENCY" | "FAQ" | "OTHER",\n'
    '  "outcome": "RESOLVED" | "CALLBACK_REQUIRED" | "ESCALATED" | "UNRESOLVED",\n'
    '  "summary": "<two-sentence summary of the call>",\n'
    '  "follow_up_required": true | false,\n'
    '  "urgency": "LOW" | "MEDIUM" | "HIGH"\n'
    "}\n\n"
    "Output JSON only. No prose."
)


def summarize_call(transcript: str, caller_phone: str) -> dict:
    """Single-agent call to produce a structured CRM record."""
    payload = {
        "agent_config": {
            "agent_name": "Call Summary Agent",
            "system_prompt": SUMMARY_SYSTEM_PROMPT,
            "model_name": "gpt-4.1-mini",
            "max_tokens": 512,
            "temperature": 0.1,
        },
        "task": (
            f"Caller phone: {caller_phone}\n\n"
            f"Full transcript:\n{transcript}"
        ),
    }

    response = requests.post(
        f"{SWARMS_BASE_URL}/v1/agent/completions",
        headers=swarms_headers,
        json=payload,
        timeout=60,
    )
    response.raise_for_status()
    data = response.json()

    import json
    output = data.get("output") or data.get("outputs") or ""
    if isinstance(output, list):
        for item in reversed(output):
            if isinstance(item, dict) and item.get("role") in ("assistant", "Call Summary Agent"):
                output = item.get("content", "")
                break
    try:
        return json.loads(str(output).strip())
    except json.JSONDecodeError:
        return {"intent": "OTHER", "outcome": "UNRESOLVED", "summary": str(output)[:500]}


def write_to_airtable(record: dict, call_id: str) -> None:
    """POST the structured summary to Airtable."""
    airtable_url = (
        f"https://api.airtable.com/v0/{os.getenv('AIRTABLE_BASE_ID')}/Calls"
    )
    headers = {
        "Authorization": f"Bearer {os.getenv('AIRTABLE_API_KEY')}",
        "Content-Type": "application/json",
    }
    body = {
        "fields": {
            "Call ID": call_id,
            "Caller Name": record.get("caller_name") or "",
            "Callback Number": record.get("callback_number") or "",
            "Intent": record.get("intent"),
            "Outcome": record.get("outcome"),
            "Summary": record.get("summary"),
            "Urgency": record.get("urgency"),
            "Follow Up": bool(record.get("follow_up_required")),
        }
    }
    requests.post(airtable_url, headers=headers, json=body, timeout=15)


async def handle_end_of_call(payload: dict) -> dict:
    message = payload.get("message", {})
    call = message.get("call", {})
    call_id = call.get("id", "unknown")
    caller_phone = call.get("customer", {}).get("number", "unknown")

    # Vapi places the full transcript on the end-of-call report
    transcript = message.get("artifact", {}).get("transcript") or message.get("transcript", "")

    summary = summarize_call(transcript=transcript, caller_phone=caller_phone)
    write_to_airtable(summary, call_id)

    return {"received": True}