Aioagi
======
Async agi client/server framework.
The project based on "aiohttp" framework.
Key Features
============
- Supports both client and server side of AGI protocol.
- AGI-server has middlewares and pluggable routing.
Getting started
===============
Server
------
Simple AGI server:
.. code-block:: python
import asyncio
from aiohttp.web import Application, AppRunner, TCPSite, Response
from aioagi import runner
from aioagi.app import AGIApplication
from aioagi.log import agi_server_logger
from aioagi.urldispathcer import AGIView
from aiohttp.web_runner import GracefulExit
async def hello(request):
message = await request.agi.stream_file('hello-world')
await request.agi.verbose('Hello handler: {}.'.format(request.rel_url.query))
agi_server_logger.debug(message)
async def http_hello(request):
return Response(text="Hello, world")
class HelloView(AGIView):
async def sip(self):
message = await self.request.agi.stream_file('hello-world')
await self.request.agi.verbose('HelloView handler: {}.'.format(self.request.rel_url.query))
agi_server_logger.debug(message)
if __name__ == '__main__':
app = AGIApplication()
app.router.add_route('SIP', '/', hello)
runner.run_app(app)
# OR
if __name__ == '__main__':
apps = []
app = AGIApplication()
app.router.add_route('SIP', '/', hello)
http_app = Application()
http_app.router.add_route('GET', '/', http_hello)
loop = asyncio.get_event_loop()
runners = []
sites = []
for _app in [app, http_app]:
app_runner = AppRunner(_app)
loop.run_until_complete(app_runner.setup())
if isinstance(_app, AGIApplication):
sites.append(runner.AGISite(app_runner, port=8081))
else:
sites.append(TCPSite(app_runner, port=8080))
runners.append(app_runner)
for site in sites:
loop.run_until_complete(site.start())
uris = sorted(str(s.name) for runner in runners for s in runner.sites)
print("======== Running on {} ========\n"
"(Press CTRL+C to quit)".format(', '.join(uris)))
try:
loop.run_forever()
except (GracefulExit, KeyboardInterrupt): # pragma: no cover
pass
finally:
for runner in reversed(runners):
loop.run_until_complete(runner.cleanup())
if hasattr(loop, 'shutdown_asyncgens'):
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
Client
------
To set AGI connection as Asterisk:
.. code-block:: python
import asyncio
import logging.config
from aioagi.log import agi_client_logger
from aioagi.client import AGIClientSession
from aioagi.parser import AGIMessage, AGICode
async def test_request(loop):
headers = {
'agi_channel': 'SIP/100-00000001',
'agi_language': 'ru',
'agi_uniqueid': '1532375920.8',
'agi_version': '14.0.1',
'agi_callerid': '100',
'agi_calleridname': 'test',
'agi_callingpres': '0',
'agi_callingani2': '0',
'agi_callington': '0',
'agi_callingtns': '0',
'agi_dnid': '101',
'agi_rdnis': 'unknown',
'agi_context': 'from-internal',
'agi_extension': '101',
'agi_priority': '1',
'agi_enhanced': '0.0',
'agi_accountcode': '',
'agi_threadid': '139689736754944',
}
async with AGIClientSession(headers=headers, loop=loop) as session:
async with session.sip('agi://localhost:8080/hello/?a=test1&b=var1') as response:
async for message in response:
client_logger.debug(message)
await response.send(AGIMessage(AGICode.OK, '0', {}))
async with session.sip('agi://localhost:8080/hello-view/?a=test2&b=var2') as response:
async for message in response:
client_logger.debug(message)
await response.send(AGIMessage(AGICode.OK, '0', {}))
.. note:: Session request headers are set automatically for ``session.sip('agi://localhost:8080/hello/?a=test1&b=var1')`` request:
.. code-block:: bash
agi_type: SIP
agi_network: yes
agi_network_script: hello/
agi_request: agi://localhost:8080/hello/
AMI
---
.. code-block:: python
import asyncio
from aioagi.ami.action import AMIAction
from aioagi.ami.manager import AMIManager
async def callback(manager, message):
print(message)
async def main(app):
manager = AMIManager(
app=app, title='myasterisk',
host='127.0.0.1',
port=5038,
username='username',
secret='secret',
)
manager.register_event('*', callback)
app['manager'] = manager
await manager.connect()
await asyncio.sleep(2)
message = await manager.send_action(AMIAction({
'Action': 'Command',
'Command': 'database show',
}))
print(message)
print(message.body)
async def cleanup(app):
app['manager'].close()
if __name__ == '__main__':
app = {}
_loop = asyncio.get_event_loop()
try:
_loop.run_until_complete(main(app))
except KeyboardInterrupt:
_loop.run_until_complete(cleanup(app))
_loop.close()
Install
=======
``pip install aioagi``
Thanks
======
* Gael Pasgrimaud - panoramisk_
.. _panoramisk: https://github.com/gawel/panoramisk
Raw data
{
"_id": null,
"home_page": "https://gitlab.com/VadimShakurov/aioagi.git",
"name": "aioagi-ik",
"maintainer": null,
"docs_url": null,
"requires_python": null,
"maintainer_email": null,
"keywords": "aiogi asyncio asterisk telephony voip",
"author": "Shakurov Vadim Vladimirovich",
"author_email": "apelsinsd@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/11/bb/6656f134f0535f1538fb0aef8301a0bbfdcd3bd8a865f991df66e64df777/aioagi_ik-1.2.2.tar.gz",
"platform": null,
"description": "Aioagi\n======\n\nAsync agi client/server framework.\nThe project based on \"aiohttp\" framework.\n\nKey Features\n============\n\n- Supports both client and server side of AGI protocol.\n- AGI-server has middlewares and pluggable routing.\n\nGetting started\n===============\n\n\nServer\n------\n\nSimple AGI server:\n\n.. code-block:: python\n\n import asyncio\n\n from aiohttp.web import Application, AppRunner, TCPSite, Response\n\n from aioagi import runner\n from aioagi.app import AGIApplication\n from aioagi.log import agi_server_logger\n from aioagi.urldispathcer import AGIView\n from aiohttp.web_runner import GracefulExit\n\n\n async def hello(request):\n message = await request.agi.stream_file('hello-world')\n await request.agi.verbose('Hello handler: {}.'.format(request.rel_url.query))\n agi_server_logger.debug(message)\n\n\n async def http_hello(request):\n return Response(text=\"Hello, world\")\n\n\n class HelloView(AGIView):\n async def sip(self):\n message = await self.request.agi.stream_file('hello-world')\n await self.request.agi.verbose('HelloView handler: {}.'.format(self.request.rel_url.query))\n agi_server_logger.debug(message)\n\n\n if __name__ == '__main__':\n app = AGIApplication()\n app.router.add_route('SIP', '/', hello)\n runner.run_app(app)\n\n # OR\n if __name__ == '__main__':\n apps = []\n\n app = AGIApplication()\n app.router.add_route('SIP', '/', hello)\n\n http_app = Application()\n http_app.router.add_route('GET', '/', http_hello)\n\n loop = asyncio.get_event_loop()\n\n runners = []\n sites = []\n for _app in [app, http_app]:\n app_runner = AppRunner(_app)\n loop.run_until_complete(app_runner.setup())\n if isinstance(_app, AGIApplication):\n sites.append(runner.AGISite(app_runner, port=8081))\n else:\n sites.append(TCPSite(app_runner, port=8080))\n\n runners.append(app_runner)\n\n for site in sites:\n loop.run_until_complete(site.start())\n\n uris = sorted(str(s.name) for runner in runners for s in runner.sites)\n print(\"======== Running on {} ========\\n\"\n \"(Press CTRL+C to quit)\".format(', '.join(uris)))\n\n try:\n loop.run_forever()\n except (GracefulExit, KeyboardInterrupt): # pragma: no cover\n pass\n\n finally:\n for runner in reversed(runners):\n loop.run_until_complete(runner.cleanup())\n\n if hasattr(loop, 'shutdown_asyncgens'):\n loop.run_until_complete(loop.shutdown_asyncgens())\n loop.close()\n\n\nClient\n------\n\nTo set AGI connection as Asterisk:\n\n.. code-block:: python\n\n import asyncio\n import logging.config\n\n from aioagi.log import agi_client_logger\n from aioagi.client import AGIClientSession\n from aioagi.parser import AGIMessage, AGICode\n\n\n async def test_request(loop):\n headers = {\n 'agi_channel': 'SIP/100-00000001',\n 'agi_language': 'ru',\n 'agi_uniqueid': '1532375920.8',\n 'agi_version': '14.0.1',\n 'agi_callerid': '100',\n 'agi_calleridname': 'test',\n 'agi_callingpres': '0',\n 'agi_callingani2': '0',\n 'agi_callington': '0',\n 'agi_callingtns': '0',\n 'agi_dnid': '101',\n 'agi_rdnis': 'unknown',\n 'agi_context': 'from-internal',\n 'agi_extension': '101',\n 'agi_priority': '1',\n 'agi_enhanced': '0.0',\n 'agi_accountcode': '',\n 'agi_threadid': '139689736754944',\n }\n async with AGIClientSession(headers=headers, loop=loop) as session:\n async with session.sip('agi://localhost:8080/hello/?a=test1&b=var1') as response:\n async for message in response:\n client_logger.debug(message)\n await response.send(AGIMessage(AGICode.OK, '0', {}))\n\n async with session.sip('agi://localhost:8080/hello-view/?a=test2&b=var2') as response:\n async for message in response:\n client_logger.debug(message)\n await response.send(AGIMessage(AGICode.OK, '0', {}))\n\n.. note:: Session request headers are set automatically for ``session.sip('agi://localhost:8080/hello/?a=test1&b=var1')`` request:\n\n.. code-block:: bash\n\n agi_type: SIP\n agi_network: yes\n agi_network_script: hello/\n agi_request: agi://localhost:8080/hello/\n\n\nAMI\n---\n\n.. code-block:: python\n\n import asyncio\n\n from aioagi.ami.action import AMIAction\n from aioagi.ami.manager import AMIManager\n\n\n async def callback(manager, message):\n print(message)\n\n\n async def main(app):\n manager = AMIManager(\n app=app, title='myasterisk',\n host='127.0.0.1',\n port=5038,\n username='username',\n secret='secret',\n )\n manager.register_event('*', callback)\n app['manager'] = manager\n await manager.connect()\n\n await asyncio.sleep(2)\n\n message = await manager.send_action(AMIAction({\n 'Action': 'Command',\n 'Command': 'database show',\n }))\n print(message)\n print(message.body)\n\n\n async def cleanup(app):\n app['manager'].close()\n\n\n if __name__ == '__main__':\n app = {}\n _loop = asyncio.get_event_loop()\n try:\n _loop.run_until_complete(main(app))\n except KeyboardInterrupt:\n _loop.run_until_complete(cleanup(app))\n _loop.close()\n\n\nInstall\n=======\n\n``pip install aioagi``\n\n\nThanks\n======\n* Gael Pasgrimaud - panoramisk_\n\n.. _panoramisk: https://github.com/gawel/panoramisk\n",
"bugtrack_url": null,
"license": "Apache 2",
"summary": "Async agi client/server framework (asyncio)",
"version": "1.2.2",
"project_urls": {
"Homepage": "https://gitlab.com/VadimShakurov/aioagi.git"
},
"split_keywords": [
"aiogi",
"asyncio",
"asterisk",
"telephony",
"voip"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "308720da2ca215db4dbd6445ae899bc8b0d50e2a1703db8c5a41dff169eb6ad5",
"md5": "1e67c74fcc77c394352690a766550117",
"sha256": "6329aa48a325058c88307556ec9d78e8ffc655bb87e93986492fc77287cece2b"
},
"downloads": -1,
"filename": "aioagi_ik-1.2.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "1e67c74fcc77c394352690a766550117",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 40059,
"upload_time": "2025-01-14T19:21:13",
"upload_time_iso_8601": "2025-01-14T19:21:13.309305Z",
"url": "https://files.pythonhosted.org/packages/30/87/20da2ca215db4dbd6445ae899bc8b0d50e2a1703db8c5a41dff169eb6ad5/aioagi_ik-1.2.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "11bb6656f134f0535f1538fb0aef8301a0bbfdcd3bd8a865f991df66e64df777",
"md5": "33a1df38b4860f8689dabab498047a64",
"sha256": "b7c572eea697de6aa5ab29efbf3e733b0cf54f6a870930426d8c36807ad53c6e"
},
"downloads": -1,
"filename": "aioagi_ik-1.2.2.tar.gz",
"has_sig": false,
"md5_digest": "33a1df38b4860f8689dabab498047a64",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 35843,
"upload_time": "2025-01-14T19:21:18",
"upload_time_iso_8601": "2025-01-14T19:21:18.363346Z",
"url": "https://files.pythonhosted.org/packages/11/bb/6656f134f0535f1538fb0aef8301a0bbfdcd3bd8a865f991df66e64df777/aioagi_ik-1.2.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-14 19:21:18",
"github": false,
"gitlab": true,
"bitbucket": false,
"codeberg": false,
"gitlab_user": "VadimShakurov",
"gitlab_project": "aioagi",
"lcname": "aioagi-ik"
}
JSON