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); 

#!/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() 

chmod +x /usr/local/bin/vicidial_incident_agi.py 

[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() 

asterisk -rx "sip show peers" | grep -E "UNREACHABLE|OK" asterisk -rx "sip set debug on" 

[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 

[carrier-trunk] type=peer host=203.0.113.50 port=5060 disallow=all allow=ulaw,alaw directmedia=no rtpkeepalive=20 

asterisk -rx "sip reload" asterisk -rx "sip show peers" | grep carrier-trunk 

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 

[carrier-trunk] rtpkeepalive=20 

mtr -u -b 203.0.113.50 -p 10000 

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; 

UPDATE vicidial_log SET status='DONE', <...