How to set up queue dialing via SIP and Python

This note analyzes the architecture of an automated outbound calling system: how the processing pipeline is organized, how dialing via Asterisk AMI works, and how the cycle closes from speech generation to result recording.

Architecture: a pipeline of queues

The system is built as a pipeline with explicit stages — audio generation, call creation, storage, response recognition. Each client goes through them sequentially, but the stages run in parallel with each other: while one call is in progress, audio for others is already being prepared and results are being written.

Each stage has its own queue and a pool of workers:

self.queues  = { "generate_voice_message": DelayedQueue(), "create_call": DelayedQueue(), ... }
self.workers = { "generate_voice_message": 5, "create_call": 1, "recognition": 3, "store_data": 10 }
self.stages  = {
    CallStatus.CREATED:   'generate_voice_message',
    CallStatus.GENERATED: 'create_call',
    CallStatus.PENDING:   'recognition'
}

Transitions between stages are controlled via CallStatus: the call status determines which queue it will enter next. DelayedQueue allows postponing retries without a separate scheduler — the delay is provided at the moment the task is enqueued.

Connecting to Asterisk AMI

The service starts a panoramisk.Manager and subscribes to channel events — Hangup and BridgeEnter. The calling process does not start while AMI is unavailable: before workers are started there is a loop checking availability via Ping.

From text to call

The first stage is audio generation. The message text (individual or a template from the DB) is sent to Yandex Cloud SpeechKit, resulting in a file named uuid4().hex, which is saved on the server in a directory accessible to Asterisk. The file name then travels with the call object and at the appropriate moment is passed as a dialplan variable — Asterisk will find and play it to the recipient.

When the audio is ready, the create_call worker forms an AMI Originate:

action = {
    'Action':   'Originate',
    'Channel':  f'PJSIP/11{cl.client.phone}@{trunk}',
    'Context':  'applications',
    'Exten':    2200101,
    'Priority': 1,
    'Async':    'true',
    'Variable': f'file_name={cl.generated},file_record={cl.recorded},lang={lang_value}'
}

Three things are passed into the dialplan via Variable: which file to play, where to write the call recording, and which language branch of the script to use. Async: true means AMI is not blocked waiting for the callee’s response — the command is sent and immediately returns a response.

Waiting for completion and retries

After a successful Originate the worker saves the channel identifier and waits: it loops with asyncio.sleep(1) until the Hangup event handler resets that identifier to an empty string. This is a simple but working way to synchronize the asynchronous AMI stream with the worker’s logic.

On Failure in the response — the call is re-enqueued into the same queue via DelayedQueue with a delay from the config. The limit is 6 attempts, after which the status is recorded as FAILED.

Storage, recognition and the finish

After each status change store_data performs an upsert into the DB and, based on the status value, decides what next: a final status (SUCCESSFUL / FAILED) — update the row in the Google Sheet; any other — put into the next queue according to the self.stages table.

The final stage — recognition — reads the WAV recording file and sends it to SpeechKit STT. If the file is not found or is too small, it is immediately marked FAILED. Otherwise — based on the recognition result the final status is recorded, and the cycle for that client is closed.

When this applies

The described approach fits well any task of mass outbound notifications: appointment reminders, order confirmations, debt collection calls, voice mailings to a segment. The key requirement is having Asterisk with AMI and a configured SIP trunk. Everything else — Python, queues and cloud TTS/STT — can be replaced with a specific stack.

The approach based on a state machine and delayed queues proves reliable where resilience to failures is important: TTS went down — the task will wait and be retried; the callee didn’t answer — the call will be retried after the required interval without manual intervention. All logic for transitions, retries and parallelism is concentrated in three structures — self.stages, self.workers and DelayedQueue — and can be read easily as a single system diagram.