mvezina/homeassistant-mcp-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[FIX] Type Alias error

Browse Source
This commit is contained in:
Martin 2026-02-10 16:56:48 -05:00
parent 8a3025081a
commit 76bd1582e7
1 changed files with 23 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
custom_components/mcp_bridge/__init__.py
View File

@ -1,20 +1,11 @@
"""MCP Bridge integration for Home Assistant.
This integration provides services to expose filtered entities and scripts
to MCP (Model Context Protocol) servers for AI agent control.
"""
"""MCP Bridge integration for Home Assistant."""
from __future__ import annotations
import logging
from typing import Any
from homeassistant.const import CONF_ENTITY_ID
from homeassistant.core import (
HomeAssistant,
ServiceCall,
ServiceResponse,
SupportsResponse,
)
from homeassistant.core import HomeAssistant, ServiceCall, SupportsResponse
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import area_registry as ar
from homeassistant.helpers.typing import ConfigType
@ -26,15 +17,13 @@ MCP_ACCESSIBLE_LABEL = "mcp_accessible"
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the MCP Bridge integration."""
_LOGGER.info("Setting up MCP Bridge integration")
async def get_exposed_entities(call: ServiceCall) -> ServiceResponse:
"""Return entities exposed to conversation."""
async def get_exposed_entities(call: ServiceCall) -> dict[str, Any]:
entity_reg = er.async_get(hass)
area_reg = ar.async_get(hass)
exposed_entities: list[dict[str, Any]] = []
entities: list[dict[str, Any]] = []
for state in hass.states.async_all():
entity_entry = entity_reg.async_get(state.entity_id)
@ -52,7 +41,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
if area:
area_name = area.name
exposed_entities.append(
entities.append(
{
"entity_id": state.entity_id,
"state": state.state,
@ -71,17 +60,14 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
}
)
return ServiceResponse(
{
"entities": exposed_entities,
"count": len(exposed_entities),
}
)
return {
"entities": entities,
"count": len(entities),
}
async def get_exposed_scripts(call: ServiceCall) -> ServiceResponse:
"""Return scripts marked as MCP-accessible."""
async def get_exposed_scripts(call: ServiceCall) -> dict[str, Any]:
entity_reg = er.async_get(hass)
exposed_scripts: list[dict[str, Any]] = []
scripts: list[dict[str, Any]] = []
for entity_id in hass.states.async_entity_ids("script"):
entity_entry = entity_reg.async_get(entity_id)
@ -95,7 +81,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
if not state:
continue
exposed_scripts.append(
scripts.append(
{
"entity_id": entity_id,
"friendly_name": state.attributes.get(
@ -106,32 +92,22 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
}
)
return ServiceResponse(
{
"scripts": exposed_scripts,
"count": len(exposed_scripts),
}
)
return {
"scripts": scripts,
"count": len(scripts),
}
async def get_entity_metadata(call: ServiceCall) -> ServiceResponse:
"""Return detailed metadata for a single entity."""
async def get_entity_metadata(call: ServiceCall) -> dict[str, Any]:
entity_id = call.data.get(CONF_ENTITY_ID)
if not entity_id:
return ServiceResponse(
{"error": "entity_id is required"},
success=False,
)
return {"error": "entity_id is required"}
entity_reg = er.async_get(hass)
area_reg = ar.async_get(hass)
state = hass.states.get(entity_id)
if not state:
return ServiceResponse(
{"error": f"Entity {entity_id} not found"},
success=False,
)
return {"error": f"Entity {entity_id} not found"}
entity_entry = entity_reg.async_get(entity_id)
@ -141,7 +117,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
if area:
area_name = area.name
metadata: dict[str, Any] = {
data: dict[str, Any] = {
"entity_id": entity_id,
"state": state.state,
"friendly_name": state.attributes.get(
@ -159,12 +135,12 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
}
if entity_entry:
metadata["labels"] = list(entity_entry.labels)
metadata["is_exposed_to_conversation"] = entity_entry.options.get(
data["labels"] = list(entity_entry.labels)
data["is_exposed_to_conversation"] = entity_entry.options.get(
"conversation", {}
).get("should_expose", False)
return ServiceResponse(metadata)
return data
hass.services.async_register(
DOMAIN,