refactor: move Home Assistant out of core into the homeassistant catalog plugin

The Home Assistant gateway platform (plugins/platforms/homeassistant) and the
homeassistant toolset (tools/homeassistant_tool.py) now ship as the official
catalog plugin `homeassistant` (NousResearch/hermes-homeassistant), with the
same platform name, toolset/tool names, config keys and env vars.

Migration (hermes_cli/left_core_migration.py, a generic "left core" table
reusing the memory-provider migration's helpers): every profile home that
used Home Assistant (HASS_TOKEN in its .env, platforms.homeassistant enabled
or holding a token, or the homeassistant toolset selected) gets the plugin
installed from the catalog at its reviewed pin by `hermes update`, and lazily
once per process at agent start / gateway start (honours
security.allow_lazy_installs). Every outcome reaches terminal, Desktop and
chat; a gateway-start outcome is handed to that home's first agent.

Core HA special cases become generic PlatformEntry seams the plugin uses:
trusted_inbound (allowlist bypass for event-bus platforms), display_tier
(built-in display defaults) and shared_env_prefixes (profile-clone ownership
rule for env shared with the plugin's tools). Cron delivery uses the existing
cron_deliver_env_var seam; HASS_TOKEN stays stripped from subprocesses via the
installed plugin's manifest secrets.
This commit is contained in:
teknium1 authored and Teknium committed 2026-10-04 00:08:20 -07:00
1 parent 0f7a95f16d
commit fb9fc8a3dd
65 files changed
+663 -1679

No files matched your search

-1
View File
@@ -45,7 +45,6 @@ _POLISHED_TOOLS = {
"vision_analyze", "image_generate", "text_to_speech",
# Schedulers / platform integrations
"cronjob", "send_message", "clarify", "discord", "discord_admin",
"ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service",
"feishu_doc_read", "feishu_drive_list_comments", "feishu_drive_list_comment_replies",
"feishu_drive_reply_comment", "feishu_drive_add_comment",
"kanban_create", "kanban_show", "kanban_comment", "kanban_complete",
+7
View File
@@ -1121,6 +1121,13 @@ def _init_fallback_chain(agent, fallback_model):
def _load_tools(agent, enabled_toolsets, disabled_toolsets):
# A feature that left core for a catalog plugin (Home Assistant) is installed for a home that
# used it, once per process, before discovery so its tools are in this agent's snapshot.
try:
from hermes_cli.left_core_migration import recover_at_startup as _recover_left_core
_recover_left_core(say=agent._emit_startup_warning)
except Exception:
logger.debug("left-core plugin migration skipped", exc_info=True)
# A multiplexed gateway may have switched HERMES_HOME since model_tools was imported;
# make sure this profile's plugins are discovered before the tool snapshot.
try:
-3
View File
@@ -32,9 +32,6 @@ _NEVER_PARALLEL_TOOLS = frozenset({"clarify", "manage_connections", "manage_cata
# Read-only tools with no shared mutable session state.
_PARALLEL_SAFE_TOOLS = frozenset({
"connectors__execute", # pure remote batches have per-dispatch idempotency keys
"ha_get_state",
"ha_list_entities",
"ha_list_services",
"image_generate",
"read_file",
"search_files",
-2
View File
@@ -1405,7 +1405,6 @@ gateway:
# whatsapp: hermes-whatsapp (same as telegram)
# slack: hermes-slack (same as telegram)
# signal: hermes-signal (same as telegram)
# homeassistant: hermes-homeassistant (same as telegram)
# qqbot: hermes-qqbot (same as telegram)
# teams: hermes-teams (same as telegram)
# google_chat: hermes-google_chat (same as telegram)
@@ -1417,7 +1416,6 @@ platform_toolsets:
whatsapp: [hermes-whatsapp]
slack: [hermes-slack]
signal: [hermes-signal]
homeassistant: [hermes-homeassistant]
qqbot: [hermes-qqbot]
yuanbao: [hermes-yuanbao]
teams: [hermes-teams]
+1 -1
View File
@@ -29,7 +29,7 @@ logger = logging.getLogger("cron.scheduler")
# Validates user-supplied delivery platform names, preventing env-var enumeration via crafted names.
_KNOWN_DELIVERY_PLATFORMS = frozenset({
"telegram", "discord", "slack", "whatsapp", "signal",
"matrix", "mattermost", "homeassistant", "dingtalk", "feishu",
"matrix", "mattermost", "dingtalk", "feishu",
"wecom", "wecom_callback", "weixin", "sms", "email", "webhook", "bluebubbles",
"qqbot", "yuanbao"})
+3 -2
View File
@@ -628,8 +628,9 @@ class GatewayAuthorizationMixin:
def _principal_authorized(self, source: SessionSource, *, allow_adapter_delegation: bool) -> bool:
"""The allowlist verdict alone, before the bot loop guard."""
# HA events are system-generated (HASS_TOKEN); webhook events are HMAC-verified.
if source.platform in {Platform.HOMEASSISTANT, Platform.WEBHOOK}:
# Webhook events are HMAC-verified; a ``trusted_inbound`` platform's events come from the
# service its adapter authenticated to (no human sender to allowlist).
if source.platform == Platform.WEBHOOK or getattr(_registry_entry(source.platform), "trusted_inbound", False) is True:
return True
adapter_profile = self._adapter_profile_for_source(source)
-1
View File
@@ -226,7 +226,6 @@ class Platform(Enum):
SIGNAL = "signal"
MATTERMOST = "mattermost"
MATRIX = "matrix"
HOMEASSISTANT = "homeassistant"
EMAIL = "email"
SMS = "sms"
DINGTALK = "dingtalk"
-2
View File
@@ -44,7 +44,6 @@ _ENV_ENABLE_CREDENTIALS: dict = {
Platform.SIGNAL: ("SIGNAL_HTTP_URL",),
Platform.MATTERMOST: ("MATTERMOST_TOKEN",),
Platform.MATRIX: ("MATRIX_ACCESS_TOKEN", "MATRIX_PASSWORD"),
Platform.HOMEASSISTANT: ("HASS_TOKEN",),
Platform.EMAIL: ("EMAIL_ADDRESS", "EMAIL_PASSWORD", "EMAIL_IMAP_HOST", "EMAIL_SMTP_HOST"),
Platform.SMS: ("TWILIO_ACCOUNT_SID",),
Platform.DINGTALK: ("DINGTALK_CLIENT_ID", "DINGTALK_CLIENT_SECRET"),
@@ -555,7 +554,6 @@ _ENV_STEPS: tuple = (
then=_matrix_e2ee,
),
_Home(Platform.MATRIX, "MATRIX_HOME_ROOM"),
_Cred(Platform.HOMEASSISTANT, ("HASS_TOKEN",), token="HASS_TOKEN", optional=(("url", "HASS_URL"),)),
_Cred(
Platform.EMAIL, ("EMAIL_ADDRESS", "EMAIL_PASSWORD", "EMAIL_IMAP_HOST", "EMAIL_SMTP_HOST"),
fixed=(("address", "EMAIL_ADDRESS"), ("imap_host", "EMAIL_IMAP_HOST"), ("smtp_host", "EMAIL_SMTP_HOST")),
+19 -3
View File
@@ -1,7 +1,8 @@
"""Per-platform display/verbosity resolver (``resolve_display_setting``).
Resolution order, first non-None wins: ``display.platforms.<platform>.<key>`` →
``display.<key>`` → ``_PLATFORM_DEFAULTS[platform][key]`` → ``_GLOBAL_DEFAULTS[key]``.
``display.<key>`` → ``_PLATFORM_DEFAULTS[platform][key]`` (a plugin platform: its registered
``display_tier``) → ``_GLOBAL_DEFAULTS[key]``.
Exception: ``display.streaming`` is CLI-only; gateway streaming follows the top-level
``streaming`` config unless a per-platform override sets it. Legacy
``display.tool_progress_overrides`` is still read as a ``tool_progress`` fallback.
@@ -71,10 +72,25 @@ _PLATFORM_DEFAULTS: dict[str, dict[str, Any]] = {
"email": _TIER_MINIMAL,
"sms": _TIER_MINIMAL,
"webhook": _TIER_MINIMAL,
"homeassistant": _TIER_MINIMAL,
"api_server": {**_TIER_HIGH, "tool_preview_length": 0},
}
_TIERS = {"high": _TIER_HIGH, "medium": _TIER_MEDIUM, "low": _TIER_LOW, "minimal": _TIER_MINIMAL}
def _platform_defaults(platform_key: str) -> dict[str, Any]:
"""Built-in defaults for *platform_key*: the table above, else the ``display_tier`` a plugin
platform registered (``PlatformEntry.display_tier``)."""
if platform_key in _PLATFORM_DEFAULTS:
return _PLATFORM_DEFAULTS[platform_key]
try:
from gateway.platform_registry import platform_registry
entry = platform_registry.get(platform_key)
except Exception:
return {}
return _TIERS.get(str(getattr(entry, "display_tier", "") or "").lower(), {})
# Canonical set of per-platform overrideable keys (for validation).
OVERRIDEABLE_KEYS = frozenset(_GLOBAL_DEFAULTS.keys())
@@ -88,7 +104,7 @@ def resolve_display_setting(user_config: dict, platform_key: str, setting: str,
configured = _configured_display_value(user_config, platform_key, setting)
if configured is not None:
return _normalise(setting, configured)
val = _PLATFORM_DEFAULTS.get(platform_key, {}).get(setting)
val = _platform_defaults(platform_key).get(setting)
if val is None:
val = _GLOBAL_DEFAULTS.get(setting)
return fallback if val is None else val
+10
View File
@@ -108,6 +108,16 @@ class PlatformEntry:
# ``async (pconfig, chat_id, message, *, thread_id=None, media_files=None, force_document=False)
# -> {"success": True, "message_id": ...} | {"error": str}``.
standalone_sender_fn: Optional[Callable[..., Awaitable[dict]]] = None
# Every inbound event is produced by the service the adapter authenticated to with its own
# credential (an event bus, not a person), so user allowlists and pairing do not apply. Never
# set it for a chat platform: any sender would reach the agent. Consumer: Home Assistant.
trusted_inbound: bool = False
# Built-in display defaults tier ("high" | "medium" | "low" | "minimal", see
# gateway/display_config.py) used below the user's display.* overrides; "" = global defaults.
display_tier: str = ""
# Env prefixes shared with a non-channel capability (the same plugin's tools): profile clones
# strip keys under them only when the source profile runs the adapter (profile_channels.py).
shared_env_prefixes: tuple = ()
class PlatformRegistry:
+1 -1
View File
@@ -51,7 +51,7 @@ _UNPARSEABLE = object()
_BUILTIN_DELIVER_PLATFORMS = {
"telegram", "discord", "slack", "signal", "sms", "whatsapp", "matrix", "mattermost",
"homeassistant", "email", "dingtalk", "feishu", "wecom", "wecom_callback", "weixin",
"email", "dingtalk", "feishu", "wecom", "wecom_callback", "weixin",
"bluebubbles", "qqbot", "yuanbao"}
# ``None`` → aiohttp binds BOTH address families. "0.0.0.0" is IPv4-only (unreachable on IPv6-only
+8
View File
@@ -5862,6 +5862,14 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
_start_gateway_configure_logging(verbosity)
# A platform that left core for a catalog plugin (Home Assistant) must be installed before the
# runner loads platform config; the outcome is logged and handed to this home's first agent.
try:
from hermes_cli.left_core_migration import recover_at_startup as _recover_left_core
await asyncio.to_thread(_recover_left_core)
except Exception:
logger.debug("left-core plugin migration skipped", exc_info=True)
runner = GatewayRunner(config)
# Multiplex: swap the launch-home file handlers for per-profile routers so each profile's records
# land in its own logs/. Must run after the runner resolved (possibly None) config and setup_logging.
+1 -1
View File
@@ -201,7 +201,7 @@ root). Profiles are independent
islands by design — no live config inheritance; `--clone` copies at creation, minus messaging
channels (`profile_channels.py`: ownership-based inventory evaluated in the SOURCE's plugin scope —
adapter-declared keys + canonical/alias prefixes + `GATEWAY_ALLOW*`/`GATEWAY_RELAY_*`; prefixes shared
with tools (`HASS_`/`TWILIO_`/`EMAIL_`) are stripped only when the source runs that adapter; never a hand
with tools (`TWILIO_`/`EMAIL_`, plus a plugin platform's `shared_env_prefixes`) are stripped only when the source runs that adapter; never a hand
list). `--clone-channels` opts in and its live-multiplexer refusal lives in `create_profile` (CLI, REST
and TUI all go through it). Clones are built in `profiles/.<name>.staging-<pid>` (hidden → invisible to
`_iter_named_profile_dirs` and the hot-serve rescan) and published by one `os.rename` after the strip;
+1 -1
View File
@@ -138,7 +138,7 @@ _PLATFORM_ENV_VARS = {
"telegram": "TELEGRAM_BOT_TOKEN", "discord": "DISCORD_BOT_TOKEN", "slack": "SLACK_BOT_TOKEN",
"whatsapp": "WHATSAPP_ENABLED", "signal": "SIGNAL_HTTP_URL", "email": "EMAIL_ADDRESS",
"sms": "TWILIO_ACCOUNT_SID", "matrix": "MATRIX_HOMESERVER_URL", "mattermost": "MATTERMOST_URL",
"homeassistant": "HASS_TOKEN", "dingtalk": "DINGTALK_CLIENT_ID", "feishu": "FEISHU_APP_ID",
"dingtalk": "DINGTALK_CLIENT_ID", "feishu": "FEISHU_APP_ID",
"wecom": "WECOM_BOT_ID", "wecom_callback": "WECOM_CALLBACK_CORP_ID", "weixin": "WEIXIN_ACCOUNT_ID",
"qqbot": "QQ_APP_ID",
}
+258
View File
@@ -0,0 +1,258 @@
"""Move a home onto the catalog plugin of a feature that left core.
A feature (gateway platform, toolset) that moves from core into a standalone catalog plugin keeps its
names, config keys and env vars, so migrating is only "install the plugin". Each row of
:data:`LEFT_CORE` names the catalog plugin and a read-only predicate "does this home use it". The
contract is the one memory providers established (``memory_provider_migration``, whose helpers this
reuses; memory is not a row because its plugin name comes from ``memory.provider``, not a table):
* ``hermes update`` installs the plugin for every profile home sharing the venv that uses the feature.
* Agent start and gateway start retry once per process for the active home (Desktop users update
through the app and never run ``hermes update``), honouring ``security.allow_lazy_installs``.
Installs go through the normal catalog install path at the reviewed pin (kill list, dependency
constraints, enable). Unattended dependency consent covers only these rows: the feature shipped in
core, so its users already accepted its dependencies. Every outcome reaches the user (terminal,
Desktop, chat); a gateway-start outcome waits for the home's first agent to deliver it.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Optional
from hermes_cli.memory_provider_migration import (
_home_consent, _home_label, _install_command, _interactive, _unattended_consent,
)
logger = logging.getLogger(__name__)
def _read_config(home: Path) -> dict:
import utils
try:
data = utils.fast_safe_load((home / "config.yaml").read_text(encoding="utf-8-sig"))
except Exception: # missing/unreadable/invalid YAML: nothing we can see is in use
return {}
return data if isinstance(data, dict) else {}
def _platform_block(config: dict, name: str) -> dict:
"""``platforms.<name>`` (either nesting, like gateway/config_loader.py)."""
gateway = config.get("gateway")
gateway = gateway if isinstance(gateway, dict) else {}
for section in (config.get("platforms"), gateway.get("platforms")):
block = section.get(name) if isinstance(section, dict) else None
if isinstance(block, dict):
return block
return {}
def _toolset_listed(config: dict, names: frozenset[str]) -> bool:
"""A toolset in *names* is selected in ``platform_toolsets`` or the top-level ``toolsets`` list."""
selections = list((config.get("platform_toolsets") or {}).values()) if isinstance(
config.get("platform_toolsets"), dict) else []
selections.append(config.get("toolsets"))
return any(isinstance(sel, list) and any(str(item) in names for item in sel) for sel in selections)
def homeassistant_in_use(home: Path, *, process_env: bool = False) -> bool:
"""What made core run Home Assistant for *home*: ``HASS_TOKEN`` in its ``.env`` (it enabled both
the gateway platform and the tools), ``platforms.homeassistant`` enabled or holding a token in
config.yaml, or the ``homeassistant`` / ``hermes-homeassistant`` toolset selected for a
platform. *process_env* (the active home at startup only) also counts a ``HASS_TOKEN`` the
process received from its environment (systemd unit, Docker, shell export)."""
from agent.secret_scope import load_env_file
if (load_env_file(home / ".env").get("HASS_TOKEN") or "").strip():
return True
if process_env:
try:
from agent.secret_scope import get_secret
if (get_secret("HASS_TOKEN", "") or "").strip():
return True
except Exception:
pass
config = _read_config(home)
block = _platform_block(config, "homeassistant")
if block.get("enabled") is True or (block.get("enabled") is not False and str(block.get("token") or "").strip()):
return True
return _toolset_listed(config, frozenset({"homeassistant", "hermes-homeassistant"}))
@dataclass(frozen=True)
class LeftCoreFeature:
plugin: str # catalog entry name == installed plugin name
label: str # user-facing feature name
in_use: Callable[..., bool] # (home, *, process_env=False) -> bool, read-only
unchanged: str # what migrated users keep, shown on success
LEFT_CORE: tuple[LeftCoreFeature, ...] = (
LeftCoreFeature(
plugin="homeassistant", label="Home Assistant", in_use=homeassistant_in_use,
unchanged="HASS_TOKEN/HASS_URL, platforms.homeassistant and the ha_* tool names are unchanged",
),
)
_attempted: set[str] = set()
_undelivered: dict[str, list[str]] = {}
def plugin_present(plugin: str, home: Path) -> bool:
"""Installed in *home* (enabled or not: a user who disabled it chose to). Read-only."""
return (home / "plugins" / plugin).is_dir()
def _pending(home: Path, *, say: Callable[[str], None], process_env: bool = False) -> list[LeftCoreFeature]:
"""Rows *home* uses whose plugin is not installed and that the catalog ships (a catalog miss is
reported through *say*). Read-only."""
from hermes_cli.memory_provider_migration import catalog_source
out = []
for feature in LEFT_CORE:
if plugin_present(feature.plugin, home) or not feature.in_use(home, process_env=process_env):
continue
if catalog_source(feature.plugin) is None:
say(f" ⚠ {feature.label} moved out of core into the '{feature.plugin}' plugin, which this "
f"Hermes cannot find in the plugin catalog yet. Run `{_install_command(feature.plugin, home)}` "
f"once it is listed.")
continue
out.append(feature)
return out
def _install_into(home: Path) -> Callable[[str], dict]:
def _install(name: str) -> dict:
from hermes_cli.plugins_cmd import dashboard_install_plugin
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
token = set_hermes_home_override(home)
try:
return dashboard_install_plugin("", force=False, enable=True, catalog_name=name,
assume_deps_consent=_unattended_consent())
finally:
reset_hermes_home_override(token)
return _install
def _install_one(home: Path, feature: LeftCoreFeature, *, install: Callable[[str], dict],
say: Callable[[str], None]) -> bool:
try:
result = install(feature.plugin)
except Exception as exc: # network, uv, kill list — report, do not raise
result = {"ok": False, "error": str(exc)}
if result.get("ok"):
say(f" ✓ {feature.label} moved out of core — installed the '{feature.plugin}' plugin from the "
f"catalog ({feature.unchanged}).")
return True
error = str(result.get("error") or "unknown error").rstrip(". ")
say(f" ⚠ {feature.label} moved out of core and its '{feature.plugin}' plugin could not be installed "
f"automatically: {error}. Run `{_install_command(feature.plugin, home)}`.")
return False
def migrate_home(home: Path, *, install: Callable[[str], dict], say: Callable[[str], None] = print,
process_env: bool = False) -> list[str]:
"""Install every left-core plugin *home* uses and lacks. Returns installed plugin names; never raises."""
installed = []
for feature in _pending(home, say=say, process_env=process_env):
if _install_one(home, feature, install=install, say=say):
installed.append(feature.plugin)
return installed
def migrate_all_homes(*, say: Callable[[str], None] = print) -> list[str]:
"""``hermes update`` hook: every profile home sharing this venv. Grouped by plugin and each home's
unattended consent like the memory migration: homes of one group share dependency answers, and a
failure names the rest of its group in one line instead of failing them one by one."""
from hermes_cli.plugins_cmd_install import shared_dependency_answers
from pm.plugins_state import dependency_homes
def labelled(home: Path) -> Callable[[str], None]:
return lambda message: say(f" [{_home_label(home)}] {message.lstrip()}")
pending: dict[tuple[str, bool], tuple[LeftCoreFeature, list[Path]]] = {}
for home in dependency_homes():
try:
features = _pending(home, say=labelled(home))
consent = bool(features) and _home_consent(home)
except Exception as exc:
logger.debug("left-core migration skipped for %s: %s", home, exc)
continue
for feature in features:
pending.setdefault((feature.plugin, consent), (feature, []))[1].append(home)
installed: list[str] = []
try:
for feature, homes in pending.values():
if len(homes) > 1 and _interactive():
say(f" {feature.label} is used in {len(homes)} profiles "
f"({', '.join(_home_label(h) for h in homes)}); your answers to its dependency "
f"questions apply to all of them.")
with shared_dependency_answers():
for index, home in enumerate(homes):
if _install_one(home, feature, install=_install_into(home), say=labelled(home)):
installed.append(feature.plugin)
continue
rest = homes[index + 1:]
if rest:
say(f" ⚠ The '{feature.plugin}' plugin was not installed for "
f"{', '.join(_home_label(h) for h in rest)} either. Run "
+ ", ".join(f"`{_install_command(feature.plugin, h)}`" for h in rest) + ".")
break
except KeyboardInterrupt:
say(" ⚠ Plugin migration cancelled. Profiles already migrated keep their plugin; run "
"`hermes plugins install <name>` (with `-p <profile>`) for the rest.")
return installed
def recover_at_startup(*, say: Optional[Callable[[str], None]] = None) -> list[str]:
"""Agent/gateway start hook for the active home: one attempt per process per home. With *say*
(an agent's startup-warning sink) outcomes are delivered now, together with any a gateway-start
attempt queued for this home; without it they are logged and queued for the home's first agent.
Returns installed plugin names."""
from hermes_constants import get_hermes_home, hermes_home_key
home = Path(get_hermes_home())
key = hermes_home_key(home)
def deliver(message: str) -> None:
if say is None:
_undelivered.setdefault(key, []).append(message)
return
try:
say(message)
except Exception:
logger.debug("left-core migration notification failed", exc_info=True)
if say is not None:
for message in _undelivered.pop(key, []):
deliver(message)
if key in _attempted:
return []
_attempted.add(key)
def report(message: str) -> None:
message = message.strip()
logger.warning(message)
deliver(message)
try:
features = _pending(home, say=report, process_env=True)
if not features:
return []
from pm.install import lazy_installs_allowed
if not lazy_installs_allowed():
for feature in features:
report(f"⚠ {feature.label} moved out of core and its '{feature.plugin}' plugin is not "
f"installed, so it is off. security.allow_lazy_installs is off, so Hermes did not "
f"fetch it: run `{_install_command(feature.plugin, home)}`.")
return []
installed = []
for feature in features:
if _install_one(home, feature, install=_install_into(home), say=report):
installed.append(feature.plugin)
return installed
except Exception as exc: # never take agent/gateway start down
logger.warning("left-core plugin migration failed: %s", exc, exc_info=True)
return []
@@ -3252,7 +3252,6 @@
"hermes-email",
"hermes-feishu",
"hermes-gateway",
"hermes-homeassistant",
"hermes-matrix",
"hermes-mattermost",
"hermes-qqbot",
@@ -3266,7 +3265,6 @@
"hermes-weixin",
"hermes-whatsapp",
"hermes-yuanbao",
"homeassistant",
"image_gen",
"kanban",
"memory",
@@ -4256,10 +4254,6 @@
"feishu_drive_reply_comment",
"focus_pane",
"gui_tour",
"ha_call_service",
"ha_get_state",
"ha_list_entities",
"ha_list_services",
"image_generate",
"kanban_attach",
"kanban_attach_url",
@@ -1486,7 +1486,7 @@ _TOOL_CATEGORY_PREFIXES = (
("mcp", "mcp"),
("browser", "browser"),
(("image", "tts", "video", "vision"), "media"),
("homeassistant", "home_automation"),
("homeassistant", "home_automation"), # the homeassistant catalog plugin's toolset
(("discord", "email", "feishu", "hermes-yuanbao", "slack", "sms"), "communication"),
)
-1
View File
@@ -21,7 +21,6 @@ PLATFORMS: OrderedDict[str, PlatformInfo] = OrderedDict([
("signal", PlatformInfo(label="📡 Signal", default_toolset="hermes-signal")),
("bluebubbles", PlatformInfo(label="💙 BlueBubbles", default_toolset="hermes-bluebubbles")),
("email", PlatformInfo(label="📧 Email", default_toolset="hermes-email")),
("homeassistant", PlatformInfo(label="🏠 Home Assistant", default_toolset="hermes-homeassistant")),
("mattermost", PlatformInfo(label="💬 Mattermost", default_toolset="hermes-mattermost")),
("matrix", PlatformInfo(label="💬 Matrix", default_toolset="hermes-matrix")),
("dingtalk", PlatformInfo(label="💬 DingTalk", default_toolset="hermes-dingtalk")),
+21 -10
View File
@@ -11,8 +11,8 @@ env-override table ``gateway.config_env._ENV_STEPS`` / ``_ENV_ENABLE_CREDENTIALS
under its canonical ``<PLATFORM>_`` prefix and its historical alias prefixes. Gateway-wide channel
policy (``GATEWAY_ALLOW_ALL_USERS`` / ``GATEWAY_ALLOWED_USERS``) and relay enrollment identity
(``GATEWAY_RELAY_*``) are channel settings too. A prefix an adapter SHARES with a non-channel
capability (``HASS_*`` is also the Home Assistant tool, ``TWILIO_*`` the telephony skill, ``EMAIL_*``
mail-sending scripts) is stripped only when the source actually runs that adapter — the credential is
capability (``TWILIO_*`` is also the telephony skill, ``EMAIL_*`` mail-sending scripts, and a plugin
platform's ``shared_env_prefixes`` its own tools) is stripped only when the source actually runs that adapter — the credential is
then the bot's identity; otherwise it is a tool key and survives. Model/provider keys, tool keys,
memory and general config are never touched.
"""
@@ -32,7 +32,6 @@ logger = logging.getLogger(__name__)
# ``<PLATFORM>_`` prefix (``WECOM_DM_POLICY`` and ``SMS_WEBHOOK_PORT`` are wecom / sms keys too).
_PLATFORM_ENV_PREFIX_ALIASES: dict[str, tuple[str, ...]] = {
"email": ("EMAIL_",),
"homeassistant": ("HASS_",),
"qqbot": ("QQ_",),
"relay": ("GATEWAY_RELAY_",),
"sms": ("TWILIO_",),
@@ -46,7 +45,6 @@ _GATEWAY_POLICY_KEYS = ("GATEWAY_ALLOW_ALL_USERS", "GATEWAY_ALLOWED_USERS")
# Prefixes a messaging adapter shares with a NON-channel capability. Their credentials belong to the
# channel only while the source runs that adapter; the adapter's policy keys are channel-only always.
_SHARED_WITH_TOOLS: dict[str, tuple[str, ...]] = {
"homeassistant": ("HASS_",), # tools/homeassistant_tool.py reads HASS_TOKEN / HASS_URL
"sms": ("TWILIO_",), # telephony skill reads TWILIO_ACCOUNT_SID / TWILIO_AUTH_TOKEN
"email": ("EMAIL_",), # mail-sending skills read EMAIL_ADDRESS / EMAIL_PASSWORD / EMAIL_SMTP_*
}
@@ -89,6 +87,17 @@ def _registry_entries() -> list:
return []
def _shared_with_tools(source_dir: Optional[Path] = None) -> Dict[str, Tuple[str, ...]]:
"""``_SHARED_WITH_TOOLS`` plus every plugin platform's ``shared_env_prefixes`` in ``source_dir``'s scope."""
shared = dict(_SHARED_WITH_TOOLS)
with _plugin_scope(source_dir):
for entry in _registry_entries():
prefixes = tuple(getattr(entry, "shared_env_prefixes", ()) or ())
if prefixes:
shared[entry.name] = prefixes
return shared
def platform_ids(source_dir: Optional[Path] = None) -> List[str]:
"""Every messaging platform id: built-in ``Platform`` members plus the plugin adapters registered
in ``source_dir``'s scope (ambient scope when ``None``)."""
@@ -227,12 +236,12 @@ def _explicit_enabled(raw: dict, pid: str) -> Optional[bool]:
return None
def _shared_adapters_active(source_dir: Optional[Path]) -> Set[str]:
def _shared_adapters_active(source_dir: Optional[Path], shared: Dict[str, Tuple[str, ...]]) -> Set[str]:
"""Shared-prefix platforms the SOURCE runs as a channel: explicitly enabled in its config.yaml, or
auto-enabled by a complete credential set in its ``.env`` and not explicitly disabled — the same
gate ``gateway.config_env._Cred`` applies at gateway start."""
if source_dir is None:
return set(_SHARED_WITH_TOOLS) # no source to consult: the historical (strip) behaviour
return set(shared) # no source to consult: the historical (strip) behaviour
raw: dict = {}
if (source_dir / "config.yaml").is_file():
from hermes_cli.config import read_user_config_raw
@@ -248,7 +257,7 @@ def _shared_adapters_active(source_dir: Optional[Path]) -> Set[str]:
for entry in _registry_entries():
creds_by_platform.setdefault(entry.name, set(entry.required_env))
active: Set[str] = set()
for pid in _SHARED_WITH_TOOLS:
for pid in shared:
explicit = _explicit_enabled(raw, pid)
if explicit is not None:
if explicit:
@@ -269,9 +278,11 @@ class ChannelKeyIndex:
self.platforms = platform_ids(source_dir)
self.declared = declared_channel_env_keys(source_dir)
self.policy = _policy_env_keys(source_dir)
self.shared_active = _shared_adapters_active(source_dir)
self.shared = _shared_with_tools(source_dir)
self.shared_active = _shared_adapters_active(source_dir, self.shared)
self._prefixes: List[Tuple[str, str]] = sorted(
((prefix, pid) for pid in self.platforms for prefix in platform_env_prefixes(pid)),
((prefix, pid) for pid in self.platforms
for prefix in dict.fromkeys((*platform_env_prefixes(pid), *self.shared.get(pid, ())))),
key=lambda item: -len(item[0]), # longest prefix wins: WECOM_CALLBACK_ before WECOM_
)
@@ -279,7 +290,7 @@ class ChannelKeyIndex:
pid = self.declared.get(key) or next((pid for prefix, pid in self._prefixes if key.startswith(prefix)), None)
if pid is None:
return None
shared = _SHARED_WITH_TOOLS.get(pid)
shared = self.shared.get(pid)
if shared and key.startswith(shared) and pid not in self.shared_active and not self._is_policy(key):
return None # a tool credential the source never used as a bot: keep it
return pid
+1 -5
View File
@@ -167,10 +167,6 @@ def _modal_row(config, feats):
return None
def _home_assistant_row(config, feats):
return ("Smart Home (Home Assistant)", True, None) if _setup.get_env_value("HASS_TOKEN") else None
def _spotify_row(config, feats):
# OAuth via hermes auth spotify — check auth.json, not env vars
try:
@@ -196,7 +192,7 @@ def _always_on_rows(config, feats):
_TOOL_ROW_BUILDERS = (
_vision_row, _web_row, _browser_row, _image_gen_row, _video_gen_row, _tts_row, _stt_row,
_modal_row, _home_assistant_row, _spotify_row, _skills_hub_row, _always_on_rows)
_modal_row, _spotify_row, _skills_hub_row, _always_on_rows)
def _print_cmd_rows(rows):
+8
View File
@@ -168,6 +168,14 @@ def build_update_products(project_root: Path, *, desktop: bool) -> None:
migrate_all_homes()
except Exception as exc:
print(f" ⚠ Memory provider migration skipped: {exc}")
# Same for a gateway platform / toolset that left core (Home Assistant): every home that used
# it gets its catalog plugin (hermes_cli/left_core_migration.py).
try:
from hermes_cli.left_core_migration import migrate_all_homes as migrate_left_core
migrate_left_core()
except Exception as exc:
print(f" ⚠ Plugin migration skipped: {exc}")
if __name__ == "__main__":
+3 -29
View File
@@ -87,7 +87,6 @@ CONFIGURABLE_TOOLSETS = [
("clarify", "❓ Clarifying Questions", "clarify"),
("delegation", "👥 Task Delegation", "delegate_task"),
("cronjob", "⏰ Cron Jobs", "create/list/update/pause/resume/run, with optional attached skills"),
("homeassistant", "🏠 Home Assistant", "smart home device control"),
("spotify", "🎵 Spotify", "playback, search, playlists, library"),
("discord", "💬 Discord (read/participate)", "fetch messages, search members, create thread"),
("discord_admin", "🛡️ Discord Server Admin", "list channels/roles, pin, assign roles"),
@@ -107,8 +106,8 @@ def gui_toolset_label(label: str) -> str:
# OFF by default for new installs (still in _HERMES_CORE_TOOLS; the checklist won't pre-select them). x_search
# auto-enables when xAI creds exist (mirrors HASS_TOKEN → homeassistant); its check_fn still gates the schema.
_DEFAULT_OFF_TOOLSETS = {"homeassistant", "spotify", "discord", "discord_admin", "video", "video_gen", "x_search", "a2a", "kanban"}
# auto-enables when xAI creds exist; its check_fn still gates the schema.
_DEFAULT_OFF_TOOLSETS = {"spotify", "discord", "discord_admin", "video", "video_gen", "x_search", "a2a", "kanban"}
# Config-only capabilities: provider setup in `hermes tools` (TOOL_CATEGORIES) but not model toolsets — zero
# schemas, own switch (``stt.enabled``), never in ``platform_toolsets`` or the per-platform checklist.
@@ -133,15 +132,6 @@ def _xai_credentials_present() -> bool:
return bool(str(get_secret("XAI_API_KEY") or "").strip())
def _homeassistant_credentials_present() -> bool:
"""Return whether the active profile has a Home Assistant token."""
try:
from agent.secret_scope import get_secret
return bool((get_secret("HASS_TOKEN", "") or "").strip())
except Exception:
return False
def _toolset_configuration_platform(ts_key: str, default: str = "cli") -> str:
"""Platform a platform-less configuration UI should target: a toolset restricted away from ``default``
must be configured on a supported platform, else the save helper drops it and the UI reports a no-op."""
@@ -355,14 +345,6 @@ TOOL_CATEGORIES = {
post_setup="browser_use_cli"),
],
},
"homeassistant": {
"name": "Smart Home", "icon": "🏠",
"providers": [
_row("Home Assistant", tag="REST API integration",
env_vars=[_key("HASS_TOKEN", "Home Assistant Long-Lived Access Token"),
_key("HASS_URL", "Home Assistant URL", default="http://homeassistant.local:8123")]),
],
},
"spotify": {
"name": "Spotify", "icon": "🎵",
"providers": [_row("Spotify Web API", tag="PKCE OAuth — opens the setup wizard", post_setup="spotify")],
@@ -490,20 +472,12 @@ def _configurable_subset_of(tool_names: Set[str], platform: str) -> Set[str]:
def _default_off_toolsets(platform: str, explicitly_configured: bool) -> Set[str]:
"""Toolsets to strip from an implicit (composite-derived) enable set. A platform named after a default-off
toolset (``homeassistant``) keeps it, except platform-restricted ones (``discord`` on discord stays OFF); a
configured HASS_TOKEN is an explicit opt-in that must survive platforms resolving without a saved list.
toolset keeps it, except platform-restricted ones (``discord`` on discord stays OFF).
Platform-native default-off toolsets (``discord`` on discord) are off for unconfigured platforms as a
security opt-in — an explicitly saved list IS that opt-in and lets them through."""
default_off = set(_DEFAULT_OFF_TOOLSETS)
if platform in default_off and platform not in _TOOLSET_PLATFORM_RESTRICTIONS:
default_off.remove(platform)
# Home Assistant is already runtime-gated by its check_fn (requires HASS_TOKEN to register any tools).
# When a user has configured HASS_TOKEN, they've explicitly opted in — don't also strip it via
# _DEFAULT_OFF_TOOLSETS, which would silently drop HA from platforms (e.g. cron) that run through
# _get_platform_tools without an explicit saved toolset list. Without this, Norbert's HA cron jobs
# regressed after #14798 made cron honor per-platform tool config.
if "homeassistant" in default_off and _homeassistant_credentials_present():
default_off.remove("homeassistant")
if explicitly_configured:
default_off -= {ts for ts in default_off if platform in (_TOOLSET_PLATFORM_RESTRICTIONS.get(ts) or ())}
return default_off
+1 -3
View File
@@ -62,7 +62,7 @@ _probe_gateway_health = late("_probe_gateway_health", "hermes_cli.web_server_gat
get_running_pid_cached = late("get_running_pid_cached", "gateway.status")
get_runtime_status_running_pid = late("get_runtime_status_running_pid", "gateway.status")
_GATEWAY_HEALTH_URL = LateState("_GATEWAY_HEALTH_URL")
# Display labels for env vars not in OPTIONAL_ENV_VARS (bridge toggles, Twilio, HASS, Email, ...)
# Display labels for env vars not in OPTIONAL_ENV_VARS (bridge toggles, Twilio, Email, ...)
# so the UI can still render a friendly label. Rows: (key, description, prompt, extra flags).
_MESSAGING_ENV_FALLBACKS: dict[str, dict[str, Any]] = {
key: {"description": description, "prompt": prompt, **extra}
@@ -74,8 +74,6 @@ _MESSAGING_ENV_FALLBACKS: dict[str, dict[str, Any]] = {
("WHATSAPP_MODE", "WhatsApp bridge mode", "WhatsApp mode", {"advanced": True}),
("WHATSAPP_DM_POLICY", "How WhatsApp direct messages are authorized", "WhatsApp DM policy", {"advanced": True}),
("WHATSAPP_ALLOWED_USERS", "Comma-separated WhatsApp users allowed to use the bot", "Allowed WhatsApp users", {}),
("HASS_URL", "Home Assistant base URL, e.g. https://homeassistant.local:8123", "Home Assistant URL", {}),
("HASS_TOKEN", "Long-lived access token from Home Assistant (Profile → Security)", "Home Assistant access token", {"password": True}),
("EMAIL_ADDRESS", "Email address to send and receive from", "Email address", {}),
("EMAIL_PASSWORD", "Email account password or app password", "Email password", {"password": True}),
("EMAIL_IMAP_HOST", "IMAP server host (e.g. imap.gmail.com)", "IMAP host", {}),
+1 -7
View File
@@ -71,12 +71,6 @@ _PLATFORM_OVERRIDES: dict[str, dict[str, Any]] = {
),
"required_env": (),
},
"homeassistant": {
"name": "Home Assistant",
"description": "Control your smart home from Hermes via Home Assistant.",
"docs_url": "https://www.home-assistant.io/docs/authentication/",
"env_vars": ("HASS_URL", "HASS_TOKEN"), "required_env": ("HASS_URL", "HASS_TOKEN"),
},
"email": {
"name": "Email", "description": "Talk to Hermes through an IMAP/SMTP mailbox.",
"docs_url": "https://hermes-agent.nousresearch.com/docs/user-guide/messaging/",
@@ -221,7 +215,7 @@ _PLATFORM_OVERRIDES: dict[str, dict[str, Any]] = {
# the end alphabetically.
_PLATFORM_ORDER: tuple[str, ...] = (
"telegram", "discord", "slack", "mattermost", "matrix", "whatsapp", "signal", "bluebubbles",
"homeassistant", "email", "sms", "dingtalk", "feishu", "google_chat", "wecom", "wecom_callback",
"email", "sms", "dingtalk", "feishu", "google_chat", "wecom", "wecom_callback",
"weixin", "qqbot", "yuanbao", "api_server", "webhook",
)
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent skakel af"
standalone_quit: "afgelewer"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Eenmalig goedkeur"
confirm_always_approve: "Altyd goedkeur"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "جارٍ إيقاف تشغيل Hermes Agent"
standalone_quit: "تم التسليم"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "الموافقة مرة"
confirm_always_approve: "الموافقة دائمًا"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent wird heruntergefahren"
standalone_quit: "zugestellt"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Einmalig genehmigen"
confirm_always_approve: "Immer genehmigen"
-2
View File
@@ -1519,8 +1519,6 @@ platform:
quit_message: "Hermes Agent shutting down"
standalone_quit: "delivered"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Approve Once"
confirm_always_approve: "Always Approve"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent se está apagando"
standalone_quit: "entregado"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Aprobar una vez"
confirm_always_approve: "Aprobar siempre"
-2
View File
@@ -1700,8 +1700,6 @@ platform:
irc:
quit_message: "Hermes Agent s'arrête"
standalone_quit: "livré"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Approuver une fois"
confirm_always_approve: "Toujours approuver"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent ag múchadh"
standalone_quit: "seachadta"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Ceadaigh uair amháin"
confirm_always_approve: "Ceadaigh i gcónaí"
-2
View File
@@ -1465,8 +1465,6 @@ platform:
irc:
quit_message: "Hermes Agent leáll"
standalone_quit: "kézbesítve"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Jóváhagyás egyszer"
confirm_always_approve: "Mindig jóváhagy"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent in arresto"
standalone_quit: "consegnato"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Approva una volta"
confirm_always_approve: "Approva sempre"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent をシャットダウンしています"
standalone_quit: "配信済み"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "一度だけ承認"
confirm_always_approve: "常に承認"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent를 종료하는 중"
standalone_quit: "전송됨"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "한 번 승인"
confirm_always_approve: "항상 승인"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent encerrando"
standalone_quit: "entregue"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Aprovar uma vez"
confirm_always_approve: "Aprovar sempre"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent завершает работу"
standalone_quit: "доставлено"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Одобрить один раз"
confirm_always_approve: "Одобрять всегда"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent kapatılıyor"
standalone_quit: "teslim edildi"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Bir Kez Onayla"
confirm_always_approve: "Her Zaman Onayla"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent завершує роботу"
standalone_quit: "доставлено"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "Схвалити один раз"
confirm_always_approve: "Завжди схвалювати"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent 正在關閉"
standalone_quit: "已傳送"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "批准一次"
confirm_always_approve: "永遠批准"
-2
View File
@@ -1448,8 +1448,6 @@ platform:
irc:
quit_message: "Hermes Agent 正在关闭"
standalone_quit: "已送达"
homeassistant:
notification_title: "Hermes Agent"
relay:
confirm_approve_once: "批准一次"
confirm_always_approve: "始终批准"
+23
View File
@@ -0,0 +1,23 @@
name: homeassistant
repo: https://github.com/NousResearch/hermes-homeassistant
sha: f272c96b9a538a2327b23023341f571e563ea475
description: 'Home Assistant gateway platform (state-change events in, persistent notifications and cron notify
delivery out) plus the homeassistant toolset (ha_list_entities, ha_get_state, ha_list_services, ha_call_service).
Formerly built into Hermes core; existing users are migrated automatically.'
maintainer: NousResearch
tier: official
category: platform
requires_hermes: ">=0.21.5"
version: "2.0.0"
docs_url: https://github.com/NousResearch/hermes-homeassistant
platforms: []
capabilities:
provides_tools:
- ha_list_entities
- ha_get_state
- ha_list_services
- ha_call_service
provides_hooks: []
provides_middleware: []
requires_env:
- HASS_TOKEN
@@ -1,3 +0,0 @@
from .adapter import register
__all__ = ["register"]
-374
View File
@@ -1,374 +0,0 @@
"""Home Assistant adapter: WS ``state_changed`` events -> MessageEvents; outbound = persistent notifications.
Requires aiohttp, HASS_TOKEN (Long-Lived Access Token) and HASS_URL (default http://homeassistant.local:8123).
"""
import asyncio
import errno
import json
import logging
import sys
import time
import uuid
from datetime import datetime
from typing import Any, Dict, Optional, Set
try:
import aiohttp
AIOHTTP_AVAILABLE = True
except ImportError:
AIOHTTP_AVAILABLE = False
aiohttp = None # type: ignore[assignment]
from gateway.restart import is_supervised_gateway_launch
from gateway.config import Platform, PlatformConfig
from agent.i18n import t
from gateway.platforms.base import gateway_trust_env, BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType
from gateway.platforms._shared import (
env_is_connected as _env_is_connected, get_scoped_secret as _get_scoped_secret, send_error
)
logger = logging.getLogger(__name__)
def check_ha_requirements() -> bool:
"""Check if Home Assistant runtime dependencies are available."""
return AIOHTTP_AVAILABLE
def validate_ha_config(config: PlatformConfig) -> bool:
"""True when Home Assistant has enough credential config to connect."""
return bool((getattr(config, "token", None) or _get_scoped_secret("HASS_TOKEN", "")).strip())
def _domain_of(entity_id: str) -> str:
return entity_id.split(".")[0] if "." in entity_id else ""
def _auth_headers(token: str) -> Dict[str, str]:
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
# domain -> description template; see ``_format_state_change`` for the fields.
_TURNED = "[Home Assistant] {name}: turned {on_off}"
_DOMAIN_TEMPLATES = {
"climate": (
"[Home Assistant] {name}: HVAC mode changed from "
"'{old}' to '{new}' (current: {temp}, target: {target})"
),
"sensor": "[Home Assistant] {name}: changed from {old}{unit} to {new}{unit}",
"binary_sensor": "[Home Assistant] {name}: {new_trig} (was {old_trig})",
"light": _TURNED,
"switch": _TURNED,
"fan": _TURNED,
"alarm_control_panel": "[Home Assistant] {name}: alarm state changed from '{old}' to '{new}'",
}
_DEFAULT_TEMPLATE = "[Home Assistant] {name} ({entity_id}): changed from '{old}' to '{new}'"
_TRIGGERED = ("cleared", "triggered") # binary_sensor wording, indexed by ``state == "on"``
def _connect_error_detail(exc: BaseException) -> str:
"""Annotate macOS Local Network Privacy denials so launchd HA failures are actionable (#71206).
Only a supervised (launchd) gateway on macOS can be denied this way; the same errno from a
Terminal-run gateway or on another OS is a genuinely unreachable host.
"""
text = str(exc)
os_error = getattr(exc, "os_error", None) or exc.__cause__ or exc
if (sys.platform == "darwin" and is_supervised_gateway_launch()
and getattr(os_error, "errno", None) == errno.EHOSTUNREACH):
return (
f"{text} — macOS Local Network Privacy is blocking this launchd gateway from the LAN. "
"Run `hermes gateway install` to regenerate the launchd job, then `hermes gateway restart`. "
"https://github.com/NousResearch/hermes-agent/issues/71206"
)
return text
class HomeAssistantAdapter(BasePlatformAdapter):
"""``state_changed`` -> MessageEvents with domain/entity filtering and per-entity cooldowns."""
MAX_MESSAGE_LENGTH = 4096
_BACKOFF_STEPS = [5, 10, 30, 60] # reconnect backoff (seconds)
def __init__(self, config: PlatformConfig):
super().__init__(config, Platform.HOMEASSISTANT)
self._session: Optional["aiohttp.ClientSession"] = None
self._ws: Optional["aiohttp.ClientWebSocketResponse"] = None
self._rest_session: Optional["aiohttp.ClientSession"] = None
self._listen_task: Optional[asyncio.Task] = None
self._msg_id: int = 0
extra = config.extra or {}
# URL is scoped like the token below: a secondary's HASS_TOKEN must never be posted to the
# DEFAULT profile's HA instance (os.environ under multiplex).
self._hass_url: str = (extra.get("url") or _get_scoped_secret("HASS_URL", "http://homeassistant.local:8123")).rstrip("/")
self._hass_token: str = config.token or _get_scoped_secret("HASS_TOKEN", "")
self._watch_domains: Set[str] = set(extra.get("watch_domains", []))
self._watch_entities: Set[str] = set(extra.get("watch_entities", []))
self._ignore_entities: Set[str] = set(extra.get("ignore_entities", []))
self._watch_all: bool = bool(extra.get("watch_all", False))
self._cooldown_seconds: int = int(extra.get("cooldown_seconds", 30))
self._last_event_time: Dict[str, float] = {} # entity_id -> last event ts
def _next_id(self) -> int:
self._msg_id += 1
return self._msg_id
@staticmethod
def _new_session() -> "aiohttp.ClientSession":
return aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30), trust_env=gateway_trust_env())
# -- Connection lifecycle -----------------------------------------------
async def connect(self, *, is_reconnect: bool = False) -> bool:
"""Connect to HA WebSocket API and subscribe to events."""
if not AIOHTTP_AVAILABLE:
logger.warning("[%s] aiohttp not installed. Run: pip install aiohttp", self.name)
return False
if not self._hass_token:
logger.warning("[%s] No HASS_TOKEN configured", self.name)
return False
try:
if not await self._ws_connect():
return False
self._rest_session = self._new_session() # dedicated REST session for send()
if not (self._watch_domains or self._watch_entities or self._watch_all):
logger.warning(
"[%s] No watch_domains, watch_entities, or watch_all configured. "
"All state_changed events will be dropped. Configure filters in "
"your HA platform config to receive events.",
self.name)
self._listen_task = asyncio.create_task(self._listen_loop())
self._running = True
logger.info("[%s] Connected to %s", self.name, self._hass_url)
self._wire_plugin_handlers(None)
return True
except Exception as e:
logger.error("[%s] Failed to connect: %s", self.name, _connect_error_detail(e))
return False
async def _ws_connect(self) -> bool:
"""Open the WebSocket, authenticate, and subscribe to ``state_changed``."""
ws_url = self._hass_url.replace("https://", "wss://").replace("http://", "ws://")
self._session = self._new_session()
self._ws = await self._session.ws_connect(f"{ws_url}/api/websocket", heartbeat=30, timeout=30)
msg = await self._ws.receive_json()
if msg.get("type") != "auth_required":
return await self._handshake_failed("Expected auth_required, got: %s", msg.get("type"))
await self._ws.send_json({"type": "auth", "access_token": self._hass_token})
msg = await self._ws.receive_json()
if msg.get("type") != "auth_ok":
return await self._handshake_failed("Auth failed: %s", msg)
await self._ws.send_json({"id": self._next_id(), "type": "subscribe_events", "event_type": "state_changed"})
msg = await self._ws.receive_json()
if not msg.get("success"):
return await self._handshake_failed("Failed to subscribe to events: %s", msg)
return True
async def _handshake_failed(self, fmt: str, detail: Any) -> bool:
logger.error(fmt, detail)
await self._cleanup_ws()
return False
@staticmethod
async def _close(obj) -> None:
if obj and not obj.closed:
await obj.close()
async def _cleanup_ws(self) -> None:
await self._close(self._ws)
self._ws = None
await self._close(self._session)
self._session = None
async def disconnect(self) -> None:
self._running = False
if self._listen_task:
self._listen_task.cancel()
try:
await self._listen_task
except asyncio.CancelledError:
pass
self._listen_task = None
await self._cleanup_ws()
await self._close(self._rest_session)
self._rest_session = None
logger.info("[%s] Disconnected", self.name)
# -- Event listener -----------------------------------------------------
async def _listen_loop(self) -> None:
"""Main event loop with automatic reconnection."""
backoff_idx = 0
while self._running:
try:
await self._read_events()
except asyncio.CancelledError:
return
except Exception as e:
logger.warning("[%s] WebSocket error: %s", self.name, _connect_error_detail(e))
if not self._running:
return
delay = self._BACKOFF_STEPS[min(backoff_idx, len(self._BACKOFF_STEPS) - 1)]
logger.info("[%s] Reconnecting in %ds...", self.name, delay)
await asyncio.sleep(delay)
backoff_idx += 1
try:
await self._cleanup_ws()
if await self._ws_connect():
backoff_idx = 0
logger.info("[%s] Reconnected", self.name)
except Exception as e:
logger.warning("[%s] Reconnection failed: %s", self.name, _connect_error_detail(e))
async def _read_events(self) -> None:
"""Read events from WebSocket until disconnected."""
if self._ws is None or self._ws.closed:
return
async for ws_msg in self._ws:
if ws_msg.type in {aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR}:
break
if ws_msg.type != aiohttp.WSMsgType.TEXT:
continue
try:
data = json.loads(ws_msg.data)
except json.JSONDecodeError:
logger.debug("Invalid JSON from HA WS: %s", ws_msg.data[:200])
continue
if data.get("type") == "event":
await self._handle_ha_event(data.get("event", {}))
def _passes_filters(self, entity_id: str) -> bool:
"""Closed by default: requires watch_domains, watch_entities, or watch_all."""
if entity_id in self._ignore_entities:
return False
if self._watch_domains or self._watch_entities:
return _domain_of(entity_id) in self._watch_domains or entity_id in self._watch_entities
return self._watch_all
async def _handle_ha_event(self, event: Dict[str, Any]) -> None:
"""Process a state_changed event from Home Assistant."""
event_data = event.get("data", {})
entity_id: str = event_data.get("entity_id", "")
if not entity_id or not self._passes_filters(entity_id):
return
now = time.time()
if (now - self._last_event_time.get(entity_id, 0)) < self._cooldown_seconds:
return
self._last_event_time[entity_id] = now
message = self._format_state_change(
entity_id, event_data.get("old_state", {}), event_data.get("new_state", {}))
if not message:
return
source = self.build_source(
chat_id="ha_events", chat_name="Home Assistant Events", chat_type="channel",
user_id="homeassistant", user_name="Home Assistant")
await self.handle_message(MessageEvent(
text=message, message_type=MessageType.TEXT, source=source,
message_id=f"ha_{entity_id}_{int(now)}", timestamp=datetime.now()))
@staticmethod
def _format_state_change(entity_id: str, old_state: Dict[str, Any], new_state: Dict[str, Any]) -> Optional[str]:
"""Convert a state_changed event into a human-readable description."""
if not new_state:
return None
old_val = old_state.get("state", "unknown") if old_state else "unknown"
new_val = new_state.get("state", "unknown")
if old_val == new_val:
return None
attrs = new_state.get("attributes", {})
template = _DOMAIN_TEMPLATES.get(_domain_of(entity_id), _DEFAULT_TEMPLATE)
return template.format(
name=attrs.get("friendly_name", entity_id), entity_id=entity_id, old=old_val, new=new_val,
temp=attrs.get("current_temperature", "?"), target=attrs.get("temperature", "?"),
unit=attrs.get("unit_of_measurement", ""), on_off="on" if new_val == "on" else "off",
new_trig=_TRIGGERED[new_val == "on"], old_trig=_TRIGGERED[old_val == "on"])
# -- Outbound messaging -------------------------------------------------
async def send(
self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Send a notification via HA REST API (persistent_notification.create).
REST rather than the WebSocket, to avoid racing the listener loop that
reads from the same WS connection.
"""
url = f"{self._hass_url}/api/services/persistent_notification/create"
payload = {"title": t("platform.homeassistant.notification_title"), "message": content[:self.MAX_MESSAGE_LENGTH]}
async def _post(session) -> SendResult:
async with session.post(
url, headers=_auth_headers(self._hass_token), json=payload, timeout=aiohttp.ClientTimeout(total=10),
) as resp:
if resp.status < 300:
return SendResult(success=True, message_id=uuid.uuid4().hex[:12])
return SendResult(success=False, error=f"HTTP {resp.status}: {await resp.text()}")
try:
if self._rest_session:
return await _post(self._rest_session)
async with aiohttp.ClientSession(trust_env=gateway_trust_env()) as session:
return await _post(session)
except asyncio.TimeoutError:
return SendResult(success=False, error="Timeout sending notification to HA")
except Exception as e:
return SendResult(success=False, error=str(e))
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
return {"name": "Home Assistant Events", "type": "channel", "url": self._hass_url}
# -- Standalone (out-of-process) sender — cron deliver=homeassistant ---------
# ────────────────────────────────────────────────────────────────────────── Plugin migration glue (#41112 /
# #3823) Added when the Email adapter moved from gateway/platforms/email.py into this bundled plugin.
# register() exposes the platform via the registry, replacing the Platform.EMAIL elif in gateway/run.py, the
# _PLATFORM_CONNECTED_CHECKERS entry in gateway/config.py, the _PLATFORMS["email"] static dict in
# hermes_cli/gateway.py, and the _send_email dispatch in tools/send_message_tool.py. EMAIL_*
# env→PlatformConfig seeding stays in core.
# ──────────────────────────────────────────────────────────────────────────
async def _standalone_send(
pconfig, chat_id: str, message: str, *,
thread_id: Optional[str] = None, media_files: Optional[list] = None, force_document: bool = False,
) -> Dict[str, Any]:
"""Send via the HA ``notify.notify`` service without a live gateway adapter.
Token: ``pconfig.token`` then ``HASS_TOKEN``; URL: ``pconfig.extra["url"]`` then ``HASS_URL``.
``thread_id``/``media_files``/``force_document`` are signature parity only (HA has no threads/attachments).
"""
if not AIOHTTP_AVAILABLE:
return send_error("aiohttp not installed. Run: pip install aiohttp")
extra = getattr(pconfig, "extra", {}) or {}
hass_url = (extra.get("url") or _get_scoped_secret("HASS_URL", "")).rstrip("/")
token = (getattr(pconfig, "token", None) or _get_scoped_secret("HASS_TOKEN", "")).strip()
if not hass_url or not token:
return send_error("Home Assistant standalone send: HASS_URL and HASS_TOKEN must both be set")
url = f"{hass_url}/api/services/notify/notify"
payload = {"message": message, "target": chat_id}
try:
async with HomeAssistantAdapter._new_session() as session:
async with session.post(url, headers=_auth_headers(token), json=payload) as resp:
if resp.status not in {200, 201}:
return send_error(f"Home Assistant API error ({resp.status}): {await resp.text()}")
return {"success": True, "platform": "homeassistant", "chat_id": chat_id}
except asyncio.TimeoutError:
return send_error("Timeout sending notification to Home Assistant")
except Exception as e:
return send_error(f"Home Assistant send failed: {e}")
_is_connected = _env_is_connected("HASS_TOKEN")
def register(ctx) -> None:
"""Plugin entry point — called by the Hermes plugin system."""
ctx.register_platform(
name="homeassistant", label="Home Assistant", adapter_factory=HomeAssistantAdapter,
check_fn=check_ha_requirements, validate_config=validate_ha_config, is_connected=_is_connected,
required_env=["HASS_TOKEN"], install_hint="pip install aiohttp",
standalone_sender_fn=_standalone_send, # out-of-process cron delivery via notify.notify
max_message_length=HomeAssistantAdapter.MAX_MESSAGE_LENGTH, emoji="🏠", allow_update_command=True)
@@ -1,22 +0,0 @@
name: homeassistant-platform
label: Home Assistant
kind: platform
version: 1.0.0
description: >
Home Assistant gateway adapter for Hermes Agent.
Subscribes to HA's WebSocket event bus and forwards state-change events
(with per-entity cooldowns and domain/entity filtering) to the agent.
Outbound messages are delivered as HA persistent notifications via the
REST API. Out-of-process cron delivery via the ``notify.notify``
service is also supported.
author: NousResearch
requires_env:
- name: HASS_TOKEN
description: "Home Assistant Long-Lived Access Token"
prompt: "Home Assistant Long-Lived Access Token"
password: true
optional_env:
- name: HASS_URL
description: "Home Assistant base URL (default: http://homeassistant.local:8123)"
prompt: "Home Assistant URL"
password: false
-3
View File
@@ -406,7 +406,6 @@ mcp = [
# Backwards-compatible no-op alias. Relay is a core dependency on supported
# wheel targets and intentionally unavailable on other platforms.
nemo-relay = []
homeassistant = ["aiohttp==3.14.3"]
sms = ["aiohttp==3.14.3"]
teams = [
"microsoft-teams-apps==2.0.13.4",
@@ -455,7 +454,6 @@ termux-all = [
# first use, so they're no longer eager-installed here.
"hermes-agent[termux]",
"hermes-agent[google]",
"hermes-agent[homeassistant]",
"hermes-agent[sms]",
"hermes-agent[web]",
"hermes-agent[pty]",
@@ -526,7 +524,6 @@ all = [
"hermes-agent[pty]",
"hermes-agent[mcp]",
"hermes-agent[uvloop]",
"hermes-agent[homeassistant]",
"hermes-agent[sms]",
"hermes-agent[acp]",
"hermes-agent[google]",
@@ -33,7 +33,6 @@ MIGRATED_ADAPTER_MODULES = [
("plugins.platforms.teams.adapter", "TEAMS_CLIENT_SECRET"),
("plugins.platforms.mattermost.adapter", "MATTERMOST_TOKEN"),
("plugins.platforms.ntfy.adapter", "NTFY_TOKEN"),
("plugins.platforms.homeassistant.adapter", "HASS_TOKEN"),
("plugins.platforms.sms.adapter", "TWILIO_AUTH_TOKEN"),
("plugins.platforms.dingtalk.adapter", "DINGTALK_CLIENT_SECRET"),
("plugins.platforms.feishu.adapter", "FEISHU_APP_SECRET"),
@@ -3,7 +3,7 @@ in config.yaml must survive ``_apply_env_overrides`` when that platform's
credentials are present in the environment.
Before the fix, twelve credential-presence branches (weixin, whatsapp_cloud,
homeassistant, email, sms, dingtalk, feishu, wecom, wecom_callback, bluebubbles,
email, sms, dingtalk, feishu, wecom, wecom_callback, bluebubbles,
qqbot, yuanbao) force-set ``enabled = True`` unconditionally, while Telegram /
Discord / Slack routed through ``_enable_from_env`` and honored the
``_enabled_explicit`` marker. These tests drive the real ``load_gateway_config``
@@ -27,7 +27,6 @@ CRED_ENV = {
"WHATSAPP_CLOUD_PHONE_NUMBER_ID": "1234567890",
"WHATSAPP_CLOUD_ACCESS_TOKEN": "EAAB-test-access-token",
},
"homeassistant": {"HASS_TOKEN": "hass-long-lived-token"},
# flag-driven, not credential-driven: WHATSAPP_ENABLED=true must not beat an explicit YAML disable (#73289)
"whatsapp": {"WHATSAPP_ENABLED": "true"},
"email": {
@@ -151,16 +150,16 @@ def test_explicit_disable_with_env_credentials_warns_once(platform, tmp_path, mo
@pytest.mark.usefixtures("_fresh_warn_dedup")
def test_no_warning_when_yaml_has_no_opinion_or_is_enabled(tmp_path, monkeypatch, caplog):
hermes_home = _isolate(monkeypatch, tmp_path, {**CRED_ENV["weixin"], **CRED_ENV["homeassistant"]})
hermes_home = _isolate(monkeypatch, tmp_path, {**CRED_ENV["weixin"], **CRED_ENV["qqbot"]})
(hermes_home / "config.yaml").write_text(
"platforms:\n homeassistant:\n enabled: true\n", encoding="utf-8"
"platforms:\n qqbot:\n enabled: true\n", encoding="utf-8"
)
with caplog.at_level(logging.WARNING, logger="gateway.config"):
config = load_gateway_config()
assert config.platforms[Platform.WEIXIN].enabled is True
assert config.platforms[Platform.HOMEASSISTANT].enabled is True
assert config.platforms[Platform.QQBOT].enabled is True
assert not [r for r in caplog.records if "explicitly disabled" in r.getMessage()]
@pytest.mark.usefixtures("_fresh_warn_dedup")
-316
View File
@@ -1,316 +0,0 @@
"""Tests for the Home Assistant gateway adapter.
Tests real logic: state change formatting, event filtering pipeline,
cooldown behavior, config integration, and adapter initialization.
"""
import errno
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agent.i18n import t
from gateway.config import (
Platform,
PlatformConfig,
)
from plugins.platforms.homeassistant.adapter import (
HomeAssistantAdapter,
check_ha_requirements,
validate_ha_config,
)
# ---------------------------------------------------------------------------
# check_ha_requirements
# ---------------------------------------------------------------------------
class TestCheckRequirements:
@patch("plugins.platforms.homeassistant.adapter.AIOHTTP_AVAILABLE", False)
def test_returns_false_without_aiohttp(self, monkeypatch):
monkeypatch.setenv("HASS_TOKEN", "test-token")
assert check_ha_requirements() is False
def test_validate_config_accepts_platform_token(self, monkeypatch):
monkeypatch.delenv("HASS_TOKEN", raising=False)
config = PlatformConfig(enabled=True, token="config-token")
assert validate_ha_config(config) is True
class TestValidateConfig:
def test_returns_false_without_token_in_config_or_env(self, monkeypatch):
monkeypatch.delenv("HASS_TOKEN", raising=False)
assert validate_ha_config(PlatformConfig(enabled=True)) is False
# ---------------------------------------------------------------------------
# _format_state_change - pure function, all domain branches
# ---------------------------------------------------------------------------
class TestFormatStateChange:
@staticmethod
def fmt(entity_id, old_state, new_state):
return HomeAssistantAdapter._format_state_change(entity_id, old_state, new_state)
def test_climate_includes_temperatures(self):
msg = self.fmt(
"climate.thermostat",
{"state": "off"},
{"state": "heat", "attributes": {
"friendly_name": "Main Thermostat",
"current_temperature": 21.5,
"temperature": 23,
}},
)
assert "Main Thermostat" in msg
assert "'off'" in msg and "'heat'" in msg
assert "21.5" in msg and "23" in msg
def test_sensor_includes_unit(self):
msg = self.fmt(
"sensor.temperature",
{"state": "22.5"},
{"state": "25.1", "attributes": {
"friendly_name": "Living Room Temp",
"unit_of_measurement": "C",
}},
)
assert "22.5C" in msg and "25.1C" in msg
assert "Living Room Temp" in msg
# ---------------------------------------------------------------------------
# Adapter initialization from config
# ---------------------------------------------------------------------------
class TestAdapterInit:
def test_url_and_token_from_config_extra(self, monkeypatch):
monkeypatch.delenv("HASS_URL", raising=False)
monkeypatch.delenv("HASS_TOKEN", raising=False)
config = PlatformConfig(
enabled=True,
token="config-token",
extra={"url": "http://192.168.1.50:8123"},
)
adapter = HomeAssistantAdapter(config)
assert adapter._hass_token == "config-token"
assert adapter._hass_url == "http://192.168.1.50:8123"
def test_watch_filters_parsed(self):
config = PlatformConfig(
enabled=True, token="***",
extra={
"watch_domains": ["climate", "binary_sensor"],
"watch_entities": ["sensor.special"],
"ignore_entities": ["sensor.uptime", "sensor.cpu"],
"cooldown_seconds": 120,
},
)
adapter = HomeAssistantAdapter(config)
assert adapter._watch_domains == {"climate", "binary_sensor"}
assert adapter._watch_entities == {"sensor.special"}
assert adapter._ignore_entities == {"sensor.uptime", "sensor.cpu"}
assert adapter._watch_all is False
assert adapter._cooldown_seconds == 120
# ---------------------------------------------------------------------------
# Event filtering pipeline (_handle_ha_event)
#
# We mock handle_message (not our code, it's the base class pipeline) to
# capture the MessageEvent that _handle_ha_event produces.
# ---------------------------------------------------------------------------
def _make_adapter(**extra) -> HomeAssistantAdapter:
config = PlatformConfig(enabled=True, token="tok", extra=extra)
adapter = HomeAssistantAdapter(config)
adapter.handle_message = AsyncMock()
return adapter
def _make_event(entity_id, old_state, new_state, old_attrs=None, new_attrs=None):
return {
"data": {
"entity_id": entity_id,
"old_state": {"state": old_state, "attributes": old_attrs or {}},
"new_state": {"state": new_state, "attributes": new_attrs or {"friendly_name": entity_id}},
}
}
class TestEventFilteringPipeline:
@pytest.mark.asyncio
async def test_ignored_entity_not_forwarded(self):
adapter = _make_adapter(watch_all=True, ignore_entities=["sensor.uptime"])
await adapter._handle_ha_event(_make_event("sensor.uptime", "100", "101"))
adapter.handle_message.assert_not_called()
@pytest.mark.asyncio
async def test_unwatched_domain_not_forwarded(self):
adapter = _make_adapter(watch_domains=["climate"])
await adapter._handle_ha_event(_make_event("light.bedroom", "off", "on"))
adapter.handle_message.assert_not_called()
@pytest.mark.asyncio
async def test_watched_domain_forwarded(self):
adapter = _make_adapter(watch_domains=["climate"], cooldown_seconds=0)
await adapter._handle_ha_event(
_make_event("climate.thermostat", "off", "heat",
new_attrs={"friendly_name": "Thermostat", "current_temperature": 20, "temperature": 22})
)
adapter.handle_message.assert_called_once()
# Verify the actual MessageEvent text content
msg_event = adapter.handle_message.call_args[0][0]
assert "Thermostat" in msg_event.text
assert "heat" in msg_event.text
assert msg_event.source.platform == Platform.HOMEASSISTANT
assert msg_event.source.chat_id == "ha_events"
# ---------------------------------------------------------------------------
# Cooldown behavior
# ---------------------------------------------------------------------------
class TestCooldown:
@pytest.mark.asyncio
async def test_cooldown_expires(self):
adapter = _make_adapter(watch_all=True, cooldown_seconds=1)
event = _make_event("sensor.temp", "20", "21",
new_attrs={"friendly_name": "Temp"})
await adapter._handle_ha_event(event)
assert adapter.handle_message.call_count == 1
# Simulate time passing beyond cooldown
adapter._last_event_time["sensor.temp"] = time.time() - 2
event2 = _make_event("sensor.temp", "21", "22",
new_attrs={"friendly_name": "Temp"})
await adapter._handle_ha_event(event2)
assert adapter.handle_message.call_count == 2
# ---------------------------------------------------------------------------
# Config integration (env overrides, round-trip)
# ---------------------------------------------------------------------------
class TestConfigIntegration:
def test_env_override_creates_ha_platform(self, monkeypatch):
monkeypatch.setenv("HASS_TOKEN", "env-token")
monkeypatch.setenv("HASS_URL", "http://10.0.0.5:8123")
# Clear other platform tokens
for v in ["TELEGRAM_BOT_TOKEN", "DISCORD_BOT_TOKEN", "SLACK_BOT_TOKEN"]:
monkeypatch.delenv(v, raising=False)
from gateway.config import load_gateway_config
config = load_gateway_config()
assert Platform.HOMEASSISTANT in config.platforms
ha = config.platforms[Platform.HOMEASSISTANT]
assert ha.enabled is True
assert ha.token == "env-token"
assert ha.extra["url"] == "http://10.0.0.5:8123"
# ---------------------------------------------------------------------------
# send() via REST API
# ---------------------------------------------------------------------------
class TestSendViaRestApi:
"""send() uses REST API (not WebSocket) to avoid race conditions."""
@staticmethod
def _mock_aiohttp_session(response_status=200, response_text="OK"):
"""Build a mock aiohttp session + response for async-with patterns.
aiohttp.ClientSession() is a sync constructor whose return value
is used as ``async with session:``. ``session.post(...)`` returns a
context-manager (not a coroutine), so both layers use MagicMock for
the call and AsyncMock only for ``__aenter__`` / ``__aexit__``.
"""
mock_response = MagicMock()
mock_response.status = response_status
mock_response.text = AsyncMock(return_value=response_text)
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
mock_response.__aexit__ = AsyncMock(return_value=False)
mock_session = MagicMock()
mock_session.post = MagicMock(return_value=mock_response)
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock(return_value=False)
return mock_session
@pytest.mark.asyncio
async def test_send_success(self):
adapter = _make_adapter()
mock_session = self._mock_aiohttp_session(200)
with patch("plugins.platforms.homeassistant.adapter.aiohttp") as mock_aiohttp:
mock_aiohttp.ClientSession = MagicMock(return_value=mock_session)
mock_aiohttp.ClientTimeout = lambda total: total
result = await adapter.send("ha_events", "Test notification")
assert result.success is True
# Verify the REST API was called with correct payload
call_args = mock_session.post.call_args
assert "/api/services/persistent_notification/create" in call_args[0][0]
assert call_args[1]["json"]["title"] == t("platform.homeassistant.notification_title")
assert call_args[1]["json"]["message"] == "Test notification"
assert "Bearer tok" in call_args[1]["headers"]["Authorization"]
# ---------------------------------------------------------------------------
# Toolset integration
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# WebSocket URL construction
# ---------------------------------------------------------------------------
class TestLocalNetworkConnectHint:
def test_ehostunreach_outside_launchd_is_a_plain_unreachable_host(self, monkeypatch):
"""The same errno from a Terminal-run gateway (or another OS) must not be blamed on macOS."""
from plugins.platforms.homeassistant.adapter import _connect_error_detail
monkeypatch.delenv("HERMES_SUPERVISED_CHILD", raising=False)
monkeypatch.delenv("XPC_SERVICE_NAME", raising=False)
err = OSError(errno.EHOSTUNREACH, "No route to host")
assert _connect_error_detail(err) == str(err)
assert _connect_error_detail(RuntimeError("auth failed")) == "auth failed"
@pytest.mark.platforms("macos")
def test_ehostunreach_under_launchd_names_the_remedy(self, monkeypatch):
"""Only the launchd-supervised gateway can be denied by Local Network Privacy (#71206)."""
from plugins.platforms.homeassistant.adapter import _connect_error_detail
monkeypatch.setenv("HERMES_SUPERVISED_CHILD", "1")
err = OSError(errno.EHOSTUNREACH, "No route to host")
detail = _connect_error_detail(err)
assert detail.startswith(str(err))
assert len(detail) > len(str(err)) # a remedy hint is appended
@@ -20,7 +20,6 @@ from tests.gateway._plugin_adapter_loader import load_plugin_adapter
DEFAULT_ENV = {
"MATRIX_HOMESERVER": "https://default.matrix.example", "MATRIX_USER_ID": "@default:example",
"MATRIX_DEVICE_ID": "DEFAULTDEV", "MATRIX_ACCESS_TOKEN": "default-matrix-token",
"HASS_URL": "http://default-ha.example:8123", "HASS_TOKEN": "default-ha-token",
"TEAMS_CLIENT_ID": "default-client-id", "TEAMS_TENANT_ID": "default-tenant",
"TEAMS_CLIENT_SECRET": "default-teams-secret", "TEAMS_HOME_CHANNEL": "default-conv",
"DINGTALK_CLIENT_ID": "default-ding-id", "DINGTALK_CLIENT_SECRET": "default-ding-secret",
@@ -30,7 +29,6 @@ DEFAULT_ENV = {
SECONDARY = {
"MATRIX_HOMESERVER": "https://bot2.matrix.example", "MATRIX_USER_ID": "@bot2:example",
"MATRIX_DEVICE_ID": "BOT2DEV", "MATRIX_ACCESS_TOKEN": "bot2-matrix-token",
"HASS_URL": "http://bot2-ha.example:8123", "HASS_TOKEN": "bot2-ha-token",
"TEAMS_CLIENT_ID": "bot2-client-id", "TEAMS_TENANT_ID": "bot2-tenant",
"TEAMS_CLIENT_SECRET": "bot2-teams-secret", "TEAMS_HOME_CHANNEL": "bot2-conv",
"DINGTALK_CLIENT_ID": "bot2-ding-id", "DINGTALK_CLIENT_SECRET": "bot2-ding-secret",
@@ -86,37 +84,6 @@ async def test_matrix_standalone_send_posts_scoped_token_to_scoped_homeserver(se
assert DEFAULT_ENV["MATRIX_HOMESERVER"] not in seen["url"]
def test_homeassistant_url_follows_the_scoped_token(secondary_scope):
from plugins.platforms.homeassistant.adapter import HomeAssistantAdapter
adapter = HomeAssistantAdapter(PlatformConfig(enabled=True))
assert (adapter._hass_url, adapter._hass_token) == (SECONDARY["HASS_URL"], SECONDARY["HASS_TOKEN"])
@pytest.mark.asyncio
async def test_homeassistant_standalone_send_targets_scoped_url(secondary_scope, monkeypatch):
import aiohttp
from plugins.platforms.homeassistant import adapter as ha
seen = {}
class _Resp:
status = 200
async def __aenter__(self): return self
async def __aexit__(self, *a): return False
async def text(self): return ""
class _Sess:
def __init__(self, *a, **k): pass
async def __aenter__(self): return self
async def __aexit__(self, *a): return False
def post(self, url, **kw): seen["url"] = url; return _Resp()
monkeypatch.setattr(aiohttp, "ClientSession", _Sess)
await ha._standalone_send(PlatformConfig(enabled=True), "x", "hi")
assert seen["url"].startswith(SECONDARY["HASS_URL"] + "/")
def test_teams_credentials_pair_scoped_secret_with_scoped_app_identity(secondary_scope):
teams = load_plugin_adapter("teams")
assert teams._credentials(PlatformConfig(enabled=True)) == (
@@ -27,7 +27,6 @@ def test_all_builtins_have_checker_or_generic_token_path():
Platform.SLACK,
Platform.MATRIX,
Platform.MATTERMOST,
Platform.HOMEASSISTANT,
}}
# Platforms with a bespoke checker
@@ -0,0 +1,71 @@
"""PlatformEntry seams a plugin platform uses instead of a core special case.
``trusted_inbound`` (no human sender: allowlists/pairing do not apply) and ``display_tier`` (built-in
display defaults) were Home Assistant branches in core; the homeassistant catalog plugin is their
consumer. Driven through the real registry, authz and display resolver.
"""
from __future__ import annotations
import pytest
from gateway.config import Platform
from gateway.display_config import resolve_display_setting
from gateway.platform_registry import PlatformEntry, platform_registry
from gateway.session import SessionSource
@pytest.fixture
def plugin_platform():
names = []
def _register(name: str, **fields):
platform_registry.register(PlatformEntry(
name=name, label=name.title(), adapter_factory=lambda cfg: None, check_fn=lambda: True,
source="plugin", **fields))
names.append(name)
return Platform(name)
yield _register
for name in names:
platform_registry.unregister(name)
def _runner():
from gateway.run import GatewayRunner
runner = object.__new__(GatewayRunner)
runner.pairing_store = None
return runner
def _event_source(platform):
return SessionSource(platform=platform, user_id="event-bus", chat_id="events", user_name="bus", chat_type="dm")
def test_trusted_inbound_platform_bypasses_allowlists(plugin_platform, monkeypatch):
for key in ("GATEWAY_ALLOWED_USERS", "GATEWAY_ALLOW_ALL_USERS"):
monkeypatch.delenv(key, raising=False)
trusted = plugin_platform("seamtrusted", trusted_inbound=True)
plain = plugin_platform("seamplain")
runner = _runner()
assert runner._is_user_authorized(_event_source(trusted)) is True
# Control: the default stays default-deny for an unknown sender.
assert runner._is_user_authorized(_event_source(plain)) is False
def test_display_tier_supplies_builtin_defaults_below_user_overrides(plugin_platform):
plugin_platform("seamminimal", display_tier="minimal")
plugin_platform("seamnotier")
assert resolve_display_setting({}, "seamminimal", "tool_progress") == "off"
assert resolve_display_setting({}, "seamminimal", "tool_preview_length") == 0
assert resolve_display_setting({}, "seamminimal", "streaming") is False
# The user's per-platform override still wins over the registered tier.
user = {"display": {"platforms": {"seamminimal": {"tool_progress": "all"}}}}
assert resolve_display_setting(user, "seamminimal", "tool_progress") == "all"
# No tier: the global defaults, as before.
assert resolve_display_setting({}, "seamnotier", "tool_progress") == resolve_display_setting({}, "nosuchplatform", "tool_progress")
def test_platform_entry_seam_defaults_are_inert():
entry = PlatformEntry(name="x", label="X", adapter_factory=lambda cfg: None, check_fn=lambda: True)
assert (entry.trusted_inbound, entry.display_tier, entry.shared_env_prefixes) == (False, "", ())
@@ -188,7 +188,7 @@ class TestA2AClientToolsInCliProcess:
def test_a2a_appears_in_the_hermes_tools_checklist(self):
"""`a2a` is in _DEFAULT_OFF_TOOLSETS, so it must be tickable.
Every other member of that set (homeassistant, spotify, video_gen,
Every other member of that set (spotify, video_gen,
x_search, ...) renders a checkbox; a2a rendered nothing, so the
documented opt-in path had nothing to tick.
"""
@@ -0,0 +1,165 @@
"""``hermes_cli.left_core_migration``: homes that used a feature that left core get its catalog plugin.
Real config/.env files under a temp home and the in-tree catalog; only the network install is a
recording stand-in (the live install is exercised end to end outside the unit suite).
"""
from __future__ import annotations
from pathlib import Path
import pytest
import hermes_cli.left_core_migration as lcm
@pytest.fixture(autouse=True)
def _fresh(monkeypatch):
monkeypatch.setattr(lcm, "_attempted", set())
monkeypatch.setattr(lcm, "_undelivered", {})
monkeypatch.delenv("HASS_TOKEN", raising=False)
# The in-tree catalog (this checkout's plugin-catalog/), never the network.
import hermes_cli.plugin_catalog as pc
monkeypatch.setattr(pc, "fetch_live_catalog", lambda **_: None)
def _home(tmp_path: Path, name: str = "home", *, env: str = "", config: str = "") -> Path:
home = tmp_path / name
home.mkdir(parents=True)
if env:
(home / ".env").write_text(env, encoding="utf-8")
if config:
(home / "config.yaml").write_text(config, encoding="utf-8")
return home
@pytest.mark.parametrize(("env", "config"), [
("HASS_TOKEN=abc\n", ""),
("", "platforms:\n homeassistant:\n enabled: true\n"),
("", "gateway:\n platforms:\n homeassistant:\n enabled: true\n"),
("", "platforms:\n homeassistant:\n token: inline-token\n"),
("", "platform_toolsets:\n cli: [hermes-cli, homeassistant]\n"),
("", "platform_toolsets:\n homeassistant: [hermes-homeassistant]\n"),
])
def test_homeassistant_in_use_matches_what_core_activated(tmp_path, env, config):
assert lcm.homeassistant_in_use(_home(tmp_path, env=env, config=config)) is True
@pytest.mark.parametrize(("env", "config"), [
("", ""),
("HASS_TOKEN=\n", ""),
("", "platforms:\n homeassistant:\n enabled: false\n token: t\n"),
("", "platform_toolsets:\n cli: [hermes-cli]\n"),
("OPENAI_API_KEY=sk\n", "platforms:\n telegram:\n enabled: true\n"),
])
def test_homeassistant_not_in_use(tmp_path, env, config):
assert lcm.homeassistant_in_use(_home(tmp_path, env=env, config=config)) is False
def test_process_env_token_counts_only_for_the_active_home(tmp_path, monkeypatch):
home = _home(tmp_path)
monkeypatch.setenv("HASS_TOKEN", "from-systemd")
assert lcm.homeassistant_in_use(home) is False
assert lcm.homeassistant_in_use(home, process_env=True) is True
def test_migrate_home_installs_the_catalog_plugin_once(tmp_path):
home = _home(tmp_path, env="HASS_TOKEN=abc\n")
calls, said = [], []
def install(name):
calls.append(name)
(home / "plugins" / name).mkdir(parents=True)
return {"ok": True}
assert lcm.migrate_home(home, install=install, say=said.append) == ["homeassistant"]
assert calls == ["homeassistant"]
assert "✓ Home Assistant moved out of core" in said[0]
# Installed now: a second pass is a no-op and silent.
said.clear()
assert lcm.migrate_home(home, install=install, say=said.append) == []
assert calls == ["homeassistant"] and said == []
def test_migrate_home_reports_a_failed_install_with_the_command(tmp_path):
home = _home(tmp_path, env="HASS_TOKEN=abc\n")
said = []
assert lcm.migrate_home(home, install=lambda n: {"ok": False, "error": "network down"}, say=said.append) == []
assert "could not be installed automatically: network down" in said[0]
assert "plugins install homeassistant" in said[0]
def test_home_without_homeassistant_gets_nothing_and_no_notice(tmp_path):
home = _home(tmp_path, env="OPENAI_API_KEY=sk\n")
said = []
assert lcm.migrate_home(home, install=lambda n: pytest.fail("installed"), say=said.append) == []
assert said == []
def test_a_disabled_or_present_plugin_is_left_alone(tmp_path):
home = _home(tmp_path, env="HASS_TOKEN=abc\n")
(home / "plugins" / "homeassistant").mkdir(parents=True)
assert lcm.migrate_home(home, install=lambda n: pytest.fail("reinstalled"), say=print) == []
def test_catalog_miss_is_reported_not_installed(tmp_path, monkeypatch):
import hermes_cli.memory_provider_migration as mpm
monkeypatch.setattr(mpm, "catalog_source", lambda name: None)
home = _home(tmp_path, env="HASS_TOKEN=abc\n")
said = []
assert lcm.migrate_home(home, install=lambda n: pytest.fail("installed"), say=said.append) == []
assert "cannot find in the plugin catalog" in said[0]
def test_migrate_all_homes_migrates_only_the_profile_that_used_it(tmp_path, monkeypatch):
import pm.plugins_state as ps
a = _home(tmp_path, "a", env="HASS_TOKEN=abc\n")
b = _home(tmp_path, "b", env="OPENAI_API_KEY=sk\n")
monkeypatch.setattr(ps, "dependency_homes", lambda: [a, b])
installed_into = []
def install_into(home):
def _install(name):
installed_into.append((home, name))
return {"ok": True}
return _install
monkeypatch.setattr(lcm, "_install_into", install_into)
said = []
assert lcm.migrate_all_homes(say=said.append) == ["homeassistant"]
assert installed_into == [(a, "homeassistant")]
assert len(said) == 1 and str(a) in said[0]
def test_startup_honours_lazy_install_opt_out(tmp_path, monkeypatch):
home = _home(tmp_path, env="HASS_TOKEN=abc\n")
monkeypatch.setenv("HERMES_HOME", str(home))
import pm.install
monkeypatch.setattr(pm.install, "lazy_installs_allowed", lambda: False)
monkeypatch.setattr(lcm, "_install_into", lambda h: pytest.fail("installed"))
said = []
assert lcm.recover_at_startup(say=said.append) == []
assert "allow_lazy_installs is off" in said[0] and "plugins install homeassistant" in said[0]
# Once per process per home.
assert lcm.recover_at_startup(say=said.append) == [] and len(said) == 1
def test_gateway_start_outcome_waits_for_the_first_agent(tmp_path, monkeypatch):
home = _home(tmp_path, env="HASS_TOKEN=abc\n")
monkeypatch.setenv("HERMES_HOME", str(home))
import pm.install
monkeypatch.setattr(pm.install, "lazy_installs_allowed", lambda: True)
monkeypatch.setattr(lcm, "_install_into", lambda h: (lambda name: {"ok": True}))
assert lcm.recover_at_startup() == ["homeassistant"] # gateway start: nobody to tell yet
said = []
assert lcm.recover_at_startup(say=said.append) == [] # first agent: delivered, not re-attempted
assert len(said) == 1 and said[0].startswith("✓ Home Assistant moved out of core")
assert lcm.recover_at_startup(say=said.append) == [] and len(said) == 1
def test_in_tree_catalog_ships_the_homeassistant_entry():
from hermes_cli.plugin_catalog import get_catalog_entry
entry = get_catalog_entry("homeassistant")
assert entry is not None and entry.category == "platform" and entry.tier == "official"
assert set(entry.capabilities.provides_tools) == {
"ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service"}
@@ -202,29 +202,57 @@ _SHARED_ENV = (
"GATEWAY_ALLOW_ALL_USERS=true\nGATEWAY_ALLOWED_USERS=1,2\n"
"GATEWAY_RELAY_ID=gw-1\nGATEWAY_RELAY_SECRET=s\nGATEWAY_RELAY_DELIVERY_KEY=k\n"
"WECOM_DM_POLICY=open\nSMS_WEBHOOK_PORT=8700\n"
"HASS_TOKEN=hass-tool-token\nHASS_URL=http://ha.local\n"
"TWILIO_ACCOUNT_SID=AC1\nTWILIO_AUTH_TOKEN=tw\nTWILIO_PHONE_NUMBER=+1\n"
"EMAIL_ADDRESS=a@b\nEMAIL_PASSWORD=p\nEMAIL_SMTP_HOST=smtp\nEMAIL_IMAP_HOST=imap\nEMAIL_ALLOWED_USERS=x@y\n"
)
def test_ownership_inventory_strips_policy_relay_and_aliases_but_keeps_tool_credentials(home):
"""Gateway-wide policy, relay identity and alias-prefixed keys are channel settings. HASS/TWILIO/EMAIL
"""Gateway-wide policy, relay identity and alias-prefixed keys are channel settings. TWILIO/EMAIL
credentials are shared with tools: they leave with the channel only when the source's gateway would
run that adapter (email here — complete creds, not disabled); an explicitly disabled channel means
the key is a tool credential and stays. Its policy keys (allowlists, ports) go regardless."""
(home / ".env").write_text(_SHARED_ENV, encoding="utf-8")
(home / "config.yaml").write_text(
yaml.safe_dump({"model": {"default": "gpt-5", "provider": "openai"},
"platforms": {"homeassistant": {"enabled": False}, "sms": {"enabled": False}}}),
"platforms": {"sms": {"enabled": False}}}),
encoding="utf-8")
profile_dir = create_profile("bot3", clone_config=True, no_alias=True)
keys = {line.split("=", 1)[0] for line in (profile_dir / ".env").read_text(encoding="utf-8").splitlines()
if "=" in line and not line.startswith("#")}
assert keys == {"OPENAI_API_KEY", "HASS_TOKEN", "HASS_URL",
"TWILIO_ACCOUNT_SID", "TWILIO_AUTH_TOKEN", "TWILIO_PHONE_NUMBER"}
assert keys == {"OPENAI_API_KEY", "TWILIO_ACCOUNT_SID", "TWILIO_AUTH_TOKEN", "TWILIO_PHONE_NUMBER"}
_SHARED_PLUGIN_MANIFEST = "name: sharedplat\nkind: platform\nversion: 1.0.0\nrequires_env:\n - name: SHP_TOKEN\n password: true\n"
_SHARED_PLUGIN_INIT = (
"from gateway.platform_registry import PlatformEntry\n"
"def register(ctx):\n"
" ctx.register_platform(name='sharedplat', label='Shared', adapter_factory=lambda cfg: None,\n"
" check_fn=lambda: True, required_env=['SHP_TOKEN'], shared_env_prefixes=('SHP_',))\n"
)
@pytest.mark.parametrize(("enabled", "stripped"), [(False, False), (True, True)])
def test_plugin_platform_shared_env_prefixes_follow_the_ownership_rule(home, enabled, stripped):
"""A plugin platform's ``shared_env_prefixes`` (Home Assistant's ``HASS_``) get the shared-with-tools
rule: the keys leave with the channel only when the source runs that adapter."""
plugin = home / "plugins" / "sharedplat"
plugin.mkdir(parents=True)
(plugin / "plugin.yaml").write_text(_SHARED_PLUGIN_MANIFEST, encoding="utf-8")
(plugin / "__init__.py").write_text(_SHARED_PLUGIN_INIT, encoding="utf-8")
(home / ".env").write_text("OPENAI_API_KEY=sk\nSHP_TOKEN=tool-token\nSHP_URL=http://x\n", encoding="utf-8")
(home / "config.yaml").write_text(
yaml.safe_dump({"model": {"default": "gpt-5", "provider": "openai"}, "plugins": {"enabled": ["sharedplat"]},
"platforms": {"sharedplat": {"enabled": enabled}}}),
encoding="utf-8")
profile_dir = create_profile(f"shp{int(enabled)}", clone_config=True, no_alias=True)
keys = {line.split("=", 1)[0] for line in (profile_dir / ".env").read_text(encoding="utf-8").splitlines()
if "=" in line and not line.startswith("#")}
assert keys == ({"OPENAI_API_KEY"} if stripped else {"OPENAI_API_KEY", "SHP_TOKEN", "SHP_URL"})
def test_clone_all_drops_directory_shaped_channel_state(home):
-35
View File
@@ -129,41 +129,6 @@ def test_malformed_list_string_platform_toolsets_warns_then_falls_back(caplog):
def test_get_platform_tools_homeassistant_toolset_enabled_for_cron_when_hass_token_set(monkeypatch):
"""HA toolset is runtime-gated by check_fn (requires HASS_TOKEN).
When HASS_TOKEN is set, the user has explicitly opted in — _DEFAULT_OFF_TOOLSETS
shouldn't also strip HA from platforms (like cron) that run through
_get_platform_tools without an explicit saved toolset list.
Regression guard for Norbert's HA cron breakage after #14798 made cron
honor per-platform tool config.
"""
monkeypatch.setenv("HASS_TOKEN", "fake-test-token")
cron_enabled = _get_platform_tools({}, "cron")
assert "homeassistant" in cron_enabled
# moa must stay off — the original goal of #14798
assert "moa" not in cron_enabled
cli_enabled = _get_platform_tools({}, "cli")
assert "homeassistant" in cli_enabled
def test_get_platform_tools_homeassistant_uses_active_profile_token(monkeypatch):
from agent import secret_scope
monkeypatch.delenv("HASS_TOKEN", raising=False)
secret_scope.set_multiplex_active(True)
token = secret_scope.set_secret_scope({"HASS_TOKEN": "profile-token"})
try:
assert "homeassistant" in _get_platform_tools({}, "cron")
assert "homeassistant" in _get_platform_tools({}, "cli")
finally:
secret_scope.reset_secret_scope(token)
secret_scope.set_multiplex_active(False)
# ─── #35527: platform-restricted default-off toolsets (discord/discord_admin)
# are stripped by _DEFAULT_OFF_TOOLSETS even when the user explicitly opts in
# via the platform's native composite. The composite ``hermes-discord``
-344
View File
@@ -1,344 +0,0 @@
"""Tests for the Home Assistant tool module.
Tests real logic: entity filtering, payload building, response parsing,
handler validation, and availability gating.
"""
import json
from unittest.mock import AsyncMock, patch
import pytest
from tools.homeassistant_tool import (
_check_ha_available,
_filter_and_summarize,
_build_service_payload,
_parse_service_response,
_get_headers,
_handle_get_state,
_handle_call_service,
_BLOCKED_DOMAINS,
_ENTITY_ID_RE,
_SERVICE_NAME_RE,
)
# ---------------------------------------------------------------------------
# Sample HA state data (matches real HA /api/states response shape)
# ---------------------------------------------------------------------------
SAMPLE_STATES = [
{"entity_id": "light.bedroom", "state": "on", "attributes": {"friendly_name": "Bedroom Light", "brightness": 200}},
{"entity_id": "light.kitchen", "state": "off", "attributes": {"friendly_name": "Kitchen Light"}},
{"entity_id": "switch.fan", "state": "on", "attributes": {"friendly_name": "Living Room Fan"}},
{"entity_id": "sensor.temperature", "state": "22.5", "attributes": {"friendly_name": "Kitchen Temperature", "unit_of_measurement": "C"}},
{"entity_id": "climate.thermostat", "state": "heat", "attributes": {"friendly_name": "Main Thermostat", "current_temperature": 21}},
{"entity_id": "binary_sensor.motion", "state": "off", "attributes": {"friendly_name": "Hallway Motion"}},
{"entity_id": "sensor.humidity", "state": "55", "attributes": {"friendly_name": "Bedroom Humidity", "area": "bedroom"}},
]
# ---------------------------------------------------------------------------
# Entity filtering and summarization
# ---------------------------------------------------------------------------
class TestFilterAndSummarize:
def test_no_filters_returns_all(self):
result = _filter_and_summarize(SAMPLE_STATES)
assert result["count"] == 7
ids = {e["entity_id"] for e in result["entities"]}
assert "light.bedroom" in ids
assert "climate.thermostat" in ids
def test_domain_filter_lights(self):
result = _filter_and_summarize(SAMPLE_STATES, domain="light")
assert result["count"] == 2
for e in result["entities"]:
assert e["entity_id"].startswith("light.")
def test_missing_attributes_handled(self):
states = [{"entity_id": "light.x", "state": "on"}]
result = _filter_and_summarize(states)
assert result["count"] == 1
assert result["entities"][0]["friendly_name"] == ""
# ---------------------------------------------------------------------------
# Service payload building
# ---------------------------------------------------------------------------
class TestBuildServicePayload:
def test_entity_id_only(self):
payload = _build_service_payload(entity_id="light.bedroom")
assert payload == {"entity_id": "light.bedroom"}
def test_entity_id_param_takes_precedence_over_data(self):
payload = _build_service_payload(
entity_id="light.a",
data={"entity_id": "light.b"},
)
# explicit entity_id parameter wins over data["entity_id"]
assert payload["entity_id"] == "light.a"
# ---------------------------------------------------------------------------
# Service response parsing
# ---------------------------------------------------------------------------
class TestParseServiceResponse:
def test_list_response_extracts_entities(self):
ha_response = [
{"entity_id": "light.bedroom", "state": "on", "attributes": {}},
{"entity_id": "light.kitchen", "state": "on", "attributes": {}},
]
result = _parse_service_response("light", "turn_on", ha_response)
assert result["success"] is True
assert result["service"] == "light.turn_on"
assert len(result["affected_entities"]) == 2
assert result["affected_entities"][0]["entity_id"] == "light.bedroom"
# ---------------------------------------------------------------------------
# Handler validation (no mocks - these paths don't reach the network)
# ---------------------------------------------------------------------------
class TestHandlerValidation:
def test_get_state_missing_entity_id(self):
result = json.loads(_handle_get_state({}))
assert "error" in result
assert "entity_id" in result["error"]
def test_call_service_empty_strings(self):
result = json.loads(_handle_call_service({"domain": "", "service": ""}))
assert "error" in result
# ---------------------------------------------------------------------------
# Security: domain blocklist
# ---------------------------------------------------------------------------
class TestDomainBlocklist:
"""Verify dangerous HA service domains are blocked."""
@pytest.mark.parametrize("domain", sorted(_BLOCKED_DOMAINS))
def test_blocked_domain_rejected(self, domain):
result = json.loads(_handle_call_service({
"domain": domain, "service": "any_service"
}))
assert "error" in result
assert "blocked" in result["error"].lower()
@patch("tools.homeassistant_tool._async_call_service", new_callable=AsyncMock)
def test_safe_domain_not_blocked(self, mock_call_service):
"""Safe domains like ``light`` reach the service-call layer."""
mock_call_service.return_value = {"success": True}
result = json.loads(_handle_call_service({
"domain": "light", "service": "turn_on", "entity_id": "light.test"
}))
assert result["result"]["success"] is True
mock_call_service.assert_awaited_once_with(
"light",
"turn_on",
"light.test",
None,
)
def test_blocked_domains_include_shell_command(self):
assert "shell_command" in _BLOCKED_DOMAINS
def test_blocked_domains_include_hassio(self):
assert "hassio" in _BLOCKED_DOMAINS
# ---------------------------------------------------------------------------
# Security: entity_id validation
# ---------------------------------------------------------------------------
class TestEntityIdValidation:
"""Verify entity_id format validation prevents path traversal."""
def test_valid_entity_id_accepted(self):
assert _ENTITY_ID_RE.match("light.bedroom")
assert _ENTITY_ID_RE.match("sensor.temperature_1")
assert _ENTITY_ID_RE.match("binary_sensor.motion")
assert _ENTITY_ID_RE.match("climate.main_thermostat")
def test_path_traversal_rejected(self):
assert _ENTITY_ID_RE.match("../../config") is None
assert _ENTITY_ID_RE.match("light/../../../etc/passwd") is None
assert _ENTITY_ID_RE.match("../api/config") is None
@patch("tools.homeassistant_tool._async_call_service", new_callable=AsyncMock)
def test_call_service_allows_no_entity_id(self, mock_call_service):
"""Some services (like scene.turn_on) don't need entity_id."""
mock_call_service.return_value = {"success": True}
result = json.loads(_handle_call_service({
"domain": "scene", "service": "turn_on"
}))
assert result["result"]["success"] is True
mock_call_service.assert_awaited_once_with(
"scene",
"turn_on",
None,
None,
)
# ---------------------------------------------------------------------------
# String-data deserialization (XML tool calling workaround)
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Security: domain/service name format validation
# ---------------------------------------------------------------------------
class TestServiceNameValidation:
"""Verify domain/service format validation prevents path traversal in URL.
The domain and service parameters are interpolated into
/api/services/{domain}/{service}, so allowing arbitrary strings would
enable SSRF via path traversal or blocked-domain bypass.
"""
def test_valid_domain_names(self):
assert _SERVICE_NAME_RE.match("light")
assert _SERVICE_NAME_RE.match("switch")
assert _SERVICE_NAME_RE.match("climate")
assert _SERVICE_NAME_RE.match("shell_command")
assert _SERVICE_NAME_RE.match("media_player")
def test_path_traversal_in_domain_rejected(self):
assert _SERVICE_NAME_RE.match("../../api/config") is None
assert _SERVICE_NAME_RE.match("light/../../../etc") is None
assert _SERVICE_NAME_RE.match("../config") is None
def test_path_traversal_in_service_rejected(self):
assert _SERVICE_NAME_RE.match("../../api/config") is None
assert _SERVICE_NAME_RE.match("turn_on/../../config") is None
def test_blocked_domain_bypass_via_traversal_rejected(self):
"""Ensure shell_command/../light is rejected, not just checked against blocklist."""
assert _SERVICE_NAME_RE.match("shell_command/../light") is None
assert _SERVICE_NAME_RE.match("python_script/../scene") is None
assert _SERVICE_NAME_RE.match("hassio/../automation") is None
def test_special_chars_rejected(self):
assert _SERVICE_NAME_RE.match("light;rm") is None
assert _SERVICE_NAME_RE.match("light&cmd") is None
assert _SERVICE_NAME_RE.match("light cmd") is None
def test_handler_rejects_traversal_domain(self):
"""_handle_call_service must reject domain with path traversal."""
result = json.loads(_handle_call_service({
"domain": "../../api/config",
"service": "turn_on",
}))
assert "error" in result
assert "Invalid domain" in result["error"]
def test_handler_rejects_traversal_service(self):
"""_handle_call_service must reject service with path traversal."""
result = json.loads(_handle_call_service({
"domain": "light",
"service": "../../api/config",
}))
assert "error" in result
assert "Invalid service" in result["error"]
# ---------------------------------------------------------------------------
# Availability check
# ---------------------------------------------------------------------------
class TestCheckAvailable:
def test_unavailable_without_token(self, monkeypatch):
monkeypatch.delenv("HASS_TOKEN", raising=False)
assert _check_ha_available() is False
def test_empty_token_is_unavailable(self, monkeypatch):
monkeypatch.setenv("HASS_TOKEN", "")
assert _check_ha_available() is False
def test_multiplex_scope_does_not_fall_back_to_another_profile(self, monkeypatch):
from agent import secret_scope
monkeypatch.setenv("HASS_TOKEN", "default-profile-token")
secret_scope.set_multiplex_active(True)
token = secret_scope.set_secret_scope({})
try:
assert _check_ha_available() is False
finally:
secret_scope.reset_secret_scope(token)
secret_scope.set_multiplex_active(False)
def test_multiplex_scope_supplies_profile_url_and_token(self, monkeypatch):
from agent import secret_scope
from tools.homeassistant_tool import _get_config
monkeypatch.setenv("HASS_URL", "http://default-profile:8123")
monkeypatch.setenv("HASS_TOKEN", "default-profile-token")
secret_scope.set_multiplex_active(True)
token = secret_scope.set_secret_scope({
"HASS_URL": "http://secondary-profile:8123/",
"HASS_TOKEN": "secondary-profile-token",
})
try:
assert _get_config() == (
"http://secondary-profile:8123",
"secondary-profile-token",
)
finally:
secret_scope.reset_secret_scope(token)
secret_scope.set_multiplex_active(False)
# ---------------------------------------------------------------------------
# Auth headers
# ---------------------------------------------------------------------------
class TestGetHeaders:
def test_bearer_token_format(self, monkeypatch):
monkeypatch.setattr("tools.homeassistant_tool._get_config",
lambda: ("http://ha.local:8123", "my-secret-token"))
headers = _get_headers()
assert headers["Authorization"] == "Bearer my-secret-token"
assert headers["Content-Type"] == "application/json"
# ---------------------------------------------------------------------------
# Registry integration
# ---------------------------------------------------------------------------
class TestRegistration:
def test_check_fn_includes_when_token_set(self, monkeypatch):
"""Registry should include HA tools when HASS_TOKEN is set."""
from tools.registry import invalidate_check_fn_cache, registry
monkeypatch.setenv("HASS_TOKEN", "test-token")
invalidate_check_fn_cache()
defs = registry.get_definitions({"ha_list_entities", "ha_get_state", "ha_call_service"})
assert len(defs) == 3
+1 -3
View File
@@ -43,7 +43,7 @@ DISCORD_HOME_CHANNEL_NAME DISCORD_REQUIRE_MENTION DISCORD_FREE_RESPONSE_CHANNELS
DISCORD_AUTO_THREAD SLACK_HOME_CHANNEL SLACK_HOME_CHANNEL_NAME SLACK_ALLOWED_USERS
WHATSAPP_ENABLED WHATSAPP_MODE WHATSAPP_ALLOWED_USERS SIGNAL_HTTP_URL SIGNAL_ACCOUNT
SIGNAL_ALLOWED_USERS SIGNAL_GROUP_ALLOWED_USERS SIGNAL_HOME_CHANNEL SIGNAL_HOME_CHANNEL_NAME
SIGNAL_IGNORE_STORIES HASS_TOKEN HASS_URL EMAIL_ADDRESS EMAIL_PASSWORD EMAIL_IMAP_HOST
SIGNAL_IGNORE_STORIES EMAIL_ADDRESS EMAIL_PASSWORD EMAIL_IMAP_HOST
EMAIL_SMTP_HOST EMAIL_HOME_ADDRESS EMAIL_HOME_ADDRESS_NAME HERMES_DASHBOARD_SESSION_TOKEN
HERMES_DASHBOARD_BASIC_AUTH_PASSWORD HERMES_DASHBOARD_BASIC_AUTH_SECRET HERMES_DASHBOARD_DRAIN_SECRET
HERMES_DASHBOARD_OIDC_CLIENT_SECRET HERMES_ANON_API_SECRET HERMES_MEET_REALTIME_KEY
@@ -1477,8 +1477,6 @@ class TestBlocklistCoverage:
"SIGNAL_HOME_CHANNEL",
"SIGNAL_HOME_CHANNEL_NAME",
"SIGNAL_IGNORE_STORIES",
"HASS_TOKEN",
"HASS_URL",
"EMAIL_ADDRESS",
"EMAIL_PASSWORD",
"EMAIL_IMAP_HOST",
+11 -11
View File
@@ -38,30 +38,30 @@ class TestRegisterServerTools:
def test_colliding_static_toolset_name_merges_both_tool_sets(self, mock_registry):
"""An MCP server named after a built-in toolset must not be shadowed.
Regression: an MCP server registered as `homeassistant` (colliding
with the static `homeassistant` toolset) had its tools silently
Regression: an MCP server registered under a built-in toolset's name
(originally `browser`; `browser` here) had its tools silently
dropped because get_toolset() returned the static definition without
consulting the alias registered by _register_server_tools().
"""
from toolsets import TOOLSETS, get_toolset, resolve_toolset
assert "homeassistant" in TOOLSETS # collision premise
static_tools = set(TOOLSETS["homeassistant"]["tools"])
assert "browser" in TOOLSETS # collision premise
static_tools = set(TOOLSETS["browser"]["tools"])
server = MCPServerTask("homeassistant")
server._tools = [_make_mcp_tool("get_entities", "List HA entities")]
server = MCPServerTask("browser")
server._tools = [_make_mcp_tool("get_entities", "List entities")]
server.session = MagicMock()
with patch("tools.registry.registry", mock_registry):
registered = _register_server_tools("homeassistant", server, {})
assert "mcp__homeassistant__get_entities" in registered
registered = _register_server_tools("browser", server, {})
assert "mcp__browser__get_entities" in registered
ts = get_toolset("homeassistant")
ts = get_toolset("browser")
# Static built-ins are still present...
assert static_tools <= set(ts["tools"])
# ...and the MCP server's tools are no longer shadowed.
assert "mcp__homeassistant__get_entities" in ts["tools"]
assert "mcp__homeassistant__get_entities" in resolve_toolset("homeassistant")
assert "mcp__browser__get_entities" in ts["tools"]
assert "mcp__browser__get_entities" in resolve_toolset("browser")
class TestRefreshTools:
"""Tests for MCPServerTask._refresh_tools nuke-and-repave cycle."""
@@ -1,4 +1,4 @@
"""Tests for _send_mattermost, _send_matrix, _send_homeassistant, _send_dingtalk."""
"""Tests for _send_mattermost, _send_matrix, _send_dingtalk."""
import asyncio
import os
@@ -46,22 +46,6 @@ async def _send_mattermost(token, extra, chat_id, message):
return await _mattermost_standalone_send(pconfig, chat_id, message)
# ``_send_homeassistant`` moved into the homeassistant plugin
# (``plugins/platforms/homeassistant/adapter.py::_standalone_send``). Same
# shim pattern as ``_send_mattermost`` above.
from plugins.platforms.homeassistant.adapter import (
_standalone_send as _homeassistant_standalone_send,
)
async def _send_homeassistant(token, extra, chat_id, message):
"""Pre-migration ``(token, extra, chat_id, message)`` shim around the
plugin's ``_standalone_send(pconfig, chat_id, message)``.
"""
pconfig = SimpleNamespace(token=token, extra=extra or {})
return await _homeassistant_standalone_send(pconfig, chat_id, message)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@@ -191,42 +175,6 @@ class TestSendMatrix:
assert txn_ids[0] != txn_ids[1]
# ---------------------------------------------------------------------------
# _send_homeassistant
# ---------------------------------------------------------------------------
class TestSendHomeAssistant:
def test_success(self):
resp = _make_aiohttp_resp(200)
session_ctx, session = _make_aiohttp_session(resp)
with patch("aiohttp.ClientSession", return_value=session_ctx), \
patch.dict(os.environ, {"HASS_URL": "", "HASS_TOKEN": ""}, clear=False):
extra = {"url": "https://hass.example.com"}
result = asyncio.run(_send_homeassistant("hass-tok", extra, "mobile_app_phone", "alert!"))
assert result == {"success": True, "platform": "homeassistant", "chat_id": "mobile_app_phone"}
session.post.assert_called_once()
call_kwargs = session.post.call_args
assert call_kwargs[0][0] == "https://hass.example.com/api/services/notify/notify"
assert call_kwargs[1]["headers"]["Authorization"] == "Bearer hass-tok"
assert call_kwargs[1]["json"] == {"message": "alert!", "target": "mobile_app_phone"}
def test_env_var_fallback(self):
resp = _make_aiohttp_resp(200)
session_ctx, session = _make_aiohttp_session(resp)
with patch("aiohttp.ClientSession", return_value=session_ctx), \
patch.dict(os.environ, {"HASS_URL": "https://hass.env.com", "HASS_TOKEN": "env-tok"}, clear=False):
result = asyncio.run(_send_homeassistant("", {}, "notify_target", "hi"))
assert result["success"] is True
url = session.post.call_args[0][0]
assert "hass.env.com" in url
# ---------------------------------------------------------------------------
# _send_dingtalk
# ---------------------------------------------------------------------------
+1 -1
View File
@@ -67,7 +67,7 @@ Rules for tool code:
## Toolsets (`toolsets.py`)
Single `TOOLSETS` dict. Keys today: `browser, clarify, code_execution, cronjob, debugging,
delegation, discord, discord_admin, feishu_doc, feishu_drive, file, homeassistant, image_gen,
delegation, discord, discord_admin, feishu_doc, feishu_drive, file, image_gen,
kanban, memory, messaging, moa, rl, safe, search, session_search, skills, spotify, terminal, todo,
tts, video, vision, web, yuanbao` (don't assert the list in tests). Per-platform enable/disable via
`hermes tools` (curses) or `tools.<platform>.enabled/disabled` in config.yaml. `browser_exec`
+2 -2
View File
@@ -32,7 +32,7 @@ _STATIC_PROVIDER_ENV_BLOCKLIST = frozenset({
"SLACK_HOME_CHANNEL_NAME", "SLACK_ALLOWED_USERS", "WHATSAPP_ENABLED",
"WHATSAPP_MODE", "WHATSAPP_ALLOWED_USERS", "SIGNAL_HTTP_URL", "SIGNAL_ACCOUNT",
"SIGNAL_ALLOWED_USERS", "SIGNAL_GROUP_ALLOWED_USERS", "SIGNAL_HOME_CHANNEL",
"SIGNAL_HOME_CHANNEL_NAME", "SIGNAL_IGNORE_STORIES", "HASS_TOKEN", "HASS_URL",
"SIGNAL_HOME_CHANNEL_NAME", "SIGNAL_IGNORE_STORIES",
"EMAIL_ADDRESS", "EMAIL_PASSWORD", "EMAIL_IMAP_HOST", "EMAIL_SMTP_HOST",
"EMAIL_HOME_ADDRESS", "EMAIL_HOME_ADDRESS_NAME", "HERMES_DASHBOARD_SESSION_TOKEN",
"GATEWAY_ALLOWED_USERS", "GH_TOKEN", "GITHUB_APP_ID", "GITHUB_APP_PRIVATE_KEY_PATH",
@@ -374,7 +374,7 @@ _ALWAYS_STRIP_KEYS: frozenset[str] = frozenset({
# _is_hermes_internal_secret, but _ID has no secret suffix, so it must be
# enumerated here to stay stripped on the inherit_credentials=True path.
"GATEWAY_RELAY_ID", "GATEWAY_RELAY_SECRET", "GATEWAY_RELAY_DELIVERY_KEY",
"HASS_TOKEN", "EMAIL_PASSWORD", "HERMES_DASHBOARD_SESSION_TOKEN",
"EMAIL_PASSWORD", "HERMES_DASHBOARD_SESSION_TOKEN",
# Dashboard auth: the basic-auth password and session-signing secret, the OIDC client
# secret and the drain bearer. They let a holder mint or forge dashboard sessions, and no
# child (credentialed CLIs included) consumes them.
-337
View File
@@ -1,337 +0,0 @@
"""Home Assistant tool for controlling smart home devices via REST API.
Registers ``ha_list_entities``, ``ha_get_state``, ``ha_list_services``, ``ha_call_service``.
Auth is a Long-Lived Access Token (``HASS_TOKEN``); the instance URL comes from
``HASS_URL`` (default http://homeassistant.local:8123).
"""
import asyncio
import json
import logging
import re
from typing import Any, Dict, Optional
from agent.secret_scope import get_secret
from tools.registry import registry, tool_error
logger = logging.getLogger(__name__)
def _get_config():
"""Return the active profile's Home Assistant URL and token."""
return (
(get_secret("HASS_URL", "http://homeassistant.local:8123") or "").rstrip("/"),
get_secret("HASS_TOKEN", "") or "")
# Valid HA entity_id (e.g. "light.living_room", "sensor.temperature_1").
_ENTITY_ID_RE = re.compile(r"^[a-z_][a-z0-9_]*\.[a-z0-9_]+$")
# Domain/service names are interpolated into /api/services/{domain}/{service}, so only
# [a-z0-9_] is allowed: anything else enables SSRF via path traversal
# (domain="../../api/config") or blocklist bypass (domain="shell_command/../light").
_SERVICE_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*$")
# Domains that allow arbitrary code/command execution on the HA host or SSRF on the
# local network. HA has zero service-level access control; all safety lives here.
_BLOCKED_DOMAINS = frozenset({
"shell_command", # arbitrary shell commands as root in HA container
"command_line", # sensors/switches that execute shell commands
"python_script", # sandboxed but can escalate via hass.services.call()
"pyscript", # scripting integration with broader access
"hassio", # addon control, host shutdown/reboot, stdin to containers
"rest_command", # HTTP requests from HA server (SSRF vector)
})
def _get_headers(token: str = "") -> Dict[str, str]:
"""Return authorization headers for HA REST API."""
if not token:
_, token = _get_config()
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
async def _api_json(method: str, path: str, timeout: float, payload: Any = None) -> Any:
"""One HA REST call (GET or POST JSON) that raises on HTTP errors and returns the JSON body."""
import aiohttp
hass_url, hass_token = _get_config()
kwargs: Dict[str, Any] = {"headers": _get_headers(hass_token), "timeout": aiohttp.ClientTimeout(total=timeout)}
if method == "POST":
kwargs["json"] = payload
async with aiohttp.ClientSession() as session:
async with session.request(method, f"{hass_url}{path}", **kwargs) as resp:
resp.raise_for_status()
return await resp.json()
# ── async helpers (called from sync handlers via _run_async) ─────────────────
def _filter_and_summarize(states: list, domain: Optional[str] = None, area: Optional[str] = None) -> Dict:
"""Filter raw HA states by domain/area (area matches friendly_name or area attr) and compact them."""
if domain:
states = [s for s in states if s.get("entity_id", "").startswith(f"{domain}.")]
if area:
area_lower = area.lower()
states = [
s for s in states
if area_lower in (s.get("attributes", {}).get("friendly_name", "") or "").lower()
or area_lower in (s.get("attributes", {}).get("area", "") or "").lower()]
entities = [
{
"entity_id": s["entity_id"], "state": s["state"],
"friendly_name": s.get("attributes", {}).get("friendly_name", "")}
for s in states]
return {"count": len(entities), "entities": entities}
async def _async_list_entities(domain: Optional[str] = None, area: Optional[str] = None) -> Dict[str, Any]:
return _filter_and_summarize(await _api_json("GET", "/api/states", 15), domain, area)
async def _async_get_state(entity_id: str) -> Dict[str, Any]:
data = await _api_json("GET", f"/api/states/{entity_id}", 10)
return {
"entity_id": data["entity_id"], "state": data["state"], "attributes": data.get("attributes", {}),
"last_changed": data.get("last_changed"), "last_updated": data.get("last_updated")}
def _build_service_payload(entity_id: Optional[str] = None, data: Optional[Dict[str, Any]] = None) -> Dict:
"""JSON payload for a HA service call; ``entity_id`` overrides data["entity_id"]."""
payload: Dict[str, Any] = dict(data or {})
if entity_id:
payload["entity_id"] = entity_id
return payload
def _parse_service_response(domain: str, service: str, result: Any) -> Dict[str, Any]:
affected = []
if isinstance(result, list):
affected = [{"entity_id": s.get("entity_id", ""), "state": s.get("state", "")} for s in result]
return {"success": True, "service": f"{domain}.{service}", "affected_entities": affected}
async def _async_call_service(
domain: str, service: str, entity_id: Optional[str] = None, data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
result = await _api_json(
"POST", f"/api/services/{domain}/{service}", 15, _build_service_payload(entity_id, data))
return _parse_service_response(domain, service, result)
async def _async_list_services(domain: Optional[str] = None) -> Dict[str, Any]:
"""Available services, optionally filtered by domain, compacted for context."""
services = await _api_json("GET", "/api/services", 15)
if domain:
services = [s for s in services if s.get("domain") == domain]
result = []
for svc_domain in services:
domain_services = {}
for svc_name, svc_info in svc_domain.get("services", {}).items():
svc_entry: Dict[str, Any] = {"description": svc_info.get("description", "")}
fields = svc_info.get("fields", {})
if fields:
svc_entry["fields"] = {
k: v.get("description", "") for k, v in fields.items() if isinstance(v, dict)}
domain_services[svc_name] = svc_entry
result.append({"domain": svc_domain.get("domain", ""), "services": domain_services})
return {"count": len(result), "domains": result}
# ── sync wrappers (handler signature: (args, **kw) -> str) ───────────────────
def _run_async(coro):
"""Run a coroutine from a sync handler; hops to a thread if a loop is already running."""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = None
if loop and loop.is_running(): # already inside a loop: asyncio.run() needs its own thread
import concurrent.futures
import contextvars
# Carry the caller's Context: the coroutine reads HASS_* via get_secret, which is profile-scoped.
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(contextvars.copy_context().run, asyncio.run, coro).result(timeout=30)
return asyncio.run(coro)
def _dispatch(coro, log_name: str, fail_msg: str) -> str:
"""Run ``coro`` and wrap as ``{"result": ...}``; on error log and return tool_error."""
try:
return json.dumps({"result": _run_async(coro)})
except Exception as e:
logger.error("%s error: %s", log_name, e)
return tool_error(f"{fail_msg}: {e}")
def _handle_get_state(args: dict, **kw) -> str:
entity_id = args.get("entity_id", "")
if not entity_id:
return tool_error("Missing required parameter: entity_id")
if not _ENTITY_ID_RE.match(entity_id):
return tool_error(f"Invalid entity_id format: {entity_id}")
return _dispatch(_async_get_state(entity_id), "ha_get_state", f"Failed to get state for {entity_id}")
def _handle_call_service(args: dict, **kw) -> str:
domain = args.get("domain", "")
service = args.get("service", "")
if not domain or not service:
return tool_error("Missing required parameters: domain and service")
# Format check BEFORE the blocklist: rejects "shell_command/../light" style bypasses.
if not _SERVICE_NAME_RE.match(domain):
return tool_error(f"Invalid domain format: {domain!r}")
if not _SERVICE_NAME_RE.match(service):
return tool_error(f"Invalid service format: {service!r}")
if domain in _BLOCKED_DOMAINS:
return tool_error(
f"Service domain '{domain}' is blocked for security. "
f"Blocked domains: {', '.join(sorted(_BLOCKED_DOMAINS))}")
entity_id = args.get("entity_id")
if entity_id and not _ENTITY_ID_RE.match(entity_id):
return tool_error(f"Invalid entity_id format: {entity_id}")
data = args.get("data")
if isinstance(data, str): # XML tool-calling mode delivers data as a JSON string
try:
data = json.loads(data) if data.strip() else None
except json.JSONDecodeError as e:
return tool_error(f"Invalid JSON string in 'data' parameter: {e}")
return _dispatch(
_async_call_service(domain, service, entity_id, data),
"ha_call_service", f"Failed to call {domain}.{service}")
def _check_ha_available() -> bool:
"""Tool is only available when HASS_TOKEN is set."""
return bool(get_secret("HASS_TOKEN"))
# ── tool schemas ─────────────────────────────────────────────────────────────
HA_LIST_ENTITIES_SCHEMA = {
"name": "ha_list_entities",
"description": (
"List Home Assistant entities. Optionally filter by domain "
"(light, switch, climate, sensor, binary_sensor, cover, fan, etc.) "
"or by area name (living room, kitchen, bedroom, etc.)."
),
"parameters": {
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": (
"Entity domain to filter by (e.g. 'light', 'switch', 'climate', "
"'sensor', 'binary_sensor', 'cover', 'fan', 'media_player'). "
"Omit to list all entities."
),
},
"area": {
"type": "string",
"description": (
"Area/room name to filter by (e.g. 'living room', 'kitchen'). "
"Matches against entity friendly names. Omit to list all."
),
},
},
"required": [],
},
}
HA_GET_STATE_SCHEMA = {
"name": "ha_get_state",
"description": (
"Get the detailed state of a single Home Assistant entity, including all "
"attributes (brightness, color, temperature setpoint, sensor readings, etc.)."
),
"parameters": {
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": (
"The entity ID to query (e.g. 'light.living_room', "
"'climate.thermostat', 'sensor.temperature')."
),
},
},
"required": ["entity_id"],
},
}
HA_LIST_SERVICES_SCHEMA = {
"name": "ha_list_services",
"description": (
"List available Home Assistant services (actions) for device control. "
"Shows what actions can be performed on each device type and what "
"parameters they accept. Use this to discover how to control devices "
"found via ha_list_entities."
),
"parameters": {
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": (
"Filter by domain (e.g. 'light', 'climate', 'switch'). "
"Omit to list services for all domains."
),
},
},
"required": [],
},
}
HA_CALL_SERVICE_SCHEMA = {
"name": "ha_call_service",
"description": (
"Call a Home Assistant service to control a device. Use ha_list_services "
"to discover available services and their parameters for each domain."
),
"parameters": {
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": (
"Service domain (e.g. 'light', 'switch', 'climate', "
"'cover', 'media_player', 'fan', 'scene', 'script')."
),
},
"service": {
"type": "string",
"description": (
"Service name (e.g. 'turn_on', 'turn_off', 'toggle', "
"'set_temperature', 'set_hvac_mode', 'open_cover', "
"'close_cover', 'set_volume_level')."
),
},
"entity_id": {
"type": "string",
"description": (
"Target entity ID (e.g. 'light.living_room'). "
"Some services (like scene.turn_on) may not need this."
),
},
"data": {
"type": "string",
"description": (
"Additional service data as a JSON string. Examples: "
'{"brightness": 255, "color_name": "blue"} for lights, '
'{"temperature": 22, "hvac_mode": "heat"} for climate, '
'{"volume_level": 0.5} for media players.'
),
},
},
"required": ["domain", "service"],
},
}
for _schema, _handler in (
(HA_LIST_ENTITIES_SCHEMA, lambda args, **kw: _dispatch(
_async_list_entities(domain=args.get("domain"), area=args.get("area")),
"ha_list_entities", "Failed to list entities")),
(HA_GET_STATE_SCHEMA, _handle_get_state),
(HA_LIST_SERVICES_SCHEMA, lambda args, **kw: _dispatch(
_async_list_services(domain=args.get("domain")), "ha_list_services", "Failed to list services")),
(HA_CALL_SERVICE_SCHEMA, _handle_call_service)):
registry.register(
name=_schema["name"], toolset="homeassistant", schema=_schema, handler=_handler,
check_fn=_check_ha_available, emoji="🏠")
+4 -8
View File
@@ -27,7 +27,6 @@ _HERMES_CORE_TOOLS = [
"clarify",
"execute_code", "delegate_task",
"cronjob_manage",
"ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service",
"kanban_show", "kanban_list",
"kanban_complete", "kanban_block", "kanban_request_review",
"kanban_request_changes",
@@ -42,7 +41,6 @@ _HERMES_CORE_TOOLS = [
# Webhook payloads are untrusted third-party content: no file/system execution.
_HERMES_WEBHOOK_SAFE_TOOLS = ["web_search", "web_extract", "vision_analyze", "clarify"]
_HA_TOOLS = ["ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service"]
_FEISHU_TOOLS = [
"feishu_doc_read", "feishu_drive_list_comments", "feishu_drive_list_comment_replies",
"feishu_drive_reply_comment", "feishu_drive_add_comment",
@@ -66,8 +64,8 @@ def _core_without(*excluded, kanban=True):
# Coding posture: everything you reach for while pairing on code; drops messaging,
# tts, image_gen, home-assistant, cron, kanban and computer-use.
_CODING_TOOLS = _core_without("image_generate", "text_to_speech", "cronjob_manage", "computer_use", *_HA_TOOLS, kanban=False)
# tts, image_gen, cron, kanban and computer-use.
_CODING_TOOLS = _core_without("image_generate", "text_to_speech", "cronjob_manage", "computer_use", kanban=False)
# Toolsets a CLIENT adds to its own sessions (tui_gateway/server.py::_gui_surface_toolsets), never
# config: another surface lacking them made no configuration choice.
@@ -159,7 +157,6 @@ TOOLSETS = {
"clarify": _ts("Ask the user clarifying questions (multiple-choice or open-ended)", ["clarify"]),
"code_execution": _ts("Run Python scripts that call tools programmatically (reduces LLM round trips)", ["execute_code"]),
"delegation": _ts("Spawn subagents with isolated context for complex subtasks", ["delegate_task"]),
"homeassistant": _ts("Home Assistant smart home control and monitoring", _HA_TOOLS),
"kanban": _ts(
"Kanban multi-agent coordination — only active when the agent is spawned by "
"the kanban dispatcher (HERMES_KANBAN_TASK env set). The dispatcher runs "
@@ -224,7 +221,6 @@ TOOLSETS = {
"hermes-slack": _bundle("Slack bot toolset - full access for workspace use (terminal has safety checks)"),
"hermes-signal": _bundle("Signal bot toolset - encrypted messaging platform (full access)"),
"hermes-bluebubbles": _bundle("BlueBubbles iMessage bot toolset - Apple iMessage via local BlueBubbles server"),
"hermes-homeassistant": _bundle("Home Assistant bot toolset - smart home event monitoring and control"),
"hermes-email": _bundle("Email bot toolset - interact with Hermes via email (IMAP/SMTP)"),
"hermes-mattermost": _bundle("Mattermost bot toolset - self-hosted team messaging (full access)"),
"hermes-matrix": _bundle("Matrix bot toolset - decentralized encrypted messaging (full access)"),
@@ -247,7 +243,7 @@ TOOLSETS = {
[],
includes=[
"hermes-telegram", "hermes-discord", "hermes-whatsapp", "hermes-slack",
"hermes-signal", "hermes-bluebubbles", "hermes-homeassistant", "hermes-email",
"hermes-signal", "hermes-bluebubbles", "hermes-email",
"hermes-sms", "hermes-mattermost", "hermes-matrix", "hermes-dingtalk",
"hermes-feishu", "hermes-wecom", "hermes-wecom-callback", "hermes-weixin",
"hermes-qqbot", "hermes-webhook", "hermes-yuanbao",
@@ -306,7 +302,7 @@ def get_toolset(name: str, *, include_registry: bool = True) -> Optional[Dict[st
if toolset:
merged_tools = set(toolset.get("tools", [])) | set(registry.get_tool_names_for_toolset(name))
# An MCP server named like a built-in toolset ("homeassistant", "browser") registers a bare
# An MCP server named like a built-in toolset ("browser", "memory") registers a bare
# alias to its `mcp-<name>` toolset; without this union the static entry shadows it and the
# server's tools never reach the model even though discovery registered them.
alias_target = registry.get_toolset_alias_target(name)
Generated
+1 -7
View File
@@ -2479,9 +2479,6 @@ google-meet = [
{ name = "playwright" },
{ name = "websockets" },
]
homeassistant = [
{ name = "aiohttp" },
]
kittentts = [
{ name = "kittentts", marker = "(platform_machine != 'x86_64' and sys_platform == 'darwin') or (platform_machine != 'ARM64' and sys_platform == 'win32') or (sys_platform != 'darwin' and sys_platform != 'win32')" },
{ name = "misaki", extra = ["en"], marker = "(platform_machine != 'x86_64' and sys_platform == 'darwin') or (platform_machine != 'ARM64' and sys_platform == 'win32') or (sys_platform != 'darwin' and sys_platform != 'win32')" },
@@ -2651,7 +2648,6 @@ test = [
requires-dist = [
{ name = "agent-client-protocol", marker = "extra == 'acp'", specifier = "==0.9.0" },
{ name = "aiohttp", marker = "sys_platform == 'linux' and extra == 'matrix'", specifier = "==3.14.3" },
{ name = "aiohttp", marker = "extra == 'homeassistant'", specifier = "==3.14.3" },
{ name = "aiohttp", marker = "extra == 'messaging'", specifier = "==3.14.3" },
{ name = "aiohttp", marker = "extra == 'slack'", specifier = "==3.14.3" },
{ name = "aiohttp", marker = "extra == 'sms'", specifier = "==3.14.3" },
@@ -2699,8 +2695,6 @@ requires-dist = [
{ name = "hermes-agent", extras = ["cron"], marker = "extra == 'termux'" },
{ name = "hermes-agent", extras = ["google"], marker = "extra == 'all'" },
{ name = "hermes-agent", extras = ["google"], marker = "extra == 'termux-all'" },
{ name = "hermes-agent", extras = ["homeassistant"], marker = "extra == 'all'" },
{ name = "hermes-agent", extras = ["homeassistant"], marker = "extra == 'termux-all'" },
{ name = "hermes-agent", extras = ["mcp"], marker = "extra == 'all'" },
{ name = "hermes-agent", extras = ["mcp"], marker = "extra == 'termux'" },
{ name = "hermes-agent", extras = ["pty"], marker = "extra == 'all'" },
@@ -2811,7 +2805,7 @@ requires-dist = [
{ name = "winrt-windows-services-store", marker = "python_full_version >= '3.14' and sys_platform == 'win32'", specifier = ">=3.2.1,<4" },
{ name = "youtube-transcript-api", marker = "extra == 'youtube'", specifier = "==1.2.4" },
]
provides-extras = ["uvloop", "anthropic", "exa", "firecrawl", "parallel-web", "ddgs", "fal", "edge-tts", "neutts", "kittentts", "piper", "modal", "daytona", "vercel", "google-meet", "messaging", "cron", "slack", "matrix", "wecom", "tts-premium", "voice", "wake", "telegram", "discord", "stt-whisper", "audio-io", "silk", "wake-openwakeword", "wake-sherpa", "wake-porcupine", "google-chat", "doc-extract", "trace-upload", "mem0", "vision", "pty", "mcp", "nemo-relay", "homeassistant", "sms", "teams", "computer-use", "acp", "mistral", "otlp", "langfuse", "bedrock", "vertex", "azure-identity", "termux", "termux-all", "dingtalk", "feishu", "google", "youtube", "web", "all"]
provides-extras = ["uvloop", "anthropic", "exa", "firecrawl", "parallel-web", "ddgs", "fal", "edge-tts", "neutts", "kittentts", "piper", "modal", "daytona", "vercel", "google-meet", "messaging", "cron", "slack", "matrix", "wecom", "tts-premium", "voice", "wake", "telegram", "discord", "stt-whisper", "audio-io", "silk", "wake-openwakeword", "wake-sherpa", "wake-porcupine", "google-chat", "doc-extract", "trace-upload", "mem0", "vision", "pty", "mcp", "nemo-relay", "sms", "teams", "computer-use", "acp", "mistral", "otlp", "langfuse", "bedrock", "vertex", "azure-identity", "termux", "termux-all", "dingtalk", "feishu", "google", "youtube", "web", "all"]
[package.metadata.requires-dev]
dev = [