| """Authenticated Streamable HTTP bridge to Home Assistant.""" |
| import os |
| import secrets |
| from typing import Any |
|
|
| import httpx |
| from mcp.server.fastmcp import FastMCP |
| from mcp.server.transport_security import TransportSecuritySettings |
| from starlette.responses import JSONResponse |
|
|
| HA_URL = os.getenv("HA_URL", "http://127.0.0.1:8123").rstrip("/") |
| HA_TOKEN = os.getenv("HA_TOKEN", "") |
| MCP_API_KEY = os.getenv("MCP_API_KEY", "") |
| mcp = FastMCP("Home Assistant", stateless_http=True, json_response=True, streamable_http_path="/mcp", |
| transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False)) |
|
|
| async def ha_request(method: str, path: str, data: dict | None = None) -> Any: |
| if not HA_TOKEN: |
| raise ValueError("Set the HA_TOKEN Space secret to a Home Assistant long-lived token.") |
| async with httpx.AsyncClient(timeout=45, trust_env=False) as client: |
| response = await client.request(method, HA_URL + "/api/" + path, |
| headers={"Authorization": "Bearer " + HA_TOKEN}, json=data) |
| if response.is_error: |
| raise ValueError(f"Home Assistant returned HTTP {response.status_code}.") |
| return response.json() |
|
|
| @mcp.tool() |
| async def list_entities(domain: str | None = None) -> list[dict]: |
| """List actual entity IDs, states and attributes. Optionally filter by domain (light, climate, etc.).""" |
| states = await ha_request("GET", "states") |
| return [s for s in states if domain is None or s["entity_id"].startswith(domain + ".")] |
|
|
| @mcp.tool() |
| async def get_entity_state(entity_id: str) -> dict: |
| """Read an entity's current state and attributes using its actual entity ID.""" |
| validate_name(entity_id, dotted=True) |
| return await ha_request("GET", "states/" + entity_id) |
|
|
| @mcp.tool() |
| async def list_services() -> list[dict]: |
| """Discover available HA services and their parameters before controlling devices.""" |
| return await ha_request("GET", "services") |
|
|
| @mcp.tool() |
| async def call_service(domain: str, service: str, service_data: dict[str, Any] | None = None) -> Any: |
| """Execute a Home Assistant service. This changes devices or automations. Use actual service names and entity IDs discovered by the read tools; service_data is the HA JSON body, e.g. {"entity_id":"light.room"}. No device whitelist is applied.""" |
| validate_name(domain) |
| validate_name(service) |
| return await ha_request("POST", f"services/{domain}/{service}", service_data or {}) |
|
|
| def validate_name(value: str, dotted: bool = False): |
| import re |
| pattern = r"[a-z0-9_]+\.[a-z0-9_]+" if dotted else r"[a-z0-9_]+" |
| if not re.fullmatch(pattern, value): |
| raise ValueError("Invalid Home Assistant identifier.") |
|
|
| @mcp.custom_route("/mcp/health", methods=["GET"]) |
| async def health(request): |
| """Authenticated routing check; does not test HA token validity or HA readiness.""" |
| return JSONResponse({"service": "home-assistant-mcp", "transport": "streamable-http", |
| "endpoint": "/mcp", "ha_token_configured": bool(HA_TOKEN)}) |
|
|
|
|
| class BearerAuth: |
| def __init__(self, app): |
| self.app = app |
|
|
| async def __call__(self, scope, receive, send): |
| if scope["type"] == "http": |
| if scope["path"] == "/mcp/": |
| scope = {**scope, "path": "/mcp", "raw_path": b"/mcp"} |
| if not MCP_API_KEY: |
| response = JSONResponse({"error": "MCP is disabled until MCP_API_KEY is configured"}, status_code=503) |
| return await response(scope, receive, send) |
| headers = dict(scope["headers"]) |
| actual = headers.get(b"authorization", b"") |
| expected = ("Bearer " + MCP_API_KEY).encode() |
| if not secrets.compare_digest(actual, expected): |
| response = JSONResponse({"error": "Unauthorized"}, status_code=401, |
| headers={"WWW-Authenticate": "Bearer"}) |
| return await response(scope, receive, send) |
| await self.app(scope, receive, send) |
|
|
| app = BearerAuth(mcp.streamable_http_app()) |
|
|
