Own minimal MCP server exposing the live state of the HA instances in
services/home-assistant/instances.yaml to Claude Code over stdio — the
phase-2 "MCP read-only" gate in services/home-assistant/DESIGN.md.
Operator decision 2026-07-30: build our own rather than adopt hass-mcp,
so the tools reuse scripts/ha/lib/{ha_api,ha_ws}.py (one token-handling
story for the whole HA toolchain) and can answer from the repo and from
instances.yaml, which a generic server cannot.
Seven tools, all read-only, default instance `ken`: list_entities,
get_state, get_areas, find_entities_by_description, read_automation,
list_automations, instance_status.
Read-only by construction, not by policy: REST goes through ha_api.Client
(get/get_raw_text only — no POST method exists), WebSocket commands are
checked against a three-entry *_list allowlist before being sent, and
read_automation reads services/home-assistant/config/<instance>/ rather
than /api/config. Tests assert all three, including a grep guard that
fails if requests.post/call_service ever appears in the package. The
write path stays repo + scripts/ha/deploy.sh.
Details that follow from how this instance actually behaves:
- unavailable is never silent — every entity view carries unavailable +
unavailable_since, every list a count. The 2026-07-23 audit traced ~15
silently dead automations to conditions sitting on dead sensors.
- chelsty-ha (status: offline in instances.yaml) is answered from the
file, never dialed — no 5s timeout for a known-offline LTE site.
- areas come from the WS registries (entity area_id > device area_id) with
a storage-export fallback; area_source/area_note say which was used and
what the offline export cannot resolve.
- PL->EN fuzzy matching, since the house is Polish and the entity_ids are
transliterated English: "czujnik temperatury salon" ->
sensor.thsalon_temperature, each hit explaining why it matched.
- 5s timeouts and errors returned as {"error": ...} inside a normal tool
result — a missing token or an unreachable instance never crashes the
server or hangs the agent.
Registered for Claude Code in the repo-root .mcp.json (new file) as `ha`,
via services/ha-mcp/run.sh (prefers the venv, falls back to system
python3). The mcp SDK lives in services/ha-mcp/.venv — rationale for venv
over --break-system-packages is in the README.
Tests: 42 offline (no network, no HA, no token) + a live read-only smoke
against ken — HA 2026.7.2, 1647 entities, 377 unavailable, 115
automations, 13 areas.
117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
"""Offline test rig: a fake backend fed from JSON fixtures.
|
|
|
|
Same idea as scripts/ha/tests/test_import_api_offline.sh — no network, no HA
|
|
instance, no token. The fixtures under tests/fixtures/ are shaped exactly
|
|
like real `/api/states` and WebSocket registry payloads (trimmed down);
|
|
read_automation tests read the real repo automations, which is offline too.
|
|
"""
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
SERVICE_DIR = Path(__file__).resolve().parents[1]
|
|
REPO_ROOT = SERVICE_DIR.parents[1]
|
|
FIXTURES = Path(__file__).resolve().parent / "fixtures"
|
|
|
|
sys.path.insert(0, str(SERVICE_DIR / "src"))
|
|
|
|
from ha_mcp import backend as backend_mod # noqa: E402
|
|
|
|
|
|
def load_fixture(name):
|
|
with open(FIXTURES / name, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
class FakeBackend:
|
|
"""Stands in for LiveBackend: canned states/config/registries, real repo files."""
|
|
|
|
def __init__(self, instance="ken", states=None, registries=None, config=None,
|
|
area_source="websocket", automations_instance=None):
|
|
self.instance = instance
|
|
self.cfg = {
|
|
"adapter": "api",
|
|
"base_url": "http://ha.example:8123",
|
|
"status": "active",
|
|
"site": "ken",
|
|
"token_path": "/nonexistent",
|
|
}
|
|
self._states = load_fixture("states.json") if states is None else states
|
|
reg = load_fixture("registries.json") if registries is None else registries
|
|
self._config = config or {
|
|
"version": "2026.7.2",
|
|
"location_name": "KEN",
|
|
"time_zone": "Europe/Warsaw",
|
|
}
|
|
self._index = backend_mod._index_from_registries(
|
|
reg.get("areas"), reg.get("entities"), reg.get("devices"), area_source
|
|
)
|
|
self._automations_instance = automations_instance or instance
|
|
|
|
@property
|
|
def status(self):
|
|
return self.cfg.get("status") or "unknown"
|
|
|
|
def states(self):
|
|
return self._states
|
|
|
|
def ha_config(self):
|
|
return self._config
|
|
|
|
def area_index(self):
|
|
return self._index
|
|
|
|
def automation_files(self):
|
|
return backend_mod.automation_files(self._automations_instance, REPO_ROOT)
|
|
|
|
|
|
class UnreachableBackend(FakeBackend):
|
|
"""Live calls fail; repo-backed calls still work (read_automation offline path)."""
|
|
|
|
def states(self):
|
|
raise backend_mod.BackendError("instance 'ken' unreachable at http://ha.example:8123")
|
|
|
|
def ha_config(self):
|
|
raise backend_mod.BackendError("instance 'ken' unreachable at http://ha.example:8123")
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_backend():
|
|
return FakeBackend()
|
|
|
|
|
|
@pytest.fixture
|
|
def unreachable_backend():
|
|
return UnreachableBackend()
|
|
|
|
|
|
@pytest.fixture
|
|
def repo_root():
|
|
return REPO_ROOT
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def real_ken_backend():
|
|
"""FakeBackend over the newest committed /api/states snapshot of `ken`.
|
|
|
|
Still offline (the fixture is a file in the repo), but ~1650 real
|
|
entities instead of ten — ranking behaves differently at that scale.
|
|
Areas come from the storage-export fallback, same as a live run with the
|
|
WebSocket unavailable.
|
|
"""
|
|
import yaml
|
|
|
|
from ha_mcp import backend as bm
|
|
|
|
snapshots = sorted((REPO_ROOT / "services" / "home-assistant" / "fixtures").glob("ken-states-*.yaml"))
|
|
if not snapshots:
|
|
pytest.skip("no ken-states-*.yaml fixture in the repo")
|
|
with open(snapshots[-1], encoding="utf-8") as f:
|
|
states = yaml.safe_load(f)
|
|
|
|
backend = FakeBackend(states=states)
|
|
backend._index = bm.load_offline_area_index("ken", REPO_ROOT)
|
|
return backend
|