Skip to main content
The Event System is the backbone of Home Assistant’s architecture, enabling loose coupling between components through an event-driven communication model. All significant actions in Home Assistant trigger events that other components can listen to and react upon.

​
EventBus Overview

The EventBus manages event distribution across the entire system:
homeassistant/core.py
class EventBus:
    """Allow the firing of and listening for events."""
    
    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize a new event bus."""
        self._listeners: defaultdict[
            EventType[Any] | str, list[_FilterableJobType[Any]]
        ] = defaultdict(list)
        self._match_all_listeners: list[_FilterableJobType[Any]] = []
        self._listeners[MATCH_ALL] = self._match_all_listeners
        self._hass = hass
The EventBus uses a dictionary to map event types to listener lists, enabling O(1) lookup for event dispatch. The MATCH_ALL special key allows listeners to receive all events.

​
Core Event Types

Home Assistant defines several core event types in homeassistant/const.py:
homeassistant/const.py
# Lifecycle Events
EVENT_HOMEASSISTANT_START = EventType("homeassistant_start")
EVENT_HOMEASSISTANT_STARTED = EventType("homeassistant_started")
EVENT_HOMEASSISTANT_STOP = EventType("homeassistant_stop")
EVENT_HOMEASSISTANT_FINAL_WRITE = EventType("homeassistant_final_write")
EVENT_HOMEASSISTANT_CLOSE = EventType("homeassistant_close")

# State Events
EVENT_STATE_CHANGED: EventType[EventStateChangedData] = EventType("state_changed")
EVENT_STATE_REPORTED: EventType[EventStateReportedData] = EventType("state_reported")

# Service Events  
EVENT_CALL_SERVICE = "call_service"
EVENT_SERVICE_REGISTERED = "service_registered"
EVENT_SERVICE_REMOVED = "service_removed"

# Configuration Events
EVENT_CORE_CONFIG_UPDATE = "core_config_updated"

​
Event Type Categories

Lifecycle Events

Events marking major transitions in Home Assistant’s lifecycle

State Events

Events fired when entity states change or are reported

Service Events

Events related to service registration and execution

Configuration Events

Events signaling configuration changes

​
Event Structure

Every event is represented by the Event class:
homeassistant/core.py
class Event(Generic[_DataT]):
    """Representation of an event within the bus."""
    
    def __init__(
        self,
        event_type: EventType[_DataT] | str,
        data: _DataT | None = None,
        origin: EventOrigin = EventOrigin.local,
        time_fired_timestamp: float | None = None,
        context: Context | None = None,
    ) -> None:
        """Initialize a new event."""
        self.event_type = event_type
        self.data: _DataT = data or {}
        self.origin = origin
        self.time_fired_timestamp = time_fired_timestamp or time.time()
        if not context:
            context = Context(id=ulid_at_time(self.time_fired_timestamp))
        self.context = context
        if not context.origin_event:
            context.origin_event = self

​
Event Components

  • event_type: String or typed identifier for the event
  • data: Dictionary containing event-specific information
  • origin: Whether the event originated locally or remotely
  • time_fired_timestamp: Unix timestamp when the event was fired
  • context: Context tracking the chain of events

​
Firing Events

​
Internal Events

For performance-critical internal events, use async_fire_internal:
homeassistant/core.py
@callback
def async_fire_internal(
    self,
    event_type: EventType[_DataT] | str,
    event_data: _DataT | None = None,
    origin: EventOrigin = EventOrigin.local,
    context: Context | None = None,
    time_fired: float | None = None,
) -> None:
    """Fire an event, for internal use only."""
    if self._debug:
        _LOGGER.debug(
            "Bus:Handling %s", 
            _event_repr(event_type, origin, event_data)
        )
    
    listeners = self._listeners.get(event_type, EMPTY_LIST)
    if event_type not in EVENTS_EXCLUDED_FROM_MATCH_ALL:
        match_all_listeners = self._match_all_listeners
    else:
        match_all_listeners = EMPTY_LIST
    
    event: Event[_DataT] | None = None
    for job, event_filter in listeners + match_all_listeners:
        if event_filter is not None:
            try:
                if event_data is None or not event_filter(event_data):
                    continue
            except Exception:
                _LOGGER.exception("Error in event filter")
                continue
        
        if not event:
            event = Event(
                event_type, event_data, origin, time_fired, context
            )
        
        try:
            self._hass.async_run_hass_job(job, event)
        except Exception:
            _LOGGER.exception("Error running job: %s", job)
async_fire_internal is for internal Home Assistant use only and bypasses certain checks. Integrations should use async_fire instead.

​
Public Events

For regular event firing, use the public API: 
# Async context (event loop)
await hass.bus.async_fire(
    "my_custom_event",
    {"key": "value"},
    context=context
)

# Synchronous context (external thread)
hass.bus.fire(
    "my_custom_event",
    {"key": "value"},
    context=context
)
1

Event Type Validated

The event type length is validated (max 64 characters).
2

Event Created

An Event object is created with the provided data and context.
3

Listeners Retrieved

All listeners for this event type and MATCH_ALL listeners are retrieved.
4

Filters Applied

Event filters are evaluated to determine which listeners should receive the event.
5

Jobs Executed

Each matching listener’s job is executed asynchronously.

​
Listening for Events

​
Basic Listener

Register a listener for a specific event type: 
from homeassistant.core import callback, Event

@callback
def handle_my_event(event: Event) -> None:
    """Handle my custom event."""
    data = event.data
    _LOGGER.info("Received event with data: %s", data)

# Register the listener
remove_listener = hass.bus.async_listen(
    "my_custom_event",
    handle_my_event
)

# Later, to unregister:
remove_listener()
Always decorate event listeners with @callback to indicate they’re safe to run in the event loop without creating a task. This improves performance significantly.

​
Event Filters

Filters allow you to selectively receive events: 
from homeassistant.core import callback

@callback
def event_filter(event_data: dict) -> bool:
    """Filter events based on data."""
    return event_data.get("importance") == "high"

@callback  
def handle_important_event(event: Event) -> None:
    """Handle only high-importance events."""
    process_important_event(event.data)

remove_listener = hass.bus.async_listen(
    "my_event",
    handle_important_event,
    event_filter=event_filter
)
Event filters must be decorated with @callback and should execute quickly. Slow filters will block event processing for all listeners.

​
Listen Once

For one-time event handling: 
@callback
def handle_startup(event: Event) -> None:
    """Handle the startup event once."""
    _LOGGER.info("Home Assistant has started")
    initialize_integration()

remove_listener = hass.bus.async_listen_once(
    EVENT_HOMEASSISTANT_STARTED,
    handle_startup
)

​
State Change Events

State changes trigger the most common events in Home Assistant:
homeassistant/core.py
class EventStateChangedData(EventStateEventData):
    """EVENT_STATE_CHANGED data.
    
    A state changed event is fired when on state write 
    the state is changed.
    """
    
    entity_id: str
    new_state: State | None
    old_state: State | None

​
Listening for State Changes

from homeassistant.const import EVENT_STATE_CHANGED
from homeassistant.core import callback, Event, EventStateChangedData

@callback
def handle_state_change(event: Event[EventStateChangedData]) -> None:
    """Handle state change event."""
    entity_id = event.data["entity_id"]
    old_state = event.data["old_state"]
    new_state = event.data["new_state"]
    
    if old_state is None:
        _LOGGER.info("New entity added: %s", entity_id)
    elif new_state is None:
        _LOGGER.info("Entity removed: %s", entity_id)
    else:
        _LOGGER.info(
            "Entity %s changed from %s to %s",
            entity_id,
            old_state.state,
            new_state.state
        )

hass.bus.async_listen(EVENT_STATE_CHANGED, handle_state_change)

​
Optimized State Tracking

For better performance when tracking specific entities, use the event helpers: 
from homeassistant.helpers.event import async_track_state_change_event

@callback
def handle_light_change(event: Event[EventStateChangedData]) -> None:
    """Handle changes to specific light."""
    new_state = event.data["new_state"]
    if new_state and new_state.state == "on":
        _LOGGER.info("Light turned on")

# This is much faster than listening to all state changes
remove_listener = async_track_state_change_event(
    hass,
    ["light.living_room", "light.bedroom"],
    handle_light_change
)
async_track_state_change_event uses an optimized index that routes events directly to listeners interested in specific entities, avoiding the need to process all state change events.

​
Match All Listeners

Listen to every event (use sparingly for debugging/monitoring): 
from homeassistant.const import MATCH_ALL

@callback
def log_all_events(event: Event) -> None:
    """Log every event for debugging."""
    _LOGGER.debug(
        "Event: %s - %s",
        event.event_type,
        event.data
    )

remove_listener = hass.bus.async_listen(MATCH_ALL, log_all_events)
MATCH_ALL listeners receive every single event, which can significantly impact performance. Use only for temporary debugging or specialized monitoring tools.

​
Event Context

Every event includes a context tracking its origin:
homeassistant/core.py
class Context:
    """The context that triggered something."""
    
    def __init__(
        self,
        user_id: str | None = None,
        parent_id: str | None = None,
        id: str | None = None,
    ) -> None:
        self.id = id or ulid_now()
        self.user_id = user_id
        self.parent_id = parent_id
        self.origin_event: Event[Any] | None = None

​
Using Context

# Check who triggered an event
@callback
def handle_light_change(event: Event[EventStateChangedData]) -> None:
    if event.context.user_id:
        user = await hass.auth.async_get_user(event.context.user_id)
        _LOGGER.info("Light changed by user: %s", user.name)
    else:
        _LOGGER.info("Light changed by automation or integration")

# Propagate context in automation chains
context = Context(parent_id=trigger_event.context.id)
hass.bus.async_fire("my_event", {}, context=context)

​
Event Origin

Events can originate locally or from remote systems:
homeassistant/core.py
class EventOrigin(enum.Enum):
    """Represent the origin of an event."""
    
    local = "LOCAL"   # Event originated on this Home Assistant instance
    remote = "REMOTE" # Event came from a remote instance
This is primarily used in multi-instance setups or when integrating with external systems.

​
Performance Considerations

Use @callback: Always decorate listeners with @callback when possible
Specific Events: Listen to specific events rather than MATCH_ALL
Event Filters: Use filters to reduce unnecessary job creation
Fast Execution: Keep listener code fast; offload heavy work to tasks

​
Best Practices

  1. Always unregister listeners when your component unloads
  2. Use typed event data with TypedDict for type safety
  3. Handle None states in state change events (entity added/removed)
  4. Propagate context to maintain audit trails
  5. Avoid blocking operations in event listeners

Build docs developers (and LLMs) love

Get started for freeTalk to us