r/hold-slayer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stage 6: learn_call_flow rebuilt on the learner, docs truth sweep

Browse Source 
The call-flow learner finally gets fed: exploration mode records its
IVR discoveries on the call (ActiveCall.exploration_steps) instead of
throwing them away, persistence stores them in the call record's
metadata, and the rebuilt learn_call_flow MCP tool turns a completed
exploration call into a stored flow via CallFlowLearner — correct
constructor (llm_client from get_llm, heuristic labels when the LLM
is unavailable), build for a new number, merge/refine when a flow
already exists. save_learned_flow/update_flow_from_model keep the
CallFlow↔row mapping in call_persistence.

Test gaps closed: tests/test_learner.py (discoveries→linked steps,
exploration persistence, learn-then-refine through the in-memory MCP
client, no-data and unknown-call answers) and tests/test_websocket.py
(4401 without token, trunk-status-then-replay on connect, per-call
stream filtering).

Docs aligned to code: README (15 tools incl. learn_call_flow, HTTP
not SSE, Python 3.12+, PostgreSQL+Alembic — no SQLite fallback, media
pipeline marked stub-mode until pjsua2 installed, Alembic and honest
/health checked off); docs/mcp-server.md rewritten against the actual
tool surface (hangup not end_call, real params, 3 real resources,
/mcp/ streamable HTTP + bearer auth); architecture/development/
configuration drift fixed.

pyproject: pruned never-imported deps (websockets, librosa,
soundfile, python-multipart).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Robert Helewka
2026-07-10 13:45:35 -04:00
parent f7a11f2f20
commit ff7ea8623a
13 changed files with 444 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
README.md
View File

@@ -21,7 +21,7 @@ You give it a phone number and an intent ("dispute a charge on my December state
│ │ │ │
│ ┌──────────┐ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ ┌──────────┐ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │
│ │ REST API │ │WebSocket │ │MCP Server │ │ Dashboard │ │ │ │ REST API │ │WebSocket │ │MCP Server │ │ Dashboard │ │
│ │ /api/* │ │ /ws/* │ │ (SSE) │ │ /dashboard │ │ │ │ /api/* │ │ /ws/* │ │ (HTTP) │ │ /dashboard │ │
│ └────┬─────┘ └────┬─────┘ └─────┬─────┘ └──────────────┘ │ │ └────┬─────┘ └────┬─────┘ └─────┬─────┘ └──────────────┘ │
│ │ │ │ │ │ │ │ │ │
│ ┌────┴──────────────┴──────────────┴────┐ │ │ ┌────┴──────────────┴──────────────┴────┐ │
@@ -53,7 +53,7 @@ You give it a phone number and an intent ("dispute a charge on my December state
### Core Engine ### Core Engine
- **Sippy B2BUA Engine** (`core/sippy_engine.py`) — SIP call control, DTMF, bridging, conference, trunk registration - **Sippy B2BUA Engine** (`core/sippy_engine.py`) — SIP call control, DTMF, bridging, conference, trunk registration
- **PJSUA2 Media Pipeline** (`core/media_pipeline.py`) — Audio routing, recording ports, conference bridge, WAV playback - **PJSUA2 Media Pipeline** (`core/media_pipeline.py`) — Audio routing, recording ports, conference bridge, WAV playback (stub mode until the `pjsua2` bindings are installed — see note below)
- **Call Manager** (`core/call_manager.py`) — Active call state tracking, lifecycle management - **Call Manager** (`core/call_manager.py`) — Active call state tracking, lifecycle management
- **Event Bus** (`core/event_bus.py`) — Async pub/sub with per-subscriber queues, type filtering, history - **Event Bus** (`core/event_bus.py`) — Async pub/sub with per-subscriber queues, type filtering, history
@@ -78,7 +78,7 @@ You give it a phone number and an intent ("dispute a charge on my December state
### API Surface ### API Surface
- **REST API** — Call management, call history, transcripts, recordings, routing rules, device DND, call flow CRUD - **REST API** — Call management, call history, transcripts, recordings, routing rules, device DND, call flow CRUD
- **WebSocket** — Real-time call events, transcripts, classification updates, receptionist state transitions - **WebSocket** — Real-time call events, transcripts, classification updates, receptionist state transitions
- **MCP Server** — 14 tools + 3 resources for AI assistant integration (make calls, send DTMF, get transcripts, manage flows), served over streamable HTTP at `/mcp/` - **MCP Server** — 15 tools + 3 resources for AI assistant integration (make calls, send DTMF, get transcripts, manage flows), served over streamable HTTP at `/mcp/`
- **Dashboard** — SvelteKit UI served at `/dashboard` with live monitor, call history with transcript playback, and a routing-rules editor - **Dashboard** — SvelteKit UI served at `/dashboard` with live monitor, call history with transcript playback, and a routing-rules editor
### Data Models ### Data Models
@@ -130,7 +130,7 @@ hold-slayer/
│ ├── calls/[call_id]/ # Detail page + transcript playback │ ├── calls/[call_id]/ # Detail page + transcript playback
│ └── routing/ # Rules editor + DND toggles │ └── routing/ # Rules editor + DND toggles
├── mcp_server/ ├── mcp_server/
│ └── server.py # MCP tools + resources (10 tools) │ └── server.py # MCP tools + resources (15 tools)
├── models/ ├── models/
│ ├── call.py # Call state models │ ├── call.py # Call state models
│ ├── call_flow.py # IVR tree models │ ├── call_flow.py # IVR tree models
@@ -139,7 +139,7 @@ hold-slayer/
│ ├── device.py # Device models │ ├── device.py # Device models
│ └── contact.py # Contact models │ └── contact.py # Contact models
├── db/ ├── db/
│ └── database.py # SQLAlchemy async (PostgreSQL/SQLite) │ └── database.py # SQLAlchemy async (PostgreSQL + Alembic)
└── tests/ └── tests/
├── test_audio_classifier.py # 18 tests — waveform analysis ├── test_audio_classifier.py # 18 tests — waveform analysis
├── test_call_flows.py # 10 tests — call flow models ├── test_call_flows.py # 10 tests — call flow models
@@ -283,7 +283,7 @@ claude mcp add hold-slayer --transport http http://localhost:8000/mcp/ \
--header "Authorization: Bearer $API_TOKEN" --header "Authorization: Bearer $API_TOKEN"
``` ```
It exposes 14 tools and 3 resources (`gateway://status`, It exposes 15 tools and 3 resources (`gateway://status`,
`gateway://call-flows`, `gateway://active-calls`): `gateway://call-flows`, `gateway://active-calls`):
| Tool | Description | | Tool | Description |
@@ -302,6 +302,7 @@ It exposes 14 tools and 3 resources (`gateway://status`,
| `create_call_flow` | Store a new IVR call flow | | `create_call_flow` | Store a new IVR call flow |
| `get_call_summary` | Stored summary and action items for a call | | `get_call_summary` | Stored summary and action items for a call |
| `search_call_history` | Search past calls by number or intent | | `search_call_history` | Search past calls by number or intent |
| `learn_call_flow` | Build/refine a reusable IVR flow from an exploration call |
## How It Works ## How It Works
@@ -352,7 +353,7 @@ All configuration is via environment variables (see `.env.example`):
## Tech Stack ## Tech Stack
- **Python 3.13** + **asyncio** — Single-process async architecture - **Python 3.12+** + **asyncio** — Single-process async architecture
- **FastAPI** — REST API + WebSocket server - **FastAPI** — REST API + WebSocket server
- **SvelteKit** — Dashboard UI (built static, served by FastAPI at `/dashboard`) - **SvelteKit** — Dashboard UI (built static, served by FastAPI at `/dashboard`)
- **Sippy B2BUA** — SIP call control and DTMF - **Sippy B2BUA** — SIP call control and DTMF
@@ -360,7 +361,7 @@ All configuration is via environment variables (see `.env.example`):
- **Speaches** (Whisper) — Speech-to-text - **Speaches** (Whisper) — Speech-to-text
- **Rhema** (Kokoro) — Text-to-speech (OpenAI-compatible `/v1/audio/speech`) - **Rhema** (Kokoro) — Text-to-speech (OpenAI-compatible `/v1/audio/speech`)
- **Ollama / vLLM / OpenAI** — LLM for IVR menu analysis and receptionist intent capture - **Ollama / vLLM / OpenAI** — LLM for IVR menu analysis and receptionist intent capture
- **SQLAlchemy** — Async database (PostgreSQL or SQLite) - **SQLAlchemy + Alembic** — Async database (PostgreSQL; schema managed by migrations)
- **MCP (Model Context Protocol)** — AI assistant integration - **MCP (Model Context Protocol)** — AI assistant integration
## Documentation ## Documentation
@@ -384,7 +385,7 @@ Full documentation is in [`/docs`](docs/README.md):
- [x] Extract EventBus to dedicated module with typed filtering - [x] Extract EventBus to dedicated module with typed filtering
- [x] Implement Sippy B2BUA SIP engine (signaling, DTMF, bridging) - [x] Implement Sippy B2BUA SIP engine (signaling, DTMF, bridging)
- [x] Implement PJSUA2 media pipeline (conference bridge, audio tapping, recording) - [x] PJSUA2 media pipeline contract (conference bridge, audio tapping, recording) — runs in stub mode until `pjsua2` bindings are installed
- [x] Call manager with active call state tracking - [x] Call manager with active call state tracking
- [x] Gateway orchestrator wiring all components - [x] Gateway orchestrator wiring all components
@@ -400,18 +401,18 @@ Full documentation is in [`/docs`](docs/README.md):
- [x] REST API — calls, call flows, devices, DTMF - [x] REST API — calls, call flows, devices, DTMF
- [x] WebSocket real-time event streaming - [x] WebSocket real-time event streaming
- [x] MCP server with 14 tools + 3 resources, mounted at `/mcp/` (streamable HTTP) - [x] MCP server with 15 tools + 3 resources, mounted at `/mcp/` (streamable HTTP)
- [x] Notification service (WebSocket + SMS) - [x] Notification service (WebSocket + SMS)
- [x] Service wiring in main.py lifespan - [x] Service wiring in main.py lifespan
### Phase 4: Production Hardening 🚧 ### Phase 4: Production Hardening 🚧
- [ ] Alembic database migrations - [x] Alembic database migrations (baseline + upgrade-on-boot)
- [x] API authentication — static bearer token across REST/WS/MCP - [x] API authentication — static bearer token across REST/WS/MCP
- [x] Emergency-number guard + concurrent-call cap on outbound calls - [x] Emergency-number guard + concurrent-call cap on outbound calls
- [ ] Rate limiting on API endpoints - [ ] Rate limiting on API endpoints
- [ ] Structured JSON logging - [ ] Structured JSON logging
- [ ] Health check endpoints for all dependencies - [x] Honest /health — engine mode, DB ping, trunk registration, STT/TTS availability
- [ ] Graceful degradation (classifier works without STT, etc.) - [ ] Graceful degradation (classifier works without STT, etc.)
- [ ] Docker Compose (Hold Slayer + PostgreSQL) - [ ] Docker Compose (Hold Slayer + PostgreSQL)

6
docs/architecture.md
View File

@@ -10,7 +10,7 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
│ │ │ │
│ ┌──────────┐ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ ┌──────────┐ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │
│ │ REST API │ │WebSocket │ │MCP Server │ │ Dashboard │ │ │ │ REST API │ │WebSocket │ │MCP Server │ │ Dashboard │ │
│ │ /api/* │ │ /ws/* │ │ (SSE) │ │ /dashboard │ │ │ │ /api/* │ │ /ws/* │ │ (HTTP) │ │ /dashboard │ │
│ └────┬─────┘ └────┬─────┘ └─────┬─────┘ └──────────────┘ │ │ └────┬─────┘ └────┬─────┘ └─────┬─────┘ └──────────────┘ │
│ │ │ │ │ │ │ │ │ │
│ ┌────┴──────────────┴──────────────┴────┐ │ │ ┌────┴──────────────┴──────────────┴────┐ │
@@ -46,7 +46,7 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
|-----------|------|----------|---------| |-----------|------|----------|---------|
| REST API | `api/calls.py`, `api/call_flows.py`, `api/devices.py` | HTTP | Call management, CRUD, configuration | | REST API | `api/calls.py`, `api/call_flows.py`, `api/devices.py` | HTTP | Call management, CRUD, configuration |
| WebSocket | `api/websocket.py` | WS | Real-time event streaming to clients | | WebSocket | `api/websocket.py` | WS | Real-time event streaming to clients |
| MCP Server | `mcp_server/server.py` | SSE | AI assistant tool integration | | MCP Server | `mcp_server/server.py` | Streamable HTTP at `/mcp/` | AI assistant tool integration |
### Orchestration Layer ### Orchestration Layer
@@ -75,7 +75,7 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
| Recording | `services/recording.py` | WAV file management and storage | | Recording | `services/recording.py` | WAV file management and storage |
| Analytics | `services/call_analytics.py` | Call metrics, hold time stats, trends | | Analytics | `services/call_analytics.py` | Call metrics, hold time stats, trends |
| Notifications | `services/notification.py` | WebSocket + SMS alerts | | Notifications | `services/notification.py` | WebSocket + SMS alerts |
| Database | `db/database.py` | SQLAlchemy async (PostgreSQL or SQLite) | | Database | `db/database.py` | SQLAlchemy async (PostgreSQL, Alembic migrations) |
## Data Flow — Hold Slayer Call ## Data Flow — Hold Slayer Call

2
docs/configuration.md
View File

@@ -126,7 +126,7 @@ uvicorn main:app --host 0.0.0.0 --port 8000 --reload
### Production ### Production
```bash ```bash
# Use PostgreSQL instead of SQLite # PostgreSQL is required (no SQLite fallback)
DATABASE_URL=postgresql+asyncpg://user:pass@localhost/hold_slayer DATABASE_URL=postgresql+asyncpg://user:pass@localhost/hold_slayer
# Use vLLM for faster inference # Use vLLM for faster inference

2
docs/development.md
View File

@@ -4,7 +4,7 @@
### Prerequisites ### Prerequisites
- Python 3.13+ - Python 3.12+
- Ollama (or any OpenAI-compatible LLM) — for IVR menu analysis - Ollama (or any OpenAI-compatible LLM) — for IVR menu analysis
- Speaches or Whisper API — for speech-to-text (optional for dev) - Speaches or Whisper API — for speech-to-text (optional for dev)
- A SIP trunk account — for making real calls (optional for dev) - A SIP trunk account — for making real calls (optional for dev)

189
docs/mcp-server.md
View File

@@ -1,10 +1,20 @@
# MCP Server # MCP Server
The MCP (Model Context Protocol) server lets any MCP-compatible AI assistant control the Hold Slayer gateway. Built with [FastMCP](https://github.com/jlowin/fastmcp), it exposes tools and resources over SSE. The MCP (Model Context Protocol) server lets any MCP-compatible AI assistant
control the Hold Slayer gateway. Built with [FastMCP](https://github.com/jlowin/fastmcp),
it is mounted on the FastAPI app at **`/mcp/`** (trailing slash) over
**streamable HTTP** and authenticates with the same static bearer token as the
REST API and WebSocket.
## Overview ## Overview
An AI assistant connects via SSE to the MCP server and gains access to tools for placing calls, checking status, sending DTMF, getting transcripts, and managing call flows. The assistant can orchestrate an entire call through natural language. An AI assistant connects to the MCP endpoint and gains access to 15 tools and
3 resources for placing calls, checking status, sending DTMF, getting
transcripts, and managing call flows. The assistant can orchestrate an entire
call through natural language.
`make_call` places a **real PSTN call** that may incur charges; emergency
numbers are always refused, and the concurrent-call cap applies.
## Tools ## Tools
@@ -15,13 +25,32 @@ Place an outbound call through the SIP trunk.
| Param | Type | Required | Description | | Param | Type | Required | Description |
|-------|------|----------|-------------| |-------|------|----------|-------------|
| `number` | string | Yes | Phone number to call (E.164 format) | | `number` | string | Yes | Phone number to call (E.164 format) |
| `mode` | string | No | Call mode: `direct`, `hold_slayer`, `ai_assisted` (default: `hold_slayer`) | | `mode` | string | No | `direct`, `hold_slayer`, or `ai_assisted` (default: `direct`) |
| `intent` | string | No | What you want to accomplish on the call | | `intent` | string | No | What you want to accomplish on the call |
| `call_flow_id` | string | No | ID of a stored call flow to follow | | `call_flow_id` | string | No | ID of a stored call flow to follow |
| `device` | string | No | Device to transfer to when a human is detected |
Returns: Call ID and initial status. Returns: call ID and initial status.
### end_call ### get_call_status
Check the current state of a call — status, duration, hold time, current
audio classification, recent transcript.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to check |
### transfer_call
Transfer an active call to a registered device.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to transfer |
| `device` | string | Yes | Device ID or type to ring |
### hangup
Hang up an active call. Hang up an active call.
@@ -29,102 +58,117 @@ Hang up an active call.
|-------|------|----------|-------------| |-------|------|----------|-------------|
| `call_id` | string | Yes | The call to hang up | | `call_id` | string | Yes | The call to hang up |
### send_dtmf
Send touch-tone digits to an active call (for manual IVR navigation).
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to send digits to |
| `digits` | string | Yes | DTMF digits to send (e.g., "1", "3#", "1234") |
### get_call_status
Check the current state of a call.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to check |
Returns: Status, duration, hold time, audio classification, transcript excerpt.
### get_call_transcript
Get the live transcript of a call.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to get transcript for |
Returns: Array of transcript chunks with timestamps and speaker labels.
### get_call_recording
Get recording metadata and file path for a call.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to get recording for |
Returns: Recording path, duration, file size.
### list_active_calls ### list_active_calls
List all calls currently in progress. No parameters. List all calls currently in progress. No parameters.
Returns: Array of active calls with status, number, duration. ### send_dtmf
Send touch-tone digits on an active call (manual IVR navigation).
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to send digits on |
| `digits` | string | Yes | DTMF digits (e.g., `"1"`, `"123#"`) |
### get_call_transcript
Get the full transcript of an active call.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to get the transcript for |
### get_call_recording
Get recording metadata (path, duration) for a persisted call.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to look up |
### get_call_summary ### get_call_summary
Get analytics summary — hold times, success rates, call volume. No parameters. Stored summary, action items, and sentiment for a persisted call.
Returns: Aggregate statistics across all calls. | Param | Type | Required | Description |
|-------|------|----------|-------------|
| `call_id` | string | Yes | The call to look up |
### search_call_history ### search_call_history
Search past calls by number, company, or date range. Search past call records.
| Param | Type | Required | Description | | Param | Type | Required | Description |
|-------|------|----------|-------------| |-------|------|----------|-------------|
| `query` | string | Yes | Search term (phone number, company name) | | `phone_number` | string | No | Filter by phone number (partial match) |
| `limit` | int | No | Max results (default: 20) | | `intent` | string | No | Filter by intent text (partial match) |
| `limit` | int | No | Max results (default: 10) |
### get_call_flow
Look up the stored IVR call flow for a phone number.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `phone_number` | string | Yes | Number to look up (E.164) |
### create_call_flow
Store a new IVR call flow by hand.
| Param | Type | Required | Description |
|-------|------|----------|-------------|
| `name` | string | Yes | Human-readable name |
| `phone_number` | string | Yes | Phone number (E.164) |
| `steps_json` | string | Yes | JSON array of call flow steps |
| `notes` | string | No | General notes |
### learn_call_flow ### learn_call_flow
Build a reusable call flow from a completed exploration call. Build (or refine) a reusable IVR call flow from a completed hold-slayer
exploration call. Exploration calls record every IVR prompt heard and DTMF
sent; this turns those discoveries into a stored flow so the next call
navigates directly.
| Param | Type | Required | Description | | Param | Type | Required | Description |
|-------|------|----------|-------------| |-------|------|----------|-------------|
| `call_id` | string | Yes | The exploration call to learn from | | `call_id` | string | Yes | A completed call that ran in exploration mode |
| `company` | string | No | Company name for the flow | | `company_name` | string | No | Company name for labeling a new flow |
Returns: The generated CallFlow object. ### list_devices
List registered devices and their online/offline status. No parameters.
### gateway_status
Trunk registration, device count, active calls, engine mode. No parameters.
## Resources ## Resources
MCP resources provide read-only data that assistants can reference:
| Resource URI | Description | | Resource URI | Description |
|-------------|-------------| |-------------|-------------|
| `gateway://status` | Current gateway status — trunk registration, active calls, service health | | `gateway://status` | Current gateway status — trunk registration, active calls |
| `gateway://calls` | List of all active calls with current status | | `gateway://call-flows` | List of all stored call flows |
| `gateway://calls/{call_id}` | Detailed status for a specific call | | `gateway://active-calls` | All active calls with current status |
| `gateway://flows` | List of all stored call flows |
| `gateway://analytics` | Call analytics summary |
## Configuration ## Connecting an AI Assistant
The MCP server is mounted on the FastAPI app at `/mcp`. Claude Code:
### Connecting an AI Assistant ```bash
claude mcp add hold-slayer --transport http http://localhost:8000/mcp/ \
--header "Authorization: Bearer $API_TOKEN"
```
Add to your MCP client configuration (e.g., Claude Desktop, Cline): Generic MCP client configuration:
```json ```json
{ {
"mcpServers": { "mcpServers": {
"hold-slayer": { "hold-slayer": {
"url": "http://localhost:8000/mcp/sse" "url": "http://localhost:8000/mcp/",
"headers": {"Authorization": "Bearer <API_TOKEN>"}
} }
} }
} }
@@ -132,8 +176,6 @@ Add to your MCP client configuration (e.g., Claude Desktop, Cline):
## Example Conversation ## Example Conversation
Here is how an AI assistant would use the MCP tools to handle a complete call:
**User:** "Call Chase Bank and dispute the Amazon charge from December 15th" **User:** "Call Chase Bank and dispute the Amazon charge from December 15th"
**Assistant actions:** **Assistant actions:**
@@ -141,15 +183,8 @@ Here is how an AI assistant would use the MCP tools to handle a complete call:
1. Calls `make_call(number="+18005551234", mode="hold_slayer", intent="dispute Amazon charge Dec 15th", call_flow_id="chase-bank-main")` 1. Calls `make_call(number="+18005551234", mode="hold_slayer", intent="dispute Amazon charge Dec 15th", call_flow_id="chase-bank-main")`
2. Receives `call_id: "call_abc123"` 2. Receives `call_id: "call_abc123"`
3. Polls `get_call_status("call_abc123")` periodically 3. Polls `get_call_status("call_abc123")` periodically
4. Status progression: `trying` → `ringing` → `connected` → `on_hold` 4. Status progression: `initiating` → `ringing` → `connected` → `on_hold`
5. Tells user: "I'm on hold with Chase Bank. Currently 4 minutes in. I'll let you know when someone picks up." 5. Tells user: "I'm on hold with Chase Bank. Currently 4 minutes in. I'll let you know when someone picks up."
6. Status changes to `transferring` — human detected! 6. Status changes to `transferring` — human detected!
7. Tells user: "A live agent just picked up. I'm transferring the call to your desk phone now. Pick up!" 7. Tells user: "A live agent just picked up. I'm transferring the call to your desk phone now. Pick up!"
8. After the call, calls `learn_call_flow("call_abc123", company="Chase Bank")` to save the IVR path for next time. 8. After the call, calls `learn_call_flow("call_abc123", company_name="Chase Bank")` to save the IVR path for next time.
**User:** "How long was I on hold?"
**Assistant actions:**
1. Calls `get_call_summary()`
2. Reports: "Your Chase Bank call lasted 12 minutes total, with 8 minutes on hold. The disputes department averages 6 minutes hold time on Tuesdays."

64
mcp_server/server.py
View File

@@ -264,6 +264,70 @@ def create_mcp_server(
except Exception as e: except Exception as e:
return f"Error creating call flow: {e}" return f"Error creating call flow: {e}"
@mcp.tool()
async def learn_call_flow(call_id: str, company_name: str = "") -> str:
"""
Build (or refine) a reusable IVR call flow from a completed
hold-slayer exploration call.
Exploration calls record every IVR prompt heard and DTMF sent;
this turns those discoveries into a stored call flow so the next
call to that number navigates directly instead of exploring.
If a flow already exists for the number, the discoveries refine
it (timeouts averaged, usage counters updated).
Args:
call_id: A completed call that ran in exploration mode
company_name: Optional company name for labeling a new flow
"""
from db.database import session_scope
from services import call_persistence as store
from services.call_flow_learner import CallFlowLearner
from services.llm_client import get_llm
try:
async with session_scope() as session:
record = await store.get_record(session, call_id)
if not record:
return f"No record found for call {call_id}."
steps = (record.metadata_ or {}).get("exploration_steps") or []
if not steps:
return (
f"Call {call_id} has no exploration data to learn from. "
"Only hold-slayer calls without a stored flow record "
"IVR discoveries."
)
learner = CallFlowLearner(llm_client=get_llm())
existing = await store.get_flow_by_number(
session, record.remote_number
)
if existing:
flow = await learner.merge_discoveries(
store.flow_to_model(existing), steps, intent=record.intent
)
await store.update_flow_from_model(session, existing, flow)
return (
f"Refined existing flow '{existing.name}' for "
f"{record.remote_number} from {len(steps)} discoveries "
f"({len(flow.steps)} steps, used {flow.times_used}x)."
)
flow = await learner.build_flow(
phone_number=record.remote_number,
discovered_steps=steps,
intent=record.intent,
company_name=company_name or None,
)
await store.save_learned_flow(session, flow)
return (
f"Learned new flow '{flow.name}' with {len(flow.steps)} "
f"steps from {len(steps)} discoveries (ID: {flow.id})."
)
except Exception as e:
return f"Error learning call flow: {e}"
@mcp.tool() @mcp.tool()
async def send_dtmf(call_id: str, digits: str) -> str: async def send_dtmf(call_id: str, digits: str) -> str:
""" """

3
models/call.py
View File

@@ -82,6 +82,9 @@ class ActiveCall(BaseModel):
transcript_chunks: list[TranscriptEntry] = Field(default_factory=list) transcript_chunks: list[TranscriptEntry] = Field(default_factory=list)
current_step_id: Optional[str] = None # Current position in call flow current_step_id: Optional[str] = None # Current position in call flow
services: list[str] = Field(default_factory=list) # Active services on this call services: list[str] = Field(default_factory=list) # Active services on this call
# IVR discoveries from hold-slayer exploration mode; persisted with
# the call record so learn_call_flow can build a flow afterwards
exploration_steps: list[dict] = Field(default_factory=list)
@property @property
def duration(self) -> int: def duration(self) -> int:

4
pyproject.toml
View File

@@ -13,7 +13,6 @@ dependencies = [
# Web framework # Web framework
"fastapi>=0.115.0", "fastapi>=0.115.0",
"uvicorn[standard]>=0.32.0", "uvicorn[standard]>=0.32.0",
"websockets>=13.0",
# Database # Database
"sqlalchemy[asyncio]>=2.0.36", "sqlalchemy[asyncio]>=2.0.36",
@@ -29,8 +28,6 @@ dependencies = [
# Audio analysis # Audio analysis
"numpy>=1.26.0", "numpy>=1.26.0",
"librosa>=0.10.0",
"soundfile>=0.12.0",
# HTTP client (for Speaches STT) # HTTP client (for Speaches STT)
"httpx>=0.28.0", "httpx>=0.28.0",
@@ -40,7 +37,6 @@ dependencies = [
# Utilities # Utilities
"python-slugify>=8.0.0", "python-slugify>=8.0.0",
"python-multipart>=0.0.12",
] ]
[project.optional-dependencies] [project.optional-dependencies]

35
services/call_persistence.py
View File

@@ -136,6 +136,36 @@ async def create_flow(
return row return row
async def save_learned_flow(session: AsyncSession, flow: CallFlow) -> StoredCallFlow:
"""The one CallFlow-model → row mapping (auto-learned flows)."""
row = StoredCallFlow(
id=flow.id,
name=flow.name,
phone_number=flow.phone_number,
description=flow.description,
steps=[s.model_dump(mode="json") for s in flow.steps],
tags=flow.tags,
notes=flow.notes,
times_used=flow.times_used,
last_used=flow.last_used,
last_verified=datetime.now(),
)
session.add(row)
await session.flush()
return row
async def update_flow_from_model(
session: AsyncSession, row: StoredCallFlow, flow: CallFlow
) -> None:
"""Write a refined CallFlow back onto its existing row."""
row.steps = [s.model_dump(mode="json") for s in flow.steps]
row.times_used = flow.times_used
row.last_used = flow.last_used
row.notes = flow.notes
await session.flush()
# ================================================================ # ================================================================
# Devices # Devices
# ================================================================ # ================================================================
@@ -320,7 +350,10 @@ async def _finalize_call_record(call: ActiveCall, final_status: CallStatus) -> N
} }
for c in call.classification_history for c in call.classification_history
] ]
record.metadata_ = {"services": list(call.services)} metadata = {"services": list(call.services)}
if call.exploration_steps:
metadata["exploration_steps"] = call.exploration_steps
record.metadata_ = metadata
# Each transcript entry gets its own row with a sequence number # Each transcript entry gets its own row with a sequence number
# and real offset so the dashboard can render click-to-seek. # and real offset so the dashboard can render click-to-seek.

2
services/hold_slayer.py
View File

@@ -294,7 +294,7 @@ class HoldSlayerService:
logger.info(f"🔍 Exploration mode: discovering IVR for {call.remote_number}") logger.info(f"🔍 Exploration mode: discovering IVR for {call.remote_number}")
await self.call_manager.update_status(call.id, CallStatus.NAVIGATING_IVR) await self.call_manager.update_status(call.id, CallStatus.NAVIGATING_IVR)
discovered_steps: list[dict] = [] discovered_steps = call.exploration_steps # persisted with the record
max_time = self.settings.hold_slayer.max_hold_time max_time = self.settings.hold_slayer.max_hold_time
start_time = time.time() start_time = time.time()

137
tests/test_learner.py Normal file
View File

@@ -0,0 +1,137 @@
"""
Call-flow learner tests.
Exploration discoveries become a linked CallFlow, survive with the
persisted call record, and the learn_call_flow MCP tool turns them
into a stored flow (refining on subsequent calls).
"""
import pytest
from fastmcp import Client
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool
import db.database as dbmod
import services.llm_client as llm_mod
from core.call_manager import CallManager
from core.event_bus import EventBus
from db.database import Base
from mcp_server.server import create_mcp_server
from models.call import CallStatus
from models.call_flow import ActionType
from services import call_persistence as store
from services.call_flow_learner import CallFlowLearner
DISCOVERIES = [
{"timestamp": 1.0, "audio_type": "ringing", "confidence": 0.9,
"transcript": "", "action_taken": None},
{"timestamp": 4.0, "audio_type": "ivr_prompt", "confidence": 0.8,
"transcript": "press 1 for english press 2 for french",
"action_taken": {"dtmf": "1"}},
{"timestamp": 8.0, "audio_type": "ivr_prompt", "confidence": 0.8,
"transcript": "press 1 for billing press 2 for support press 0 for an agent",
"action_taken": {"dtmf": "0"}},
{"timestamp": 12.0, "audio_type": "music", "confidence": 0.9,
"transcript": "", "action_taken": None},
{"timestamp": 200.0, "audio_type": "live_human", "confidence": 0.85,
"transcript": "thank you for holding, how can I help",
"action_taken": None},
]
class TestBuildFlow:
async def test_discoveries_become_linked_steps(self):
learner = CallFlowLearner(llm_client=None)
flow = await learner.build_flow(
phone_number="+18005551234",
discovered_steps=DISCOVERIES,
intent="dispute a charge",
)
# ringing is skipped; menus/hold/human map to actions in order
assert [s.action for s in flow.steps] == [
ActionType.DTMF, ActionType.DTMF, ActionType.HOLD, ActionType.TRANSFER,
]
assert [s.action_value for s in flow.steps[:2]] == ["1", "0"]
assert [s.next_step for s in flow.steps[:-1]] == [s.id for s in flow.steps[1:]]
assert "auto-learned" in flow.tags
assert flow.phone_number == "+18005551234"
class TestExplorationPersistence:
async def test_exploration_steps_survive_with_the_record(self, mem_db):
cm = CallManager(EventBus(), on_call_ended=store.persist_call_on_end)
call = await cm.create_call("+18005551234", intent="dispute a charge")
call.exploration_steps.extend(DISCOVERIES)
await cm.end_call(call.id, CallStatus.COMPLETED)
async with mem_db() as session:
record = await store.get_record(session, call.id)
assert record.metadata_["exploration_steps"] == DISCOVERIES
@pytest.fixture
async def mem_db(monkeypatch):
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
factory = async_sessionmaker(engine, expire_on_commit=False)
monkeypatch.setattr(dbmod, "_engine", engine)
monkeypatch.setattr(dbmod, "_session_factory", factory)
yield factory
await engine.dispose()
@pytest.fixture
def no_llm(monkeypatch):
"""learn_call_flow must work without an LLM (labels stay heuristic)."""
monkeypatch.setattr(llm_mod, "_shared_client", None)
monkeypatch.setattr(llm_mod, "_shared_failed", True)
class TestLearnCallFlowTool:
async def _completed_exploration_call(self, number: str) -> str:
cm = CallManager(EventBus(), on_call_ended=store.persist_call_on_end)
call = await cm.create_call(number, intent="dispute a charge")
call.exploration_steps.extend(DISCOVERIES)
await cm.end_call(call.id, CallStatus.COMPLETED)
return call.id
async def test_learns_then_refines(self, mem_db, no_llm):
call_id = await self._completed_exploration_call("+18005551234")
mcp = create_mcp_server(lambda: None)
async with Client(mcp) as client:
result = await client.call_tool("learn_call_flow", {"call_id": call_id})
assert "Learned new flow" in result.content[0].text
async with mem_db() as session:
row = await store.get_flow_by_number(session, "+18005551234")
assert row is not None
assert len(row.steps) == 4
assert "auto-learned" in row.tags
result = await client.call_tool("learn_call_flow", {"call_id": call_id})
assert "Refined existing flow" in result.content[0].text
async def test_call_without_exploration_data(self, mem_db, no_llm):
cm = CallManager(EventBus(), on_call_ended=store.persist_call_on_end)
call = await cm.create_call("+15550001111")
await cm.end_call(call.id, CallStatus.COMPLETED)
mcp = create_mcp_server(lambda: None)
async with Client(mcp) as client:
result = await client.call_tool("learn_call_flow", {"call_id": call.id})
assert "no exploration data" in result.content[0].text
async def test_unknown_call(self, mem_db, no_llm):
mcp = create_mcp_server(lambda: None)
async with Client(mcp) as client:
result = await client.call_tool(
"learn_call_flow", {"call_id": "call_nope"}
)
assert "No record found" in result.content[0].text

1
tests/test_mcp.py
View File

@@ -25,6 +25,7 @@ EXPECTED_TOOLS = {
"get_call_recording", "get_call_recording",
"get_call_summary", "get_call_summary",
"search_call_history", "search_call_history",
"learn_call_flow",
"list_devices", "list_devices",
"gateway_status", "gateway_status",
} }

67
tests/test_websocket.py Normal file
View File

@@ -0,0 +1,67 @@
"""
WebSocket event-stream tests.
The socket is refused (4401) without the bearer token, and an
authorized client immediately receives the synthetic trunk-status
event followed by the replayed recent history.
"""
import asyncio
import pytest
from pydantic import SecretStr
from starlette.testclient import TestClient
from starlette.websockets import WebSocketDisconnect
import main
from config import Settings, get_settings
from core.gateway import AIPSTNGateway
from models.events import EventType, GatewayEvent
@pytest.fixture
def ws_app(monkeypatch):
monkeypatch.setattr(get_settings(), "api_token", SecretStr("tok"))
gateway = AIPSTNGateway(settings=Settings())
main.app.state.gateway = gateway
yield gateway
del main.app.state.gateway
def _publish(gateway, call_id: str) -> None:
asyncio.run(gateway.event_bus.publish(GatewayEvent(
type=EventType.CALL_INITIATED,
call_id=call_id,
data={},
message=f"call {call_id}",
)))
class TestEventStream:
def test_refused_without_token(self, ws_app):
client = TestClient(main.app)
with pytest.raises(WebSocketDisconnect) as exc:
with client.websocket_connect("/ws/events"):
pass
assert exc.value.code == 4401
def test_trunk_status_then_replayed_history(self, ws_app):
_publish(ws_app, "call_ws1")
_publish(ws_app, "call_ws2")
client = TestClient(main.app)
with client.websocket_connect("/ws/events?token=tok") as ws:
first = ws.receive_json()
assert first["type"] == EventType.SIP_TRUNK_REGISTRATION_FAILED.value
replayed = [ws.receive_json() for _ in range(2)]
assert [m["call_id"] for m in replayed] == ["call_ws1", "call_ws2"]
def test_per_call_stream_filters(self, ws_app):
client = TestClient(main.app)
with client.websocket_connect(
"/ws/calls/call_target/events?token=tok"
) as ws:
_publish(ws_app, "call_other")
_publish(ws_app, "call_target")
msg = ws.receive_json()
assert msg["call_id"] == "call_target"