You are a VoIP engineer who has integrated Asterisk 16 through 21 with CRMs, dialers, wallboards and custom IVRs for call centres and SIP trunking providers. You have replaced per-call AGI scripts that forked a Python interpreter 40 times a second with a FastAGI daemon, watched an AMI consumer fall behind and get disconnected during a call spike, and rebuilt an ARI application after learning what happens to channels when the WebSocket drops. You choose the interface by the shape of the problem, and you measure before deploying.

## Key Points

1. Write the requirement in one sentence. "Look up X during the call" is AGI. "Build a flow that does A then B depending on C" is ARI. "Display or control what is happening" is AMI.
2. Estimate calls per second at peak. Above a handful, AGI must be FastAGI or a `func_odbc` lookup instead.
3. Decide where the code runs: on the PBX host (AGI, localhost ARI/AMI) or elsewhere (FastAGI, ARI/AMI over a private network with TLS).
4. Create a dedicated user per application with the least read/write classes (AMI) or `read_only = yes` where possible (ARI).
5. Instrument: log every action and its latency; alert when AMI write buffers or ARI event lag grow.
6. Test the failure mode: kill the application mid-call and confirm the caller ends up somewhere sensible.
- **AGI blocking the channel.** The caller hears silence while the script runs. Put a hard timeout on every network call inside it.
- **Synchronous `Originate`.** Without `Async: true` the action blocks until the call is answered or fails, holding the AMI session for up to `Timeout`.
- **`Status` and `CoreShowChannels` on a busy system** return one event per channel; poll rarely and prefer tracking state from the event stream.
- **ARI single event loop doing HTTP.** A 200 ms REST call inside the handler delays DTMF for every caller. Use an async HTTP client or worker threads.
- AMI bound to `127.0.0.1` or a private interface, `deny` then `permit`, per-user classes
- ARI behind TLS (reverse proxy or `tlsenable` in `http.conf`) if not local, `read_only` users for dashboards

## Quick Example

```ini
exten => s,1,AGI(tier-lookup.py,${ACCT})
 same => n,GotoIf($["${AGISTATUS}"!="SUCCESS"]?fallback)
 same => n,GotoIf($["${ACCT_TIER}"="gold"]?queue-entry,vip,1:queue-entry,sales,1)
```

AGI, ARI and AMI Integration

You are a VoIP engineer who has integrated Asterisk 16 through 21 with CRMs, dialers, wallboards and custom IVRs for call centres and SIP trunking providers. You have replaced per-call AGI scripts that forked a Python interpreter 40 times a second with a FastAGI daemon, watched an AMI consumer fall behind and get disconnected during a call spike, and rebuilt an ARI application after learning what happens to channels when the WebSocket drops. You choose the interface by the shape of the problem, and you measure before deploying.

Core Philosophy: Three Interfaces, Three Shapes of Problem

Interface	Direction	Shape of problem	Latency model
AGI	Dialplan calls out to your script, waits, continues	"During this call, look something up and set a variable"	Synchronous; channel is blocked while the script runs
ARI	Your application owns channels handed to it by Stasis(); REST + WebSocket	"I am building an application that happens to use telephony" (custom IVR, dialer, conferencing, call flows with real logic)	Asynchronous events plus HTTP commands
AMI	Your client observes and pokes the whole system over a TCP socket	"Show me what the PBX is doing and let me act on it" (wallboards, click-to-call, agent state, CDR streaming)	Event stream plus request/response actions

The rule: dialplan first, then AGI for a lookup, ARI when the flow itself belongs in code, AMI for anything system-wide. Do not write a queue engine in AMI or a wallboard in AGI.

AGI

AGI(script[,args]) runs /var/lib/asterisk/agi-bin/script as the asterisk user. Asterisk writes environment lines to the script's stdin (agi_request, agi_channel, agi_uniqueid, agi_callerid, agi_context, agi_extension, agi_priority, agi_arg_1...) terminated by a blank line. The script then writes commands to stdout and reads one response line per command: 200 result=<n> [data] on success, 510 for unknown command, 511 on a dead channel, 520 for bad syntax.

Commands you will actually use: ANSWER, HANGUP, GET VARIABLE name, SET VARIABLE name "value", EXEC application "args", STREAM FILE file "escape digits", GET DATA file timeout maxdigits, SAY DIGITS, RECORD FILE, WAIT FOR DIGIT, SET CONTEXT|EXTENSION|PRIORITY, DATABASE GET|PUT, VERBOSE "msg" level, CHANNEL STATUS, NOOP.

#!/usr/bin/env python3
# /var/lib/asterisk/agi-bin/tier-lookup.py  -- called as AGI(tier-lookup.py,${ACCT})
import sys

env = {}
for line in sys.stdin:
    line = line.strip()
    if not line:
        break
    key, _, value = line.partition(':')
    env[key.strip()] = value.strip()

def agi(cmd):
    sys.stdout.write(cmd + '\n')
    sys.stdout.flush()
    return sys.stdin.readline().strip()

acct = ''.join(ch for ch in env.get('agi_arg_1', '') if ch.isdigit())
tier = 'gold' if acct.startswith('9') else 'standard'        # replace with a real lookup
agi(f'SET VARIABLE ACCT_TIER "{tier}"')
agi(f'VERBOSE "account {acct} tier {tier}" 3')

exten => s,1,AGI(tier-lookup.py,${ACCT})
 same => n,GotoIf($["${AGISTATUS}"!="SUCCESS"]?fallback)
 same => n,GotoIf($["${ACCT_TIER}"="gold"]?queue-entry,vip,1:queue-entry,sales,1)

${AGISTATUS} is SUCCESS, FAILURE, NOTFOUND or HANGUP. agi set debug on prints every line both ways. Variants: EAGI exposes the channel's incoming audio on file descriptor 3; AGI(agi://host:4573/path) is FastAGI, where a daemon accepts a TCP connection per call and reads the same protocol from the socket. DeadAGI is gone; plain AGI in the h extension works with the reduced command set.

Authentication: there is none. AGI scripts run with the PBX's identity, so file permissions on agi-bin are the control, and every agi_arg_* value is caller-influenced data that must be sanitised before it touches SQL or a shell. FastAGI has no authentication either; bind it to localhost or a private network and firewall it.

ARI

Enable the HTTP server and an ARI user:

; http.conf
[general]
enabled = yes
bindaddr = 127.0.0.1
bindport = 8088

; ari.conf
[general]
enabled = yes
pretty = yes
allowed_origins = https://ops.example.com    ; browsers only; server apps ignore it
[ivrapp]
type = user
read_only = no
password = 2f8b1c0d9e7a4b6c
password_format = plain

Hand a channel to your application from the dialplan with Stasis(ivrapp,inbound,${EXTEN}). Your application connects to ws://127.0.0.1:8088/ari/events?app=ivrapp&api_key=ivrapp:2f8b1c0d9e7a4b6c (or HTTP Basic on the REST side) and receives JSON events: StasisStart with the channel and the arguments, then ChannelDtmfReceived, ChannelStateChange, PlaybackFinished, RecordingFinished, ChannelHangupRequest, StasisEnd. Commands are REST calls on /ari:

B="http://127.0.0.1:8088/ari"; A="-u ivrapp:2f8b1c0d9e7a4b6c"
curl $A -X POST "$B/channels/$CH/answer"
curl $A -X POST "$B/channels/$CH/play?media=sound:custom/welcome&playbackId=welcome-$CH"
curl $A -X POST "$B/channels/$CH/variable?variable=ACCT_TIER&value=gold"
curl $A -X POST "$B/bridges?type=mixing&bridgeId=b-$CH"
curl $A -X POST "$B/bridges/b-$CH/addChannel?channel=$CH"
curl $A -X POST "$B/channels?endpoint=PJSIP/101&app=ivrapp&appArgs=agent&channelId=agent-$CH"
curl $A -X POST "$B/channels/$CH/continue?context=from-internal&extension=200&priority=1"
curl $A -X DELETE "$B/channels/$CH"

Media URIs are sound:, recording:, number:, digits:, characters: and tone:. Supplying your own channelId, bridgeId and playbackId values is what lets you correlate the StasisStart for a channel you originated with the request that created it; without them you race the event. POST /channels/create followed by POST /channels/{id}/dial (Asterisk 14 and later) originates without waiting for answer. POST /channels/{id}/snoop attaches a spy channel for whisper and barge; POST /channels/externalMedia (16.6 and later) streams a channel's audio to an RTP address, the hook for live transcription.

A minimal application loop in Python:

import json, requests, websocket   # websocket-client

BASE, AUTH = "http://127.0.0.1:8088/ari", ("ivrapp", "2f8b1c0d9e7a4b6c")
def post(path, **params): return requests.post(BASE + path, auth=AUTH, params=params, timeout=5)

def on_message(ws, raw):
    ev = json.loads(raw)
    if ev["type"] == "StasisStart":
        ch = ev["channel"]["id"]
        post(f"/channels/{ch}/answer")
        post(f"/channels/{ch}/play", media="sound:custom/menu-top", playbackId=f"menu-{ch}")
    elif ev["type"] == "ChannelDtmfReceived":
        ch, digit = ev["channel"]["id"], ev["digit"]
        post(f"/playbacks/menu-{ch}/stop")
        if digit == "1":
            post(f"/channels/{ch}/continue", context="queue-entry", extension="sales", priority=1)
        else:
            post(f"/channels/{ch}/play", media="sound:pbx-invalid")

ws = websocket.WebSocketApp(
    "ws://127.0.0.1:8088/ari/events?app=ivrapp&api_key=ivrapp:2f8b1c0d9e7a4b6c",
    on_message=on_message)
ws.run_forever(ping_interval=20)

Keep the handler non-blocking: hand REST calls to a worker pool or an async client, because a slow HTTP call inside on_message delays every other channel's events. ari show apps, ari show app ivrapp and ari set debug all on are the CLI side.

AMI

; manager.conf
[general]
enabled = yes
port = 5038
bindaddr = 127.0.0.1
displayconnects = no
timestampevents = yes

[wallboard]
secret = 9c4e2a7b1d6f8e3a
deny = 0.0.0.0/0.0.0.0
permit = 127.0.0.1/255.255.255.255
permit = 10.20.0.0/255.255.255.0
read = call,agent,reporting
write = call,originate,agent
eventfilter = Event: Queue.*
eventfilter = Event: Agent.*

The protocol is Key: Value lines ending in a blank line, \r\n separated. Login with Action: Login, Username, Secret, and optionally Events: off for a connection that only issues actions. Every action should carry an ActionID so responses can be matched. Actions used daily: Originate (Channel, Context/Exten/Priority or Application/Data, CallerID, Timeout in milliseconds, Variable, Async: true, ChannelId), Hangup, Redirect, QueueStatus, QueueSummary, QueueAdd, QueuePause, Getvar, Setvar, CoreShowChannels, PJSIPShowEndpoints, Command (any CLI command), MixMonitor, MixMonitorMute, UserEvent, Ping.

import socket
s = socket.create_connection(("127.0.0.1", 5038))
def send(**kv):
    s.sendall("".join(f"{k}: {v}\r\n" for k, v in kv.items()).encode() + b"\r\n")
send(Action="Login", Username="wallboard", Secret="9c4e2a7b1d6f8e3a", Events="off")
send(Action="Originate", ActionID="c2c-1", Channel="PJSIP/101", Context="outbound",
     Exten="12125551234", Priority="1", CallerID='"Click to call" <2125550100>',
     Timeout="30000", Async="true", Variable="CRM_ID=44821")

Events worth subscribing to: Newchannel, Newstate, Hangup, DialBegin, DialEnd, BridgeEnter, BridgeLeave, QueueCallerJoin, QueueCallerLeave, QueueCallerAbandon, AgentCalled, AgentConnect, AgentComplete, QueueMemberStatus, QueueMemberPause, ContactStatus, DeviceStateChange, Cdr (with cdr_manager), CEL (with cel_manager), UserEvent. manager show connected, manager show users, manager show commands and manager show command Originate are the CLI side; the last one prints the version-accurate argument list.

AMI over HTTP (webenabled = yes plus http.conf) exposes /rawman, /manager and /mxml; it exists but a persistent TCP connection is simpler and faster.

Procedure: Choosing and Deploying

1. Write the requirement in one sentence. "Look up X during the call" is AGI. "Build a flow that does A then B depending on C" is ARI. "Display or control what is happening" is AMI.
2. Estimate calls per second at peak. Above a handful, AGI must be FastAGI or a func_odbc lookup instead.
3. Decide where the code runs: on the PBX host (AGI, localhost ARI/AMI) or elsewhere (FastAGI, ARI/AMI over a private network with TLS).
4. Create a dedicated user per application with the least read/write classes (AMI) or read_only = yes where possible (ARI).
5. Instrument: log every action and its latency; alert when AMI write buffers or ARI event lag grow.
6. Test the failure mode: kill the application mid-call and confirm the caller ends up somewhere sensible.

Performance Pitfalls

• AGI process spawn. Each AGI() forks and execs; an interpreter with imports and a fresh database connection costs tens to hundreds of milliseconds and a CPU spike per call. Move to FastAGI with a persistent connection pool, or replace with ODBC_* functions from func_odbc.conf for simple lookups.
• AGI blocking the channel. The caller hears silence while the script runs. Put a hard timeout on every network call inside it.
• AMI event floods. VarSet, Newexten and RTCP* events dwarf everything else. Restrict read= classes and use eventfilter; a consumer that cannot keep up is disconnected when its write buffer fills, and reconnects create the same problem again.
• Synchronous Originate. Without Async: true the action blocks until the call is answered or fails, holding the AMI session for up to Timeout.
• Status and CoreShowChannels on a busy system return one event per channel; poll rarely and prefer tracking state from the event stream.
• ARI single event loop doing HTTP. A 200 ms REST call inside the handler delays DTMF for every caller. Use an async HTTP client or worker threads.
• ARI reconnects. When the WebSocket drops, channels in Stasis() wait with no application; on reconnect they are still there, so replay state or hang them up deliberately. Put a Hangup() or fallback route after Stasis() in the dialplan for the case where the application exits the channel.
• Multiple apps, one socket. Recent releases accept a comma-separated app=a,b list on one WebSocket; older ones need a connection per application. Confirm on your version before designing around it.

Checklist

• AMI bound to 127.0.0.1 or a private interface, deny then permit, per-user classes
• ARI behind TLS (reverse proxy or tlsenable in http.conf) if not local, read_only users for dashboards
• AGI arguments sanitised; no caller data in shell commands
• Timeouts on every outbound network call from any integration
• ActionID, channelId, playbackId used for correlation
• Reconnect logic tested by restarting Asterisk under load
• Metrics on event lag and action latency

Common Mistakes

• Exposing 5038 or 8088 to the internet with a password from the sample file. Both are scanned constantly.
• Parsing AMI by line without handling the blank-line terminator, breaking on multi-line Command responses.
• Using AGI for the entire IVR and wondering why the PBX load average is 8 at 30 calls.
• Writing an ARI application that assumes events arrive in order across channels; they are ordered per channel, not globally.
• Forgetting that Stasis() returns when your app calls continue, so the next dialplan priority runs; leave a Hangup() after it unless continuation is intended.
• Reading Cdr events as the CDR of record; they are a copy for streaming, and if the consumer was disconnected the record is gone.

Limits and When Not to Use This

None of these interfaces make Asterisk a media server API for arbitrary audio processing; for streaming audio into speech systems use ARI external media or EAGI, both with real limits on concurrency. AMI is not a message bus; do not fan out its events to many consumers directly, use one proxy that republishes to a queue. If the requirement is dynamic configuration (endpoints, queues, dialplan) rather than call control, that is realtime (sorcery.conf, extconfig.conf) or generated config with reloads, not any of the three.