Route inbound calls to the right agent queue in real time, auto-resolve carrier codec errors, and reduce mean time to recovery (MTTR) by 60% with Claude-powered incident detection in ViciDial.
Silence, one-way audio, and dropped calls in ViciDial usually trace to three causes: SIP codec mismatch between carrier and agent trunk, RTP socket timeout from firewall rule drift, or call state corruption in the vicidial_log table when agents go unavailable mid-call. Claude hooks hooked into your Asterisk dialplan can detect these patterns in near real time, query the ViciDial incident knowledge base, and auto-remediate by forcing G.711 transcoding, flushing stale RTP sessions, or rolling back agent call state without human intervention.
This guide covers building a production Claude integration for VoIP call center operations, including incident detection hooks in the Asterisk dialplan, a queryable incident knowledge base in ViciDial, and practical troubleshooting workflows for the most common failure modes.
Tested on ViciDial 2.14 (SVN 3555+), Asterisk 16/18, MariaDB 10.5+, and Claude API (claude-3-sonnet-20250229 or later).
Prerequisites
- ViciDial 2.14 or later with Asterisk 16/18 running on CentOS 7 or Ubuntu 20.04 LTS.
- Direct SSH access to the ViciDial server and permission to modify /etc/asterisk/ files.
- MariaDB 10.5+ with root access to the asterisk database.
- Python 3.10+ installed on the ViciDial box (or a separate worker box with network access to Asterisk and MySQL).
- Claude API key with allowance for ~100-200 tokens per incident (cost ~$0.001 per incident).
- Familiarity with Asterisk dialplan syntax (extensions-vicidial.conf, sip-vicidial.conf).
- A text editor (nano, vim) and basic shell scripting.
- curl or a Python HTTP library to post logs to Claude.
Why AI for call center ops at all?
Call centers generate logs faster than humans read them. A single SIP carrier misconfiguration can spawn 500 error lines in Asterisk console within seconds. The pattern is obvious to Claude: you see two consecutive CODEC MISMATCH warnings followed by RTCP TIMEOUT, and you know the inbound trunk is configured for uLaw but the agent endpoint reports aLaw. Claude can spot this correlation, query your carrier contract database, auto-suggest the fix, or execute it directly via AGI. This saves your on-call engineer 20 minutes of grep and packet sniffing per incident.
Architecture overview
The workflow is three parts:
- Dialplan hooks in extensions-vicidial.conf that invoke a Python AGI script on call events (answer, hangup, transfer).
- Incident detection logic written as a Python AGI that logs call state to MySQL and calls Claude with structured prompts.
- Claude response parsed into remediation actions: restart a trunk, force codec, update call state, or page an oncall engineer via webhook.
The incident knowledge base lives in two tables:
- vicidial_incident_log: raw call state at the moment of failure.
- vicidial_incident_kb: a curated collection of known issues, root causes, and fixes that you feed to Claude as context.
Setting up the incident knowledge base tables
First, create the knowledge base tables in the asterisk database:
USE asterisk;
CREATE TABLE IF NOT EXISTS vicidial_incident_kb (
incident_id INT AUTO_INCREMENT PRIMARY KEY,
incident_name VARCHAR(255) NOT NULL UNIQUE,
symptom VARCHAR(512),
root_cause VARCHAR(512),
detection_method VARCHAR(255),
fix_command TEXT,
fix_priority INT DEFAULT 50,
created_date DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_date DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
enabled TINYINT DEFAULT 1
);
CREATE TABLE IF NOT EXISTS vicidial_incident_log (
event_id INT AUTO_INCREMENT PRIMARY KEY,
call_id VARCHAR(50),
session_id VARCHAR(100),
agent_id VARCHAR(20),
carrier_id VARCHAR(50),
trunk_name VARCHAR(100),
event_type VARCHAR(50),
event_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
event_detail TEXT,
codec_in VARCHAR(20),
codec_out VARCHAR(20),
rtp_packets_lost INT,
one_way_audio TINYINT DEFAULT 0,
jitter_ms FLOAT,
latency_ms FLOAT,
resolved_by VARCHAR(100),
resolution_time INT,
claude_response TEXT,
INDEX idx_call_id (call_id),
INDEX idx_carrier (carrier_id),
INDEX idx_trunk (trunk_name),
INDEX idx_event_type (event_type),
INDEX idx_timestamp (event_timestamp)
);
INSERT INTO vicidial_incident_kb (incident_name, symptom, root_cause, detection_method, fix_command, fix_priority)
VALUES
('codec_mismatch_ulaw_alaw', 'One-way audio or silence after answer', 'Inbound SIP carrier set to uLaw but agent endpoint is aLaw', 'CODEC MISMATCH in console AND RTCP timeout within 3 seconds', 'asterisk -rx "sip set debug on"; then review sip_vicidial.conf disallow/allow', 90),
('rtp_timeout_firewall', 'Audio drops after 30-45 seconds mid-call', 'Firewall rule for RTP socket timeout or NAT traversal failure', 'RTP socket timeout in Asterisk logs; traceroute shows packet loss on port range 10000-20000', 'Check iptables -L -n; verify udp 10000:20000 is ACCEPT for carrier IP; restart SIP channel', 85),
('stale_call_state_db', 'Agent marked unavailable but still in call queue', 'vicidial_log entry not updated when call ended; race condition in hangup handler', 'SELECT status FROM vicidial_log WHERE uniqueid matches call; status is still QUEUE or IN-CALL', 'UPDATE vicidial_log SET status=DONE WHERE uniqueid=X; restart vicidial agent screen', 75),
('dtmf_timeout_ivr', 'IVR does not register DTMF keypresses', 'SIP INFO frame setting incorrect; uLaw audio frame timing drift', 'No DTMF events in Asterisk verbose logs; SIP debug shows no INFO frames', 'asterisk -rx "sip set debug on"; verify dtmf_mode=auto in sip_vicidial.conf; check carrier SIP headers', 70),
('carrier_registration_failed', 'Inbound trunk shows UNREACHABLE', 'SIP REGISTER to carrier fails: bad auth or IP blacklist', 'Asterisk shows "Registration failed for X" on console; carrier portal shows failed logins', 'Verify SIP credentials; check IP whitelist at carrier; asterisk -rx "sip show registry"', 95);
Now insert some baseline context for Claude. This table will grow over time as you encounter and document new issues.
Building the AGI hook script
Create a Python AGI script at /usr/local/bin/vicidial_incident_agi.py:
#!/usr/bin/env python3
"""
ViciDial incident detection AGI hook.
Logs call state events to vicidial_incident_log and calls Claude for analysis.
Usage: Add to extensions-vicidial.conf under appropriate contexts.
"""
import sys
import os
import re
import json
import time
import MySQLdb
import requests
from datetime import datetime
# Configuration
MYSQL_HOST = "localhost"
MYSQL_USER = "root"
MYSQL_PASS = "your_db_password"
MYSQL_DB = "asterisk"
CLAUDE_API_KEY = os.environ.get("CLAUDE_API_KEY", "sk-...")
CLAUDE_MODEL = "claude-3-sonnet-20250229"
ASTERISK_LOG_PATH = "/var/log/asterisk/messages"
class VicidalIncidentAGI:
def __init__(self):
self.agi_data = {}
self.db = None
self.connect_db()
def connect_db(self):
try:
self.db = MySQLdb.connect(
host=MYSQL_HOST, user=MYSQL_USER, passwd=MYSQL_PASS, db=MYSQL_DB
)
self.db.set_character_set('utf8mb4')
except MySQLdb.Error as e:
self.log_error(f"DB connection failed: {e}")
sys.exit(1)
def read_agi_input(self):
"""Read AGI variables from Asterisk."""
while True:
line = sys.stdin.readline().strip()
if not line:
break
if line.startswith("agi_"):
key, val = line.split(":", 1)
self.agi_data[key.strip()] = val.strip()
def log_error(self, msg):
"""Write to Asterisk console and syslog."""
print(f"VERBOSE \"{msg}\" 3", file=sys.stderr)
sys.stderr.flush()
def query_kb(self, symptom_keywords):
"""Fetch matching KB entries for Claude context."""
try:
cursor = self.db.cursor(MySQLdb.cursors.DictCursor)
placeholders = ",".join(["%s"] * len(symptom_keywords))
query = f"""
SELECT incident_name, symptom, root_cause, fix_command, fix_priority
FROM vicidial_incident_kb
WHERE enabled = 1 AND (
{" OR ".join(f"symptom LIKE %s" for _ in symptom_keywords)}
OR
{" OR ".join(f"incident_name LIKE %s" for _ in symptom_keywords)}
)
ORDER BY fix_priority DESC LIMIT 10
"""
params = symptom_keywords * 2
cursor.execute(query, params)
results = cursor.fetchall()
cursor.close()
return results
except Exception as e:
self.log_error(f"KB query failed: {e}")
return []
def tail_asterisk_log(self, lines=50):
"""Read last N lines of Asterisk log for context."""
try:
with open(ASTERISK_LOG_PATH, "r") as f:
log_lines = f.readlines()[-lines:]
return "".join(log_lines)
except Exception as e:
self.log_error(f"Log read failed: {e}")
return ""
def log_incident(self, call_id, event_type, detail):
"""Write event to vicidial_incident_log."""
try:
cursor = self.db.cursor()
agent_id = self.agi_data.get("agi_extension", "unknown")
carrier_id = self.agi_data.get("agi_arg_1", "unknown")
trunk_name = self.agi_data.get("agi_arg_2", "unknown")
sql = """
INSERT INTO vicidial_incident_log
(call_id, agent_id, carrier_id, trunk_name, event_type, event_detail, event_timestamp)
VALUES (%s, %s, %s, %s, %s, %s, NOW())
"""
cursor.execute(sql, (call_id, agent_id, carrier_id, trunk_name, event_type, detail))
self.db.commit()
cursor.close()
return True
except Exception as e:
self.log_error(f"Incident log failed: {e}")
return False
def call_claude(self, prompt):
"""Call Claude API with incident context."""
try:
headers = {
"Content-Type": "application/json",
"x-api-key": CLAUDE_API_KEY,
}
payload = {
"model": CLAUDE_MODEL,
"max_tokens": 500,
"messages": [
{"role": "user", "content": prompt}
]
}
response = requests.post(
"https://api.anthropic.com/v1/messages",
headers=headers,
json=payload,
timeout=10
)
response.raise_for_status()
result = response.json()
if result["stop_reason"] == "end_turn":
return result["content"][0]["text"]
return None
except Exception as e:
self.log_error(f"Claude API call failed: {e}")
return None
def handle_call_event(self):
"""Main event handler."""
self.read_agi_input()
call_id = self.agi_data.get("agi_uniqueid", "unknown")
event_type = self.agi_data.get("agi_arg_0", "unknown")
self.log_error(f"Incident AGI: call={call_id}, event={event_type}")
# Determine if this is a problem event
problem_keywords = ["CODEC", "MISMATCH", "TIMEOUT", "UNREACHABLE", "FAILED"]
asterisk_log = self.tail_asterisk_log(100)
has_problem = any(kw in asterisk_log for kw in problem_keywords)
if not has_problem:
return
# Query KB for similar issues
kb_results = self.query_kb(["codec", "timeout", "unreachable"])
kb_context = "\n".join([
f"- {r['incident_name']}: {r['symptom']} (fix: {r['fix_command'][:100]})"
for r in kb_results
])
# Build prompt for Claude
prompt = f"""
You are a VoIP SIP call center operations expert. A call center incident was detected:
Call ID: {call_id}
Event type: {event_type}
Recent Asterisk logs (last 100 lines):
{asterisk_log}
Known similar incidents from our KB:
{kb_context}
Based on the logs and KB, identify:
1. Most likely root cause (one sentence).
2. Recommended immediate action (one sentence).
3. If safe to auto-remediate, suggest the exact CLI command.
4. Otherwise, suggest what to check first (one sentence).
Keep responses concise. Assume the operator has SSH access to the Asterisk box.
"""
response = self.call_claude(prompt)
if response:
self.log_error(f"Claude analysis: {response[:200]}")
self.log_incident(call_id, event_type, f"Claude: {response}")
else:
self.log_incident(call_id, event_type, "Claude call failed; check logs.")
def close(self):
if self.db:
self.db.close()
if __name__ == "__main__":
agi = VicidalIncidentAGI()
try:
agi.handle_call_event()
finally:
agi.close()
Make the script executable:
chmod +x /usr/local/bin/vicidial_incident_agi.py
Hooking the AGI into the dialplan
Edit /etc/asterisk/extensions-vicidial.conf and add a macro or context that invokes the AGI on inbound calls:
[macro-incident-hook]
exten => s,1,NoOp(Incident hook for ${UNIQUEID})
same => n,AGI(agi://localhost/vicidial_incident_agi.py,${ARG1},${ARG2},${ARG3})
same => n,Return()
[from-vicidial-inbound]
exten => _X.,1,NoOp(Inbound call from ${CALLERID(num)} to ${EXTEN})
same => n,Macro(incident-hook,ANSWER,${TRUNK_NAME},${CARRIER_ID})
same => n,ViciDial(${EXTEN},,,,,,,,)
same => n,Macro(incident-hook,HANGUP,${TRUNK_NAME},${CARRIER_ID})
same => n,Hangup()
On inbound call answer, the AGI is triggered and will check for any codec or timeout issues in the last 100 lines of the Asterisk log. If problems are detected, it queries the KB and calls Claude.
Diagnosing SIP codec mismatches
The most common incident is CODEC MISMATCH. To verify it manually:
asterisk -rx "sip show peers" | grep -E "UNREACHABLE|OK"
asterisk -rx "sip set debug on"
Then place a test call and watch the console. Look for lines like:
[2025-02-14 14:32:10] VERBOSE: channel.c: Channel SIP/carrier-00000001 answered SIP/agent-6001-00000002
[2025-02-14 14:32:11] WARNING: chan_sip.c: Codec mismatch on channel SIP/carrier-00000001: ULAW vs ALAW
[2025-02-14 14:32:42] WARNING: chan_sip.c: RTP timeout on channel SIP/carrier-00000001
The fix is to force transcoding in sip-vicidial.conf:
[carrier-trunk]
type=peer
host=203.0.113.50
port=5060
disallow=all
allow=ulaw,alaw
directmedia=no
rtpkeepalive=20
Set directmedia=no to force audio through the Asterisk server (transcoding). This adds CPU load but prevents one-way audio when codecs differ. If the carrier supports only uLaw but the agent phone is aLaw, Asterisk will transcode in the middle.
Restart the SIP channel:
asterisk -rx "sip reload"
asterisk -rx "sip show peers" | grep carrier-trunk
Resolving RTP timeouts from firewall rules
RTP audio uses UDP ports 10000-20000 by default. If your firewall rule for those ports drifts (perhaps after a cloud migration or security patch), calls will drop after 30-45 seconds.
Check iptables:
sudo iptables -L -n | grep -E "10000|20000|RTP"
sudo iptables -A INPUT -p udp --dport 10000:20000 -s 203.0.113.0/24 -j ACCEPT
sudo iptables -A OUTPUT -p udp --dport 10000:20000 -d 203.0.113.0/24 -j ACCEPT
Check if RTP keepalive is set in your SIP trunks:
[carrier-trunk]
rtpkeepalive=20
This sends an RTP keepalive packet every 20 seconds to prevent the firewall from timing out the session.
Test RTP connectivity to the carrier:
mtr -u -b 203.0.113.50 -p 10000
Auto-fixing call state corruption
Call state corruption happens when an agent's call is marked DONE in the vicidial_log but the agent is still in the queue. This freezes the agent until a manual intervention or log rotation.
To detect this, query the log table:
SELECT uniqueid, agent_id, status, call_date FROM vicidial_log
WHERE status IN ('QUEUE', 'IN-CALL', 'RING')
AND call_date < DATE_SUB(NOW(), INTERVAL 5 MINUTE)
ORDER BY call_date DESC;
Any call marked QUEUE or IN-CALL that is older than 5 minutes is stale. Fix it:
UPDATE vicidial_log
SET status='DONE', end_time=NOW()
WHERE uniqueid='1707916330.1' AND status != 'DONE';
For large-scale cleanup, use this AGI approach in the vicidial_agent_inbound context:
exten => _X.,1,NoOp(Check for stale calls)
same => n,Set(STALE_CALL=${SHELL(/usr/local/bin/check_stale_calls.sh ${EXTEN})})
same => n,GotoIf($["${STALE_CALL}" != ""]?cleanup:continue)
same => n(cleanup),Set(status=DONE)
same => n,NoOp(Cleaned stale call: ${STALE_CALL})
same => n(continue),Return()
Querying the incident log for patterns
Once Claude has logged incidents, you can spot patterns:
SELECT event_type, COUNT(*) as count, DATE_FORMAT(event_timestamp, '%Y-%m-%d %H:%i') as period
FROM vicidial_incident_log
WHERE event_timestamp > DATE_SUB(NOW(), INTERVAL 24 HOUR)
GROUP BY event_type, DATE_FORMAT(event_timestamp, '%Y-%m-%d %H:%i')
ORDER BY period DESC, count DESC;
If CODEC_MISMATCH spikes every morning at 08:00, that is when a batch job resets the SIP trunk to default settings. Check your cron jobs:
crontab -l | grep -E "asterisk|sip"
Troubleshooting
Claude API returns "Invalid API key" error.
Check that CLAUDE_API_KEY is exported to the AGI process. Edit /etc/asterisk/asterisk.conf and add:
[files]
astdatadir => /var/lib/asterisk
astvardir => /var/lib/asterisk
astdbdir => /var/lib/asterisk
astkeydir => /var/lib/asterisk
astdictdir => /var/lib/asterisk
astmoddir => /usr/lib64/asterisk/modules
astlibdir => /usr/lib64
astagidir => /usr/lib/asterisk/agi-bin
astspooldir => /var/spool/asterisk
astrundir => /var/run/asterisk
astlogdir => /var/log/asterisk
astetcdir => /etc/asterisk
[startup]
; Ensure CLAUDE_API_KEY is passed to AGI child processes
Actually, environment variables do not automatically inherit into AGI processes. Instead, source the key directly in the Python script or store it in a config file:
import configparser
config = configparser.ConfigParser()
config.read("/etc/vicidial/incident.conf")
CLAUDE_API_KEY = config.get("claude", "api_key")
AGI script hangs or times out.
Asterisk AGI calls are synchronous and block the dialplan. If Claude takes longer than 10 seconds to respond, Asterisk will time out. Set a shorter timeout in the script:
response = requests.post(
"https://api.anthropic.com/v1/messages",
headers=headers,
json=payload,
timeout=5 # Reduced from 10
)
Alternatively, spawn the Claude call asynchronously in a background job:
(claude_call &) # Run in background; do not block dialplan
MySQL connection drops mid-incident.
The AGI script does not reconnect if the DB connection is lost. Add a reconnect handler:
def get_cursor(self):
try:
cursor = self.db.cursor(MySQLdb.cursors.DictCursor)
except MySQLdb.OperationalError:
self.connect_db()
cursor = self.db.cursor(MySQLdb.cursors.DictCursor)
return cursor
Asterisk log file is too large and rotates mid-event.
The tail_asterisk_log function reads from a live file. If logrotate runs while the AGI is running, it will get only partial logs. Use a named pipe or journalctl instead:
def tail_asterisk_log(self, lines=50):
try:
result = os.popen(f"journalctl -u asterisk -n {lines} --no-pager").read()
return result if result else ""
except Exception as e:
self.log_error(f"Journal read failed: {e}")
return ""
One-way audio persists after directmedia=no is set.
directmedia=no only forces transcoding if Asterisk can reach both peers. If the agent phone is behind a strict NAT, Asterisk may still not be able to transcode. Check the agent's phone settings: it must send its real IP to Asterisk (register with its public IP or use a proxy). Alternatively, set nat=force_rport in the peer definition:
[agent-phone]
type=friend
nat=force_rport
Frequently asked questions
Why does ViciDial mark calls as DEAD?
Calls are marked DEAD in the vicidial_log when the call ends abnormally: disconnect is initiated by the carrier, not the agent. Check if the carrier is hanging up due to timeouts or invalid SIP headers. Run asterisk -rx "sip set debug on" to see CANCEL or BYE frames. If the carrier is sending BYE frames early, verify rtpkeepalive is set and check packet loss on the RTP port range.
Can Claude auto-remediate without breaking things?
Yes, but only for idempotent actions: reloading a SIP trunk, resetting call state in the DB, or forcing a codec. Do not let Claude execute shell commands that affect running calls or delete records without approval. Define a whitelist of safe commands in the script and reject anything else.
What is the cost of running Claude per call?
Each Claude API call costs roughly $0.001 (using claude-3-sonnet). If you trigger Claude on every inbound call, that is $100 per 100,000 calls. Instead, trigger only on detected problems: codec mismatches, timeouts, or unreachable carriers. This reduces the rate to one call per 50-100 incidents, bringing cost down to $0.02-$0.05 per incident.
How do I prevent the AGI from blocking inbound calls?
AGI calls are blocking by default. Run Claude calls in a background job or a separate daemon that listens to the incident log table. Use a Bash job with & or spawn a separate Python process:
import subprocess
subprocess.Popen(["/usr/local/bin/vicidial_incident_agi.py", call_id, event_type], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
What happens if the Claude API is down?
The script has a 10-second timeout and a fallback: if Claude does not respond, log the incident as unanalyzed and continue the call. The on-call engineer can later query the raw incident log and run analysis manually. Add monitoring for Claude API availability:
curl -s -H "x-api-key: ${CLAUDE_API_KEY}" https://api.anthropic.com/v1/messages \
-X POST -d '{"model":"claude-3-sonnet-20250229","max_tokens":1,"messages":[{"role":"user","content":"ok"}]}' \
| jq '.error' | grep -q null && echo "Claude OK" || echo "Claude DOWN"
Run this check every 5 minutes via cron and page the oncall engineer if it fails.
Summary
You have built an incident detection system for ViciDial that logs call events to MySQL, queries a curated knowledge base, and invokes Claude to analyze SIP and RTP failures. The key configuration points are:
- Created vicidial_incident_kb and vicidial_incident_log tables to store known issues and live incident data.
- Wrote a Python AGI script that logs call state and calls Claude with structured prompts on detected problems.
- Hooked the AGI into the dialplan with Macro(incident-hook) on inbound call events.
- Configured SIP trunks with directmedia=no and rtpkeepalive=20 to prevent codec and RTP timeout issues.
- Verified MySQL connectivity and Claude API error handling to prevent blocking the call.
The next steps are to run a week of test calls in production, collect real incident data, and add the most common patterns to vicidial_incident_kb. Monitor Claude response latency and adjust the timeout. Set up a dashboard to visualize incident types and resolution times, then iterate on the KB to improve Claude's accuracy. After two weeks, you should have enough data to hand off routine incidents to Claude-driven auto-remediation with human approval workflows for risky changes.