Asterisk IVR Development Guide 2026 — Building Custom Interactive Voice Systems

// ARI Node.js Example: Handle StasisStart, collect DTMF
const ws = new WebSocket('ws://asterisk:8088/ari/events?app=my_app&token=secret');

ws.on('message', (event) => {
  const msg = JSON.parse(event);

  if (msg.type === 'StasisStart') {
    const channelId = msg.channel.id;
    console.log(`Call started: ${channelId}`);

    // Play greeting
    fetch(`http://asterisk:8088/ari/channels/${channelId}/play`, {
      method: 'POST',
      body: JSON.stringify({ media: 'sound:en/hello' })
    });
  }

  if (msg.type === 'ChannelDtmfReceived') {
    console.log(`DTMF: ${msg.digit}`);
    // Route based on digit...
  }
});

; extensions.conf
[from-trunk]
exten => +1234567890,1,Answer()
exten => +1234567890,n,Set(CHANNEL(language)=en)
exten => +1234567890,n,AGI(agi:///ivr_main.py,${CALLERID(num)})
exten => +1234567890,n,Hangup()

; Alternative: ARI/Stasis app
exten => +1234567890,1,Answer()
exten => +1234567890,n,Stasis(my_ivr_app,${CALLERID(num)})
exten => +1234567890,n,Hangup()

[from-internal]
exten => 5555,1,Set(QUEUE_LOG_CONTEXT=on)
exten => 5555,n,Queue(support_queue,t,,,300)
exten => 5555,n,Hangup()

# Production AGI: Async call handling with retries
import asyncio, logging, psycopg2, openai
from datetime import datetime

async def ivr_handler(channel_id, caller_id, did):
    """Main IVR entrypoint."""
    logger = logging.getLogger(__name__)

    try:
        # 1. Verify caller in database
        conn = psycopg2.connect("dbname=customers")
        cur = conn.cursor()
        cur.execute("SELECT customer_id, tier FROM callers WHERE phone=%s", (caller_id,))
        result = cur.fetchone()
        conn.close()

        if not result:
            await play_audio(channel_id, 'sound:welcome_unknown')
            await collect_input(channel_id, 5)  # 5-digit customer ID
            return

        customer_id, tier = result

        # 2. Route based on tier
        if tier == 'premium':
            await play_audio(channel_id, 'sound:welcome_premium')
            await dial_extension(channel_id, 'SIP/support_vip@trunk')
        else:
            await play_audio(channel_id, 'sound:welcome_standard')
            await queue_call(channel_id, 'support_queue')

    except Exception as e:
        logger.error(f"AGI error: {e}")
        await play_audio(channel_id, 'sound:error_try_again')

async def play_audio(channel_id, media):
    """Queue audio playback."""
    # Implementation...
    pass

async def dial_extension(channel_id, target):
    """Transfer to extension."""
    # Implementation...
    pass

async def queue_call(channel_id, queue_name):
    """Queue for agent."""
    # Implementation...
    pass