Asterisk fastapi 1.1.2

Source code Github

What is this program for??

Create web server on the same server as asterisk. This server can:

1. Provide asterisk calls history by http endpoint like start_datetime and end_datetime (mysql,postgresql,sqlite,csv)
2. Provide webhooks for asterisk. Allows you to send request to your site, for example, an incoming call that was answered or a missed call. Next, your server saves this to the database and notifies clients (browsers) through any mechanism (websockets, longpolling) that a call has arrived and, for example, by calling a pop-up window and creating a lead or opening a partner's card
3. Also provide recordngs although they are available in asterisk ari, just to address the same address. (essentially a duplication)
4. Endpoint numbers list
5. Endpoint checkup ( getting the status of the service) After launching the service, the documentation with the available endpoints will be at your_ir:8082/docs Simple http base authentication is also enabled, the username and password are taken from the config to protect your data in asterisk.

How configure Asterisk?

1. Enable the Asterisk HTTP service in /etc/asterisk/http.conf:

[general]
enabled = yes
bindaddr = 0.0.0.0
bindport = 8088

2. Configure an ARI user in /etc/asterisk/ari.conf:

[general]
enabled = yes
pretty = yes
allowed_origins = localhost:8088,http://ari.asterisk.org
channelvars = linkedid

[asterisk-supersecret]
type = user
read_only = no
password = $6$nqvAB8Bvs1dJ4V$8zCUygFXuXXp8EU3t2M8i.N8iCsY4WRchxe2AYgGOzHAQrmjIPif3DYrvdj5U2CilLLMChtmFyvFa3XHSxBlB/
password_format = crypt

How configure app?

By default, the environment data is read from the .env file. Perhaps the .env.sample file as an example will help you (just rename that). Please set your credentials to it file before work.

How start app?

1 Variant. Start on host.

On your asterist server. Setup python enviroment. Python version 3.11.0 or more. A best practice among Python developers is to use a project-specific virtual environment. Once you activate that environment, any packages you then install are isolated from other environments, including the global interpreter environment, reducing many complications that can arise from conflicting package versions. You can create non-global environments in VS Code using Venv or Anaconda

  python -m venv .venv
  python -m pip install -r requirements.txt

Start from root folder backend web server (ASGI) as service

uvicorn main:app --host 127.0.0.1 --port 8082 --log-level debug

or

python3 -m uvicorn main:app --host 127.0.0.1 --port 8082 --log-level debug --workers 2

2 Variant. Start on docker.(no tested)

On your asterist server. Setup docker enviroment.

Start from root folder backend web server (ASGI) as docker service

  docker-compose -f docker-compose.yml up

Use as a library (embed in your app)

The reusable core now lives in the importable asterisk_agent package. The bundled FastAPI service (main.py) is just a thin layer on top of it, and its behaviour is unchanged. Another application (e.g. a CRM) can install the package and embed the same code directly — listening to ARI events in-process and reading CDR straight from the database, without the HTTP webhook / REST hop.

Install (only the WS + CDR core is required; DB driver and AMI are optional extras):

pip install /path/to/asterisk_python_fastapi            # core: pydantic, httpx, websockets
pip install "/path/to/asterisk_python_fastapi[mysql]"   # + aiomysql   (or [postgres] / [sqlite])
pip install "/path/to/asterisk_python_fastapi[ami]"     # + panoramisk (only if you use AMI)

1. Listen to ARI events in-process

WebsocketEvents connects to the ARI websocket and applies the same events_ignore / events_used filtering. Pass on_event to receive each (already filtered) event and handle it yourself instead of POSTing it to a webhook URL:

import asyncio
from asterisk_agent import AriConfig, WebsocketEvents

async def handle(event: dict) -> None:
    # your own processing — e.g. call your app's webhook handler directly
    print(event["type"], event.get("channel", {}).get("id"))

ari = AriConfig(
    url="http://mypbx:8088/ari",
    wss="ws://mypbx:8088/ari/events",
    login="freepbxuser",
    password="secret",
    events_ignore=["ChannelVarset", "ChannelDialplan"],
    events_used=["ChannelStateChange", "ChannelDestroyed", "ChannelHangupRequest"],
)
ws = WebsocketEvents(
    api_key_base64="",           # not needed when on_event is provided
    api_key=f"{ari.login}:{ari.password}",
    ari_config=ari,
    on_event=handle,             # <- in-process handler (no HTTP)
)
asyncio.run(ws.run_forever())    # supervised connect + reconnect loop

Without on_event, behaviour is identical to the standalone service (HTTP POST to webhook_url).

2. Read CDR straight from the Asterisk database

Build a lightweight DbConfig (no .env needed) and use the same DB strategies the service exposes over REST:

import asyncio
from asterisk_agent import DbConfig, get_db_connector

db = get_db_connector(DbConfig(
    db_dialect="mysql",
    db_host="127.0.0.1", db_port=3306,
    db_database="asteriskcdrdb", db_user="root", db_password="secret",
    db_table_cdr_name="cdr",
))

async def main():
    await db.check_cdr_old()                              # detect calldate vs start column
    rows = await db.get_cdr_uniqueid_or_linkedid("1715866158.71448")
    history = await db.get_cdr("2024-05-16 00:00:00", "2024-05-16 23:59:59")
    print(len(history))

asyncio.run(main())

Recordings can be fetched via ARI without the service: Ari(ari_url, api_key).call_recording(filename).