Lifecycle Events

Callback interface for RealtimeAgent session lifecycle events.

Subclass this and pass an instance to RealtimeAgent via the listener parameter to react to speaking state changes, transcripts, errors, and session boundaries.

All methods are async no-ops by default — override only the ones you need.

Example

class MyListener(AgentListener):
    async def on_user_transcript(self, transcript: str) -> None:
        print(f"User: {transcript}")

    async def on_assistant_transcript(self, transcript: str) -> None:
        print(f"Assistant: {transcript}")


agent = RealtimeAgent(listener=MyListener())

on_agent_error async

on_agent_error(error: AgentError) -> None

Called when the agent or the Realtime API encounters an error.

Parameters:

Name	Type	Description	Default
error	AgentError	Structured error information including type, message, and optional code and parameter.	required

on_agent_interrupted async

on_agent_interrupted() -> None

Called when the assistant's response is interrupted by the user speaking.

on_agent_session_connected async

on_agent_session_connected() -> None

Called once the WebSocket session has been established and is ready.

on_agent_starting async

on_agent_starting() -> None

Called immediately when run() is invoked, before any I/O or WebSocket setup.

Use this to show loading states in the UI before the session is ready.

on_agent_stopped async

on_agent_stopped() -> None

Called after the agent has fully shut down and run() is about to return.

on_assistant_started_responding async

on_assistant_started_responding() -> None

Called when the assistant begins streaming an audio response.

on_assistant_stopped_responding async

on_assistant_stopped_responding() -> None

Called when the assistant has finished streaming its audio response.

on_assistant_transcript async

on_assistant_transcript(transcript: str) -> None

Called when the assistant has finished generating a response and its transcript is complete.

Parameters:

Name	Type	Description	Default
transcript	str	The full transcript of the assistant's response.	required

on_assistant_transcript_delta async

on_assistant_transcript_delta(delta: str) -> None

Called when a partial assistant text delta is streamed.

This is emitted for text output events such as response.text.delta and response.output_text.delta.

Parameters:

Name	Type	Description	Default
delta	str	Incremental transcript text chunk.	required

on_subagent_finished async

on_subagent_finished(agent_name: str) -> None

Called when a subagent finishes running.

Parameters:

Name	Type	Description	Default
agent_name	str	Name of the finished subagent.	required

on_subagent_started async

on_subagent_started(agent_name: str) -> None

Called when a subagent starts running.

Parameters:

Name	Type	Description	Default
agent_name	str	Name of the started subagent.	required

on_user_inactivity_countdown async

on_user_inactivity_countdown(
    remaining_seconds: int,
) -> None

Called each second during the countdown before the inactivity timeout fires.

Fires at remaining_seconds = 5, 4, 3, 2, 1.

Parameters:

Name	Type	Description	Default
remaining_seconds	int	Seconds remaining until the session is stopped.	required

on_user_started_speaking async

on_user_started_speaking() -> None

Called when VAD detects that the user has started speaking.

on_user_stopped_speaking async

on_user_stopped_speaking() -> None

Called when VAD detects that the user has stopped speaking.

on_user_transcript async

on_user_transcript(transcript: str) -> None

Called when the user's speech has been fully transcribed.

Only fires if a transcription_model is configured on the agent.

Parameters:

Name	Type	Description	Default
transcript	str	The finalised transcript text for the current user turn.	required

Error information received in AgentListener.on_agent_error.

This object is created internally and passed to your listener — you do not construct it yourself.

Attributes:

Name	Type	Description
type	str	OpenAI error type identifier (e.g. invalid_request_error).
message	str	Human-readable description of the error.

Example

class MyListener(AgentListener):
    async def on_agent_error(self, error: AgentError) -> None:
        if error.type == "invalid_request_error":
            print(f"Bad request: {error.message}")

message instance-attribute

message: str

Human-readable description of the error.

type instance-attribute

type: str

OpenAI error type identifier (e.g. invalid_request_error).