Skip to main content
The event system lets you react to what’s happening in your agent — participant joins, transcriptions, LLM responses, errors, and more. Subscribe to events using the @agent.events.subscribe decorator.
For a complete list of available events, see Events Reference.

​
Subscribing to Events

Use the @agent.events.subscribe decorator with a type hint to specify which event you want: 
from vision_agents.core.events import CallSessionParticipantJoinedEvent

@agent.events.subscribe
async def handle_participant_joined(event: CallSessionParticipantJoinedEvent):
    if event.participant.user.id == "agent":
        return  # Skip agent's own join event
    
    await agent.simple_response(f"Hello {event.participant.user.name}!")
The type hint determines which events trigger your handler.

​
Common Events

EventWhenImport
CallSessionParticipantJoinedEventUser joins callvision_agents.core.events
CallSessionParticipantLeftEventUser leaves callvision_agents.core.events
CallSessionStartedEventCall session beginsvision_agents.core.events
STTTranscriptEventSpeech transcribedvision_agents.core.stt.events
LLMResponseCompletedEventLLM finishes responsevision_agents.core.llm.events
PluginErrorEventPlugin encounters errorvision_agents.core.events
ToolStartEvent / ToolEndEventFunction callingvision_agents.core.llm.events

​
Example: Greeting Participants

import asyncio
from vision_agents.core.agents import Agent
from vision_agents.core.edge.types import User
from vision_agents.core.events import (
    CallSessionParticipantJoinedEvent,
    CallSessionParticipantLeftEvent,
)
from vision_agents.plugins import openai, getstream, deepgram, elevenlabs

async def main():
    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Assistant", id="agent"),
        instructions="You're a helpful voice assistant.",
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=elevenlabs.TTS(),
        stt=deepgram.STT(),
    )
    
    @agent.events.subscribe
    async def on_join(event: CallSessionParticipantJoinedEvent):
        if event.participant.user.id != "agent":
            await agent.simple_response(f"Welcome, {event.participant.user.name}!")
    
    @agent.events.subscribe
    async def on_leave(event: CallSessionParticipantLeftEvent):
        if event.participant.user.id != "agent":
            await agent.simple_response(f"Goodbye, {event.participant.user.name}!")
    
    await agent.create_user()
    call = agent.edge.client.video.call("default", "my-call")
    
    async with agent.join(call):
        await agent.finish()

if __name__ == "__main__":
    asyncio.run(main())

​
Listening to Component Events

Subscribe to events from specific components using their events property: 
from vision_agents.core.stt.events import STTTranscriptEvent
from vision_agents.core.llm.events import LLMResponseCompletedEvent

# STT transcripts
@agent.events.subscribe
async def on_transcript(event: STTTranscriptEvent):
    print(f"User said: {event.text}")

# LLM responses
@agent.events.subscribe
async def on_response(event: LLMResponseCompletedEvent):
    print(f"Agent said: {event.text}")
For Realtime LLMs, use the transcription events: 
from vision_agents.core.llm.events import (
    RealtimeUserSpeechTranscriptionEvent,
    RealtimeAgentSpeechTranscriptionEvent,
)

@agent.events.subscribe
async def on_user_speech(event: RealtimeUserSpeechTranscriptionEvent):
    print(f"User: {event.text}")

@agent.events.subscribe
async def on_agent_speech(event: RealtimeAgentSpeechTranscriptionEvent):
    print(f"Agent: {event.text}")

​
Error Handling

Subscribe to error events for monitoring and recovery: 
from vision_agents.core.events import PluginErrorEvent

@agent.events.subscribe
async def on_error(event: PluginErrorEvent):
    print(f"Error in {event.plugin_name}: {event.error_message}")
    
    if event.is_fatal:
        await agent.simple_response("I'm having technical difficulties.")

​
Multiple Event Types

Handle related events in one handler using Union: 
from typing import Union
from vision_agents.core.events import (
    CallSessionParticipantJoinedEvent,
    CallSessionParticipantLeftEvent,
)

@agent.events.subscribe
async def on_participant_change(
    event: Union[CallSessionParticipantJoinedEvent, CallSessionParticipantLeftEvent]
):
    action = "joined" if isinstance(event, CallSessionParticipantJoinedEvent) else "left"
    print(f"{event.participant.user.name} {action}")

​
Best Practices

Check the event source — Filter out the agent’s own events to avoid loops: 
if event.participant.user.id == "agent":
    return
Keep handlers focused — One handler per concern: 
# Good: separate handlers
@agent.events.subscribe
async def log_transcripts(event: STTTranscriptEvent):
    logger.info(f"Transcript: {event.text}")

@agent.events.subscribe
async def detect_keywords(event: STTTranscriptEvent):
    if "help" in event.text.lower():
        await agent.simple_response("How can I help?")
Use async handlers — Event handlers should be async for non-blocking operations.