feat: add include_knx_project to ha_get_integration for KNX group addresses

Adds an opt-in `include_knx_project` flag to `ha_get_integration` that attaches the parsed ETS project (group-address table, group-range hierarchy, project metadata) for a KNX config entry, read via the `knx/get_knx_project` WebSocket command. Per-entity GA assignments remain available via `include_diagnostics`.

Author: pkkr

site/src/data/tools.json

@@ -1913,7 +1913,7 @@
   {
     "name": "ha_get_integration",
     "title": "Get Integration",
-    "description": "Get integration (config entry) information with pagination.\n\nWithout an entry_id: Lists all configured integrations with optional filters.\nWith an entry_id: Returns detailed information including full options/configuration.\n\nEXAMPLES:\n- List all integrations: ha_get_integration()\n- Paginate: ha_get_integration(offset=50)\n- Search: ha_get_integration(query=\"zigbee\")\n- Get specific entry: ha_get_integration(entry_id=\"abc123\")\n- Get entry with editable fields: ha_get_integration(entry_id=\"abc123\", include_schema=True)\n- Get entry with diagnostics dump: ha_get_integration(entry_id=\"abc123\", include_diagnostics=True)\n- Get device-scoped diagnostics: ha_get_integration(entry_id=\"abc123\", include_diagnostics=True, device_id=\"dev123\")\n- Walk a sub-tree: ha_get_integration(entry_id=\"abc123\", include_diagnostics=True, diagnostics_data_path=\"<dotted-path>\")\n- Paginate a large list: ha_get_integration(entry_id=\"abc123\", include_diagnostics=True, diagnostics_data_path=\"<list-valued path>\", diagnostics_data_limit=10, diagnostics_data_offset=20)\n- List config subentries: ha_get_integration(entry_id=\"abc123\", include_subentries=True)\n- Inspect subentry create schema: ha_get_integration(entry_id=\"abc123\", include_subentry_schema=True, subentry_type=\"conversation\")\n- Inspect subentry reconfigure schema: ha_get_integration(entry_id=\"abc123\", include_subentry_schema=True, subentry_type=\"conversation\", subentry_id=\"sub123\")\n- List template entries: ha_get_integration(domain=\"template\")\n\nSTATES: 'loaded', 'setup_error', 'setup_retry', 'not_loaded',\n'failed_unload', 'migration_error'.\n\nEach entry carries:\n\n- ``log_level``: the canonical Python logger level name\n  (``DEBUG``/``INFO``/``WARNING``/``ERROR``/``CRITICAL``) when the\n  integration has a ``logger.set_level`` override, or ``\"DEFAULT\"``\n  (uppercase sentinel) when no override is set.\n- ``log_level_raw``: the original numeric level (e.g. ``10`` for DEBUG)\n  when HA returned an int, ``None`` otherwise (no override set, or HA\n  provided a level name as a string).\n\nThis is distinct from the add-on side, where ``ha_get_addon`` returns\nSupervisor's lowercase ``\"default\"`` literal — do not cross-compare.",
+    "description": "Get integration (config entry) information with pagination.\n\nWithout an entry_id: Lists all configured integrations with optional filters.\nWith an entry_id: Returns detailed information including full options/configuration.\n\nEXAMPLES:\n- List all integrations: ha_get_integration()\n- Paginate: ha_get_integration(offset=50)\n- Search: ha_get_integration(query=\"zigbee\")\n- Get specific entry: ha_get_integration(entry_id=\"abc123\")\n- Get entry with editable fields: ha_get_integration(entry_id=\"abc123\", include_schema=True)\n- Get entry with diagnostics dump: ha_get_integration(entry_id=\"abc123\", include_diagnostics=True)\n- Get device-scoped diagnostics: ha_get_integration(entry_id=\"abc123\", include_diagnostics=True, device_id=\"dev123\")\n- Get the parsed KNX ETS project (group-address table): ha_get_integration(entry_id=\"<knx entry>\", include_knx_project=True)\n- Walk a sub-tree: ha_get_integration(entry_id=\"abc123\", include_diagnostics=True, diagnostics_data_path=\"<dotted-path>\")\n- Paginate a large list: ha_get_integration(entry_id=\"abc123\", include_diagnostics=True, diagnostics_data_path=\"<list-valued path>\", diagnostics_data_limit=10, diagnostics_data_offset=20)\n- List config subentries: ha_get_integration(entry_id=\"abc123\", include_subentries=True)\n- Inspect subentry create schema: ha_get_integration(entry_id=\"abc123\", include_subentry_schema=True, subentry_type=\"conversation\")\n- Inspect subentry reconfigure schema: ha_get_integration(entry_id=\"abc123\", include_subentry_schema=True, subentry_type=\"conversation\", subentry_id=\"sub123\")\n- List template entries: ha_get_integration(domain=\"template\")\n\nSTATES: 'loaded', 'setup_error', 'setup_retry', 'not_loaded',\n'failed_unload', 'migration_error'.\n\nEach entry carries:\n\n- ``log_level``: the canonical Python logger level name\n  (``DEBUG``/``INFO``/``WARNING``/``ERROR``/``CRITICAL``) when the\n  integration has a ``logger.set_level`` override, or ``\"DEFAULT\"``\n  (uppercase sentinel) when no override is set.\n- ``log_level_raw``: the original numeric level (e.g. ``10`` for DEBUG)\n  when HA returned an int, ``None`` otherwise (no override set, or HA\n  provided a level name as a string).\n\nThis is distinct from the add-on side, where ``ha_get_addon`` returns\nSupervisor's lowercase ``\"default\"`` literal — do not cross-compare.",
     "inputSchema": {
       "properties": {
         "entry_id": {
@@ -1972,6 +1972,10 @@
           "type": "Annotated[bool, Field(description=\"When entry_id is set, also fetch the integration's diagnostics dump — integration-defined JSON (commonly includes redacted config, device list, state snapshots; exact top-level keys vary by integration). The canonical artifact users grab via Settings → Devices & Services → [integration] → ⋯ → Download diagnostics. Use when triaging integration bugs or filing ha_report_issue for a specific integration. Payloads can be large (Hue ~290 KB, ZHA/MQTT/ESPHome several MB) — pair with diagnostics_fields or diagnostics_truncate_at_bytes to fit the LLM context budget.\", default=False)]",
           "default": false
         },
+        "include_knx_project": {
+          "type": "Annotated[bool, Field(description='When entry_id is a KNX config entry, also return the parsed ETS project: the full group-address table (address, name, DPT, description) under knx_project.group_addresses, plus the group-range hierarchy and project metadata. This is the parsed-project GA table that is NOT in the diagnostics dump; per-entity GA assignments are already covered by include_diagnostics (config_store / configuration_yaml). Ignored (with a warning) when the entry is not a KNX integration. The KNX integration exposes a single project, so the result is the same regardless of which KNX entry_id is used.', default=False)]",
+          "default": false
+        },
         "device_id": {
           "type": "Annotated[str | None, Field(description='Optional. When set with include_diagnostics=True, returns the device-scoped diagnostics dump for that specific device under the integration (rather than the full integration dump). Some integrations only expose config-entry-level dumps; others expose both.', default=None)]",
           "default": null

src/ha_mcp/tools/tools_integrations.py

@@ -445,6 +445,25 @@ async def ha_get_integration(
                 default=False,
             ),
         ] = False,
+        include_knx_project: Annotated[
+            bool,
+            Field(
+                description=(
+                    "When entry_id is a KNX config entry, also return the parsed "
+                    "ETS project: the full group-address table (address, name, "
+                    "DPT, description) under knx_project.group_addresses, plus the "
+                    "group-range hierarchy and project metadata. This is the "
+                    "parsed-project GA table that is NOT in the diagnostics dump; "
+                    "per-entity GA assignments are already covered by "
+                    "include_diagnostics (config_store / configuration_yaml). "
+                    "Ignored (with a warning) when the entry is not a KNX "
+                    "integration. The KNX integration exposes a single project, "
+                    "so the result is the same regardless of which KNX entry_id "
+                    "is used."
+                ),
+                default=False,
+            ),
+        ] = False,
         device_id: Annotated[
             str | None,
             Field(
@@ -553,6 +572,7 @@ async def ha_get_integration(
         - Get entry with editable fields: ha_get_integration(entry_id="abc123", include_schema=True)
         - Get entry with diagnostics dump: ha_get_integration(entry_id="abc123", include_diagnostics=True)
         - Get device-scoped diagnostics: ha_get_integration(entry_id="abc123", include_diagnostics=True, device_id="dev123")
+        - Get the parsed KNX ETS project (group-address table): ha_get_integration(entry_id="<knx entry>", include_knx_project=True)
         - Walk a sub-tree: ha_get_integration(entry_id="abc123", include_diagnostics=True, diagnostics_data_path="<dotted-path>")
         - Paginate a large list: ha_get_integration(entry_id="abc123", include_diagnostics=True, diagnostics_data_path="<list-valued path>", diagnostics_data_limit=10, diagnostics_data_offset=20)
         - List config subentries: ha_get_integration(entry_id="abc123", include_subentries=True)
@@ -580,6 +600,7 @@ async def ha_get_integration(
             include_opts = include_options
             include_schema_bool = include_schema
             include_diagnostics_bool = include_diagnostics
+            include_knx_project_bool = include_knx_project
             include_subentries_bool = include_subentries
             include_subentry_schema_bool = include_subentry_schema
             show_advanced_options_bool = show_advanced_options
@@ -639,6 +660,8 @@ async def ha_get_integration(
                         "device_id was provided but ignored because "
                         "include_diagnostics=False"
                     )
+                if include_knx_project_bool:
+                    await self._attach_knx_project(resp, entry_id)
                 return resp
 
             # List mode - get all config entries
@@ -648,6 +671,8 @@ async def ha_get_integration(
             ignored_detail_params = []
             if include_diagnostics_bool:
                 ignored_detail_params.append("include_diagnostics")
+            if include_knx_project_bool:
+                ignored_detail_params.append("include_knx_project")
             if device_id is not None:
                 ignored_detail_params.append("device_id")
             if fields_list is not None:
@@ -777,6 +802,69 @@ async def _get_single_entry(
             )
             return None  # unreachable: exception_to_structured_error raises
 
+    async def _attach_knx_project(self, resp: dict[str, Any], entry_id: str) -> None:
+        """Attach the parsed KNX ETS project to a single-entry response.
+
+        Reads ``knx/get_knx_project`` (the same command the KNX panel uses) and
+        attaches the group-address table, group-range hierarchy, and project
+        metadata under ``resp["knx_project"]``. The KNX integration exposes one
+        project globally, so the command takes no entry scope; this is gated on
+        the entry actually being a KNX integration to avoid attaching unrelated
+        data to a non-KNX entry.
+
+        Failures are surfaced as warnings rather than raised — the primary
+        entry read already succeeded, mirroring how diagnostics fetch failures
+        are handled. The "KNX integration not loaded" case collapses into
+        ha_get_integration's existing no-such-entry handling (no KNX entry → no
+        entry_id to pass here), so it needs no special casing.
+        """
+        entry = resp.get("entry") if isinstance(resp.get("entry"), dict) else {}
+        domain = entry.get("domain") if isinstance(entry, dict) else None
+        if domain != "knx":
+            resp.setdefault("warnings", []).append(
+                f"include_knx_project ignored: entry {entry_id} is not a KNX "
+                f"integration (domain={domain!r})"
+            )
+            return
+
+        result = await self._client.send_websocket_message(
+            {"type": "knx/get_knx_project"}
+        )
+        if not isinstance(result, dict) or not result.get("success"):
+            error_msg = websocket_error_message(
+                result.get("error", "Unknown error")
+                if isinstance(result, dict)
+                else result
+            )
+            resp.setdefault("warnings", []).append(
+                f"Failed to fetch KNX project: {error_msg}"
+            )
+            return
+
+        project = result.get("result")
+        if not project:
+            # No ETS project uploaded yet — get_knxproject() returns None.
+            resp["knx_project"] = {
+                "count": 0,
+                "group_addresses": {},
+                "group_ranges": {},
+                "info": {},
+                "note": (
+                    "The KNX integration is loaded but no ETS project has been "
+                    "uploaded yet. Upload a .knxproj file via the KNX panel to "
+                    "populate the group-address table."
+                ),
+            }
+            return
+
+        group_addresses = project.get("group_addresses", {})
+        resp["knx_project"] = {
+            "count": len(group_addresses),
+            "group_addresses": group_addresses,
+            "group_ranges": project.get("group_ranges", {}),
+            "info": project.get("info", {}),
+        }
+
     async def _fetch_config_subentries(self, entry_id: str) -> list[dict[str, Any]]:
         """Fetch config subentries for a detailed entry response."""
         result = await self._client.list_config_subentries(entry_id)

tests/src/e2e/tools/test_tools_knx.py

@@ -0,0 +1,50 @@
+"""
+E2E smoke test for the KNX project fold into ha_get_integration.
+
+The test container has no KNX integration configured, so there is no KNX
+config entry to pass include_knx_project against. This exercises the
+not-loaded path the maintainer asked us to keep: requesting
+include_knx_project on a non-KNX entry must degrade cleanly (a warning, no
+crash, no KNX data attached) over the real WebSocket/REST plumbing. The
+parsed-project success path is covered by the unit tests, which would
+otherwise require a live HA with the KNX integration and an uploaded ETS
+project.
+"""
+
+import logging
+
+import pytest
+
+from ..utilities.assertions import assert_mcp_success
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.asyncio
+async def test_include_knx_project_on_non_knx_entry_degrades_cleanly(mcp_client):
+    """include_knx_project against a non-KNX entry returns a warning, not data.
+
+    Picks the first available config entry (none are KNX in the test
+    container) and asserts the response succeeds with no knx_project key and a
+    warning explaining the flag was ignored.
+    """
+    listing = await mcp_client.call_tool("ha_get_integration", {})
+    raw_list = assert_mcp_success(listing, "list integrations")
+    entries = raw_list.get("data", raw_list).get("entries", [])
+    if not entries:
+        pytest.skip("no config entries available in the test container")
+    entry_id = entries[0]["entry_id"]
+
+    result = await mcp_client.call_tool(
+        "ha_get_integration",
+        {"entry_id": entry_id, "include_knx_project": True},
+    )
+    raw = assert_mcp_success(result, "get integration with include_knx_project")
+    data = raw.get("data", raw)
+
+    assert "knx_project" not in data
+    warnings = data.get("warnings", [])
+    assert any("KNX" in w for w in warnings), (
+        f"expected an include_knx_project warning, got warnings={warnings}"
+    )
+    logger.info("include_knx_project degraded cleanly on a non-KNX entry")