Event Handling

A Call produces discrete events (for example hangup and errors) and continuous audio streams. Register connector-level handlers before run_forever(), and attach call-level handlers after process_call.

Connector-level notifications

Register on SessionManager before run_forever():

@sm.on_incoming_call
async def on_call(noti: IncomingCallNotification):
    # noti.call_id, noti.subscriber, noti.participant
    ...

Each inbound call handler typically runs in its own task (concurrent calls). Notifications that arrive between connect and run_forever() are buffered, not dropped.

Call events

from agentduet import CallEvent
@call.on_hangup
def on_hangup(evt):
    # payload is always None
    print("Call ended", call.id)
@call.on_call_event(CallEvent.ERROR)
def on_error(evt):
    # dict with error_code and optional error_message
    print(evt["error_code"], evt.get("error_message"))

on_hangup is shorthand for on_call_event(CallEvent.HANGUP). Handlers take exactly one argument. Sync handlers run on a worker thread so they do not stall audio delivery.

CallEvent in 1.0.0: HANGUP, ERROR.

You can also register without a decorator:

call.on_hangup(my_handler)

Audio streams

from agentduet import CallClosedError
try:
    async for audio_chunk in call.caller.audio_stream():
        await call.send_audio(await process(audio_chunk))
except CallClosedError:
    pass  # hangup while a send was in flight - normal

Once the call terminates, send_audio / clear_send_audio_buffer raise CallClosedError. A hangup can land mid-response - catch it as the stop signal and cancel model tasks in your hangup handler.

Ordering and concurrency

• Notifications are connector-wide competing-consumer: concurrent and possibly out of order. Correlate by (subscriber, participant) or call_id.
• Dedup redelivered notifications (at-least-once).
• Do not assume audio chunks and hangup events are ordered relative to your model’s network I/O - always clear the buffer on interrupt and cancel tasks on hangup.

Next Step