Event System

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.

How Events Work

The event system is built on three properties worth understanding before you write a handler.

Fire-and-forget dispatch. When something inside the agent emits an event, agent.events.send(...) schedules each handler as its own asyncio.Task and returns immediately. The caller does not wait for handlers to finish — so don’t rely on a handler having run before the next line of agent code executes.

Fanout. All handlers subscribed to a given event type run concurrently as independent tasks. There is no ordering guarantee between handlers, and one slow handler does not delay the others.

Error isolation. If a handler raises, the exception is caught and re-emitted as an ExceptionEvent; other handlers still run and the agent keeps going. Subscribe to ExceptionEvent if you want a single place to log handler failures.

Subscribing to Events

Use the @agent.events.subscribe decorator with a type hint to specify which event you want. Handlers must be async functions:

from vision_agents.core.edge.events import ParticipantJoinedEvent

@agent.events.subscribe
async def handle_participant_joined(event: ParticipantJoinedEvent):
    await agent.simple_response(f"Hello {event.participant.user_id}!")

Common Events

Event	When	Import
ParticipantJoinedEvent	User joins the call	vision_agents.core.edge.events
ParticipantLeftEvent	User leaves the call	vision_agents.core.edge.events
UserTranscriptEvent	Final user transcript ready	vision_agents.core.agents.events
UserTurnStartedEvent / UserTurnEndedEvent	User started / stopped speaking	vision_agents.core.agents.events
AgentTurnStartedEvent / AgentTurnEndedEvent	Agent started / stopped speaking	vision_agents.core.agents.events
LLMResponseFinalEvent	LLM finishes a response	vision_agents.core.llm.events
ToolStartEvent / ToolEndEvent	Function/tool call	vision_agents.core.llm.events

Example: Greeting Participants

from vision_agents.core import Agent, User
from vision_agents.core.edge.events import (
    ParticipantJoinedEvent,
    ParticipantLeftEvent,
)
from vision_agents.plugins import openai, getstream, deepgram, elevenlabs

agent = Agent(
    edge=getstream.Edge(),
    agent_user=User(name="Assistant", id="agent"),
    instructions="You're a helpful voice assistant.",
    llm=openai.LLM(model="gpt-5.4"),
    tts=elevenlabs.TTS(),
    stt=deepgram.STT(),
)

@agent.events.subscribe
async def on_join(event: ParticipantJoinedEvent):
    await agent.simple_response(
        f"Welcome, {event.participant.user_id}!",
        interrupt=True,
    )

@agent.events.subscribe
async def on_leave(event: ParticipantLeftEvent):
    await agent.say(f"Goodbye, {event.participant.user_id}!", interrupt=False)

Component Events

Subscribe to events from specific components. Each component (LLM, agent lifecycle, TTS, etc.) emits events when it does work:

from vision_agents.core.agents.events import UserTranscriptEvent
from vision_agents.core.llm.events import LLMResponseFinalEvent

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

@agent.events.subscribe
async def on_response(event: LLMResponseFinalEvent):
    print(f"Agent said: {event.text}")
    print(f"Model: {event.model}")

Realtime LLM Events

UserTranscriptEvent fires in both classic STT and realtime modes, so use it for the user side regardless of which LLM you’re running. For connection-state changes in a realtime session, subscribe to RealtimeConnectedEvent / RealtimeDisconnectedEvent.

Tool Execution Events

Monitor function calling with tool events:

from vision_agents.core.llm.events import ToolStartEvent, ToolEndEvent

@agent.events.subscribe
async def on_tool_start(event: ToolStartEvent):
    print(f"Calling tool: {event.tool_name}")
    print(f"Arguments: {event.arguments}")

@agent.events.subscribe
async def on_tool_end(event: ToolEndEvent):
    if event.success:
        print(f"Tool {event.tool_name} completed in {event.execution_time_ms}ms")
    else:
        print(f"Tool {event.tool_name} failed: {event.error}")

Error Handling

Each component has its own error event type, and handler exceptions become ExceptionEvents:

from vision_agents.core.events.base import ExceptionEvent
from vision_agents.core.stt.events import STTErrorEvent
from vision_agents.core.llm.events import LLMErrorEvent

@agent.events.subscribe
async def on_stt_error(event: STTErrorEvent):
    print(f"STT error: {event.error_message}")

@agent.events.subscribe
async def on_llm_error(event: LLMErrorEvent):
    print(f"LLM error: {event.error_message}")
    print(f"Context: {event.context}")

@agent.events.subscribe
async def on_handler_failure(event: ExceptionEvent):
    print(f"Handler {event.handler.__name__} raised: {event.exc}")

Multiple Event Types

Handle related events in one handler using union types:

@agent.events.subscribe
async def on_participant_change(
    event: ParticipantJoinedEvent | ParticipantLeftEvent
):
    action = "joined" if isinstance(event, ParticipantJoinedEvent) else "left"
    print(f"{event.participant.user_id} {action}")

Use the | operator (Python 3.10+) or Union from typing for older versions.

When to Use Events

A few patterns drawn from the real example apps:

Observing agent behavior. Subscribe to UserTranscriptEvent plus LLMResponseFinalEvent to log what the user said and what the agent answered — useful for debugging, replay, or building a transcript UI.

Reacting to participants. Greet on ParticipantJoinedEvent, persist call duration on ParticipantLeftEvent. See the Greeting Participants example above.

Triggering actions from vision detections. Subscribe to a video plugin’s DetectionCompletedEvent (e.g. roboflow.DetectionCompletedEvent) to fire an LLM response when something appears on camera. See the Football Commentator example for a debounced version.

Coordinating with awaitable completion. For tests or scripted flows, briefly subscribe inside a function and await asyncio.Event() to wait for a specific event to fire — for example, awaiting TTSSynthesisCompleteEvent before sending the next prompt. For normal app code, prefer the higher-level agent.simple_response(..., interrupt=...) and agent.say(..., interrupt=...).

Best Practices

Keep handlers focused — One handler per concern:

@agent.events.subscribe
async def log_transcripts(event: UserTranscriptEvent):
    logger.info(f"Transcript: {event.text}")

@agent.events.subscribe
async def detect_keywords(event: UserTranscriptEvent):
    if "help" in event.text.lower():
        await agent.simple_response("How can I help?", interrupt=True)

Don’t rely on handler completion order — Handlers for the same event run concurrently with no ordering guarantee. If one handler depends on state another sets, fold them into a single handler instead.

Use async handlers — Event handlers must be async functions. Non-async handlers raise an error at subscribe time.

Access common event fields — All events have these base fields:

• event.type — Event type identifier (e.g. "agent.user_transcript")
• event.event_id — Unique ID for this event instance
• event.timestamp — When the event was created (UTC)
• event.session_id — Current session identifier
• event.participant — Participant the event relates to (when applicable)

Next Steps

Events Reference

Complete event list

Interruption Handling

Handle user interruptions