Skip to main content
Event handling is a fundamental concept in Home Assistant. The event bus is the central nervous system that allows components to communicate asynchronously. Understanding how to work with events efficiently is crucial for building responsive integrations.

​
Understanding the Event Bus

The event bus (EventBus) allows firing and listening for events throughout Home Assistant. Every state change, service call, and custom integration action can be represented as an event.

​
Key Event Types

Home Assistant Core defines several built-in event types:
  • EVENT_STATE_CHANGED - Fired when an entity’s state changes
  • EVENT_STATE_REPORTED - Fired when state is updated but unchanged
  • EVENT_HOMEASSISTANT_START - Fired when Home Assistant starts
  • EVENT_HOMEASSISTANT_STARTED - Fired after all components are loaded
  • EVENT_HOMEASSISTANT_STOP - Fired when Home Assistant begins shutdown
  • EVENT_SERVICE_REGISTERED - Fired when a service is registered
  • EVENT_CALL_SERVICE - Fired when a service is called
Reference: homeassistant/const.py

​
Listening to Events

​
Basic Event Listener

Use async_listen to subscribe to events. This method must be run in the event loop: 
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.const import EVENT_STATE_CHANGED

@callback
def handle_event(event: Event) -> None:
    """Handle the event."""
    print(f"Event fired: {event.event_type}")
    print(f"Event data: {event.data}")

def async_setup(hass: HomeAssistant):
    """Set up the component."""
    # Register event listener
    remove_listener = hass.bus.async_listen(
        EVENT_STATE_CHANGED,
        handle_event
    )
    
    # Store the remove callback to clean up later
    return True
Reference: homeassistant/core.py:1565

​
Event Filters

Event filters allow you to pre-filter events before your handler is called, improving performance: 
from homeassistant.core import callback

@callback
def event_filter(event_data: dict) -> bool:
    """Filter events - return True to handle, False to skip."""
    # Only handle events for specific entity
    return event_data.get("entity_id") == "light.living_room"

@callback
def handle_filtered_event(event: Event) -> None:
    """Handle only filtered events."""
    print(f"Light changed: {event.data}")

remove = hass.bus.async_listen(
    EVENT_STATE_CHANGED,
    handle_filtered_event,
    event_filter=event_filter
)
Reference: homeassistant/core.py:1569
Event filters must be decorated with @callback to ensure they run synchronously in the event loop.

​
Tracking State Changes

​
Track State Change Events

For entity-specific state tracking, use the optimized async_track_state_change_event helper: 
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.core import Event, callback

@callback
def state_changed(event: Event) -> None:
    """Handle state change."""
    entity_id = event.data["entity_id"]
    old_state = event.data["old_state"]
    new_state = event.data["new_state"]
    
    if old_state and new_state:
        print(f"{entity_id}: {old_state.state} -> {new_state.state}")

# Track specific entities (more efficient than filtering manually)
remove = async_track_state_change_event(
    hass,
    ["light.living_room", "light.bedroom"],
    state_changed
)
Reference: homeassistant/helpers/event.py:309

​
Track by Domain

Track when entities are added to specific domains: 
from homeassistant.helpers.event import async_track_state_added_domain

@callback
def handle_new_light(event: Event) -> None:
    """Handle new light entity."""
    entity_id = event.data["entity_id"]
    new_state = event.data["new_state"]
    print(f"New light added: {entity_id}")

# Track when new lights are added
remove = async_track_state_added_domain(
    hass,
    "light",
    handle_new_light
)
Reference: homeassistant/helpers/event.py:653

​
Advanced Tracking Patterns

​
Filtered State Change Tracking

Use async_track_state_change_filtered for dynamic entity tracking: 
from homeassistant.helpers.event import (
    async_track_state_change_filtered,
    TrackStates,
)

@callback
def handle_change(event: Event) -> None:
    """Handle tracked state changes."""
    print(f"Tracked entity changed: {event.data['entity_id']}")

# Create tracker with initial entities and domains
track_states = TrackStates(
    all_states=False,
    entities={"light.living_room", "switch.kitchen"},
    domains={"climate"},
)

tracker = async_track_state_change_filtered(
    hass,
    track_states,
    handle_change
)

# Later, update what we're tracking
new_track_states = TrackStates(
    all_states=False,
    entities={"light.bedroom"},
    domains={"climate", "fan"},
)
tracker.async_update_listeners(new_track_states)

# Clean up when done
tracker.async_remove()
Reference: homeassistant/helpers/event.py:867

​
Template Tracking

Track template results and react to changes: 
from homeassistant.helpers.event import (
    async_track_template_result,
    TrackTemplate,
    TrackTemplateResult,
)
from homeassistant.helpers.template import Template

@callback
def template_changed(
    event: Event | None,
    updates: list[TrackTemplateResult]
) -> None:
    """Handle template result changes."""
    for update in updates:
        print(f"Template result: {update.result}")
        print(f"Previous result: {update.last_result}")

# Create template to track
template = Template(
    "{{ states('sensor.temperature') | float > 20 }}",
    hass
)

track_template = TrackTemplate(template, variables=None)

# Start tracking
info = async_track_template_result(
    hass,
    [track_template],
    template_changed
)

# Clean up
info.async_remove()
Reference: homeassistant/helpers/event.py:1343

​
Time-Based Event Tracking

​
Track Point in Time

Schedule a callback at a specific time: 
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.util import dt as dt_util
from datetime import timedelta

@callback
def timed_callback(now: datetime) -> None:
    """Called at scheduled time."""
    print(f"Timer fired at {now}")

# Schedule for 5 minutes from now
point_in_time = dt_util.utcnow() + timedelta(minutes=5)
remove = async_track_point_in_utc_time(
    hass,
    timed_callback,
    point_in_time
)

# Cancel if needed
remove()
Reference: homeassistant/helpers/event.py:1544

​
Firing Events

​
Fire Internal Events

For internal core use, fire events directly without extra validation: 
from homeassistant.core import EventOrigin

# Fire internal event (most efficient)
hass.bus.async_fire_internal(
    "my_custom_event",
    {"key": "value"},
)
Reference: homeassistant/core.py:1492

​
Fire Public Events

For integration events that may be consumed externally: 
# Fire public event with context
hass.bus.async_fire(
    "my_integration_event",
    {"device_id": "abc123", "status": "online"},
    origin=EventOrigin.local,
    context=some_context
)
Reference: homeassistant/core.py:1470

​
Event Dispatching Optimization

The event system uses several optimizations:
  1. Entity ID Indexing: State change events are indexed by entity_id for fast routing
  2. Event Filters: Pre-filtering prevents unnecessary job creation
  3. Call Soon: Events are dispatched with call_soon to ensure proper event loop iteration
  4. Job Types: Callbacks are executed directly, coroutines are scheduled efficiently
Reference: homeassistant/helpers/event.py:340-367

​
Best Practices

​
Use @callback Decorator

Always decorate synchronous event handlers with @callback: 
@callback
def my_handler(event: Event) -> None:
    """Handle event synchronously."""
    # Synchronous code only
    pass

​
Async Handlers

For async operations, use coroutine functions: 
async def my_async_handler(event: Event) -> None:
    """Handle event asynchronously."""
    await some_async_operation()

​
Clean Up Listeners

Always remove listeners when they’re no longer needed: 
class MyComponent:
    def __init__(self, hass: HomeAssistant):
        self._remove_listener = hass.bus.async_listen(
            EVENT_STATE_CHANGED,
            self.handle_event
        )
    
    async def async_will_remove_from_hass(self):
        """Clean up."""
        if self._remove_listener:
            self._remove_listener()

​
Avoid Long-Running Operations

Event handlers should be fast. For long operations, create a task: 
@callback
def handle_event(event: Event) -> None:
    """Handle event."""
    # Create background task for slow operation
    hass.async_create_task(
        slow_operation(event.data)
    )

async def slow_operation(data: dict) -> None:
    """Perform slow operation."""
    await asyncio.sleep(10)
    # Do work

​
Performance Considerations

  1. Use specific tracking helpers - async_track_state_change_event is more efficient than manual filtering
  2. Implement event filters - Pre-filter events to avoid unnecessary processing
  3. Index by entity_id - The event system automatically optimizes entity-specific tracking
  4. Batch updates - When possible, batch multiple state changes together

​
Common Pitfalls

  • Don’t block the event loop in event handlers
  • Always clean up listeners to prevent memory leaks
  • Don’t modify event data directly (it may be shared)
  • Use @callback for synchronous handlers to avoid thread issues

Build docs developers (and LLMs) love

Get started for freeTalk to us