Complete project scaffolding and core implementation of an AI-powered telephony system that calls companies, navigates IVR menus, waits on hold, and transfers to the user when a human answers. Key components: - FastAPI server with REST API, WebSocket, and MCP (SSE) interfaces - SIP/VoIP call management via PJSUA2 with RTP audio streaming - LLM-powered IVR navigation using OpenAI/Anthropic with tool calling - Hold detection service combining audio analysis and silence detection - Real-time STT (Whisper/Deepgram) and TTS (OpenAI/Piper) pipelines - Call recording with per-channel and mixed audio capture - Event bus (asyncio pub/sub) for real-time client updates - Web dashboard with live call monitoring - SQLite persistence via SQLAlchemy with call history and analytics - Notification support (email, SMS, webhook, desktop) - Docker Compose deployment with Opal VoIP and Opal Media containers - Comprehensive test suite with unit, integration, and E2E tests - Simplified .gitignore and full project documentation in README
5.4 KiB
Development
Setup
Prerequisites
- Python 3.13+
- Ollama (or any OpenAI-compatible LLM) — for IVR menu analysis
- Speaches or Whisper API — for speech-to-text (optional for dev)
- A SIP trunk account — for making real calls (optional for dev)
Install
git clone <repo-url>
cd hold-slayer
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
Dev Dependencies
The [dev] extras include:
pytest— test runnerpytest-asyncio— async test supportpytest-cov— coverage reporting
Testing
Run All Tests
pytest tests/ -v
Run Specific Test Files
pytest tests/test_audio_classifier.py -v # 18 tests — waveform analysis
pytest tests/test_call_flows.py -v # 10 tests — call flow models
pytest tests/test_hold_slayer.py -v # 20 tests — IVR nav, EventBus, CallManager
pytest tests/test_services.py -v # 27 tests — LLM, notifications, recording,
# analytics, learner, EventBus
Run with Coverage
pytest tests/ --cov=. --cov-report=term-missing
Test Architecture
Tests are organized by component:
| File | Tests | What's Covered |
|---|---|---|
test_audio_classifier.py |
18 | Silence, tone, DTMF, music, speech detection; feature extraction; classification history |
test_call_flows.py |
10 | CallFlowStep types, CallFlow navigation, serialization roundtrip, create/summary models |
test_hold_slayer.py |
20 | IVR menu navigation (6 intent scenarios), EventBus pub/sub, CallManager lifecycle, MockSIPEngine |
test_services.py |
27 | LLMClient init/stats/chat/JSON/errors/IVR analysis, NotificationService event mapping, RecordingService paths, CallAnalytics summaries, CallFlowLearner build/merge, EventBus integration |
Known Test Issues
test_complex_tone_as_music — A synthetic multi-harmonic tone is classified as LIVE_HUMAN instead of MUSIC. This is a known edge case. Real hold music has different spectral characteristics than synthetic test signals. This test documents the limitation rather than a bug.
Writing Tests
All tests use pytest-asyncio for async support. The test configuration in pyproject.toml:
[tool.pytest.ini_options]
asyncio_mode = "auto"
This means all async def test_* functions automatically run in an asyncio event loop.
Pattern for testing services:
import pytest
from services.llm_client import LLMClient
class TestLLMClient:
def test_init(self):
client = LLMClient(base_url="http://localhost:11434/v1", model="llama3")
assert client._model == "llama3"
@pytest.mark.asyncio
async def test_chat(self):
# Mock httpx for unit tests
...
Pattern for testing EventBus:
import asyncio
from core.event_bus import EventBus
from models.events import EventType, GatewayEvent
async def test_publish_receive():
bus = EventBus()
sub = bus.subscribe()
event = GatewayEvent(type=EventType.CALL_STARTED, call_id="test", data={})
await bus.publish(event)
received = await asyncio.wait_for(sub.get(), timeout=1.0)
assert received.type == EventType.CALL_STARTED
Project Conventions
Code Style
- Type hints everywhere — All function signatures have type annotations
- Pydantic models — All data structures are Pydantic BaseModel or dataclass
- Async by default — All I/O operations are async
- Logging — Every module uses
logging.getLogger(__name__) - Docstrings — Module-level docstrings explain purpose and usage
File Organization
module.py
├── Module docstring (purpose, usage examples)
├── Imports (stdlib → third-party → local)
├── Constants
├── Classes
│ ├── Class docstring
│ ├── __init__
│ ├── Public methods (async)
│ └── Private methods (_prefixed)
└── Module-level functions (if any)
Error Handling
- Services never crash the call — All service errors are caught, logged, and return sensible defaults
- LLM failures return empty string/dict — the Hold Slayer falls back to waiting
- SIP errors publish
CALL_FAILEDevents — the user is notified - HTTP errors in the API return structured error responses
Event-Driven Architecture
All components communicate through the EventBus:
- Publishers — SIP engine, Hold Slayer, classifier, services
- Subscribers — WebSocket handler, MCP server, notification service, analytics
This decouples components and makes the system extensible. Adding a new feature (e.g., Slack notifications) means subscribing to events — no changes to existing code.
Dependency Injection
The AIPSTNGateway owns all services and is injected into FastAPI routes via api/deps.py:
# api/deps.py
async def get_gateway() -> AIPSTNGateway:
return app.state.gateway
# api/calls.py
@router.post("/outbound")
async def make_call(request: CallRequest, gateway: AIPSTNGateway = Depends(get_gateway)):
...
This makes testing easy — swap the gateway for a mock in tests.
Contributing
- Create a feature branch
- Write tests for new functionality
- Ensure all tests pass:
pytest tests/ -v - Follow existing code conventions
- Update documentation in
/docsif adding new features - Submit a pull request