590 lines
19 KiB
Python
590 lines
19 KiB
Python
|
"""
|
||
|
|
Home Assistant MCP Server
|
||
|
|
=========================
|
||
|
|
MCP server for controlling and monitoring Home Assistant via the REST API.
|
||
|
|
Provides tools for managing entities, services, automations, scenes, and more.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from typing import Optional, Any
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
from mcp.server.fastmcp import FastMCP
|
||
|
|
from pydantic import BaseModel, Field
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Configuration
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
HASS_URL = os.environ.get("HASS_URL", "https://haos.wilddragoncore.online")
|
||
|
|
HASS_TOKEN = os.environ.get("HASS_TOKEN", "")
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# MCP Server
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
mcp = FastMCP("homeassistant_mcp")
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# HTTP helpers
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def _headers() -> dict:
|
||
|
|
return {
|
||
|
|
"Authorization": f"Bearer {HASS_TOKEN}",
|
||
|
|
"Content-Type": "application/json",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
async def _get(path: str, params: Optional[dict] = None) -> Any:
|
||
|
|
async with httpx.AsyncClient(verify=False, timeout=30) as client:
|
||
|
|
resp = await client.get(
|
||
|
|
f"{HASS_URL}/api/{path}", headers=_headers(), params=params
|
||
|
|
)
|
||
|
|
resp.raise_for_status()
|
||
|
|
return resp.json()
|
||
|
|
|
||
|
|
|
||
|
|
async def _post(path: str, payload: Any = None) -> Any:
|
||
|
|
async with httpx.AsyncClient(verify=False, timeout=30) as client:
|
||
|
|
resp = await client.post(
|
||
|
|
f"{HASS_URL}/api/{path}", headers=_headers(), json=payload
|
||
|
|
)
|
||
|
|
resp.raise_for_status()
|
||
|
|
try:
|
||
|
|
return resp.json()
|
||
|
|
except Exception:
|
||
|
|
return {"status": "ok"}
|
||
|
|
|
||
|
|
|
||
|
|
def _handle_error(e: Exception) -> str:
|
||
|
|
if isinstance(e, httpx.HTTPStatusError):
|
||
|
|
status = e.response.status_code
|
||
|
|
try:
|
||
|
|
body = e.response.json()
|
||
|
|
msg = body.get("message", str(body))
|
||
|
|
except Exception:
|
||
|
|
msg = e.response.text[:500]
|
||
|
|
return f"Error {status}: {msg}"
|
||
|
|
return f"Error: {str(e)}"
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Input Models
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class EmptyInput(BaseModel):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
class EntityIdInput(BaseModel):
|
||
|
|
entity_id: str = Field(description="Entity ID (e.g. light.living_room)")
|
||
|
|
|
||
|
|
|
||
|
|
class ServiceCallInput(BaseModel):
|
||
|
|
domain: str = Field(description="Service domain (e.g. light, switch, climate)")
|
||
|
|
service: str = Field(description="Service name (e.g. turn_on, turn_off, toggle)")
|
||
|
|
entity_id: Optional[str] = Field(
|
||
|
|
default=None, description="Target entity ID"
|
||
|
|
)
|
||
|
|
service_data: Optional[dict] = Field(
|
||
|
|
default=None,
|
||
|
|
description="Additional service data (e.g. brightness, temperature)",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class EntitiesByDomainInput(BaseModel):
|
||
|
|
domain: str = Field(
|
||
|
|
description="Entity domain to filter by (e.g. light, switch, sensor, climate, media_player)"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class SetStateInput(BaseModel):
|
||
|
|
entity_id: str = Field(description="Entity ID to update")
|
||
|
|
state: str = Field(description="New state value")
|
||
|
|
attributes: Optional[dict] = Field(
|
||
|
|
default=None, description="Optional attributes to set"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class AutomationInput(BaseModel):
|
||
|
|
entity_id: str = Field(
|
||
|
|
description="Automation entity ID (e.g. automation.morning_routine)"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class EventInput(BaseModel):
|
||
|
|
event_type: str = Field(description="Event type to fire")
|
||
|
|
event_data: Optional[dict] = Field(
|
||
|
|
default=None, description="Optional event data payload"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class HistoryInput(BaseModel):
|
||
|
|
entity_id: str = Field(description="Entity ID to get history for")
|
||
|
|
hours: int = Field(default=24, description="Number of hours of history (max 72)")
|
||
|
|
|
||
|
|
|
||
|
|
class LogbookInput(BaseModel):
|
||
|
|
hours: int = Field(
|
||
|
|
default=24, description="Number of hours of logbook entries (max 72)"
|
||
|
|
)
|
||
|
|
entity_id: Optional[str] = Field(
|
||
|
|
default=None, description="Optional entity ID to filter by"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools — System Info
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_get_config",
|
||
|
|
description="Get Home Assistant configuration and system info including version, location, units, and components.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_get_config(params: EmptyInput) -> str:
|
||
|
|
try:
|
||
|
|
config = await _get("")
|
||
|
|
return json.dumps(
|
||
|
|
{
|
||
|
|
"version": config.get("version"),
|
||
|
|
"location_name": config.get("location_name"),
|
||
|
|
"latitude": config.get("latitude"),
|
||
|
|
"longitude": config.get("longitude"),
|
||
|
|
"elevation": config.get("elevation"),
|
||
|
|
"unit_system": config.get("unit_system"),
|
||
|
|
"time_zone": config.get("time_zone"),
|
||
|
|
"state": config.get("state"),
|
||
|
|
"components": sorted(config.get("components", []))[:50],
|
||
|
|
},
|
||
|
|
indent=2,
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_check_api",
|
||
|
|
description="Check if Home Assistant API is reachable and responding.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_check_api(params: EmptyInput) -> str:
|
||
|
|
try:
|
||
|
|
async with httpx.AsyncClient(verify=False, timeout=10) as client:
|
||
|
|
resp = await client.get(f"{HASS_URL}/api/", headers=_headers())
|
||
|
|
resp.raise_for_status()
|
||
|
|
return json.dumps({"status": "connected", "message": resp.json().get("message", "API running")})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools — Entity Management
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_list_entities",
|
||
|
|
description="List all entities, optionally filtered by domain. Returns entity_id, state, and friendly name.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_list_entities(params: EntitiesByDomainInput) -> str:
|
||
|
|
try:
|
||
|
|
states = await _get("states")
|
||
|
|
filtered = [
|
||
|
|
{
|
||
|
|
"entity_id": s["entity_id"],
|
||
|
|
"state": s["state"],
|
||
|
|
"friendly_name": s.get("attributes", {}).get("friendly_name", ""),
|
||
|
|
}
|
||
|
|
for s in states
|
||
|
|
if s["entity_id"].startswith(f"{params.domain}.")
|
||
|
|
]
|
||
|
|
return json.dumps(
|
||
|
|
{"domain": params.domain, "count": len(filtered), "entities": filtered},
|
||
|
|
indent=2,
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_get_entity_state",
|
||
|
|
description="Get the full state and attributes of a specific entity.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_get_entity_state(params: EntityIdInput) -> str:
|
||
|
|
try:
|
||
|
|
state = await _get(f"states/{params.entity_id}")
|
||
|
|
return json.dumps(
|
||
|
|
{
|
||
|
|
"entity_id": state["entity_id"],
|
||
|
|
"state": state["state"],
|
||
|
|
"attributes": state.get("attributes", {}),
|
||
|
|
"last_changed": state.get("last_changed"),
|
||
|
|
"last_updated": state.get("last_updated"),
|
||
|
|
},
|
||
|
|
indent=2,
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_get_all_domains",
|
||
|
|
description="Get a summary of all entity domains and their counts.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_get_all_domains(params: EmptyInput) -> str:
|
||
|
|
try:
|
||
|
|
states = await _get("states")
|
||
|
|
domains: dict[str, int] = {}
|
||
|
|
for s in states:
|
||
|
|
domain = s["entity_id"].split(".")[0]
|
||
|
|
domains[domain] = domains.get(domain, 0) + 1
|
||
|
|
sorted_domains = dict(sorted(domains.items(), key=lambda x: -x[1]))
|
||
|
|
return json.dumps(
|
||
|
|
{"total_entities": len(states), "domains": sorted_domains}, indent=2
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools — Service Calls (actions)
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_call_service",
|
||
|
|
description="Call any Home Assistant service. Use for turning on/off lights, switches, setting climate, playing media, etc.",
|
||
|
|
annotations={"readOnlyHint": False, "destructiveHint": False},
|
||
|
|
)
|
||
|
|
async def hass_call_service(params: ServiceCallInput) -> str:
|
||
|
|
try:
|
||
|
|
payload = params.service_data or {}
|
||
|
|
if params.entity_id:
|
||
|
|
payload["entity_id"] = params.entity_id
|
||
|
|
|
||
|
|
result = await _post(f"services/{params.domain}/{params.service}", payload)
|
||
|
|
changed = []
|
||
|
|
if isinstance(result, list):
|
||
|
|
changed = [
|
||
|
|
{"entity_id": s["entity_id"], "state": s["state"]}
|
||
|
|
for s in result[:10]
|
||
|
|
]
|
||
|
|
return json.dumps(
|
||
|
|
{
|
||
|
|
"status": "ok",
|
||
|
|
"service": f"{params.domain}.{params.service}",
|
||
|
|
"affected_entities": changed,
|
||
|
|
},
|
||
|
|
indent=2,
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_turn_on",
|
||
|
|
description="Turn on an entity (light, switch, fan, etc.).",
|
||
|
|
annotations={"readOnlyHint": False},
|
||
|
|
)
|
||
|
|
async def hass_turn_on(params: EntityIdInput) -> str:
|
||
|
|
try:
|
||
|
|
domain = params.entity_id.split(".")[0]
|
||
|
|
result = await _post(
|
||
|
|
f"services/{domain}/turn_on", {"entity_id": params.entity_id}
|
||
|
|
)
|
||
|
|
return json.dumps({"status": "ok", "entity_id": params.entity_id, "action": "turn_on"})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_turn_off",
|
||
|
|
description="Turn off an entity (light, switch, fan, etc.).",
|
||
|
|
annotations={"readOnlyHint": False},
|
||
|
|
)
|
||
|
|
async def hass_turn_off(params: EntityIdInput) -> str:
|
||
|
|
try:
|
||
|
|
domain = params.entity_id.split(".")[0]
|
||
|
|
result = await _post(
|
||
|
|
f"services/{domain}/turn_off", {"entity_id": params.entity_id}
|
||
|
|
)
|
||
|
|
return json.dumps({"status": "ok", "entity_id": params.entity_id, "action": "turn_off"})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_toggle",
|
||
|
|
description="Toggle an entity (if on, turn off; if off, turn on).",
|
||
|
|
annotations={"readOnlyHint": False},
|
||
|
|
)
|
||
|
|
async def hass_toggle(params: EntityIdInput) -> str:
|
||
|
|
try:
|
||
|
|
domain = params.entity_id.split(".")[0]
|
||
|
|
result = await _post(
|
||
|
|
f"services/{domain}/toggle", {"entity_id": params.entity_id}
|
||
|
|
)
|
||
|
|
return json.dumps({"status": "ok", "entity_id": params.entity_id, "action": "toggle"})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools — Automations & Scenes
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_list_automations",
|
||
|
|
description="List all automations with their current state (on/off) and last triggered time.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_list_automations(params: EmptyInput) -> str:
|
||
|
|
try:
|
||
|
|
states = await _get("states")
|
||
|
|
automations = [
|
||
|
|
{
|
||
|
|
"entity_id": s["entity_id"],
|
||
|
|
"state": s["state"],
|
||
|
|
"friendly_name": s.get("attributes", {}).get("friendly_name", ""),
|
||
|
|
"last_triggered": s.get("attributes", {}).get("last_triggered"),
|
||
|
|
}
|
||
|
|
for s in states
|
||
|
|
if s["entity_id"].startswith("automation.")
|
||
|
|
]
|
||
|
|
return json.dumps({"count": len(automations), "automations": automations}, indent=2)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_trigger_automation",
|
||
|
|
description="Manually trigger an automation.",
|
||
|
|
annotations={"readOnlyHint": False},
|
||
|
|
)
|
||
|
|
async def hass_trigger_automation(params: AutomationInput) -> str:
|
||
|
|
try:
|
||
|
|
await _post(
|
||
|
|
"services/automation/trigger", {"entity_id": params.entity_id}
|
||
|
|
)
|
||
|
|
return json.dumps({"status": "ok", "triggered": params.entity_id})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_list_scenes",
|
||
|
|
description="List all available scenes.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_list_scenes(params: EmptyInput) -> str:
|
||
|
|
try:
|
||
|
|
states = await _get("states")
|
||
|
|
scenes = [
|
||
|
|
{
|
||
|
|
"entity_id": s["entity_id"],
|
||
|
|
"friendly_name": s.get("attributes", {}).get("friendly_name", ""),
|
||
|
|
}
|
||
|
|
for s in states
|
||
|
|
if s["entity_id"].startswith("scene.")
|
||
|
|
]
|
||
|
|
return json.dumps({"count": len(scenes), "scenes": scenes}, indent=2)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_activate_scene",
|
||
|
|
description="Activate a scene.",
|
||
|
|
annotations={"readOnlyHint": False},
|
||
|
|
)
|
||
|
|
async def hass_activate_scene(params: EntityIdInput) -> str:
|
||
|
|
try:
|
||
|
|
await _post("services/scene/turn_on", {"entity_id": params.entity_id})
|
||
|
|
return json.dumps({"status": "ok", "activated": params.entity_id})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools — History & Logbook
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_get_history",
|
||
|
|
description="Get state history for an entity over the last N hours.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_get_history(params: HistoryInput) -> str:
|
||
|
|
try:
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
|
||
|
|
end = datetime.now(timezone.utc)
|
||
|
|
start = end - timedelta(hours=min(params.hours, 72))
|
||
|
|
ts = start.strftime("%Y-%m-%dT%H:%M:%S+00:00")
|
||
|
|
|
||
|
|
async with httpx.AsyncClient(verify=False, timeout=30) as client:
|
||
|
|
resp = await client.get(
|
||
|
|
f"{HASS_URL}/api/history/period/{ts}",
|
||
|
|
headers=_headers(),
|
||
|
|
params={"filter_entity_id": params.entity_id, "minimal_response": ""},
|
||
|
|
)
|
||
|
|
resp.raise_for_status()
|
||
|
|
data = resp.json()
|
||
|
|
|
||
|
|
if not data or not data[0]:
|
||
|
|
return json.dumps({"entity_id": params.entity_id, "history": []})
|
||
|
|
|
||
|
|
entries = [
|
||
|
|
{
|
||
|
|
"state": h.get("state"),
|
||
|
|
"last_changed": h.get("last_changed"),
|
||
|
|
}
|
||
|
|
for h in data[0][:100]
|
||
|
|
]
|
||
|
|
return json.dumps(
|
||
|
|
{"entity_id": params.entity_id, "count": len(entries), "history": entries},
|
||
|
|
indent=2,
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_get_logbook",
|
||
|
|
description="Get logbook entries for the last N hours, optionally filtered by entity.",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_get_logbook(params: LogbookInput) -> str:
|
||
|
|
try:
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
|
||
|
|
end = datetime.now(timezone.utc)
|
||
|
|
start = end - timedelta(hours=min(params.hours, 72))
|
||
|
|
ts = start.strftime("%Y-%m-%dT%H:%M:%S+00:00")
|
||
|
|
|
||
|
|
query_params = {}
|
||
|
|
if params.entity_id:
|
||
|
|
query_params["entity"] = params.entity_id
|
||
|
|
|
||
|
|
async with httpx.AsyncClient(verify=False, timeout=30) as client:
|
||
|
|
resp = await client.get(
|
||
|
|
f"{HASS_URL}/api/logbook/{ts}",
|
||
|
|
headers=_headers(),
|
||
|
|
params=query_params,
|
||
|
|
)
|
||
|
|
resp.raise_for_status()
|
||
|
|
data = resp.json()
|
||
|
|
|
||
|
|
entries = [
|
||
|
|
{
|
||
|
|
"when": e.get("when"),
|
||
|
|
"name": e.get("name"),
|
||
|
|
"message": e.get("message", ""),
|
||
|
|
"entity_id": e.get("entity_id", ""),
|
||
|
|
"state": e.get("state", ""),
|
||
|
|
}
|
||
|
|
for e in data[:100]
|
||
|
|
]
|
||
|
|
return json.dumps({"count": len(entries), "entries": entries}, indent=2)
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools — Services Discovery
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_list_services",
|
||
|
|
description="List available services for a specific domain (e.g. light, climate).",
|
||
|
|
annotations={"readOnlyHint": True},
|
||
|
|
)
|
||
|
|
async def hass_list_services(params: EntitiesByDomainInput) -> str:
|
||
|
|
try:
|
||
|
|
services = await _get("services")
|
||
|
|
for svc in services:
|
||
|
|
if svc.get("domain") == params.domain:
|
||
|
|
service_list = []
|
||
|
|
for name, info in svc.get("services", {}).items():
|
||
|
|
service_list.append(
|
||
|
|
{
|
||
|
|
"service": f"{params.domain}.{name}",
|
||
|
|
"description": info.get("description", ""),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
return json.dumps(
|
||
|
|
{"domain": params.domain, "count": len(service_list), "services": service_list},
|
||
|
|
indent=2,
|
||
|
|
)
|
||
|
|
return json.dumps({"error": f"Domain '{params.domain}' not found"})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools — Events
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_fire_event",
|
||
|
|
description="Fire a custom event on the event bus.",
|
||
|
|
annotations={"readOnlyHint": False, "destructiveHint": False},
|
||
|
|
)
|
||
|
|
async def hass_fire_event(params: EventInput) -> str:
|
||
|
|
try:
|
||
|
|
result = await _post(f"events/{params.event_type}", params.event_data)
|
||
|
|
return json.dumps({"status": "ok", "event_type": params.event_type, "result": result})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools — Notifications
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
class NotifyInput(BaseModel):
|
||
|
|
message: str = Field(description="Notification message text")
|
||
|
|
title: Optional[str] = Field(default=None, description="Optional notification title")
|
||
|
|
target: str = Field(
|
||
|
|
default="notify",
|
||
|
|
description="Notification service target (e.g. 'notify' for default, 'mobile_app_phone' for specific device)",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool(
|
||
|
|
name="hass_send_notification",
|
||
|
|
description="Send a notification via Home Assistant's notification services.",
|
||
|
|
annotations={"readOnlyHint": False},
|
||
|
|
)
|
||
|
|
async def hass_send_notification(params: NotifyInput) -> str:
|
||
|
|
try:
|
||
|
|
payload: dict[str, Any] = {"message": params.message}
|
||
|
|
if params.title:
|
||
|
|
payload["title"] = params.title
|
||
|
|
await _post(f"services/notify/{params.target}", payload)
|
||
|
|
return json.dumps({"status": "ok", "sent_to": params.target})
|
||
|
|
except Exception as e:
|
||
|
|
return _handle_error(e)
|
||
|
|
|
||
|
|
|
||
|
|
# =========================================================================
|
||
|
|
# Entry point
|
||
|
|
# =========================================================================
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
mcp.run()
|