Last active
July 14, 2021 14:37
-
-
Save jfinstrom/4227521 to your computer and use it in GitHub Desktop.
Example of using the Asterisk Manager API in python...
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
# -*- coding: utf-8 -*- | |
# | |
# asterisk.py | |
# | |
# Copyright 2014 James Finstrom<jfinstrom at gmail> | |
# | |
# This program is free software; you can redistribute it and/or modify | |
# it under the terms of the GNU General Public License as published by | |
# the Free Software Foundation; either version 2 of the License, or | |
# (at your option) any later version. | |
# | |
# This program is distributed in the hope that it will be useful, | |
# but WITHOUT ANY WARRANTY; without even the implied warranty of | |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
# GNU General Public License for more details. | |
# | |
# You should have received a copy of the GNU General Public License | |
# along with this program; if not, write to the Free Software | |
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, | |
# MA 02110-1301, USA. | |
# | |
# | |
import os | |
import sys | |
import socket | |
import ConfigParser | |
mgrcnf = '/etc/asterisk/manager.conf' | |
mgruser = 'admin' | |
config = ConfigParser.ConfigParser() | |
config.read(mgrcnf) | |
username = mgruser | |
password = config.get( mgrusr, 'secret') | |
""" Initialize the dictionary in the global space """ | |
def make_dict(lst): | |
ret ={} | |
for i in lst: | |
i = i.strip() | |
if i and i[0] is not "#" and i[-1] is not "=": | |
var,val = i.rsplit(":",1) | |
ret[var.strip()] = val.strip() | |
return ret | |
class acli: | |
def __init__(self): | |
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
self.serverip = '127.0.0.1' | |
self.serverport = 5038 | |
self.username = '' | |
self.password = '' | |
def sendCmd(self,action,**args): | |
self.sock.send("Action: %s\r\n" % action) | |
for key, value in args.items(): | |
self.sock.send("%s: %s\r\n" % (key,value)) | |
self.sock.send("\r\n") | |
data = [] | |
while '\r\n\r\n' not in ''.join(data)[-4:]: | |
buf = self.sock.recv(1) | |
data.append(buf) | |
l = ''.join(data).split('\r\n') | |
return l | |
def conn(self): | |
self.sock.connect((self.serverip, self.serverport)) | |
ret = self.sendCmd("login", Username=self.username, Secret=self.password) | |
if 'Success' in ret[1]: | |
return True | |
else: | |
return False | |
def callCalvery(mesg, doing) | |
#put your action here | |
pass | |
def main(): | |
ampconf() | |
ast = acli() | |
ast.username = username | |
ast.password = password | |
if ast.conn(): | |
dev = ast.sendCmd('SIPShowPeer', Peer='1000') | |
value = make_dict(dev) | |
if value['Response'] == 'Success': | |
if value['Status'] == 'OK': | |
pass | |
else: | |
callCalvery(value['status'], 'peer 1000') | |
else: | |
callCalvery(value['Message'], 'api call') | |
return 0 | |
if __name__ == '__main__': | |
main() | |
Here we go, fixed up and working...
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# asterisk.py
#
# Copyright 2014 James Finstrom<jfinstrom at gmail>
# Updated & Fixed Sep 2016 Jeff T - github jefft4
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
# MA 02110-1301, USA.
#
#
## note: give the Manager user READ, WRITE on system and command (needed to run SIP commands) + read on whatever else it needs for other purposes.
import os
import sys
import socket
import ConfigParser
#Note: if you set up additional managers through FreePBX, change this to manager_additional.conf !
mgrcnf = '/etc/asterisk/manager.conf'
mgrusr = 'ami'
config = ConfigParser.ConfigParser()
config.read(mgrcnf)
username = mgrusr
password = config.get( mgrusr, 'secret')
""" Initialize the dictionary in the global space """
def make_dict(lst):
ret ={}
for i in lst:
i = i.strip()
if i and i[0] is not "#" and i[-1] is not "=":
var,val = i.rsplit(":",1)
ret[var.strip()] = val.strip()
return ret
class acli:
def __init__(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.serverip = '127.0.0.1'
self.serverport = 5038
self.username = ''
self.password = ''
def sendCmd(self,action,**args):
self.sock.send("Action: %s\r\n" % action)
for key, value in args.items():
self.sock.send("%s: %s\r\n" % (key,value))
self.sock.send("\r\n")
data = []
while '\r\n\r\n' not in ''.join(data)[-4:]:
buf = self.sock.recv(1)
data.append(buf)
l = ''.join(data).split('\r\n')
return l
def conn(self):
self.sock.connect((self.serverip, self.serverport))
#need Events OFF with the login else event text pollutes our command response
ret = self.sendCmd("login", Username=self.username, Secret=self.password, Events="OFF")
#print "Connect response: ", ret
if 'Response: Success' in ret:
#print 'Connected.'
return True
else:
#print "Connect failed!"
callCavalry(value['Message'], 'api call')
return False
def callCavalry( mesg, doing ):
#put your action here
print 'Ouch!', mesg, doing
return True
def main():
ast = acli()
ast.username = username
ast.password = password
if ast.conn():
dev = ast.sendCmd('SIPShowPeer', Peer='myvoiptrunkname')
#print "Command response: ", dev
value = make_dict(dev)
if value['Response'] == 'Success':
#print "Status = #", value['Status'], "#"
#don't test only for "OK" here, some return longer strings with ping time etc
if 'OK' in value['Status']:
print 'OK: trunk is up.'
pass
else:
callCavalry(value['Status'], 'peer myvoiptrunkname')
else:
callCavalry(value['Message'], 'api call')
return 0
if __name__ == '__main__':
main()
Here's a python3.6
version that parses the ami
response correctly:
import os, re, os, socket, traceback
def make_dict(data):
ret ={}
for line in data.split("\r\n"):
parts = re.findall("^(.*?): (.*)", line, re.IGNORECASE)
#print("len(parts)", len(parts))
#print("parts", parts)
if parts :
if len(parts[0]) == 2:
ret[parts[0][0].strip()] = parts[0][1].strip()
return ret
class acli:
def __init__(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.serverip = '127.0.0.1'
self.serverport = 5038
self.username = ''
self.password = ''
def sendCmd(self,action,args):
self.sock.send(f"Action: {action}\r\n".encode())
for key, value in args.items():
self.sock.send(f"{key}: {value}\r\n".encode())
self.sock.send("\r\n".encode())
data = ""
while not data.endswith('\r\n\r\n'):
buf = self.sock.recv(4)
data += buf.decode()
return data
def conn(self):
self.sock.connect((self.serverip, self.serverport))
#need Events OFF with the login else event text pollutes our command response
cmd = dict(Username=self.username, Secret=self.password, Events="OFF")
ret = self.sendCmd("login", cmd)
#print "Connect response: ", ret
if 'Response: Success' in ret:
return True
else:
print( "Connect failed!")
return
ast = acli()
ast.username = "user"
ast.password = "XXXXXX"
def ami_cmd(action, args):
global ast
if not ast.conn():
print("No connection")
return {}
resp = make_dict(ast.sendCmd(action, args))
try:
if 'OK' in resp['Status']:
return(resp)
except:
pass
return resp
action = "SipShowPeer"
args = dict(Peer='1')
resp = ami_cmd(action, args)
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Typo mgrusr / mgruser. I guess you didn't actually run this to test it? ;-)
Also, line 86, 'Response' raises a KeyError exception when run. The connect doesn't seem to be specifying "Events: OFF" when it logs in, so any events that come up before or while the SIPShowPeers command is being executed, become part of the return value of the sendCmd and screw up the logic based on ret/value.
Testing the CLI directly via telnet, "SIPShowPeers" wasn't recognised, which is strange.
I'll try to fix all the above and will post back updated code if I succeed...