前提と注意事項
この記事では以下を前提としています。
- Asterisk がインストール済みであること（記事内では version 20.6.0 を使用しています）
- Python 3.x がインストール済みであること
Asterisk のインストール手順については以下の記事で解説しています。
この記事のゴール
Asterisk の AMI を設定し、Python スクリプトで着信をリアルタイムに検知して HTTP リクエストを送信するところまでをゴールとします。
着信時に HTTP リクエストを送信できるようになれば、n8n や Zapier などのワークフロー自動化ツールとの連携、Slack やメールへの通知、CRM への自動記録など、さまざまな業務自動化に繋げることができます。
構成概要
全体の構成は以下のとおりです。
Asterisk（PBX）
↓ AMI イベント
Python スクリプト（着信検知）
↓ HTTP POST
外部サービス（n8n, Zapier, 自作API など）AMI を設定する
AMI（Asterisk Manager Interface）は、Asterisk を外部プログラムから制御・監視するためのインターフェースです。まず AMI を有効化し、接続用のユーザーを作成します。
manager.conf を編集する
/etc/asterisk/manager.conf（または /etc/asterisk/manager.d/ 配下のファイル）を編集し、AMI を有効化します。
[general]
enabled = yes
port = 5038
bindaddr = 127.0.0.1bindaddr = 127.0.0.1 を指定することで、ローカルからのみ AMI に接続できるようにしています。外部から接続する必要がある場合は、適切なファイアウォール設定を行ってください。
AMI ユーザーを作成する
同じファイルに AMI 接続用のユーザーを追加します。
[ami_user]
secret = your_secure_password
deny = 0.0.0.0/0.0.0.0
permit = 127.0.0.1/255.255.255.255
read = all
write = allセキュリティに関する注意
secretには十分に強力なパスワードを設定してください。 -permitで接続元 IP を必ず制限してください。 - 本番環境ではread = allではなく、必要な権限のみ（例:read = system,call）に絞ることを推奨します。
eventfilter を設定する
AMI はデフォルトで大量のイベントを送信します。eventfilter を使って、必要なイベントだけを受信するように設定します。
; 不要なイベントを除外
eventfilter = !Event: RTCPSent
eventfilter = !Event: RTCPReceived
; 必要なイベントを明示的に許可
eventfilter = Event: Newchannel
eventfilter = Event: Hangup
eventfilter = Event: SoftHangupRequest
eventfilter = Event: HangupRequest
eventfilter = Event: DialEndeventfilter の注意点
正のフィルタ（Event: XXX）を1つでも書くと、ホワイトリスト方式になります。明示的にリストされたイベントしか通過しなくなるため、着信検知に必要な Newchannel イベントを含め忘れると、着信を検知できません。
Hangup や DialEnd だけを許可して Newchannel を忘れると、AMI 接続・ログインは成功しているのにイベントが一切届かないという状態になります。
設定を反映する
以下のコマンドを実行して、設定を反映します。
$ asterisk -rx "manager reload"Asterisk を再起動しても設定は反映されます。
Python スクリプトを実装する
AMI に接続して着信を検知し、HTTP リクエストで外部に通知する Python スクリプトを実装します。
AMI プロトコルの基本
AMI はテキストベースのプロトコルで、TCP 接続して以下の流れで通信します。
- 接続:
Asterisk Call Manager/X.X.Xのバナーを受信します - ログイン:
Action: Loginを送信して認証します - イベント受信: 登録したイベントがテキスト形式で届きます
- キープアライブ: 定期的に
Action: Pingを送信して接続を維持します
AMI のメッセージはキーバリュー形式で、空行（\r\n\r\n）で区切られます。
Event: Newchannel
Channel: PJSIP/2020-00000000
CallerIDNum: 09012345678
CallerIDName:
Context: from-external
Exten: s着信検知のロジック
着信を検知するには Newchannel イベントを監視し、以下の条件でフィルタリングします。
- Context が外線着信のコンテキストであること（環境に応じて
from-externalやfrom-trunkなどを指定します） - CallerIDNum が存在すること
def handle_newchannel(self, event):
"""Newchannel イベントのハンドラ"""
context = event.get("Context", "")
caller_id = event.get("CallerIDNum", "")
channel = event.get("Channel", "")
# 監視対象のコンテキストかチェック
if context != "from-external":
return
# CallerID が空または不明の場合はスキップ
if not caller_id or caller_id == "<unknown>":
return
logger.info(f"着信検知: CallerID={caller_id}, Channel={channel}")
# HTTP リクエストで通知
self.send_notification(caller_id, channel)HTTP リクエスト送信処理
着信を検知したら、HTTP POST で外部サービスに通知します。ネットワークの一時的な問題に備えてリトライ処理を実装します。
import requests
import time
def send_notification(self, caller_id, channel, max_retries=3):
"""着信情報を HTTP POST で送信"""
payload = {
"caller_id": caller_id,
"channel": channel,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
}
for attempt in range(1, max_retries + 1):
try:
response = requests.post(
self.endpoint_url,
json=payload,
timeout=5
)
if response.status_code == 200:
logger.info(f"送信成功: {caller_id}")
return True
else:
logger.warning(
f"応答エラー: HTTP {response.status_code} "
f"(試行 {attempt}/{max_retries})"
)
except requests.RequestException as e:
logger.warning(f"送信例外: {e} (試行 {attempt}/{max_retries})")
if attempt < max_retries:
time.sleep(2)
logger.error(f"送信失敗: {caller_id} ({max_retries}回リトライ後)")
return False送信先の URL（self.endpoint_url）には、n8n や Zapier の Webhook URL、自作の API エンドポイントなど、任意の URL を指定できます。
キープアライブの実装
AMI 接続はアイドル状態が続くとタイムアウトで切断される場合があります。定期的に Ping を送信して接続を維持します。
import threading
def start_keepalive(self, interval=15):
"""定期的に Ping を送信して接続を維持"""
def ping_loop():
while self.connected:
self.send_action("Ping")
time.sleep(interval)
thread = threading.Thread(target=ping_loop, daemon=True)
thread.start()トラブルシューティング
AMI 接続は成功するがイベントが届かない
最も多い原因は eventfilter の設定ミスです。
eventfilter = Event: Newchannelが含まれているか確認してください- 正のフィルタを1つでも書くとホワイトリスト方式になることに注意してください
- 設定変更後は
asterisk -rx "manager reload"で反映してください
AMI ログイン自体が失敗する
manager.confのenabled = yesを確認してください- AMI ユーザーの
permit設定で接続元 IP が許可されているか確認してください - パスワードが正しいか確認してください
HTTP リクエストが送信先に届かない
- 送信先の URL が正しいか確認してください
- 送信先のサービスが起動しているか確認してください
- ファイアウォールで送信先へのアウトバウンド通信がブロックされていないか確認してください
まとめ
この記事では、Asterisk の AMI を使って着信をリアルタイムに検知し、Python スクリプトで HTTP リクエストを送信する仕組みの構築手順を解説しました。ポイントを整理すると以下のとおりです。
- AMI の
eventfilterにはNewchannelを必ず含めること（ホワイトリスト方式に注意） - Python スクリプトで
Newchannelイベントを監視し、コンテキストと CallerID でフィルタリングすること - HTTP リクエスト送信にはリトライ処理を実装すること
着信時に HTTP リクエストを送信できる仕組みがあれば、n8n や Zapier などのワークフロー自動化ツール、Slack や Microsoft Teams への通知、CRM への自動記録など、さまざまな外部サービスと連携して業務自動化を実現できます。