StarPy Asterisk Protocols for Twisted

StarPy is a Python + Twisted protocol that provides access to the Asterisk PBX's Manager Interface (AMI) and Fast Asterisk Gateway Interface (FastAGI). Together these allow you write both command-and-control interfaces (used, for example to generate new calls) and to customise user interactions from the dial-plan.  You can readily write applications that use the AMI and FastAGI protocol together with any of the already-available Twisted protocols.

StarPy is primarily intended to allow Twisted developers to add Asterisk connectivity to their Twisted applications.  It isn't really targeted at the normal AGI-writing populace, as it requires understanding Twisted's asynchronous programming model.  That said, if you do know Twisted, it can readily be used to write stand-alone FastAGIs.

StarPy is still early in its life cycle.  Its author is very interested in contributions, bug reports and feedback and is available for extending/customisation contracts if you need more help.

Installation

StarPy is a pure-Python distutils extension.  Simply unpack the source archive to a temporary directory and run:

python setup.py install

You will need Python 2.3+ and Twisted (Core) installed. You'll need BasicProperty as well.  If you want to check out the SVN version instead of a released version, use:

svn co https://svn.sourceforge.net/svnroot/starpy/trunk starpy

On your PythonPath.

The demonstration applications use the utilapplication module, which uses configuration-file-based setup of the AMI and FastAGI servers.  To use this, create a starpy.conf file for the current directory (directory from which to run an example script) or a ~/.starpy.conf user-global file.  Content of the configuration file(s) looks like this:

[AMI]
username=AMIUSERNAME
secret=AMIPASSWORD
server=127.0.0.1
port=5038

[FastAGI]
port=4573
interface=127.0.0.1
context=survey

Keep in mind that FastAGI applications are neither encrypting nor authenticating; you probably should not expose them on any interface other than local (127.0.0.1)!

Asterisk Manager Interface (AMI) Usage

StarPy provides most of the hooks you want to use on the protocol instances.  The AMI client is created by a client factory, as is standard for Twisted operation.  You can create a factory manually like so:

from starpy import manager
f = manager.AMIFactory(sys.argv[1], sys.argv[2])
df = f.login('server',port)

The factory takes the username and secret (password) for the Asterisk manager interface (note: do not actually pass in these values on the command-line in a real application, as this would expose the username and password to anyone on the machine).  The deferred object returned from the login call will fire when the AMI connection has been established and authenticated.  You register callbacks on the deferred to accomplish those tasks you'd like to accomplish.

You will need to configure Asterisk to have the AMI enabled and choose the username, password and allowed hosts in /etc/asterisk/manager.conf.  You will also need to be sure that the AMI user has sufficient permissions to carry out whatever AMI operations you want to perform:

[USERNAME]
secret=SECRETPASSWORD
permit=127.0.0.1
read = system,call,log,verbose,command,agent,user
write = system,call,log,verbose,command,agent,user

Please keep in mind that the AMI interface is not encrypted, so should never be run across an insecure network.  If you need to run across such a network, use ssh tunnelling or the like to prevent eavesdropping!  You will want to read up on the AMI in the voip-info Wiki.

The return value for the login() deferred is an AMIProtocol instance.  The various methods on the AMIProtocol generally handle the creation and interpretation of "Action ID" fields.  The return value for most methods is an event, message or list of events.  Messages and events are modeled as dictionaries with lower-case keys.

Perhaps the most common task desired for use with the AMI Protocol is the creation of new calls.  Here's a snippet showing such generation:

self.ami.originate( 
	self.callbackChannel,
	self.ourContext, id(self), 1,
	timeout = 15,
)

You will likely want to ignore the results of the originate, and instead use an equal timeout waiting for an AGI connection to determine whether you have connected (the AMI originate can "succeed" without a successful connection, and will not tell you what channel is created).  If you want to track whether you have returned from a particular call to originate, use a different extension for each originate call (you can use UtilApplication's waitForCallOn method to register a one-shot handler if you are using UtilApplication).

Another common task is watching for an event of a particular type, for instance a "Hangup" event.  The AMIProtocol instance has a method registerEvent that allows you to add a handler to be called whenever an event of a given type is observed.

def onChannelHangup( ami, event ):
	"""Deal with the hangup of an event"""
	if event['uniqueid'] == self.uniqueChannelId:
		log.info( """AMI Detected close of our channel: %s""", self.uniqueChannelId )
		self.stopTime = time.time()
		# give the user a few seconds to put down the hand-set
		reactor.callLater( 2, df.callback, event )
		self.ami.deregisterEvent( 'Hangup', onChannelHangup )
self.ami.registerEvent( 'Hangup', onChannelHangup )
return df.addCallback( self.onHangup, callbacks=5 )

Note that the registerEvent and deregisterEvent methods use object identity to manage the callbacks being stored, as a result, a method is not a good handler (since method objects are created and destroyed each time they are accessed) to choose.  A nested function that can be passed to deregisterHandler is generally a better choice.  Eventually we may use PyDispatcher for the registration as it has solved this problem already in a far more general way.

See the examples/connecttoivr.py and examples/calldurationcallback.py scripts for sample usage of the AMIProtocol

Note that StarPy uses floating-point seconds for all time values in all interfaces,

Fast Asterisk Gateway Interface (FastAGI) Usage

Again, most of the hooks you want to use are provided on the protocol instances.  FastAGI is a server, and is thus created by a (non-client) factory like so:

from starpy import fastagi
f = fastagi.FastAGIFactory(testFunction)
reactor.listenTCP( 4573, f, 50, '127.0.0.1')

testFunction in the example above is the operation to undertake when the Asterisk Server connects to the FastAGI server.  It takes a (connected) FastAGIProtocol instance as its only argument.

This FastAGI protocol has methods available which match those AGI functions documented in the voip-info wiki.  Each method has basic documentation in the automated reference linked above, but you will want to use the wiki documentation to understand the semantics of the calls.  Keep in mind that the execute method (known as exec (which is a Python keyword) in the AGI documentation) allows you to access Asterisk Applications as well as AGI methods.

You use a FastAGI application from your Dial Plan like this (note: arguments do not appear to be passed to FastAGI scripts in Asterisk 1.2.1, unlike regular AGI scripts):

exten => 1000,3,AGI(agi://127.0.0.1:4573)

Please keep in mind that the FastAGI interface is neither encrypted nor authenticating!  It should never be run across an insecure network and should never be run on a port that is accessible from a public network.  Also keep in mind that your FastAGI process must be running already when Asterisk tries to connect to it, you need to code your FastAGI process to be robust so that it is always available to Asterisk.

See the examples directory for examples of FastAGI scripts.

Note that StarPy uses floating-point seconds for all time values in all interfaces,

Sequential Operations

The InSequence class allows for easily setting up multiple chained deferred processes, for instance when you want to play 2 or 3 sound files sequentially.  It is used like this: 
sequence = fastagi.InSequence()
sequence.append( agi.setContext, agi.variables['agi_context'] )
sequence.append( agi.setExtension, agi.variables['agi_extension'] )
sequence.append( agi.setPriority, int(agi.variables['agi_priority'])+difference )
sequence.append( agi.finish )
return sequence()

Calling the populated sequence returns a deferred which fires when all elements finish, or any element fails (raise an exception/failure).  The InSequence class is a trivial convenience that avoids needing to define a new callable function for every operation of a many-step operation.

Menu Objects Usage

The FastAGI interface includes basic support for creating hierarchic IVR menus.  The purpose of the menuing system is to encapsulate common UI functionality at a higher level of abstraction than that seen in the raw FastAGI interface.  Menus are defined using "model" classes which describe the desired features of the menu.  An example Menu using simple single-digit Option instances:

m = menu.Menu(
	tellInvalid = False, # don't report incorrect selections
	prompt = 'atlantic',
	options = [
		menu.Option( option='0' ),
		menu.Option( option='#' ),
		menu.ExitOn( option='*' ),
	],
	maxRepetitions = 5,
)

To invoke the menu, simply call it with a FastAGI protocol instance as its first argument.  The menu will repeat up to maxRepetitions times if an invalid or null entry is chosen.  If tellInvalid is True, the menu will play an "invalid entry" message of your choosing on an unrecognised entry, otherwise it will ignore invalid choices.

If a callable option is specified, such as ExitOn or SubMenu, the result of calling that option with the AGI and the selected option will be returned.  This same mechanism allows for creating chained sub-menus like so:

menu.SubMenu( 
	option='1',
	menu = menu.Menu(
		tellInvalid = False, # don't report incorrect selections
		prompt = ['atlantic',menu.DigitsPrompt(53),menu.DateTimePrompt(time.time())],
		options = [
			menu.Option( option='0' ),
			menu.Option( option='#' ),
			menu.ExitOn( option='*' ),
		],
	),
),

which can be used as an option within a higher-level menu.

You can also specify an onSuccess callback in the Option, this will be called (and it's value returned) if and only if that specific Option is chosen by the user (it is called only if the Option is not itself callable (which regular Option instances are not)).

The return value from a Menu is a chain of [ (option, digit), ... ] pairs for the final option selected from the lowest-level menu.  An ExitOn option triggers a return to a higher-level menu; this is not reported as a "final" option selection.

The Menu module includes a CollectDigits class which may be used either as a top-level Menu or as a SubMenu-wrapped option in a higher-level menu:

menu.SubMenu(
	option='2',
	menu = menu.CollectDigits(
		soundFile = 'extension',
		maxDigits = 5,
		minDigits = 3,
	),
)

Eventually the CollectDigits class should support review/cancel options on completion.  It would also be nice to get it to use the prompt system, but as of yet I don't know of any way to make that work with multi-character entry during the various sayXXX functions.

Signalling Errors from FastAGI

The FastAGIProtocol has a method jumpOnError which is intended to be used for implementing the common Asterisk application pattern of setting priority to some large value beyond the current value in order to indicate an error in the application.  Yes, it's an ugly way to signal errors, but there it is.  To use, add jumpOnError to a deferred where you want any uncaught exception to trigger a jump and finish the AGI connection.  This would normally be the overall deferred for your entire FastAGI operation.

df.addErrback( agi.jumpOnError, 100 )

If you only want to cause a particular jump on a particular error/exception or set of exceptions, you can pass in a (tuple of) error classes in the forErrors argument to which to restrict the jump:

df.addErrback( agi.jumpOnError, 50, forErrors=error.OnUnknownUser )

Secondary Services

The utilapplication module contains a few simple classes which provide common services for writing AMI/FastAGI applications.  This includes configuration-file setup of AMI and FastAGI services and an application instance that provides methods for registering to handle incoming FastAGI extensions.

# map incoming calls to extension 's' to the given method onS
APPLICATION.handleCallsFor( 's', someObject.onS )

UtilApplication's agiSpecifier and amiSpecifier property point to automatically generated AGISpecifier and AMISpecifier instances whose parameters are loaded from configuration files.  The specifier instances provide methods for starting up instances configured by the specifier:

# tell the application to run a FastAGI server which dispatches
# to handlers registered with handleCallsFor (as above)
APPLICATION.agiSpecifier.run( APPLICATION.dispatchIncomingCall )

# tell the application to log into the configured AMI server
# to allow for further management operations
df = APPLICATION.amiSpecifier.login( 
).addCallback( self.onAMIConnect )

Simple FastAGI Application Example

The following is the hellofastagiapp sample application, it uses the starpy.conf file in the current directory to control the FastAGI setup, and shows use of the utilapplication handleCallsFor method, which allows for a single FastAGI server handling many different FastAGI scripts (though in this case we only register a handler for one extension, 's'):

#! /usr/bin/env python
"""FastAGI server using starpy and the utility application framework

This is basically identical to hellofastagi, save that it uses the application
framework to allow for configuration-file-based setup of the AGI service.
"""
from twisted.internet import reactor
from starpy import fastagi, utilapplication
import logging, time

log = logging.getLogger( 'hellofastagi' )

def testFunction( agi ):
	"""Demonstrate simplistic use of the AGI interface with sequence of actions"""
	log.debug( 'testFunction' )
	sequence = fastagi.InSequence()
	sequence.append( agi.sayDateTime, time.time() )
	sequence.append( agi.finish )
	def onFailure( reason ):
		log.error( "Failure: %s", reason.getTraceback())
		agi.finish()
	return sequence().addErrback( onFailure )

if __name__ == "__main__":
	logging.basicConfig()
	fastagi.log.setLevel( logging.DEBUG )
	APPLICATION = utilapplication.UtilApplication()
	APPLICATION.handleCallsFor( 's', testFunction )
	APPLICATION.agiSpecifier.run( APPLICATION.dispatchIncomingCall )
	reactor.run()

Changes

StarPy can be downloaded from the project's File Download area.

License

StarPy is licensed under extremely liberal terms.

Copyright (c) 2006, Michael C. Fletcher
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:

	Redistributions of source code must retain the above copyright
	notice, this list of conditions and the following disclaimer.

	Redistributions in binary form must reproduce the above
	copyright notice, this list of conditions and the following
	disclaimer in the documentation and/or other materials
	provided with the distribution.

	The name of Michael C. Fletcher, or the name of any Contributor,
	may not be used to endorse or promote products derived from this 
	software without specific prior written permission.

THIS SOFTWARE IS NOT FAULT TOLERANT AND SHOULD NOT BE USED IN ANY
SITUATION ENDANGERING HUMAN LIFE OR PROPERTY.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
COPYRIGHT HOLDERS AND CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
OF THE POSSIBILITY OF SUCH DAMAGE.

StarPy  is a SourceForge.net Logo Open-Source Project