pyajam — python Asterisk AJAM interface

The purpose of pyajam is to allow python programs interact with an asterisk server.

ajam = Pyajam(username='mspencer', password='*rocks!')
if not ajam.login():
  print "Invalid login"
  sys.exit(1)

peers = ajam.peers())          # array of SIP/IAX2 peers
peer  = ajam.sippeer('101'))   # dictionary of peer attributes

# screenprint events
def ajam_event_listener(data):
  print data

ajam.waitevent(async=False, callback=ajam_event_listener)

Modules

class pyajam.Pyajam(server='localhost', port=8088, path='/asterisk', username='admin', password='admin', autoconnect=True, asraw=False)

Pyajam initialisation.

Parameters:
  • server: adress or IP of the asterisk server to connect to (default= localhost),
  • port: ajam port (as defined in asterisk/http.conf),
  • path: url path to send ajam queries,
  • username: ajam login username (as defined in asterisk/manager.conf),
  • password: ajam login password,
  • autoconnect: Pyajam automatically (re)connect to AJAM server,
  • raw: return values as raw (e.g not pythonified)
login()

Login to Asterisk AJAM service. Use the username & password provided at class instantiation.

Returns True if login success, else False

version()

Return Asterisk version of server you are connected to. Values are None (not connected), 1.4 or 1.6.

sippeers()

Query SIP peers definition. Return list of peers (each peer is represented as a dictionary).

Useful items are:
  • objectname, the peer name
  • status, with OK value if peer is connected
  • ipaddress, device IP
New fields in asterisk 1.6:
  • textsupport
# sippeers() output sample
>>> import pprint
>>> from pyajam import Pyajam
>>> ajam = Pyajam(username='mspencer', password='*rocks!')
>>> pprint.pprint(ajam.sippeers())
[ { u'acl'           : u'no',
    u'channeltype'   : u'SIP',
    u'chanobjecttype': u'peer',
    u'dynamic'       : u'yes',
    u'event'         : u'PeerEntry',
    u'ipaddress'     : u'-none-',
    u'ipport'        : u'0',
    u'natsupport'    : u'no',
    u'objectname'    : u'102',
    u'realtimedevice': u'no',
    u'status'        : u'UNKNOWN',
    u'videosupport'  : u'no',
    u'textsupport'   : u'no'},
  { u'acl'           : u'no',
    u'channeltype'   : u'SIP',
    u'chanobjecttype': u'peer',
    u'dynamic'       : u'yes',
    u'event'         : u'PeerEntry',
    u'ipaddress'     : u'127.0.0.1',
    u'ipport'        : u'5061',
    u'latency'       : u'1',
    u'natsupport'    : u'no',
    u'objectname'    : u'101',
    u'realtimedevice': u'no',
    u'status'        : u'OK',
    u'videosupport'  : u'no',
    u'textsupport'   : u'no'}]
iaxpeers()

Query IAX peers definition. Return list of peers (each peer is represented as a dictionary).

Useful items are:
  • objectname, the peer name
  • status, with OK value if peer is connected
  • ipaddress, device IP
New fields in asterisk 1.6:
  • chanobjecttype
  • encryption
  • event
  • trunk
Notes:
  • username field is truncated if longer than 10 characters
# iaxpeers() output sample
>>> import pprint
>>> from pyajam import Pyajam
>>> ajam = Pyajam(username='mspencer', password='*rocks!')
>>> pprint.pprint(ajam.IAXpeers())
[ { u'channeltype'   : u'IAX2',
    u'dynamic'       : u'no',
    u'ipaddress'     : u'216.207.245.47',
    u'ipport'        : u'4569',
    u'objectname'    : u'demo',
    u'username'      : u'asterisk',
    u'status'        : u'Unmonitored',
    u'chanobjecttype': u'peer',
    u'encryption'    : u'no',
    u'event'         : u'PeerEntry',
    u'trunk'         : u'no'}]
peers()

Query both IAX and SIP peers. Return list of peers (each peer is represented as a dictionary). Use channeltype to distinguish between sip and iax2 devices.

# peers() output sample
>>> import pprint
>>> from pyajam import Pyajam
>>> ajam = Pyajam(username='mspencer', password='*rocks!')
>>> pprint.pprint(ajam.peers())
[ { u'channeltype'   : u'IAX2',
    u'chanobjecttype': u'peer',
    u'dynamic'       : u'no',
    u'encryption'    : u'no',
    u'event'         : u'PeerEntry',
    u'ipaddress'     : u'-none-',
    u'ipport'        : u'0',
    u'objectname'    : u'iax1',
    u'status'        : u'Unmonitored',
    u'trunk'         : u'no',
    u'username'      : u''},
  { u'acl'           : u'no',
    u'channeltype'   : u'SIP',
    u'chanobjecttype': u'peer',
    u'dynamic'       : u'yes',
    u'event'         : u'PeerEntry',
    u'ipaddress'     : u'127.0.0.1',
    u'ipport'        : u'5061',
    u'latency'       : u'1',
    u'natsupport'    : u'no',
    u'objectname'    : u'101',
    u'realtimedevice': u'no',
    u'status'        : u'OK',
    u'textsupport'   : u'no',
    u'videosupport'  : u'no'}]
sipregistry()

Query SIP devices registration status (mainly used for trunks).

New fields in asterisk 1.6:
  • event
  • registrationtime
# sipregistry() output sample
>>> import pprint
>>> from pyajam import Pyajam
>>> ajam = Pyajam(username='mspencer', password='*rocks!')
>>> pprint.pprint(ajam.sipregistry())
[ { u'event': u'RegistryEntry',
    u'host'            : u'mysipprovider.com',
    u'port'            : u'5060',
    u'refresh'         : u'120',
    u'registrationtime': u'0',
    u'state'           : u'Unregistered',
    u'username'        : u'1234'}]
sippeer(peername)
Query details about a SIP peer.
Comparing to sippeers(), here you have got detailed informations about the peer.
New fields in asterisk 1.6:
  • a lot of fields have been renamed or added between 1.4 and 1.6
# sippeer() output sample
>>> import pprint
>>> from pyajam import Pyajam
>>> ajam = Pyajam(username='mspencer', password='*rocks!')
>>> pprint.pprint(ajam.sippeer('101'))
{ 'acl'          : 'No',
  'addr->ip'     : '127.0.0.1 Port 5061',
  'ama flags'    : 'Unknown',
  'auto-framing' : 'No',
  'call limit'   : '0',
  'callerid'     : '"" <>',
  'callgroup'    : '',
  'callingpres'  : 'Presentation Allowed, Not Screened',
  'canreinvite'  : 'Yes',
  'codec order'  : '(none)',
  'codecs'       : '0x8000e (gsm|ulaw|alaw|h263)',
  'context'      : 'default',
  'def. username': '101',
  'defaddr->ip'  : '0.0.0.0 Port 5060',
  'dtmfmode'     : 'rfc2833',
  'dynamic'      : 'Yes',
  'expire'       : '1801',
  'insecure'     : 'no',
  'language'     : '',
  'lastmsg'      : '0',
  'lastmsgssent' : '32767/65535',
  'mailbox'      : '',
  'maxcallbr'    : '384 kbps',
  'md5secret'    : '<Not set>',
  'name'         : '101',
  'nat'          : 'RFC3581',
  'overlap dial' : 'No',
  'pickupgroup'  : '',
  'promiscredir' : 'No',
  'reg. contact' : 'sip:101@127.0.0.1:5061',
  'secret'       : '<Not set>',
  'send rpid'    : 'No',
  'sip options'  : '(none)',
  'status'       : 'OK (1 ms)',
  'subscr.cont.' : '<Not set>',
  'subscriptions': 'Yes',
  't38 pt udptl' : 'No',
  'tohost'       : '',
  'transfer mode': 'open',
  'trust rpid'   : 'No',
  'user=phone'   : 'No',
  'useragent'    : 'Twinkle/1.4.2',
  'video support': 'No',
  'vm extension' : 'asterisk'}
command(command, regex=None, normalizer=None, unifyin=[])
Execute an Asterisk command.
command is always returned in raw mode from asterisk.

command is the asterisk command as typed in asterisk console, regex is a python regular expression use to parse command returned value, normalizer is used to normalize returned results unifyin is either an empty list or dict, as you want to return a list or a dict

# direct call command() output sample
>>> import pprint, re
>>> from pyajam import Pyajam
>>> ajam = Pyajam(username='mspencer', password='*rocks!')
>>> 
>>> rx = re.compile('^(?P<key>[^\s]+)\s+(?P<value>.*?)\s+\s+.*$', re.M)
>>> def module_show_normalize(row):
...   if row['key'] in ('--END', 'Response:', 'Privilege:', 'Module') or \
...      row['value'] == 'modules':
...   return None
...
...   row[row['key']] = row['value']
...   del row['key']
...   del row['value']
...
...   return row
...
>>> pprint.pprint(ajam.command('module show', rx, module_show_normalize))
[{'res_features.so': 'Call Features Resource'},
 {'res_musiconhold.so': 'Music On Hold Resource'},
 {'pbx_config.so': 'Text Extension Configuration'},
 {'chan_sip.so': 'Session Initiation Protocol (SIP)'},
 {'app_ldap.so': 'LDAP directory lookup function for Aster 0'}]
waitevent(async=False, callback=None)

Manage Asterisk events.

Usage:
  • if async is False and callback is None, return the first eventlist received (blocking),
  • if async is False and callback is not None, execute the callback for each received eventlist (blocking, e.g waitevent() never returns until calling waitevent_stop()),
  • if async is True (callback is required), execute the callback for each eventlist (non-blocking, e.g waitevent() returns immediatly)
Callback function:
the callback function must take one argument, an eventlist (array of events, as dictionaries)
# waitevent() output sample
>>> import pprint, re
>>> from pyajam import Pyajam
>>> ajam = Pyajam(username='mspencer', password='*rocks!')
>>> pprint ajam.waitevent(async=False)
>>>
>>> def ajam_event_listener(evtlist):
>>>   pp.pprint(evtlist)
>>> ajam.waitevent(async=False, callback=ajam_event_listener)
[ { u'event': u'Reload',
    u'message': u'Reload Requested',
    u'privilege': u'system,all'},
   {u'event': u'WaitEventComplete'}]
waitevent_stop()

Stop event async manager.

Changelog

  • v0.1: initial release

Table Of Contents

Next topic

1. Asterisk configuration

This Page