asterisk.manager module

This module provides a Python API for interfacing with the asterisk manager.

Example

import sys
from asterisk.manager import Manager


def handle_shutdown(event, manager):
    print("Received shutdown event")
    manager.close()
    # we could analyze the event and reconnect here


def handle_event(event, manager):
    print("Received event: %s" % event.name)


manager = Manager()
try:
    # connect to the manager
    try:
        manager.connect("host")
        manager.login("user", "secret")

        # register some callbacks
        manager.register_event(
            "Shutdown", handle_shutdown
        )  # shutdown
        manager.register_event("*", handle_event)  # catch all

        # get a status report
        response = manager.status()
        manager.logoff()
    except asterisk.manager.ManagerSocketException as e:
        print(
            "Error connecting to the manager: %s" % e.strerror
        )
        sys.exit(1)
    except asterisk.manager.ManagerAuthException as e:
        print(
            "Error logging in to the manager: %s" % e.strerror
        )
        sys.exit(1)
    except asterisk.manager.ManagerException as e:
        print("Error: %s" % e.strerror)
        sys.exit(1)

finally:
    # remember to clean up
    manager.close()

Remember all header, response, and event names are case sensitive.

Not all manager actions are implemented as of yet, feel free to add them and submit patches.

Specification

class asterisk.manager.Event(message)

Bases: object

Manager interface Events, __init__ expects a ‘ManagerMsg’ message

get_action_id()

get_header(hname, defval=None)

Return the specified header

has_header(hname)

Check for a header

class asterisk.manager.Manager

Bases: object

TODO: add a docstring.

absolute_timeout(channel, timeout)

Set an absolute timeout on a channel

close()

Shutdown the connection to the manager

command(command)

Execute a command

connect(host, port=5038, buffer_size=0)

Connect to the manager interface

connected()

Check if we are connected or not.

dbdel(family, key)

dbdeltree(family, key)

dbget(family, key)

dbput(family, key, val)

event_dispatch()

This thread is responsible for dispatching events

extension_state(exten, context)

Get the state of an extension

get_actionID()

Return an unique actionID, with a shared prefix for all actionIDs generated by this Manager instance.

hangup(channel)

Hangup the specified channel

iaxregistry()

login(username, secret)

Login to the manager, throws ManagerAuthException when login falis

logoff()

Logoff from the manager

mailbox_count(mailbox)

mailbox_status(mailbox)

Get the status of the specified mailbox

message_loop()

The method for the event thread.

This actually receives all types of messages and places them in the proper queues.

next_seq()

Return the next number in the sequence, this is used for ActionID

originate(channel, exten, context='', priority='', timeout='', application='', data='', caller_id='', run_async=False, earlymedia='false', account='', variables={})

Originate a call

originate_vars: Dict = {}

ping()

Send a ping action to the manager

playdtmf(channel, digit)

Plays a dtmf digit on the specified channel

redirect(channel, exten, priority='1', extra_channel='', context='')

Redirect a channel

register_event(event, function)

Register a callback for the specified event.

If a callback function returns True, no more callbacks for that event will be executed.

reload(module)

Reloads config for a given module

send_action(cdict: Dict, **kwargs)

Send a command to the manager

If a list is passed to the cdict argument, each item in the list will be sent to asterisk under the same header in the following manner:

cdict = {"Action": "Originate", "Variable": ["var1=value", "var2=value"]}
send_action(cdict)

...

Action: Originate Variable: var1=value Variable: var2=value

sippeers()

sipshowpeer(peer)

sipshowregistry()

status(channel='')

Get a status message from asterisk

unregister_event(event, function)

Unregister a callback for the specified event.

exception asterisk.manager.ManagerAuthException

Bases: ManagerException

exception asterisk.manager.ManagerException

Bases: Exception

class asterisk.manager.ManagerMsg(response)

Bases: object

A manager interface message

get_header(hname, defval=None)

Return the specified header

has_header(hname)

Check for a header

parse(response)

Parse a manager message

exception asterisk.manager.ManagerSocketException

Bases: ManagerException