File size: 4,017 Bytes
ad2cbeb 6e96d5d ad2cbeb 6e96d5d ad2cbeb 6e96d5d ad2cbeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """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())
|