AMDY.ai Integration with Vicidial

Outbound call centers need efficiency, precision, and scale. By integrating AMDY.ai’s AI-powered Answering Machine Detection (AMD) with Vicidial, your agents can focus on real conversations while the system automatically handles voicemails. This integration ensures smarter call handling, higher productivity, and cost savings.

Why Integrate AMDY.ai with Vicidial?

• Real-time AI detection of humans vs. voicemail.
• Improved agent utilization by skipping unanswered or machine-answered calls.
• Seamless compatibility with Vicidial, Asterisk, and FreeSWITCH.
• Flexible configuration to adapt to your existing campaigns.

Integrating with Dialers

Below are the steps to integrate AMDY.ai with Vicidial / Asterisk (EAGI).

1. Update the Dialplan

Edit your extensions.conf file to add the AMD transfer script:

; VICIDIAL_auto_dialer transfer script AMD (load-balanced)
exten => 8370,1,AGI(agi://127.0.0.1:4577/call_log)
exten => 8370,n,Playback(sip-silence)
exten => 8370,n,EAGI(/var/lib/asterisk/agi-bin/amd.py)
exten => 8370,n,AGI(VD_amd.agi,${EXTEN})
exten => 8370,n,AGI(agi-VDAD_ALL_outbound.agi,NORMAL-----LB-----${CONNECTEDLINE(name)})
exten => 8370,n,Hangup()

2. Install Dependencies

Run the following on your dialer server:

apt-get install -y python3-pip
pip3 install websocket-client asterisk-agi
install -m 0755 amd.py /var/lib/asterisk/agi-bin/amd.py
chown asterisk:asterisk /var/lib/asterisk/agi-bin/amd.py

3. Configure the amd.py Script

The amd.py script handles audio streaming from the dialer to the AMD server, parses responses, and sets decision variables.

#!/usr/bin/env python3
#/var/lib/asterisk/agi-bin/amd.py

import os, fcntl, json, time
from websocket import create_connection
from asterisk.agi import AGI

AUDIO_FD = 3
WS_URL   = os.getenv("AMD_WS_URL", "ws://amdy.ai:8080/ws/amd")
SAMPLE_RATE = int(os.getenv("AMD_SAMPLE_RATE", "8000"))      # EAGI audio usually 8kHz
EARLY_MIN_CONF = float(os.getenv("AMD_EARLY_MIN_CONF", "0.90"))

def set_human(agi, cause="HUMAN", stats=None):
    agi.set_variable("AMDSTATUS", "HUMAN")
    agi.set_variable("AMDCAUSE",  cause)
    if stats: agi.set_variable("AMDSTATS", stats)

def set_machine(agi, cause="MACHINE", stats=None):
    agi.set_variable("AMDSTATUS", "AMD")
    agi.set_variable("AMDCAUSE",  cause)
    if stats: agi.set_variable("AMDSTATS", stats)

def should_stop_on(msg):
    if not isinstance(msg, dict): return None
    typ   = str(msg.get("type", "")).lower()
    label = str(msg.get("label", "")).lower()
    conf  = float(msg.get("confidence", 0.0) or 0.0)
    if typ == "final":
        return label
    if typ == "early" and conf >= EARLY_MIN_CONF:
        return label
    return None

def process_json_and_set_vars(agi, msg):
    label = str(msg.get("label", "")).upper()  # HUMAN/MACHINE
    conf  = msg.get("confidence", "")
    ph    = msg.get("proba_human", "")
    tr    = msg.get("transcript", "")
    typ   = msg.get("type", "")
    cause = f"{typ}|p_human={ph}|conf={conf}"
    stats = f"{label}|{cause}|{tr[:120]}"
    (set_human if label == "HUMAN" else set_machine if label == "MACHINE" else set_human)(
        agi, cause if label in ("HUMAN","MACHINE") else "UNKNOWN", stats
    )

def startAGI():
    agi = AGI()
    try:
        ws = create_connection(WS_URL, timeout=5)
        ws.send(json.dumps({"config": {"sample_rate": SAMPLE_RATE}}))
        agi.verbose(f"AMD: WS connected {WS_URL} (sr={SAMPLE_RATE})")
    except Exception as e:
        agi.verbose(f"AMD: WS connect error {e} → default HUMAN")
        set_human(agi, cause="NETERR"); return

    fcntl.fcntl(AUDIO_FD, fcntl.F_SETFL, os.O_NONBLOCK)
    last_msg = None
    total_bytes = 0
    buf = b""
    start = time.time()

    try:
        while True:
            try:
                time.sleep(0.2)
                chunk = os.read(AUDIO_FD, 9500)
                if not chunk:
                    ws.send(json.dumps({"type":"flush"}))
                    for _ in range(5):
                        try:
                            res = ws.recv()
                            msg = json.loads(res) if isinstance(res, str) else {}
                            last_msg = msg or last_msg
                            stop = should_stop_on(msg)
                            if stop:
                                process_json_and_set_vars(agi, msg); return
                        except Exception:
                            break
                    if last_msg: process_json_and_set_vars(agi, last_msg)
                    else: set_human(agi, cause="NOAUDIO")
                    return

                buf += chunk
                if len(buf) >= 6400:   # ~0.4s at 8kHz
                    ws.send_binary(buf)
                    total_bytes += len(buf)
                    buf = b""
                    try:
                        ws.settimeout(0.01)
                        res = ws.recv()
                        msg = json.loads(res) if isinstance(res, str) else {}
                        last_msg = msg or last_msg
                        stop = should_stop_on(msg)
                        if stop:
                            process_json_and_set_vars(agi, msg); return
                    except Exception:
                        pass
                    finally:
                        ws.settimeout(5)

                if (time.time() - start) > 5 and total_bytes == 0:
                    ws.send(json.dumps({"type":"flush"}))
                    try:
                        res = ws.recv()
                        msg = json.loads(res) if isinstance(res, str) else {}
                        process_json_and_set_vars(agi, msg)
                    except Exception:
                        set_human(agi, cause="TIMEOUT")
                    return

            except OSError as err:
                if getattr(err, "errno", None) == 11:
                    continue
                set_human(agi, cause="AUDIOERR"); return

    except Exception as e:
        set_human(agi, cause="NETERR")
    finally:
        try: ws.send(json.dumps({"type":"flush"}))
        except Exception: pass
        try: ws.close()
        except Exception: pass

if __name__ == "__main__":
    startAGI()

Key environment variables:

• AMD_WS_URL → WebSocket endpoint (default: ws://amdy.ai:8080/ws/amd)
• AMD_SAMPLE_RATE → Usually 8000 Hz (set 16000 if required)
• AMD_EARLY_MIN_CONF → Confidence threshold for early decisions (default 0.90)

The script sets three variables in the dialer:

• AMDSTATUS → HUMAN or MACHINE
• AMDCAUSE → Detailed cause string (e.g., final|p_human=0.02|conf=0.98)
• AMDSTATS → Compact log including transcript snippet

4. Call Flow Handling

Once AMDY.ai detects a voicemail, VD_amd.agi decides the next action:

• Drop a voicemail.
• Hang up.
• Transfer only human calls to an available agent.

5. Audio Configuration Notes

• If your media path is 16 kHz, set: export AMD_SAMPLE_RATE=16000
• If your AMD server expects 16 kHz but the dialer streams 8 kHz, enable server-side upsampling (preferred). Otherwise, configure resampling on the EAGI side.

Benefits for Call Centers

✅ Save time – Voicemail calls are filtered automatically.
✅ Increase efficiency – Agents only handle live customers.
✅ Boost ROI – More productive conversations, fewer wasted minutes.
✅ Flexible deployment – Works seamlessly across Vicidial, Asterisk, and FreeSWITCH.

Final Thoughts

Integrating AMDY.ai with Vicidial unlocks the true potential of outbound campaigns. With intelligent answering machine detection, you can maximize agent productivity, reduce wasted resources, and ensure every connected call adds value.

Ready to power up your Vicidial campaigns? Get started with AMDY.ai.