For AI agents: a documentation index is available at the root level at /llms.txt and /llms-full.txt. Append /llms.txt to any URL for a page-level index, or .md for the markdown version of any page.
  • Introduction
    • Overview
    • Installation
    • Developer Onboarding
    • Quick Start
  • Concepts
    • Architecture
    • Call States and Lifecycle
    • Call Commands
    • Audio Streaming
    • Event Handling
  • Integrations
    • Google GenAI SDK (Gemini Live)
    • Google ADK (Agent Development Kit)
  • Use Cases
    • After-hours Voicemail
    • Appointment Booking
    • Call Monitoring and Coaching
    • Database Lookup
    • Human Escalation
    • Interactive Notifications
  • Reference
    • API Reference
    • Error Handling
    • Advanced Topics
LogoLogo
On this page
  • Decorator-based Events
  • Available Events
  • Async Iterators
  • Combining Events and Streams
  • Next Steps
Concepts

Event Handling

||View as Markdown|
Was this page helpful?
Previous

Audio Streaming

Next

Google GenAI SDK (Gemini Live)

Built with

The SDK provides two mechanisms for handling events: decorator-based handlers for discrete events, and async iterators for continuous audio streams.

Decorator-based Events

Use decorators for discrete, one-time events:

1# Client-level event
2@client.on(events.INCOMING_CALL)
3async def handle_call(call: ActiveCall):
4 await call.answer()
5
6# Call-level event
7@call.on(events.CALL_TERMINATED)
8def on_terminated():
9 print("Call ended")

Available Events

EventLevelDescription
events.INCOMING_CALLClientA new call has arrived
events.CALL_TERMINATEDCallThe call has ended
events.CALL_ANSWEREDCallThe call was successfully answered
events.CALL_CONNECTEDCallThe connect (conference) succeeded
events.CALL_CONNECT_FAILEDCallThe connect failed
events.CALL_ANSWER_FAILEDCallThe answer attempt failed

Client-level events are registered on the AgenTaoClient instance and fire for any call.

Call-level events are registered on a specific ActiveCall instance and fire only for that call.

Async Iterators

Use async iterators for continuous audio streams:

1async for audio_chunk in call.audio_stream():
2 processed = await process_audio(audio_chunk)
3 await call.send_audio(processed)

The iterator yields audio chunks as they arrive. It terminates when the call ends or the media connection closes.

Combining Events and Streams

A typical pattern combines a call-level event with the audio stream:

1@client.on(events.INCOMING_CALL)
2async def handle_call(call: ActiveCall):
3 @call.on(events.CALL_TERMINATED)
4 def on_terminated():
5 print("Call ended")
6
7 await call.answer()
8
9 async for chunk in call.audio_stream():
10 response = await ai_model.generate(chunk)
11 await call.send_audio(response)

The public 0.24.0 package docs also mention ClientEvent and CallEvent enums as type-safe alternatives to the string constants in agentao_sdk.events.

Next Steps

  • Call States - See which events correspond to which state transitions
  • Call Commands - Commands that trigger state changes
  • API Reference - Full method signatures for event registration