livekit-receptionist/
├── agents/
│   ├── receptionist.py         # Main receptionist agent with tools
│   ├── scheduling.py           # Scheduling specialist agent
│   └── handoff.py              # Human handoff logic
├── services/
│   ├── appointments.py         # Appointment CRUD (SQLite)
│   ├── knowledge_base.py       # FAQ retrieval (ChromaDB)
│   └── call_logger.py          # Call record storage
├── server.py                   # LiveKit AgentServer entry point
├── config.py                   # Pydantic Settings
├── models.py                   # Shared data models
├── dashboard.py                # FastAPI dashboard + webhooks
├── tests/
│   ├── test_receptionist.py
│   ├── test_appointments.py
│   └── test_knowledge_base.py
├── data/
│   └── faq_documents.json      # FAQ seed data
├── .env.local
├── Dockerfile
├── docker-compose.yml
└── requirements.txt

livekit-agents[codecs]~=1.0
livekit-plugins-deepgram~=1.0
livekit-plugins-openai~=1.0
livekit-plugins-cartesia~=1.0
livekit-plugins-silero~=1.0
livekit-plugins-noise-cancellation~=1.0
livekit-plugins-turn-detector~=1.0
livekit-api~=1.0
chromadb==0.5.0
sentence-transformers==3.3.0
fastapi==0.115.0
uvicorn==0.32.0
pydantic-settings==2.6.0
python-dotenv==1.0.0
pytest==8.3.0
pytest-asyncio==0.24.0

# LiveKit
LIVEKIT_URL=ws://localhost:7880
LIVEKIT_API_KEY=your-api-key
LIVEKIT_API_SECRET=your-api-secret

# SIP Trunk (Twilio)
SIP_TRUNK_ID=ST_xxxx

# AI Providers
DEEPGRAM_API_KEY=your-deepgram-key
OPENAI_API_KEY=sk-your-openai-key
CARTESIA_API_KEY=your-cartesia-key

# Agent Settings
HUMAN_AGENT_PHONE=+14155559999
BUSINESS_NAME=Sunrise Medical Clinic
BUSINESS_HOURS=Monday-Friday 9AM-5PM

# Knowledge Base
CHROMA_COLLECTION=receptionist_faq
EMBEDDING_MODEL=all-MiniLM-L6-v2
CONFIDENCE_THRESHOLD=0.6

# Install LiveKit CLI
brew install livekit-cli

# Create inbound SIP trunk (receives calls from Twilio)
lk sip inbound create \
  --request '{
    "trunk": {
      "name": "Twilio Inbound",
      "numbers": ["+14155551234"],
      "krisp_enabled": true
    }
  }'

lk sip dispatch create \
  --request '{
    "dispatch_rule": {
      "rule": {
        "dispatchRuleIndividual": {
          "roomPrefix": "call-"
        }
      },
      "roomConfig": {
        "agents": [{
          "agentName": "receptionist"
        }]
      }
    }
  }'

from functools import lru_cache

from pydantic_settings import BaseSettings
from pydantic import Field


class Settings(BaseSettings):
    """Receptionist agent configuration from environment."""

    # LiveKit
    livekit_url: str = Field(..., alias="LIVEKIT_URL")
    livekit_api_key: str = Field(..., alias="LIVEKIT_API_KEY")
    livekit_api_secret: str = Field(..., alias="LIVEKIT_API_SECRET")

    # SIP
    sip_trunk_id: str = Field("", alias="SIP_TRUNK_ID")
    human_agent_phone: str = Field("", alias="HUMAN_AGENT_PHONE")

    # AI Providers
    openai_api_key: str = Field(..., alias="OPENAI_API_KEY")
    deepgram_api_key: str = Field(..., alias="DEEPGRAM_API_KEY")
    cartesia_api_key: str = Field("", alias="CARTESIA_API_KEY")

    # Business
    business_name: str = Field("Sunrise Medical Clinic", alias="BUSINESS_NAME")
    business_hours: str = Field("Monday-Friday 9AM-5PM", alias="BUSINESS_HOURS")

    # Knowledge Base
    chroma_collection: str = Field("receptionist_faq", alias="CHROMA_COLLECTION")
    embedding_model: str = Field("all-MiniLM-L6-v2", alias="EMBEDDING_MODEL")
    confidence_threshold: float = Field(0.6, alias="CONFIDENCE_THRESHOLD")

    model_config = {"env_file": ".env.local", "extra": "ignore"}


@lru_cache
def get_settings() -> Settings:
    return Settings()

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional


class CallOutcome(str, Enum):
    """How the call ended."""
    COMPLETED = "completed"         # Agent handled it fully
    TRANSFERRED = "transferred"     # Handed off to human
    ABANDONED = "abandoned"         # Caller hung up early
    ERROR = "error"                 # System failure


class AppointmentStatus(str, Enum):
    """Appointment lifecycle states."""
    CONFIRMED = "confirmed"
    CANCELLED = "cancelled"
    RESCHEDULED = "rescheduled"


@dataclass
class Appointment:
    """A scheduled appointment."""
    id: str
    patient_name: str
    phone: str
    date: str              # YYYY-MM-DD
    time: str              # HH:MM
    reason: str
    provider: str = ""     # Doctor/specialist name
    status: AppointmentStatus = AppointmentStatus.CONFIRMED
    created_at: str = ""

    def __post_init__(self):
        if not self.created_at:
            self.created_at = datetime.now().isoformat()


@dataclass
class CallRecord:
    """Record of a completed call."""
    call_id: str
    room_name: str
    caller_phone: str
    duration_seconds: int = 0
    outcome: CallOutcome = CallOutcome.COMPLETED
    tools_used: list[str] = field(default_factory=list)
    appointment_id: Optional[str] = None
    transcript_summary: str = ""
    started_at: str = ""
    ended_at: str = ""

import json
import logging
from pathlib import Path

import chromadb
from chromadb.utils.embedding_functions import (
    SentenceTransformerEmbeddingFunction,
)

from config import get_settings

logger = logging.getLogger(__name__)


class KnowledgeBase:
    """ChromaDB-backed FAQ retrieval with confidence scoring."""

    def __init__(self):
        settings = get_settings()
        self._client = chromadb.PersistentClient(path="./chroma_data")
        self._embedding_fn = SentenceTransformerEmbeddingFunction(
            model_name=settings.embedding_model
        )
        self._collection = self._client.get_or_create_collection(
            name=settings.chroma_collection,
            embedding_function=self._embedding_fn,
            metadata={"hnsw:space": "cosine"},
        )
        self._threshold = settings.confidence_threshold

    def load_faqs(self, path: str = "data/faq_documents.json") -> int:
        """Seed the knowledge base from a JSON file.

        Expected format: [{"id": "...", "question": "...", "answer": "..."}]
        """
        data = json.loads(Path(path).read_text())
        if not data:
            return 0

        self._collection.upsert(
            ids=[d["id"] for d in data],
            documents=[
                f"Q: {d['question']}\nA: {d['answer']}" for d in data
            ],
            metadatas=[{"question": d["question"]} for d in data],
        )
        logger.info("Loaded %d FAQ documents", len(data))
        return len(data)

    def search(self, query: str, n_results: int = 3) -> dict:
        """Search FAQs and return the best answer with confidence.

        Returns:
            {"answer": str, "confidence": float, "found": bool}
        """
        results = self._collection.query(
            query_texts=[query],
            n_results=n_results,
        )

        if not results["documents"] or not results["documents"][0]:
            return {"answer": "", "confidence": 0.0, "found": False}

        documents = results["documents"][0]
        distances = results["distances"][0] if results["distances"] else []

        # Cosine distance → similarity (0=opposite, 1=identical)
        similarities = [1 - d for d in distances] if distances else []
        best_score = max(similarities) if similarities else 0.0

        if best_score < self._threshold:
            return {
                "answer": "",
                "confidence": best_score,
                "found": False,
            }

        # Return the highest-scoring document
        best_idx = similarities.index(best_score)
        return {
            "answer": documents[best_idx],
            "confidence": best_score,
            "found": True,
        }

[
  {
    "id": "hours",
    "question": "What are your office hours?",
    "answer": "We are open Monday through Friday, 9 AM to 5 PM. We are closed on weekends and major holidays."
  },
  {
    "id": "insurance",
    "question": "What insurance do you accept?",
    "answer": "We accept most major insurance plans including Blue Cross Blue Shield, Aetna, Cigna, UnitedHealthcare, and Medicare. Please call ahead to verify your specific plan."
  },
  {
    "id": "location",
    "question": "Where are you located?",
    "answer": "We are located at 456 Oak Avenue, Suite 200, San Francisco, CA 94102. Free parking is available in the building garage."
  },
  {
    "id": "new-patient",
    "question": "How do I become a new patient?",
    "answer": "New patients can schedule an initial consultation by calling us or booking online. Please bring your insurance card, photo ID, and any relevant medical records to your first visit."
  },
  {
    "id": "cancellation",
    "question": "What is your cancellation policy?",
    "answer": "We require 24 hours notice for cancellations. Late cancellations or no-shows may incur a 50 dollar fee. We understand emergencies happen and handle those on a case-by-case basis."
  },
  {
    "id": "urgent",
    "question": "What should I do in an emergency?",
    "answer": "If you are experiencing a medical emergency, please call 911 immediately. For urgent but non-emergency concerns during office hours, call us and we will try to see you the same day."
  },
  {
    "id": "telehealth",
    "question": "Do you offer telehealth appointments?",
    "answer": "Yes, we offer telehealth appointments for follow-up visits and certain types of consultations. Ask when scheduling if your visit qualifies for telehealth."
  },
  {
    "id": "referral",
    "question": "Do I need a referral?",
    "answer": "Some insurance plans require a referral from your primary care physician. Check with your insurance provider before scheduling. We can help verify if you are unsure."
  }
]

import logging
import sqlite3
import uuid
from contextlib import contextmanager
from datetime import datetime, timedelta

from models import Appointment, AppointmentStatus

logger = logging.getLogger(__name__)

DB_PATH = "data/appointments.db"

# Available time slots (30-minute intervals, 9 AM to 4:30 PM)
SLOT_START_HOUR = 9
SLOT_END_HOUR = 17
SLOT_DURATION_MINUTES = 30


def _init_db():
    """Create the appointments table if it does not exist."""
    with _get_conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS appointments (
                id TEXT PRIMARY KEY,
                patient_name TEXT NOT NULL,
                phone TEXT NOT NULL,
                date TEXT NOT NULL,
                time TEXT NOT NULL,
                reason TEXT NOT NULL,
                provider TEXT DEFAULT '',
                status TEXT DEFAULT 'confirmed',
                created_at TEXT NOT NULL
            )
        """)


@contextmanager
def _get_conn():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()


def get_available_slots(date: str) -> list[str]:
    """Return available 30-minute slots for a given date.

    Args:
        date: Date in YYYY-MM-DD format.

    Returns:
        List of available time strings like ["09:00", "09:30", "10:00", ...]
    """
    # Generate all possible slots
    all_slots = []
    current = datetime.strptime(f"{date} {SLOT_START_HOUR:02d}:00", "%Y-%m-%d %H:%M")
    end = datetime.strptime(f"{date} {SLOT_END_HOUR:02d}:00", "%Y-%m-%d %H:%M")

    while current < end:
        all_slots.append(current.strftime("%H:%M"))
        current += timedelta(minutes=SLOT_DURATION_MINUTES)

    # Remove booked slots
    with _get_conn() as conn:
        rows = conn.execute(
            "SELECT time FROM appointments WHERE date = ? AND status = ?",
            (date, AppointmentStatus.CONFIRMED.value),
        ).fetchall()

    booked = {row["time"] for row in rows}
    available = [s for s in all_slots if s not in booked]

    return available


def book_appointment(
    patient_name: str,
    phone: str,
    date: str,
    time: str,
    reason: str,
    provider: str = "",
) -> Appointment:
    """Book a new appointment.

    Returns:
        The created Appointment object.

    Raises:
        ValueError: If the slot is already booked.
    """
    available = get_available_slots(date)
    if time not in available:
        raise ValueError(f"Time slot {time} on {date} is not available")

    appointment = Appointment(
        id=str(uuid.uuid4())[:8],
        patient_name=patient_name,
        phone=phone,
        date=date,
        time=time,
        reason=reason,
        provider=provider,
    )

    with _get_conn() as conn:
        conn.execute(
            """INSERT INTO appointments
               (id, patient_name, phone, date, time, reason, provider, status, created_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                appointment.id,
                appointment.patient_name,
                appointment.phone,
                appointment.date,
                appointment.time,
                appointment.reason,
                appointment.provider,
                appointment.status.value,
                appointment.created_at,
            ),
        )

    logger.info("Booked appointment %s for %s on %s at %s",
                appointment.id, patient_name, date, time)
    return appointment


def cancel_appointment(appointment_id: str) -> bool:
    """Cancel an existing appointment."""
    with _get_conn() as conn:
        result = conn.execute(
            "UPDATE appointments SET status = ? WHERE id = ? AND status = ?",
            (AppointmentStatus.CANCELLED.value, appointment_id,
             AppointmentStatus.CONFIRMED.value),
        )
    return result.rowcount > 0


# Initialize database on import
_init_db()

import json
import logging
import sqlite3
from contextlib import contextmanager

from models import CallRecord

logger = logging.getLogger(__name__)

DB_PATH = "data/calls.db"


def _init_db():
    with _get_conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS call_records (
                call_id TEXT PRIMARY KEY,
                room_name TEXT NOT NULL,
                caller_phone TEXT DEFAULT '',
                duration_seconds INTEGER DEFAULT 0,
                outcome TEXT DEFAULT 'completed',
                tools_used TEXT DEFAULT '[]',
                appointment_id TEXT DEFAULT '',
                transcript_summary TEXT DEFAULT '',
                started_at TEXT NOT NULL,
                ended_at TEXT DEFAULT ''
            )
        """)


@contextmanager
def _get_conn():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()


def save_call(record: CallRecord) -> None:
    """Save a completed call record."""
    with _get_conn() as conn:
        conn.execute(
            """INSERT OR REPLACE INTO call_records
               (call_id, room_name, caller_phone, duration_seconds,
                outcome, tools_used, appointment_id,
                transcript_summary, started_at, ended_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                record.call_id,
                record.room_name,
                record.caller_phone,
                record.duration_seconds,
                record.outcome.value,
                json.dumps(record.tools_used),
                record.appointment_id or "",
                record.transcript_summary,
                record.started_at,
                record.ended_at,
            ),
        )
    logger.info("Saved call record: %s (outcome=%s)", record.call_id, record.outcome.value)


def get_recent_calls(limit: int = 20) -> list[dict]:
    """Retrieve recent call records for the dashboard."""
    with _get_conn() as conn:
        rows = conn.execute(
            "SELECT * FROM call_records ORDER BY started_at DESC LIMIT ?",
            (limit,),
        ).fetchall()

    return [dict(row) for row in rows]


_init_db()

import logging
from typing import Any

from livekit.agents import Agent, RunContext, function_tool

from config import get_settings
from services.appointments import get_available_slots, book_appointment
from services.knowledge_base import KnowledgeBase

logger = logging.getLogger(__name__)

# Initialize knowledge base once
_kb = KnowledgeBase()
_kb.load_faqs()


class ReceptionistAgent(Agent):
    """AI phone receptionist for a medical clinic.

    Handles: greetings, FAQ answers, appointment scheduling,
    and transfers to human agents when needed.
    """

    def __init__(self, job_context=None) -> None:
        settings = get_settings()
        self.job_context = job_context

        super().__init__(
            instructions=f"""You are the phone receptionist for {settings.business_name}.
You answer incoming phone calls with warmth and professionalism.

RULES:
1. Be concise. Callers are LISTENING, not reading. Keep responses to 1-2 sentences.
2. Never spell out URLs, emails, or long numbers. Say "I can text you that information."
3. For factual questions (hours, insurance, location), ALWAYS use the lookup_faq tool.
   Never guess — if the tool returns no result, say "I don't have that information handy,
   let me transfer you to someone who can help."
4. For appointment scheduling, use check_availability first, then confirm with the caller
   before calling book_appointment.
5. If the caller asks for a human, a doctor, or says "transfer me," use transfer_to_human
   immediately. Do not try to convince them to stay.
6. Speak naturally. Avoid bullet points, markdown, or any text formatting.
7. If the caller sounds upset or frustrated, acknowledge their feelings before solving
   the problem.

BUSINESS INFO:
- Name: {settings.business_name}
- Hours: {settings.business_hours}
""",
        )

    async def on_enter(self) -> None:
        """Called when this agent becomes active. Greet the caller."""
        settings = get_settings()
        await self.session.generate_reply(
            instructions=f"Greet the caller warmly. Say: 'Thank you for calling "
            f"{settings.business_name}, how can I help you today?'"
        )

    @function_tool()
    async def lookup_faq(
        self,
        context: RunContext,
        question: str,
    ) -> dict[str, Any]:
        """Search the knowledge base for answers to common questions
        about office hours, insurance, location, policies, and services.

        Args:
            question: The caller's question to look up.
        """
        result = _kb.search(question)

        if result["found"]:
            logger.info("FAQ hit: %.2f confidence for '%s'",
                       result["confidence"], question[:50])
            return {
                "answer": result["answer"],
                "confidence": result["confidence"],
            }

        logger.info("FAQ miss: %.2f confidence for '%s'",
                    result["confidence"], question[:50])
        return {
            "answer": "No matching FAQ found.",
            "confidence": result["confidence"],
            "suggestion": "Offer to transfer to a staff member who can help.",
        }

    @function_tool()
    async def check_availability(
        self,
        context: RunContext,
        date: str,
    ) -> dict[str, Any]:
        """Check available appointment slots for a specific date.

        Args:
            date: The date to check in YYYY-MM-DD format.
        """
        slots = get_available_slots(date)

        if not slots:
            return {
                "available": False,
                "message": f"No slots available on {date}.",
                "suggestion": "Try the next business day.",
            }

        # Group slots for easier reading over the phone
        morning = [s for s in slots if int(s.split(":")[0]) < 12]
        afternoon = [s for s in slots if int(s.split(":")[0]) >= 12]

        return {
            "available": True,
            "date": date,
            "morning_slots": morning,
            "afternoon_slots": afternoon,
            "total": len(slots),
        }

    @function_tool()
    async def book_appointment(
        self,
        context: RunContext,
        patient_name: str,
        phone: str,
        date: str,
        time: str,
        reason: str,
    ) -> dict[str, Any]:
        """Book an appointment after confirming details with the caller.

        Args:
            patient_name: The patient's full name.
            phone: The patient's phone number for confirmation.
            date: Appointment date in YYYY-MM-DD format.
            time: Appointment time in HH:MM format.
            reason: Brief reason for the visit.
        """
        try:
            appointment = book_appointment(
                patient_name=patient_name,
                phone=phone,
                date=date,
                time=time,
                reason=reason,
            )
            return {
                "success": True,
                "confirmation_id": appointment.id,
                "date": date,
                "time": time,
                "message": f"Appointment confirmed for {patient_name}.",
            }
        except ValueError as exc:
            return {
                "success": False,
                "error": str(exc),
                "suggestion": "Check availability for a different time.",
            }

    @function_tool()
    async def transfer_to_human(
        self,
        context: RunContext,
        reason: str,
    ) -> dict[str, Any]:
        """Transfer the caller to a human staff member.
        Use this when the caller explicitly asks for a human,
        when you cannot answer their question, or when the situation
        requires human judgment.

        Args:
            reason: Brief reason for the transfer.
        """
        import os
        import uuid
        from livekit.protocol import api

        if not self.job_context:
            await self.session.say(
                "I'm sorry, I'm unable to transfer the call right now. "
                "Please try calling back."
            )
            return {"success": False, "error": "No job context available"}

        phone = os.environ.get("HUMAN_AGENT_PHONE", "")
        if not phone:
            await self.session.say(
                "I'm sorry, no staff members are available for transfer. "
                "Can I take a message instead?"
            )
            return {"success": False, "error": "No human agent phone configured"}

        sip_trunk_id = os.environ.get("SIP_TRUNK_ID", "")
        room_name = self.job_context.room.name

        try:
            # Add human agent to the same Room via SIP
            await self.job_context.api.sip.create_sip_participant(
                api.CreateSIPParticipantRequest(
                    sip_trunk_id=sip_trunk_id,
                    sip_call_to=phone,
                    room_name=room_name,
                    participant_identity=f"human_{uuid.uuid4().hex[:8]}",
                    participant_name="Staff Member",
                    krisp_enabled=True,
                )
            )

            await self.session.say(
                "I'm transferring you to a staff member now. "
                "Please hold for just a moment."
            )
            return {"success": True, "reason": reason}

        except Exception as exc:
            logger.error("Transfer failed: %s", exc)
            await self.session.say(
                "I'm sorry, I couldn't reach a staff member right now. "
                "Can I take your name and number so someone can call you back?"
            )
            return {"success": False, "error": str(exc)}

import logging

from dotenv import load_dotenv

from livekit import agents, rtc
from livekit.agents import AgentServer, AgentSession, JobContext, JobProcess, room_io
from livekit.plugins import noise_cancellation, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

from agents.receptionist import ReceptionistAgent

load_dotenv(".env.local")

logger = logging.getLogger(__name__)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(name)s %(levelname)s %(message)s",
)

server = AgentServer()


def prewarm(proc: JobProcess):
    """Pre-load heavy models once per worker process.

    VAD (Voice Activity Detection) loads a neural network from disk.
    Loading it once here and reusing across sessions saves ~2s per call.
    """
    proc.userdata["vad"] = silero.VAD.load()


server.setup_fnc = prewarm


@server.rtc_session(agent_name="receptionist")
async def handle_call(ctx: JobContext):
    """Handle one inbound phone call.

    LiveKit dispatches this function for each incoming SIP call
    that matches the dispatch rule with agentName="receptionist".
    """
    # Build the voice pipeline
    session = AgentSession(
        stt="deepgram/nova-3:multi",
        llm="openai/gpt-4.1-mini",
        tts="cartesia/sonic-3:9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
        vad=ctx.proc.userdata["vad"],
        turn_detection=MultilingualModel(),
    )

    # Create the receptionist agent with job context (needed for SIP transfers)
    agent = ReceptionistAgent(job_context=ctx)

    # Start the session with telephony-optimized audio
    await session.start(
        room=ctx.room,
        agent=agent,
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=_get_noise_cancellation,
            ),
        ),
    )

    # Connect to the room (makes the agent a participant)
    await ctx.connect()

    logger.info("Receptionist agent started in room %s", ctx.room.name)


def _get_noise_cancellation(params):
    """Select noise cancellation mode based on caller type.

    SIP callers (phone calls) get telephony-optimized cancellation
    that handles PSTN background noise and echo. Browser callers
    get standard background voice cancellation.
    """
    if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
        return noise_cancellation.BVCTelephony()
    return noise_cancellation.BVC()


if __name__ == "__main__":
    agents.cli.run_app(server)

import logging
from datetime import datetime

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from services.appointments import get_available_slots
from services.call_logger import get_recent_calls

logger = logging.getLogger(__name__)

app = FastAPI(title="Receptionist Dashboard", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/api/calls")
async def list_calls(limit: int = 20):
    """List recent call records."""
    return {"calls": get_recent_calls(limit)}


@app.get("/api/availability/{date}")
async def check_date(date: str):
    """Check available appointment slots for a date."""
    slots = get_available_slots(date)
    return {"date": date, "slots": slots, "total": len(slots)}


@app.get("/api/stats")
async def stats():
    """Basic call statistics."""
    calls = get_recent_calls(100)

    total = len(calls)
    transferred = sum(1 for c in calls if c.get("outcome") == "transferred")
    completed = sum(1 for c in calls if c.get("outcome") == "completed")

    return {
        "total_calls": total,
        "completed": completed,
        "transferred": transferred,
        "transfer_rate": transferred / total if total > 0 else 0,
        "generated_at": datetime.now().isoformat(),
    }


@app.get("/health")
async def health():
    return {"status": "ok"}

import pytest

from services.knowledge_base import KnowledgeBase


class TestKnowledgeBase:
    def setup_method(self):
        self.kb = KnowledgeBase()
        self.kb.load_faqs()

    def test_finds_hours_question(self):
        result = self.kb.search("What time do you open?")
        assert result["found"] is True
        assert result["confidence"] > 0.6
        assert "9 AM" in result["answer"] or "Monday" in result["answer"]

    def test_finds_insurance_question(self):
        result = self.kb.search("Do you take Blue Cross?")
        assert result["found"] is True
        assert "Blue Cross" in result["answer"]

    def test_returns_not_found_for_irrelevant_query(self):
        result = self.kb.search("What is the capital of France?")
        # Should not match any FAQ with high confidence
        assert result["found"] is False or result["confidence"] < 0.6

    def test_finds_location(self):
        result = self.kb.search("Where is your office?")
        assert result["found"] is True
        assert "Oak Avenue" in result["answer"] or "San Francisco" in result["answer"]

    def test_finds_cancellation_policy(self):
        result = self.kb.search("What if I need to cancel?")
        assert result["found"] is True
        assert "24 hours" in result["answer"]

import os
import pytest

# Use a test database
os.environ.setdefault("DB_PATH", ":memory:")

from services.appointments import (
    get_available_slots,
    book_appointment,
    cancel_appointment,
)


class TestAppointmentSlots:
    def test_all_slots_available_on_empty_day(self):
        slots = get_available_slots("2099-01-15")
        assert len(slots) > 0
        assert "09:00" in slots
        assert "16:30" in slots

    def test_slot_format(self):
        slots = get_available_slots("2099-01-15")
        for slot in slots:
            hour, minute = slot.split(":")
            assert 0 <= int(hour) <= 23
            assert int(minute) in (0, 30)


class TestBookAppointment:
    def test_book_and_confirm(self):
        apt = book_appointment(
            patient_name="John Doe",
            phone="+14155551111",
            date="2099-02-01",
            time="10:00",
            reason="Annual checkup",
        )
        assert apt.id is not None
        assert apt.patient_name == "John Doe"
        assert apt.date == "2099-02-01"
        assert apt.time == "10:00"

    def test_double_booking_raises(self):
        book_appointment(
            patient_name="Jane Smith",
            phone="+14155552222",
            date="2099-03-01",
            time="11:00",
            reason="Follow-up",
        )
        with pytest.raises(ValueError, match="not available"):
            book_appointment(
                patient_name="Bob Wilson",
                phone="+14155553333",
                date="2099-03-01",
                time="11:00",
                reason="Consultation",
            )

    def test_slot_removed_after_booking(self):
        book_appointment(
            patient_name="Alice Brown",
            phone="+14155554444",
            date="2099-04-01",
            time="14:00",
            reason="Lab results",
        )
        slots = get_available_slots("2099-04-01")
        assert "14:00" not in slots


class TestCancelAppointment:
    def test_cancel_existing(self):
        apt = book_appointment(
            patient_name="To Cancel",
            phone="+14155555555",
            date="2099-05-01",
            time="09:00",
            reason="Test",
        )
        assert cancel_appointment(apt.id) is True

    def test_cancel_nonexistent(self):
        assert cancel_appointment("nonexistent-id") is False

import pytest

from services.knowledge_base import KnowledgeBase


class TestKnowledgeBaseEdgeCases:
    def setup_method(self):
        self.kb = KnowledgeBase()
        self.kb.load_faqs()

    def test_empty_query_returns_low_confidence(self):
        result = self.kb.search("")
        assert result["confidence"] < 0.6 or not result["found"]

    def test_returns_dict_format(self):
        result = self.kb.search("office hours")
        assert "answer" in result
        assert "confidence" in result
        assert "found" in result
        assert isinstance(result["confidence"], float)

    def test_confidence_between_0_and_1(self):
        result = self.kb.search("Do you accept insurance?")
        assert 0.0 <= result["confidence"] <= 1.0

    def test_multiple_searches_consistent(self):
        """Same query should return the same result."""
        r1 = self.kb.search("What are your hours?")
        r2 = self.kb.search("What are your hours?")
        assert r1["answer"] == r2["answer"]
        assert abs(r1["confidence"] - r2["confidence"]) < 0.01

FROM python:3.11-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.11-slim

WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin

COPY agents/ ./agents/
COPY services/ ./services/
COPY data/ ./data/
COPY server.py config.py models.py dashboard.py ./

# Create data directory for SQLite databases
RUN mkdir -p /app/data

EXPOSE 8000

CMD ["python", "server.py", "start"]

services:
  # LiveKit Server (SFU)
  livekit-server:
    image: livekit/livekit-server:latest
    ports:
      - "7880:7880"   # WebSocket + HTTP
      - "7881:7881"   # WebRTC (TCP)
      - "50000-50100:50000-50100/udp"  # WebRTC (UDP)
    environment:
      - LIVEKIT_KEYS=devkey:secret
    command: --dev --bind 0.0.0.0

  # AI Receptionist Agent
  receptionist-agent:
    build: .
    env_file: .env.local
    depends_on:
      - livekit-server
    volumes:
      - agent_data:/app/data
    restart: unless-stopped

  # Dashboard API
  dashboard:
    build: .
    command: uvicorn dashboard:app --host 0.0.0.0 --port 8000
    ports:
      - "8000:8000"
    volumes:
      - agent_data:/app/data
    restart: unless-stopped

volumes:
  agent_data:

docker-compose up -d livekit-server
docker-compose up receptionist-agent dashboard

# Terminal 1: Start LiveKit server
docker run --rm -p 7880:7880 -p 7881:7881 \
  -p 50000-50100:50000-50100/udp \
  -e LIVEKIT_KEYS=devkey:secret \
  livekit/livekit-server --dev --bind 0.0.0.0

# Terminal 2: Start the agent
python server.py start

# Terminal 3: Start the dashboard
uvicorn dashboard:app --reload --port 8000

# Open LiveKit's web playground to test via browser
# Visit: https://agents-playground.livekit.io
# Enter your LiveKit server URL and API credentials
# Click "Connect" — your agent will activate via WebRTC instead of SIP

# View recent calls
curl http://localhost:8000/api/calls

# Check appointment availability
curl http://localhost:8000/api/availability/2026-04-15

# View call statistics
curl http://localhost:8000/api/stats

pytest tests/ -v