Skip to content

Instantly share code, notes, and snippets.

@jfinstrom
Last active July 14, 2021 14:37
Show Gist options
  • Save jfinstrom/4227521 to your computer and use it in GitHub Desktop.
Save jfinstrom/4227521 to your computer and use it in GitHub Desktop.
Example of using the Asterisk Manager API in python...
#!/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()
@jefft4
Copy link

jefft4 commented Sep 1, 2016

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...

@jefft4
Copy link

jefft4 commented Sep 1, 2016

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()

@x011
Copy link

x011 commented Apr 14, 2019

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