Subscription Methods

Bus provides typed subscription methods for each event category Home Assistant and Hassette emit. Each method returns a Subscription handle. Calling sub.cancel() removes the listener.

All registration methods are async and must be awaited. See Registration for what that guarantees.

Forgetting await registers nothing

A subscription call without await returns a coroutine object and registers no listener — the handler never fires, and no error is raised at the call site. Python logs RuntimeWarning: coroutine 'Bus.on_state_change' was never awaited when the coroutine is garbage-collected, but the message is easy to miss. When a handler never fires, check the registration is awaited, then confirm the listener exists with hassette listener --app <key>.

Shared Parameters

Every subscription method accepts these parameters. Individual method tables below list only method-specific parameters.

Parameter	Type	Default	Description
`handler`	`HandlerType`	—	The function called when the event matches. See Writing Handlers.
`name`	`str \\| None`	`None`	Required. Identifies this listener in logs and the monitoring UI. Must be unique per app instance and topic. Omitting raises `ListenerNameRequiredError`.
`on_error`	`BusErrorHandlerType \\| None`	`None`	Per-listener error handler. Overrides the app-level handler set via `bus.on_error()`. Available on `on_state_change`, `on_attribute_change`, `on_call_service`, `on_service_registered`, `on_component_loaded`, `on_app_state_changed`, and `on()`.
`timeout`	`float \\| None`	`None`	Per-listener timeout in seconds. If the handler runs longer, it is cancelled. `None` inherits `event_handler_timeout_seconds` from `hassette.toml`.
`timeout_disabled`	`bool`	`False`	Disables timeout enforcement for this listener regardless of config.
`debounce`	`float \\| None`	`None`	Delays the handler until events have been quiet for N seconds. Each new event resets the timer.
`throttle`	`float \\| None`	`None`	Limits the handler to one invocation per N seconds. Events during the cooldown are dropped.
`once`	`bool`	`False`	Fires the handler exactly once, then cancels the subscription.
`kwargs`	`Mapping \\| None`	`None`	Keyword arguments passed to the handler at invocation time.

debounce, throttle, and once are mutually exclusive. Combining any two raises ValueError.

`on_state_change(entity_id)`

Fires when a Home Assistant entity's state changes. entity_id accepts glob patterns ("light.*kitchen*").

await self.bus.on_state_change(
    "light.kitchen",
    handler=self.on_light_change,
    name="kitchen_light",
)

Parameter	Type	Default	Description
`entity_id`	`str`	—	Entity ID or glob pattern to match.
`changed`	`bool \\| ComparisonCondition`	`True`	`True` fires only when the state value changes. `False` fires on attribute-only updates too. A `ComparisonCondition` (e.g., `C.Increased()`) compares old and new values.
`changed_from`	`ChangeType`	not set	Filters on the previous state value. Accepts a raw value, callable, or condition. Compares raw HA state strings.
`changed_to`	`ChangeType`	not set	Filters on the new state value. Accepts a raw value, callable, or condition. Compares raw HA state strings.
`where`	`Predicate \\| Sequence[Predicate] \\| None`	`None`	Additional predicates applied after value filters. See Filtering & Predicates.
`immediate`	`bool`	`False`	Fires with the current state on registration, then on every subsequent change. Not supported with glob patterns.
`duration`	`float \\| None`	`None`	Fires only after the state has held for N seconds continuously. Not supported with glob patterns.

changed_from and changed_to compare raw HA state strings ("on", "off", "72.5"), not typed values from the state registry.

immediate=True and duration both raise ValueError when entity_id contains glob characters.

Compatible DI annotations

D is the dependency-injection module (from hassette import D). Annotating a handler parameter with one of these types makes Hassette extract and convert that piece of the event automatically. Each method section below lists the annotations its events support.

Annotation	Provides
`D.StateNew[T]`	New state object, converted to type `T`. Raises if absent.
`D.StateOld[T]`	Previous state object, converted to type `T`. Raises if absent.
`D.MaybeStateNew[T]`	New state object or `None` if not present.
`D.MaybeStateOld[T]`	Previous state object or `None` if not present.
`D.EntityId`	Entity ID string. Raises if absent.
`D.MaybeEntityId`	Entity ID string or missing-value sentinel.
`D.Domain`	Domain string (e.g., `"light"`). Raises if absent.
`D.MaybeDomain`	Domain string or missing-value sentinel.
`D.TypedStateChangeEvent[T]`	Full event with new/old states converted to type `T`.
`D.EventContext`	HA event context (user_id, parent_id, etc.).

Fire with the current value on registration, then on each subsequent change:

await self.bus.on_state_change(
    "sensor.outdoor_temperature",
    handler=self.on_temp,
    immediate=True,
    name="outdoor_temp_init",
)

Fire only after the state has held for a set duration:

await self.bus.on_state_change(
    "light.kitchen",
    changed_to="on",
    handler=self.on_light_on_long,
    duration=1800.0,
    name="kitchen_light_duration",
)

Fire only on a specific state transition:

await self.bus.on_state_change(
    "sensor.outdoor_temperature",
    changed_to=C.Comparison(">", 25),
    handler=self.on_temp_high,
    name="outdoor_temp_high",
)

`on_attribute_change(entity_id, attr)`

Fires when a specific attribute of an entity changes. entity_id accepts glob patterns.

attr does not support glob patterns

The attr parameter matches a single attribute name exactly. Glob characters in attr are treated as literal characters, not patterns. Predicates handle multi-attribute matching.

await self.bus.on_attribute_change(
    "media_player.living_room",
    "volume_level",
    handler=self.on_volume_change,
    name="living_room_volume",
)

Parameter	Type	Default	Description
`entity_id`	`str`	—	Entity ID or glob pattern to match.
`attr`	`str`	—	Attribute name to monitor (e.g., `"volume_level"`).
`changed`	`bool \\| ComparisonCondition`	`True`	`True` fires only when the attribute value changes. `False` fires on any state event for the entity.
`changed_from`	`ChangeType`	not set	Filters on the previous attribute value.
`changed_to`	`ChangeType`	not set	Filters on the new attribute value.
`where`	`Predicate \\| Sequence[Predicate] \\| None`	`None`	Additional predicates.
`immediate`	`bool`	`False`	Fires with the current attribute value on registration. Not supported with glob patterns.
`duration`	`float \\| None`	`None`	Fires only after the attribute has held the value for N seconds. Not supported with glob patterns.

changed_from and changed_to compare the attribute value, not the entity's main state string.

changed=False fires on every state event for the entity, even when the monitored attribute did not change. on_state_change with changed=False provides that broader behavior.

Compatible DI annotations

Same as on_state_change.

await self.bus.on_attribute_change(
    "sensor.phone_battery",
    "battery_level",
    changed_from=C.Comparison(">", 20),
    changed_to=C.Comparison("<=", 20),
    handler=self.on_battery_low,
    name="phone_battery_low",
)

await self.bus.on_attribute_change(
    "climate.living_room",
    "current_temperature",
    handler=self.on_temp_change,
    immediate=True,
    name="climate_temp_init",
)

`on_call_service(domain, service)`

Fires when Home Assistant calls a service.

from hassette import App, AppConfig, D


class LightControlApp(App[AppConfig]):
    async def on_initialize(self) -> None:
        await self.bus.on_call_service(
            "light",
            "turn_on",
            handler=self.on_light_turn_on,
            name="light_turn_on",
        )

    async def on_light_turn_on(self, entity_id: D.EntityId) -> None:
        self.logger.info("Light turned on: %s", entity_id)

Parameter	Type	Default	Description
`domain`	`str \\| None`	`None`	Service domain to match (e.g., `"light"`). `None` matches all domains.
`service`	`str \\| None`	`None`	Service name to match (e.g., `"turn_on"`). `None` matches all services in the domain.
`where`	`Predicate \\| Sequence[Predicate] \\| Mapping[str, ChangeType] \\| None`	`None`	Additional predicates, or a dict for service data matching.

where= accepts a plain dict mapping service data fields to expected values. {"entity_id": "light.kitchen"} matches only calls targeting light.kitchen. This dict form is unique to on_call_service. on_service_registered does not support it.

No changed, changed_from, changed_to, immediate, or duration parameters.

Compatible DI annotations

Annotation	Provides
`D.EntityId`	Entity ID from the service call. Raises if absent.
`D.MaybeEntityId`	Entity ID or missing-value sentinel.
`D.EventContext`	HA event context.

`on_service_registered(domain, service)`

Fires when Home Assistant registers a new service. Same parameter shape as on_call_service, with one difference. where= accepts only predicates, not a dict.

Parameter	Type	Default	Description
`domain`	`str \\| None`	`None`	Domain to match.
`service`	`str \\| None`	`None`	Service name to match.
`where`	`Predicate \\| Sequence[Predicate] \\| None`	`None`	Additional predicates.

`on_component_loaded(component)`

Fires when Home Assistant finishes loading a component.

Parameter	Type	Default	Description
`component`	`str \\| None`	`None`	Component name to match (e.g., `"light"`). `None` matches all components.
`where`	`Predicate \\| Sequence[Predicate] \\| None`	`None`	Additional predicates.

Home Assistant Lifecycle Methods

Three shorthands delegate to on_call_service("homeassistant", ...).

Method	Equivalent
`on_homeassistant_start(handler, ...)`	`on_call_service("homeassistant", "start", ...)`
`on_homeassistant_stop(handler, ...)`	`on_call_service("homeassistant", "stop", ...)`
`on_homeassistant_restart(handler, ...)`	`on_call_service("homeassistant", "restart", ...)`

All three accept handler, where, kwargs, name, and the shared parameters (debounce, throttle, once, timeout, timeout_disabled). They do not expose on_error directly. Per-registration error handling requires on_call_service directly.

`on(topic)`

Subscribes to any raw event topic string.

from typing import Any

from hassette import App, AppConfig
from hassette.events import Event


class ScriptApp(App[AppConfig]):
    async def on_initialize(self) -> None:
        await self.bus.on(
            topic="hass.event.automation_triggered",
            handler=self.on_automation,
            name="automation_triggered",
        )

    async def on_automation(self, event: Event[Any]) -> None:
        self.logger.info("Automation fired: %s", event.topic)

Parameter	Type	Default	Description
`topic`	`str`	—	The exact event topic string to subscribe to.
`where`	`Predicate \\| Sequence[Predicate] \\| None`	`None`	Additional predicates.

on() does not support immediate, duration, changed, changed_from, or changed_to. All shared timing parameters (debounce, throttle, once, timeout, timeout_disabled) are accepted. Internal topics used by Hassette shorthands (WebSocket events, app state events) are also accessible via on() for raw topic access.

App and Connection Events

`on_app_state_changed` and shorthands

on_app_state_changed fires when any app instance transitions to a new ResourceStatus (e.g., RUNNING, STOPPING, STOPPED, FAILED). Two shorthands cover the most common cases.

# Fire whenever any app's status changes.
await self.bus.on_app_state_changed(
    handler=self.on_any_app_change,
    name="any_app_status",
)

# Fire only when the sensor app reaches RUNNING.
await self.bus.on_app_running(
    app_key="sensor_monitor",
    handler=self.on_sensor_ready,
    name="sensor_monitor_running",
)

# Fire when the sensor app begins stopping.
await self.bus.on_app_stopping(
    app_key="sensor_monitor",
    handler=self.on_sensor_stopping,
    name="sensor_monitor_stopping",
)

Parameter	Type	Default	Description
`app_key`	`str \\| None`	`None`	Filters to a specific app (the identifier from `hassette.toml`). `None` matches all apps.
`status`	`ResourceStatus \\| None`	`None`	Filters to a specific status. `None` matches all status transitions.
`where`	`Predicate \\| Sequence[Predicate] \\| None`	`None`	Additional predicates.

on_app_running(app_key=...) delegates to on_app_state_changed(status=ResourceStatus.RUNNING). on_app_stopping(app_key=...) delegates to on_app_state_changed(status=ResourceStatus.STOPPING).

The shorthands do not expose on_error directly. Per-listener error handling requires on_app_state_changed with on_error= directly.

`on_websocket_connected` and `on_websocket_disconnected`

Fire when the Hassette WebSocket connection to Home Assistant opens or closes.

await self.bus.on_websocket_connected(
    handler=self.on_connected,
    name="ha_ws_connected",
)
await self.bus.on_websocket_disconnected(
    handler=self.on_disconnected,
    name="ha_ws_disconnected",
)

Both methods accept handler, where, kwargs, name, and **opts. Neither exposes on_error. Both delegate to on() internally.

`on_hassette_service_status` and shorthands

on_hassette_service_status fires when a Hassette background service (WebSocket, database, bus, scheduler) transitions to a new ResourceStatus. Most apps never need this — Hassette restarts failed services on its own. It exists for apps that pause work or alert when a service goes down. Three shorthands cover the common cases: on_hassette_service_failed (status FAILED), on_hassette_service_crashed (status CRASHED), and on_hassette_service_started (status RUNNING).

await self.bus.on_hassette_service_failed(
    handler=self.on_service_failed,
    name="service_watchdog",
)

All four accept handler, where, kwargs, name, and **opts. Service supervision explains when each status fires.

Error Handling

App-level handler

bus.on_error(handler) registers a fallback called when any listener on the bus raises. This call is synchronous — no await needed. The handler receives a BusErrorContext.

from hassette import App, AppConfig
from hassette.bus.error_context import BusErrorContext
from hassette.events import RawStateChangeEvent


class MyApp(App[AppConfig]):
    async def on_initialize(self) -> None:
        self.bus.on_error(self.on_bus_error)

        await self.bus.on_state_change("light.kitchen", handler=self.on_light_change, name="kitchen_light")

    async def on_bus_error(self, ctx: BusErrorContext) -> None:
        self.logger.error(
            "Handler failed for topic=%s: %s\n%s",
            ctx.topic,
            ctx.exception,
            ctx.traceback,
        )

    async def on_light_change(self, event: RawStateChangeEvent) -> None:
        raise ValueError("something went wrong")

Per-registration handler

on_error= on a registration overrides the app-level fallback for that listener only.

from hassette import App, AppConfig
from hassette.bus.error_context import BusErrorContext
from hassette.events import RawStateChangeEvent


class MyApp(App[AppConfig]):
    async def on_initialize(self) -> None:
        await self.bus.on_state_change(
            "sensor.temperature",
            handler=self.on_temp_change,
            on_error=self.on_temp_error,
            name="temp_sensor",
        )

    async def on_temp_error(self, ctx: BusErrorContext) -> None:
        self.logger.warning("Temperature handler failed: %s", ctx.exception)

    async def on_temp_change(self, event: RawStateChangeEvent) -> None:
        raise RuntimeError("temp sensor error")

BusErrorContext fields

Field	Type	Description
`exception`	`BaseException`	The raised exception, with `__traceback__` chain intact.
`traceback`	`str`	Full formatted traceback string. Always non-empty.
`topic`	`str`	The event topic the listener was registered on.
`listener_name`	`str`	Human-readable listener identity string.
`event`	`Event[Any]`	The event being processed when the exception occurred.
`execution_id`	`str \\| None`	UUIDv7 identifying the execution that failed, or `None`.

Error handlers run as fire-and-forget tasks. Handlers that start near app shutdown may be cancelled before they complete. Error handlers are not a reliable delivery channel during system teardown.

on_error is not available on on_homeassistant_start, on_homeassistant_stop, on_homeassistant_restart, on_app_running, on_app_stopping, on_websocket_connected, or on_websocket_disconnected. Per-registration error handling on these events requires the underlying method (on_call_service, on_app_state_changed, or on()) directly.

Timeout Configuration

timeout= overrides the global event_handler_timeout_seconds for a single listener. timeout_disabled=True removes timeout enforcement entirely for that listener.

from hassette import App, AppConfig
from hassette.events import RawStateChangeEvent


class MyApp(App[AppConfig]):
    async def on_initialize(self) -> None:
        # Override the global timeout for a slow handler
        await self.bus.on_state_change(
            "sensor.weather",
            handler=self.fetch_forecast,
            timeout=30.0,  # 30 seconds instead of the global default
            name="weather_forecast",
        )

        # Disable timeout for a handler that legitimately runs long
        await self.bus.on_state_change(
            "input_boolean.run_backup",
            handler=self.run_full_backup,
            timeout_disabled=True,
            name="backup_trigger",
        )

    async def fetch_forecast(self, event: RawStateChangeEvent) -> None: ...
    async def run_full_backup(self, event: RawStateChangeEvent) -> None: ...

The global default comes from event_handler_timeout_seconds in hassette.toml. A listener with timeout=None (the default) inherits that value. Setting timeout=30.0 overrides the global only for that listener. Other listeners are unaffected.

timeout_disabled=True is appropriate for handlers that legitimately run longer than the global limit. A backup job triggered by a boolean is a typical case. timeout= is appropriate when a specific handler needs a tighter or looser bound than the global.

Registration

`name=` requirement

Every registration method requires name=. Omitting it raises ListenerNameRequiredError at call time.

await self.bus.on_state_change(
    "binary_sensor.motion",
    handler=self.on_motion,
    name="motion_sensor_main",
)

await self.bus.on_state_change(
    "binary_sensor.motion",
    handler=self.on_motion_log,
    name="motion_sensor_log",
)

The name forms a natural key together with the app identifier, instance index, and topic. Two registrations with the same name on the same topic within a session raise DuplicateListenerError. Across sessions (app restart), the same name and topic performs an upsert — Hassette persists listener metadata to a local SQLite telemetry database, and the existing record is updated, not duplicated.

Synchronous completion

Registration completes before the awaited call returns. sub.listener.db_id is a valid integer immediately.

# Registration is synchronous — db_id is set before this line returns.
sub = await self.bus.on_state_change(
    "sensor.temperature", handler=self.on_temp, name="temp_monitor"
)

# db_id is always set immediately after the awaited call returns.
self.logger.info("Listener registered with db_id=%d", sub.listener.db_id)

Cancel-then-resubscribe

Cancelling a subscription and registering a new one is deterministic. The old handler is removed before the new registration begins. No overlap, no gap.

async def resubscribe(self) -> None:
    if self.sub is not None:
        # Cancel the old subscription — routing removal is immediate.
        self.sub.cancel()

    # Register the replacement — routing and DB persistence both complete
    # before this line returns. The old handler is guaranteed gone; no overlap.
    self.sub = await self.bus.on_state_change(
        "light.kitchen", handler=self.on_light, name="kitchen_light"
    )

Subscription Methods

Shared Parameters

on_state_change(entity_id)

on_attribute_change(entity_id, attr)

on_call_service(domain, service)

on_service_registered(domain, service)

on_component_loaded(component)

Home Assistant Lifecycle Methods

on(topic)

App and Connection Events

on_app_state_changed and shorthands

on_websocket_connected and on_websocket_disconnected

on_hassette_service_status and shorthands

Error Handling

App-level handler

Per-registration handler

Timeout Configuration

Registration

name= requirement

Synchronous completion

Cancel-then-resubscribe

See Also

`on_state_change(entity_id)`

`on_attribute_change(entity_id, attr)`

`on_call_service(domain, service)`

`on_service_registered(domain, service)`

`on_component_loaded(component)`

`on(topic)`

`on_app_state_changed` and shorthands

`on_websocket_connected` and `on_websocket_disconnected`

`on_hassette_service_status` and shorthands

`name=` requirement