server.py•81.8 kB
#!/usr/bin/env python3
"""
Home Assistant MCP Server
A Model Context Protocol server for Home Assistant integration.
Provides tools to interact with Home Assistant API including:
- Getting states of entities
- Calling services
- Managing automations
- Retrieving system info
"""
import asyncio
import json
import logging
import os
import uuid
import time
from typing import Any, Dict, List, Optional, Union, Set
from urllib.parse import urljoin
from datetime import datetime
import aiohttp
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.lowlevel import NotificationOptions
from mcp.types import (
Resource,
Tool,
TextContent,
ImageContent,
EmbeddedResource,
)
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("homeassistant-mcp")
class HomeAssistantClient:
"""Home Assistant API client"""
def __init__(self, base_url: str, token: str):
self.base_url = base_url.rstrip('/')
self.token = token
self.session: Optional[aiohttp.ClientSession] = None
self.headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
async def __aenter__(self):
self.session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=30)
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Make HTTP request to Home Assistant API"""
if not self.session:
raise RuntimeError("Client not initialized. Use async context manager.")
url = urljoin(self.base_url, endpoint)
try:
async with self.session.request(
method, url, headers=self.headers, **kwargs
) as response:
response.raise_for_status()
return await response.json()
except aiohttp.ClientError as e:
logger.error(f"HTTP request failed: {e}")
raise
except Exception as e:
logger.error(f"Request failed: {e}")
raise
async def get_states(self) -> List[Dict[str, Any]]:
"""Get all entity states"""
return await self._request("GET", "/api/states")
async def get_state(self, entity_id: str) -> Dict[str, Any]:
"""Get state of specific entity"""
return await self._request("GET", f"/api/states/{entity_id}")
async def call_service(self, domain: str, service: str, service_data: Optional[Dict] = None) -> List[Dict[str, Any]]:
"""Call a Home Assistant service"""
endpoint = f"/api/services/{domain}/{service}"
data = service_data or {}
return await self._request("POST", endpoint, json=data)
async def get_config(self) -> Dict[str, Any]:
"""Get Home Assistant configuration"""
return await self._request("GET", "/api/config")
async def get_services(self) -> List[Dict[str, Any]]:
"""Get all available services"""
return await self._request("GET", "/api/services")
async def get_events(self) -> List[Dict[str, Any]]:
"""Get event types"""
return await self._request("GET", "/api/events")
async def fire_event(self, event_type: str, event_data: Optional[Dict] = None) -> Dict[str, Any]:
"""Fire an event"""
endpoint = f"/api/events/{event_type}"
data = event_data or {}
return await self._request("POST", endpoint, json=data)
async def get_history(self, start_time: Optional[str] = None, end_time: Optional[str] = None,
filter_entity_id: Optional[str] = None, minimal_response: bool = False,
no_attributes: bool = False, significant_changes_only: bool = False) -> List[Dict[str, Any]]:
"""Get historical data"""
if start_time:
endpoint = f"/api/history/period/{start_time}"
else:
endpoint = "/api/history/period"
params = {}
if end_time:
params["end_time"] = end_time
if filter_entity_id:
params["filter_entity_id"] = filter_entity_id
if minimal_response:
params["minimal_response"] = "true"
if no_attributes:
params["no_attributes"] = "true"
if significant_changes_only:
params["significant_changes_only"] = "true"
return await self._request("GET", endpoint, params=params)
async def get_logbook(self, start_time: Optional[str] = None, end_time: Optional[str] = None,
entity: Optional[str] = None) -> List[Dict[str, Any]]:
"""Get logbook entries"""
if start_time:
endpoint = f"/api/logbook/{start_time}"
else:
endpoint = "/api/logbook"
params = {}
if end_time:
params["end_time"] = end_time
if entity:
params["entity"] = entity
return await self._request("GET", endpoint, params=params)
async def get_error_log(self) -> str:
"""Get error log"""
response = await self._request("GET", "/api/error_log")
return response if isinstance(response, str) else str(response)
async def get_calendars(self) -> List[Dict[str, Any]]:
"""Get list of calendar entities"""
return await self._request("GET", "/api/calendars")
async def get_calendar_events(self, calendar_id: str, start: str, end: str) -> List[Dict[str, Any]]:
"""Get calendar events for a specific calendar"""
endpoint = f"/api/calendars/{calendar_id}"
params = {"start": start, "end": end}
return await self._request("GET", endpoint, params=params)
async def get_camera_proxy(self, camera_entity_id: str) -> bytes:
"""Get camera image data"""
endpoint = f"/api/camera_proxy/{camera_entity_id}"
# This returns binary data, so we need special handling
if not self.session:
raise RuntimeError("Client not initialized. Use async context manager.")
url = urljoin(self.base_url, endpoint)
async with self.session.get(url, headers=self.headers) as response:
response.raise_for_status()
return await response.read()
async def set_state(self, entity_id: str, state: str, attributes: Optional[Dict] = None) -> Dict[str, Any]:
"""Set or update entity state"""
endpoint = f"/api/states/{entity_id}"
data = {"state": state}
if attributes:
data["attributes"] = attributes
return await self._request("POST", endpoint, json=data)
async def delete_state(self, entity_id: str) -> Dict[str, Any]:
"""Delete entity state"""
endpoint = f"/api/states/{entity_id}"
return await self._request("DELETE", endpoint)
async def render_template(self, template: str) -> str:
"""Render a Home Assistant template"""
endpoint = "/api/template"
data = {"template": template}
result = await self._request("POST", endpoint, json=data)
return result if isinstance(result, str) else str(result)
async def check_config(self) -> Dict[str, Any]:
"""Check configuration validity"""
return await self._request("POST", "/api/config/core/check_config")
async def handle_intent(self, intent_name: str, intent_data: Optional[Dict] = None) -> Dict[str, Any]:
"""Handle an intent"""
endpoint = "/api/intent/handle"
data = {"name": intent_name}
if intent_data:
data["data"] = intent_data
return await self._request("POST", endpoint, json=data)
async def call_service_with_response(self, domain: str, service: str, service_data: Optional[Dict] = None) -> Dict[str, Any]:
"""Call service and get response data"""
endpoint = f"/api/services/{domain}/{service}?return_response"
data = service_data or {}
return await self._request("POST", endpoint, json=data)
async def subscribe_to_events(self) -> aiohttp.ClientResponse:
"""Subscribe to Home Assistant event stream"""
if not self.session:
raise RuntimeError("Client not initialized. Use async context manager.")
url = urljoin(self.base_url, "/api/stream")
headers = {"Authorization": f"Bearer {self.token}", "Accept": "text/event-stream"}
return await self.session.get(url, headers=headers)
async def get_automations(self) -> List[Dict[str, Any]]:
"""Get all automations"""
states = await self._request("GET", "/api/states")
automations = [state for state in states if state["entity_id"].startswith("automation.")]
return automations
async def get_automation(self, automation_id: str) -> Dict[str, Any]:
"""Get specific automation details"""
return await self._request("GET", f"/api/states/{automation_id}")
async def toggle_automation(self, automation_id: str) -> List[Dict[str, Any]]:
"""Toggle an automation on/off"""
return await self.call_service("automation", "toggle", {"entity_id": automation_id})
async def turn_on_automation(self, automation_id: str) -> List[Dict[str, Any]]:
"""Turn on an automation"""
return await self.call_service("automation", "turn_on", {"entity_id": automation_id})
async def turn_off_automation(self, automation_id: str) -> List[Dict[str, Any]]:
"""Turn off an automation"""
return await self.call_service("automation", "turn_off", {"entity_id": automation_id})
async def trigger_automation(self, automation_id: str) -> List[Dict[str, Any]]:
"""Manually trigger an automation"""
return await self.call_service("automation", "trigger", {"entity_id": automation_id})
async def reload_automations(self) -> List[Dict[str, Any]]:
"""Reload all automations"""
return await self.call_service("automation", "reload")
async def create_automation(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new automation via config entry"""
# This creates an automation by writing to the automations.yaml file
# Note: This requires the automation integration to be set up to read from files
endpoint = "/api/config/automation/config"
# Get existing automations
try:
existing = await self._request("GET", endpoint)
if not isinstance(existing, list):
existing = []
except:
existing = []
# Add new automation
existing.append(config)
# Update the configuration
return await self._request("POST", endpoint, json=existing)
async def update_automation(self, automation_id: str, config: Dict[str, Any]) -> Dict[str, Any]:
"""Update an existing automation"""
endpoint = f"/api/config/automation/config/{automation_id}"
return await self._request("POST", endpoint, json=config)
async def delete_automation(self, automation_id: str) -> Dict[str, Any]:
"""Delete an automation"""
endpoint = f"/api/config/automation/config/{automation_id}"
return await self._request("DELETE", endpoint)
async def get_automation_trace(self, automation_id: str, run_id: Optional[str] = None) -> Dict[str, Any]:
"""Get automation trace information"""
if run_id:
endpoint = f"/api/trace/automation/{automation_id}/get/{run_id}"
else:
endpoint = f"/api/trace/automation/{automation_id}"
return await self._request("GET", endpoint)
async def get_scenes(self) -> List[Dict[str, Any]]:
"""Get all scenes"""
states = await self._request("GET", "/api/states")
scenes = [state for state in states if state["entity_id"].startswith("scene.")]
return scenes
async def activate_scene(self, scene_id: str) -> List[Dict[str, Any]]:
"""Activate a scene"""
return await self.call_service("scene", "turn_on", {"entity_id": scene_id})
async def create_scene(self, scene_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Create a new scene"""
return await self.call_service("scene", "create", scene_data)
# Area Management
async def get_areas(self) -> List[Dict[str, Any]]:
"""Get all areas/zones"""
return await self._request("GET", "/api/config/area_registry")
async def create_area(self, name: str, aliases: Optional[List[str]] = None) -> Dict[str, Any]:
"""Create a new area"""
data = {"name": name}
if aliases:
data["aliases"] = aliases
return await self._request("POST", "/api/config/area_registry", json=data)
async def update_area(self, area_id: str, name: str, aliases: Optional[List[str]] = None) -> Dict[str, Any]:
"""Update an existing area"""
data = {"name": name}
if aliases:
data["aliases"] = aliases
return await self._request("POST", f"/api/config/area_registry/{area_id}", json=data)
async def delete_area(self, area_id: str) -> Dict[str, Any]:
"""Delete an area"""
return await self._request("DELETE", f"/api/config/area_registry/{area_id}")
# Device Management
async def get_devices(self) -> List[Dict[str, Any]]:
"""Get all devices"""
return await self._request("GET", "/api/config/device_registry")
async def get_device(self, device_id: str) -> Dict[str, Any]:
"""Get specific device information"""
devices = await self.get_devices()
for device in devices:
if device.get("id") == device_id:
return device
raise ValueError(f"Device {device_id} not found")
async def update_device(self, device_id: str, name: Optional[str] = None, area_id: Optional[str] = None,
disabled_by: Optional[str] = None) -> Dict[str, Any]:
"""Update device configuration"""
data = {}
if name is not None:
data["name_by_user"] = name
if area_id is not None:
data["area_id"] = area_id
if disabled_by is not None:
data["disabled_by"] = disabled_by
return await self._request("POST", f"/api/config/device_registry/{device_id}", json=data)
async def get_entities_by_area(self, area_id: str) -> List[Dict[str, Any]]:
"""Get all entities in a specific area"""
entity_registry = await self._request("GET", "/api/config/entity_registry")
device_registry = await self.get_devices()
# Get devices in the area
area_devices = [d["id"] for d in device_registry if d.get("area_id") == area_id]
# Get entities from those devices or directly assigned to area
area_entities = []
for entity in entity_registry:
if (entity.get("area_id") == area_id or
entity.get("device_id") in area_devices):
area_entities.append(entity)
return area_entities
# System Management
async def restart_homeassistant(self) -> Dict[str, Any]:
"""Restart Home Assistant"""
return await self.call_service("homeassistant", "restart")
async def stop_homeassistant(self) -> Dict[str, Any]:
"""Stop Home Assistant"""
return await self.call_service("homeassistant", "stop")
async def check_config_valid(self) -> Dict[str, Any]:
"""Check if configuration is valid"""
return await self._request("POST", "/api/config/core/check_config")
async def get_system_health(self) -> Dict[str, Any]:
"""Get system health information"""
try:
return await self._request("GET", "/api/system_health/info")
except:
# Fallback if system_health is not available
return {"status": "System health API not available"}
async def get_supervisor_info(self) -> Dict[str, Any]:
"""Get supervisor information (if available)"""
try:
return await self._request("GET", "/api/hassio/supervisor/info")
except:
return {"error": "Supervisor not available (not running Home Assistant OS/Supervised)"}
async def get_system_info(self) -> Dict[str, Any]:
"""Get system information"""
try:
return await self._request("GET", "/api/hassio/host/info")
except:
# Return basic config info if supervisor not available
config = await self.get_config()
return {
"host_info": "Not available (not running Home Assistant OS)",
"version": config.get("version"),
"installation_type": config.get("installation_type", "unknown")
}
# Integration Management
async def get_integrations(self) -> List[Dict[str, Any]]:
"""Get all configured integrations"""
return await self._request("GET", "/api/config/config_entries")
async def reload_integration(self, integration_domain: str) -> Dict[str, Any]:
"""Reload a specific integration"""
return await self.call_service(integration_domain, "reload")
async def delete_integration(self, config_entry_id: str) -> Dict[str, Any]:
"""Delete/remove an integration"""
return await self._request("DELETE", f"/api/config/config_entries/{config_entry_id}")
async def disable_integration(self, config_entry_id: str) -> Dict[str, Any]:
"""Disable an integration"""
return await self._request("POST", f"/api/config/config_entries/{config_entry_id}/disable")
async def enable_integration(self, config_entry_id: str) -> Dict[str, Any]:
"""Enable an integration"""
return await self._request("POST", f"/api/config/config_entries/{config_entry_id}/enable")
async def get_integration_info(self, config_entry_id: str) -> Dict[str, Any]:
"""Get information about a specific integration"""
integrations = await self.get_integrations()
for integration in integrations:
if integration.get("entry_id") == config_entry_id:
return integration
raise ValueError(f"Integration {config_entry_id} not found")
# Notification Services
async def send_notification(self, message: str, title: Optional[str] = None,
target: Optional[str] = None, data: Optional[Dict] = None) -> List[Dict[str, Any]]:
"""Send a persistent notification"""
service_data = {"message": message}
if title:
service_data["title"] = title
if data:
service_data.update(data)
# Use notify service if target specified, otherwise persistent notification
if target:
return await self.call_service("notify", target, service_data)
else:
return await self.call_service("persistent_notification", "create", service_data)
async def get_notification_services(self) -> List[str]:
"""Get available notification services"""
services = await self.get_services()
notify_services = []
for domain, domain_services in services.items():
if domain == "notify":
notify_services.extend(domain_services.keys())
return notify_services
async def dismiss_notification(self, notification_id: str) -> List[Dict[str, Any]]:
"""Dismiss a persistent notification"""
return await self.call_service("persistent_notification", "dismiss",
{"notification_id": notification_id})
# Entity Registry Management
async def get_entity_registry(self) -> List[Dict[str, Any]]:
"""Get entity registry information"""
return await self._request("GET", "/api/config/entity_registry")
async def update_entity_registry(self, entity_id: str, name: Optional[str] = None,
disabled_by: Optional[str] = None, area_id: Optional[str] = None) -> Dict[str, Any]:
"""Update entity registry entry"""
data = {}
if name is not None:
data["name"] = name
if disabled_by is not None:
data["disabled_by"] = disabled_by
if area_id is not None:
data["area_id"] = area_id
return await self._request("POST", f"/api/config/entity_registry/{entity_id}", json=data)
async def enable_entity(self, entity_id: str) -> Dict[str, Any]:
"""Enable an entity"""
return await self.update_entity_registry(entity_id, disabled_by=None)
async def disable_entity(self, entity_id: str) -> Dict[str, Any]:
"""Disable an entity"""
return await self.update_entity_registry(entity_id, disabled_by="user")
class SSESubscription:
"""Represents a single SSE subscription"""
def __init__(self, client_id: str, events: Set[str] = None, entity_id: str = None, domain: str = None):
self.client_id = client_id
self.events = events or set()
self.entity_id = entity_id
self.domain = domain
self.created_at = datetime.now()
self.last_activity = time.time()
def matches_event(self, event_type: str, entity_id: str = None) -> bool:
"""Check if this subscription should receive the event"""
# Check if event type matches
if self.events and event_type not in self.events:
return False
# Check entity filter
if self.entity_id and entity_id != self.entity_id:
return False
# Check domain filter
if self.domain and entity_id and not entity_id.startswith(f"{self.domain}."):
return False
return True
class SSEManager:
"""Manages SSE connections and subscriptions"""
def __init__(self):
self.subscriptions: Dict[str, SSESubscription] = {}
self.active_connections: Dict[str, aiohttp.ClientResponse] = {}
self.connection_tasks: Dict[str, asyncio.Task] = {}
self.max_connections = 100
self.rate_limit = 1000 # requests per minute
self.rate_counters: Dict[str, List[float]] = {}
self.ping_interval = 30 # seconds
def add_subscription(self, events: Set[str] = None, entity_id: str = None, domain: str = None) -> str:
"""Add a new subscription and return client ID"""
if len(self.subscriptions) >= self.max_connections:
raise ValueError("Maximum connections exceeded")
client_id = str(uuid.uuid4())
self.subscriptions[client_id] = SSESubscription(client_id, events, entity_id, domain)
self.rate_counters[client_id] = []
logger.info(f"Added SSE subscription {client_id}")
return client_id
def remove_subscription(self, client_id: str):
"""Remove a subscription"""
self.subscriptions.pop(client_id, None)
self.rate_counters.pop(client_id, None)
# Clean up connection if exists
if client_id in self.active_connections:
self.active_connections[client_id].close()
self.active_connections.pop(client_id, None)
if client_id in self.connection_tasks:
self.connection_tasks[client_id].cancel()
self.connection_tasks.pop(client_id, None)
logger.info(f"Removed SSE subscription {client_id}")
def check_rate_limit(self, client_id: str) -> bool:
"""Check if client is within rate limit"""
now = time.time()
if client_id not in self.rate_counters:
self.rate_counters[client_id] = []
# Clean old requests (older than 1 minute)
self.rate_counters[client_id] = [
timestamp for timestamp in self.rate_counters[client_id]
if now - timestamp < 60
]
# Check if under limit
if len(self.rate_counters[client_id]) >= self.rate_limit:
return False
# Add current request
self.rate_counters[client_id].append(now)
return True
def get_stats(self) -> Dict[str, Any]:
"""Get current SSE statistics"""
return {
"active_connections": len(self.subscriptions),
"max_connections": self.max_connections,
"rate_limit_per_minute": self.rate_limit,
"subscriptions": [
{
"client_id": sub.client_id,
"events": list(sub.events),
"entity_id": sub.entity_id,
"domain": sub.domain,
"created_at": sub.created_at.isoformat(),
}
for sub in self.subscriptions.values()
]
}
async def start_event_stream(self, client_id: str, ha_url: str, ha_token: str):
"""Start event stream for a client"""
if client_id not in self.subscriptions:
return
async def event_stream_task():
try:
async with HomeAssistantClient(ha_url, ha_token) as client:
response = await client.subscribe_to_events()
self.active_connections[client_id] = response
async for line in response.content:
if client_id not in self.subscriptions:
break
line = line.decode('utf-8').strip()
if not line:
continue
# Parse SSE format
if line.startswith('data: '):
try:
event_data = json.loads(line[6:])
await self._handle_ha_event(client_id, event_data)
except json.JSONDecodeError:
continue
except Exception as e:
logger.error(f"Error in event stream for {client_id}: {e}")
finally:
self.remove_subscription(client_id)
self.connection_tasks[client_id] = asyncio.create_task(event_stream_task())
async def _handle_ha_event(self, client_id: str, ha_event: Dict[str, Any]):
"""Handle Home Assistant event and forward if subscription matches"""
if client_id not in self.subscriptions:
return
subscription = self.subscriptions[client_id]
event_type = ha_event.get('event_type', '')
# Extract entity_id from event data if available
entity_id = None
if 'data' in ha_event:
if 'entity_id' in ha_event['data']:
entity_id = ha_event['data']['entity_id']
elif 'new_state' in ha_event['data'] and ha_event['data']['new_state']:
entity_id = ha_event['data']['new_state'].get('entity_id')
if subscription.matches_event(event_type, entity_id):
# Transform HA event to SSE format
sse_event = self._transform_to_sse_event(ha_event)
# In a real implementation, you'd send this to the client
# For MCP, we'll store recent events that can be retrieved
logger.info(f"Event for {client_id}: {sse_event['type']}")
def _transform_to_sse_event(self, ha_event: Dict[str, Any]) -> Dict[str, Any]:
"""Transform Home Assistant event to SSE format"""
event_type = ha_event.get('event_type', '')
if event_type == 'state_changed':
new_state = ha_event['data'].get('new_state', {})
return {
"type": "state_changed",
"data": {
"entity_id": new_state.get('entity_id'),
"state": new_state.get('state'),
"attributes": new_state.get('attributes', {}),
"last_changed": new_state.get('last_changed'),
"last_updated": new_state.get('last_updated')
},
"timestamp": datetime.now().isoformat()
}
elif event_type == 'service_called':
return {
"type": "service_called",
"data": ha_event['data'],
"timestamp": datetime.now().isoformat()
}
elif event_type == 'automation_triggered':
return {
"type": "automation_triggered",
"data": ha_event['data'],
"timestamp": datetime.now().isoformat()
}
elif event_type == 'script_started':
return {
"type": "script_executed",
"data": ha_event['data'],
"timestamp": datetime.now().isoformat()
}
else:
return {
"type": event_type,
"data": ha_event.get('data', {}),
"timestamp": datetime.now().isoformat()
}
# Initialize SSE manager
sse_manager = SSEManager()
# Initialize MCP server
server = Server("homeassistant-mcp")
# Configuration
HA_URL = os.getenv("HA_URL", "http://homeassistant.local:8123")
HA_TOKEN = os.getenv("HA_TOKEN", "")
if not HA_TOKEN:
logger.warning("HA_TOKEN not set. You'll need to provide it via environment variable.")
@server.list_resources()
async def handle_list_resources() -> List[Resource]:
"""List available Home Assistant resources"""
return [
Resource(
uri="homeassistant://states",
name="All Entity States",
description="Current state of all Home Assistant entities",
mimeType="application/json",
),
Resource(
uri="homeassistant://config",
name="Home Assistant Configuration",
description="Home Assistant system configuration",
mimeType="application/json",
),
Resource(
uri="homeassistant://services",
name="Available Services",
description="List of all available Home Assistant services",
mimeType="application/json",
),
Resource(
uri="homeassistant://events",
name="Event Types",
description="List of available event types",
mimeType="application/json",
),
Resource(
uri="homeassistant://sse",
name="SSE Events",
description="Subscribe to real-time Home Assistant events",
mimeType="text/event-stream",
)
]
@server.read_resource()
async def handle_read_resource(uri: str) -> str:
"""Read Home Assistant resource data"""
if not HA_TOKEN:
return json.dumps({"error": "HA_TOKEN not configured"})
try:
async with HomeAssistantClient(HA_URL, HA_TOKEN) as client:
if uri == "homeassistant://states":
data = await client.get_states()
elif uri == "homeassistant://config":
data = await client.get_config()
elif uri == "homeassistant://services":
data = await client.get_services()
elif uri == "homeassistant://events":
data = await client.get_events()
else:
return json.dumps({"error": f"Unknown resource: {uri}"})
return json.dumps(data, indent=2)
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
return json.dumps({"error": str(e)})
@server.list_tools()
async def handle_list_tools() -> List[Tool]:
"""List available Home Assistant tools"""
return [
Tool(
name="get_entity_state",
description="Get the current state of a Home Assistant entity",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Entity ID (e.g., light.living_room, sensor.temperature)"
}
},
"required": ["entity_id"]
}
),
Tool(
name="call_service",
description="Call a Home Assistant service",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Service domain (e.g., light, switch, automation)"
},
"service": {
"type": "string",
"description": "Service name (e.g., turn_on, turn_off, toggle)"
},
"entity_id": {
"type": "string",
"description": "Target entity ID (optional)"
},
"service_data": {
"type": "object",
"description": "Additional service data (optional)"
}
},
"required": ["domain", "service"]
}
),
Tool(
name="search_entities",
description="Search for entities by name, domain, or state",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query (entity name, domain, or state)"
},
"domain": {
"type": "string",
"description": "Filter by domain (optional)"
},
"state": {
"type": "string",
"description": "Filter by state (optional)"
}
},
"required": ["query"]
}
),
Tool(
name="get_area_entities",
description="Get all entities in a specific area",
inputSchema={
"type": "object",
"properties": {
"area_name": {
"type": "string",
"description": "Name of the area"
}
},
"required": ["area_name"]
}
),
Tool(
name="fire_event",
description="Fire a custom Home Assistant event",
inputSchema={
"type": "object",
"properties": {
"event_type": {
"type": "string",
"description": "Type of event to fire"
},
"event_data": {
"type": "object",
"description": "Event data payload (optional)"
}
},
"required": ["event_type"]
}
),
Tool(
name="get_history",
description="Get historical data for entities with advanced filtering options",
inputSchema={
"type": "object",
"properties": {
"entity_ids": {
"type": "array",
"items": {"type": "string"},
"description": "List of entity IDs to get history for"
},
"start_time": {
"type": "string",
"description": "Start time (ISO format, optional)"
},
"end_time": {
"type": "string",
"description": "End time (ISO format, optional)"
},
"minimal_response": {
"type": "boolean",
"description": "Return minimal response for faster queries",
"default": False
},
"no_attributes": {
"type": "boolean",
"description": "Skip attributes in response for faster queries",
"default": False
},
"significant_changes_only": {
"type": "boolean",
"description": "Only return significant state changes",
"default": False
}
},
"required": ["entity_ids"]
}
),
Tool(
name="get_logbook",
description="Get logbook entries (event history)",
inputSchema={
"type": "object",
"properties": {
"start_time": {
"type": "string",
"description": "Start time (ISO format, optional)"
},
"end_time": {
"type": "string",
"description": "End time (ISO format, optional)"
},
"entity": {
"type": "string",
"description": "Filter to specific entity (optional)"
}
},
"required": []
}
),
Tool(
name="set_state",
description="Set or update the state of an entity (does not control devices, use call_service for that)",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Entity ID to set state for"
},
"state": {
"type": "string",
"description": "New state value"
},
"attributes": {
"type": "object",
"description": "Entity attributes (optional)"
}
},
"required": ["entity_id", "state"]
}
),
Tool(
name="delete_state",
description="Delete an entity state",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Entity ID to delete"
}
},
"required": ["entity_id"]
}
),
Tool(
name="render_template",
description="Render a Home Assistant template",
inputSchema={
"type": "object",
"properties": {
"template": {
"type": "string",
"description": "Template string to render (e.g., 'The temperature is {{ states(\"sensor.temperature\") }}')"
}
},
"required": ["template"]
}
),
Tool(
name="check_config",
description="Check Home Assistant configuration validity",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_error_log",
description="Retrieve Home Assistant error log",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_calendars",
description="Get list of calendar entities",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_calendar_events",
description="Get events from a specific calendar",
inputSchema={
"type": "object",
"properties": {
"calendar_id": {
"type": "string",
"description": "Calendar entity ID (e.g., calendar.personal)"
},
"start": {
"type": "string",
"description": "Start timestamp (ISO format)"
},
"end": {
"type": "string",
"description": "End timestamp (ISO format)"
}
},
"required": ["calendar_id", "start", "end"]
}
),
Tool(
name="get_camera_image",
description="Get image from a camera entity (returns base64 encoded image)",
inputSchema={
"type": "object",
"properties": {
"camera_entity_id": {
"type": "string",
"description": "Camera entity ID (e.g., camera.front_door)"
}
},
"required": ["camera_entity_id"]
}
),
Tool(
name="call_service_with_response",
description="Call a service that returns response data (like weather forecasts)",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Service domain"
},
"service": {
"type": "string",
"description": "Service name"
},
"service_data": {
"type": "object",
"description": "Service data payload"
}
},
"required": ["domain", "service"]
}
),
Tool(
name="handle_intent",
description="Handle a Home Assistant intent",
inputSchema={
"type": "object",
"properties": {
"intent_name": {
"type": "string",
"description": "Intent name (e.g., SetTimer, GetWeather)"
},
"intent_data": {
"type": "object",
"description": "Intent data (optional)"
}
},
"required": ["intent_name"]
}
),
Tool(
name="subscribe_events",
description="Subscribe to real-time Home Assistant events via SSE",
inputSchema={
"type": "object",
"properties": {
"events": {
"type": "array",
"items": {"type": "string"},
"description": "Event types to subscribe to (e.g., ['state_changed', 'service_called'])",
"default": []
},
"entity_id": {
"type": "string",
"description": "Filter events to specific entity (optional)"
},
"domain": {
"type": "string",
"description": "Filter events to specific domain (optional)"
}
},
"required": []
}
),
Tool(
name="get_sse_stats",
description="Get current SSE connection statistics",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_automations",
description="Get list of all automations",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_automation",
description="Get details of a specific automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "Automation entity ID (e.g., automation.morning_routine)"
}
},
"required": ["automation_id"]
}
),
Tool(
name="toggle_automation",
description="Toggle an automation on/off",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "Automation entity ID to toggle"
}
},
"required": ["automation_id"]
}
),
Tool(
name="turn_on_automation",
description="Turn on an automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "Automation entity ID to turn on"
}
},
"required": ["automation_id"]
}
),
Tool(
name="turn_off_automation",
description="Turn off an automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "Automation entity ID to turn off"
}
},
"required": ["automation_id"]
}
),
Tool(
name="trigger_automation",
description="Manually trigger an automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "Automation entity ID to trigger"
}
},
"required": ["automation_id"]
}
),
Tool(
name="create_automation",
description="Create a new automation",
inputSchema={
"type": "object",
"properties": {
"config": {
"type": "object",
"description": "Automation configuration with alias, trigger, condition, and action",
"properties": {
"alias": {
"type": "string",
"description": "Human readable name for the automation"
},
"description": {
"type": "string",
"description": "Description of what the automation does"
},
"trigger": {
"type": "object",
"description": "Trigger configuration (e.g., time, state, event)"
},
"condition": {
"type": "object",
"description": "Condition configuration (optional)"
},
"action": {
"type": "object",
"description": "Action to perform when triggered"
}
},
"required": ["alias", "trigger", "action"]
}
},
"required": ["config"]
}
),
Tool(
name="update_automation",
description="Update an existing automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "Automation entity ID to update"
},
"config": {
"type": "object",
"description": "Updated automation configuration"
}
},
"required": ["automation_id", "config"]
}
),
Tool(
name="delete_automation",
description="Delete an automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "Automation entity ID to delete"
}
},
"required": ["automation_id"]
}
),
Tool(
name="reload_automations",
description="Reload all automations from configuration",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_automation_trace",
description="Get automation execution trace information",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "Automation entity ID to get trace for"
},
"run_id": {
"type": "string",
"description": "Specific run ID to get trace for (optional)"
}
},
"required": ["automation_id"]
}
),
Tool(
name="get_scenes",
description="Get list of all scenes",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="activate_scene",
description="Activate a scene",
inputSchema={
"type": "object",
"properties": {
"scene_id": {
"type": "string",
"description": "Scene entity ID (e.g., scene.movie_time)"
}
},
"required": ["scene_id"]
}
),
Tool(
name="create_scene",
description="Create a new scene from current entity states",
inputSchema={
"type": "object",
"properties": {
"scene_data": {
"type": "object",
"description": "Scene configuration with scene_id and entities",
"properties": {
"scene_id": {
"type": "string",
"description": "ID for the new scene"
},
"entities": {
"type": "object",
"description": "Entity states to capture in the scene"
}
},
"required": ["scene_id", "entities"]
}
},
"required": ["scene_data"]
}
),
# Area Management Tools
Tool(
name="get_areas",
description="Get list of all areas/zones in Home Assistant",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="create_area",
description="Create a new area/zone",
inputSchema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Name of the area to create"
},
"aliases": {
"type": "array",
"items": {"type": "string"},
"description": "Optional aliases for the area"
}
},
"required": ["name"]
}
),
Tool(
name="update_area",
description="Update an existing area",
inputSchema={
"type": "object",
"properties": {
"area_id": {
"type": "string",
"description": "ID of the area to update"
},
"name": {
"type": "string",
"description": "New name for the area"
},
"aliases": {
"type": "array",
"items": {"type": "string"},
"description": "Updated aliases for the area"
}
},
"required": ["area_id", "name"]
}
),
Tool(
name="delete_area",
description="Delete an area/zone",
inputSchema={
"type": "object",
"properties": {
"area_id": {
"type": "string",
"description": "ID of the area to delete"
}
},
"required": ["area_id"]
}
),
# Device Management Tools
Tool(
name="get_devices",
description="Get list of all devices in Home Assistant",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_device",
description="Get information about a specific device",
inputSchema={
"type": "object",
"properties": {
"device_id": {
"type": "string",
"description": "ID of the device to get information for"
}
},
"required": ["device_id"]
}
),
Tool(
name="update_device",
description="Update device configuration (name, area assignment, etc.)",
inputSchema={
"type": "object",
"properties": {
"device_id": {
"type": "string",
"description": "ID of the device to update"
},
"name": {
"type": "string",
"description": "New name for the device"
},
"area_id": {
"type": "string",
"description": "Area ID to assign device to"
},
"disabled_by": {
"type": "string",
"description": "Disable device (set to 'user' to disable, null to enable)"
}
},
"required": ["device_id"]
}
),
Tool(
name="get_entities_by_area",
description="Get all entities in a specific area",
inputSchema={
"type": "object",
"properties": {
"area_id": {
"type": "string",
"description": "ID of the area to get entities for"
}
},
"required": ["area_id"]
}
),
# System Management Tools
Tool(
name="restart_homeassistant",
description="Restart Home Assistant system",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="stop_homeassistant",
description="Stop Home Assistant system",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="check_config_valid",
description="Check if Home Assistant configuration is valid",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_system_health",
description="Get system health information",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_supervisor_info",
description="Get Home Assistant supervisor information (if available)",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_system_info",
description="Get system/host information",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
# Integration Management Tools
Tool(
name="get_integrations",
description="Get list of all configured integrations",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="reload_integration",
description="Reload a specific integration",
inputSchema={
"type": "object",
"properties": {
"integration_domain": {
"type": "string",
"description": "Domain of the integration to reload (e.g., 'zwave_js', 'mqtt')"
}
},
"required": ["integration_domain"]
}
),
Tool(
name="disable_integration",
description="Disable an integration",
inputSchema={
"type": "object",
"properties": {
"config_entry_id": {
"type": "string",
"description": "Config entry ID of the integration to disable"
}
},
"required": ["config_entry_id"]
}
),
Tool(
name="enable_integration",
description="Enable a disabled integration",
inputSchema={
"type": "object",
"properties": {
"config_entry_id": {
"type": "string",
"description": "Config entry ID of the integration to enable"
}
},
"required": ["config_entry_id"]
}
),
Tool(
name="delete_integration",
description="Delete/remove an integration",
inputSchema={
"type": "object",
"properties": {
"config_entry_id": {
"type": "string",
"description": "Config entry ID of the integration to delete"
}
},
"required": ["config_entry_id"]
}
),
Tool(
name="get_integration_info",
description="Get detailed information about a specific integration",
inputSchema={
"type": "object",
"properties": {
"config_entry_id": {
"type": "string",
"description": "Config entry ID of the integration"
}
},
"required": ["config_entry_id"]
}
),
# Notification Services Tools
Tool(
name="send_notification",
description="Send a notification via Home Assistant notification services",
inputSchema={
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "Notification message content"
},
"title": {
"type": "string",
"description": "Optional notification title"
},
"target": {
"type": "string",
"description": "Target notification service (e.g., 'mobile_app_phone', 'persistent_notification')"
},
"data": {
"type": "object",
"description": "Additional notification data (optional)"
}
},
"required": ["message"]
}
),
Tool(
name="get_notification_services",
description="Get list of available notification services",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="dismiss_notification",
description="Dismiss a persistent notification",
inputSchema={
"type": "object",
"properties": {
"notification_id": {
"type": "string",
"description": "ID of the notification to dismiss"
}
},
"required": ["notification_id"]
}
),
# Entity Registry Management Tools
Tool(
name="get_entity_registry",
description="Get entity registry information for all entities",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="update_entity_registry",
description="Update entity registry entry (name, area, enabled/disabled status)",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Entity ID to update"
},
"name": {
"type": "string",
"description": "New name for the entity"
},
"area_id": {
"type": "string",
"description": "Area ID to assign entity to"
},
"disabled_by": {
"type": "string",
"description": "Disable entity (set to 'user' to disable, null to enable)"
}
},
"required": ["entity_id"]
}
),
Tool(
name="enable_entity",
description="Enable a disabled entity",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Entity ID to enable"
}
},
"required": ["entity_id"]
}
),
Tool(
name="disable_entity",
description="Disable an entity",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Entity ID to disable"
}
},
"required": ["entity_id"]
}
)
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls"""
if not HA_TOKEN:
return [TextContent(
type="text",
text=json.dumps({"error": "HA_TOKEN not configured"})
)]
try:
async with HomeAssistantClient(HA_URL, HA_TOKEN) as client:
if name == "get_entity_state":
entity_id = arguments["entity_id"]
result = await client.get_state(entity_id)
elif name == "call_service":
domain = arguments["domain"]
service = arguments["service"]
service_data = arguments.get("service_data", {})
# Add entity_id to service_data if provided
if "entity_id" in arguments:
service_data["entity_id"] = arguments["entity_id"]
result = await client.call_service(domain, service, service_data)
elif name == "search_entities":
all_states = await client.get_states()
query = arguments["query"].lower()
domain_filter = arguments.get("domain")
state_filter = arguments.get("state")
filtered_entities = []
for entity in all_states:
# Check if query matches entity_id or friendly_name
matches_query = (
query in entity["entity_id"].lower() or
query in entity.get("attributes", {}).get("friendly_name", "").lower()
)
# Apply domain filter
if domain_filter and not entity["entity_id"].startswith(f"{domain_filter}."):
continue
# Apply state filter
if state_filter and entity["state"].lower() != state_filter.lower():
continue
if matches_query:
filtered_entities.append(entity)
result = filtered_entities
elif name == "get_area_entities":
# This would require additional API calls to get area information
# For now, return a placeholder
result = {"message": "Area entity lookup requires additional implementation"}
elif name == "fire_event":
event_type = arguments["event_type"]
event_data = arguments.get("event_data")
result = await client.fire_event(event_type, event_data)
elif name == "get_history":
entity_ids = arguments["entity_ids"]
start_time = arguments.get("start_time")
end_time = arguments.get("end_time")
minimal_response = arguments.get("minimal_response", False)
no_attributes = arguments.get("no_attributes", False)
significant_changes_only = arguments.get("significant_changes_only", False)
result = await client.get_history(
start_time=start_time,
end_time=end_time,
filter_entity_id=",".join(entity_ids),
minimal_response=minimal_response,
no_attributes=no_attributes,
significant_changes_only=significant_changes_only
)
elif name == "get_logbook":
start_time = arguments.get("start_time")
end_time = arguments.get("end_time")
entity = arguments.get("entity")
result = await client.get_logbook(start_time, end_time, entity)
elif name == "set_state":
entity_id = arguments["entity_id"]
state = arguments["state"]
attributes = arguments.get("attributes")
result = await client.set_state(entity_id, state, attributes)
elif name == "delete_state":
entity_id = arguments["entity_id"]
result = await client.delete_state(entity_id)
elif name == "render_template":
template = arguments["template"]
result = await client.render_template(template)
elif name == "check_config":
result = await client.check_config()
elif name == "get_error_log":
result = await client.get_error_log()
elif name == "get_calendars":
result = await client.get_calendars()
elif name == "get_calendar_events":
calendar_id = arguments["calendar_id"]
start = arguments["start"]
end = arguments["end"]
result = await client.get_calendar_events(calendar_id, start, end)
elif name == "get_camera_image":
import base64
camera_entity_id = arguments["camera_entity_id"]
image_data = await client.get_camera_proxy(camera_entity_id)
result = {
"image_base64": base64.b64encode(image_data).decode('utf-8'),
"content_type": "image/jpeg"
}
elif name == "call_service_with_response":
domain = arguments["domain"]
service = arguments["service"]
service_data = arguments.get("service_data", {})
result = await client.call_service_with_response(domain, service, service_data)
elif name == "handle_intent":
intent_name = arguments["intent_name"]
intent_data = arguments.get("intent_data")
result = await client.handle_intent(intent_name, intent_data)
elif name == "subscribe_events":
# Handle SSE subscription
events = set(arguments.get("events", []))
entity_id = arguments.get("entity_id")
domain = arguments.get("domain")
try:
client_id = sse_manager.add_subscription(events, entity_id, domain)
# Start the event stream
await sse_manager.start_event_stream(client_id, HA_URL, HA_TOKEN)
result = {
"status": "subscribed",
"client_id": client_id,
"events": list(events),
"entity_id": entity_id,
"domain": domain,
"message": "Successfully subscribed to Home Assistant events"
}
except ValueError as e:
result = {"error": str(e)}
elif name == "get_sse_stats":
result = sse_manager.get_stats()
elif name == "get_automations":
result = await client.get_automations()
elif name == "get_automation":
automation_id = arguments["automation_id"]
result = await client.get_automation(automation_id)
elif name == "toggle_automation":
automation_id = arguments["automation_id"]
result = await client.toggle_automation(automation_id)
elif name == "turn_on_automation":
automation_id = arguments["automation_id"]
result = await client.turn_on_automation(automation_id)
elif name == "turn_off_automation":
automation_id = arguments["automation_id"]
result = await client.turn_off_automation(automation_id)
elif name == "trigger_automation":
automation_id = arguments["automation_id"]
result = await client.trigger_automation(automation_id)
elif name == "create_automation":
config = arguments["config"]
result = await client.create_automation(config)
elif name == "update_automation":
automation_id = arguments["automation_id"]
config = arguments["config"]
result = await client.update_automation(automation_id, config)
elif name == "delete_automation":
automation_id = arguments["automation_id"]
result = await client.delete_automation(automation_id)
elif name == "reload_automations":
result = await client.reload_automations()
elif name == "get_automation_trace":
automation_id = arguments["automation_id"]
run_id = arguments.get("run_id")
result = await client.get_automation_trace(automation_id, run_id)
elif name == "get_scenes":
result = await client.get_scenes()
elif name == "activate_scene":
scene_id = arguments["scene_id"]
result = await client.activate_scene(scene_id)
elif name == "create_scene":
scene_data = arguments["scene_data"]
result = await client.create_scene(scene_data)
# Area Management Tools
elif name == "get_areas":
result = await client.get_areas()
elif name == "create_area":
name_arg = arguments["name"]
aliases = arguments.get("aliases")
result = await client.create_area(name_arg, aliases)
elif name == "update_area":
area_id = arguments["area_id"]
name_arg = arguments["name"]
aliases = arguments.get("aliases")
result = await client.update_area(area_id, name_arg, aliases)
elif name == "delete_area":
area_id = arguments["area_id"]
result = await client.delete_area(area_id)
elif name == "get_entities_by_area":
area_id = arguments["area_id"]
result = await client.get_entities_by_area(area_id)
# Device Management Tools
elif name == "get_devices":
result = await client.get_devices()
elif name == "get_device":
device_id = arguments["device_id"]
result = await client.get_device(device_id)
elif name == "update_device":
device_id = arguments["device_id"]
name_arg = arguments.get("name")
area_id = arguments.get("area_id")
disabled_by = arguments.get("disabled_by")
result = await client.update_device(device_id, name_arg, area_id, disabled_by)
# System Management Tools
elif name == "restart_homeassistant":
result = await client.restart_homeassistant()
elif name == "stop_homeassistant":
result = await client.stop_homeassistant()
elif name == "check_config_valid":
result = await client.check_config_valid()
elif name == "get_system_health":
result = await client.get_system_health()
elif name == "get_supervisor_info":
result = await client.get_supervisor_info()
elif name == "get_system_info":
result = await client.get_system_info()
# Integration Management Tools
elif name == "get_integrations":
result = await client.get_integrations()
elif name == "reload_integration":
integration_domain = arguments["integration_domain"]
result = await client.reload_integration(integration_domain)
elif name == "disable_integration":
config_entry_id = arguments["config_entry_id"]
result = await client.disable_integration(config_entry_id)
elif name == "enable_integration":
config_entry_id = arguments["config_entry_id"]
result = await client.enable_integration(config_entry_id)
elif name == "delete_integration":
config_entry_id = arguments["config_entry_id"]
result = await client.delete_integration(config_entry_id)
elif name == "get_integration_info":
config_entry_id = arguments["config_entry_id"]
result = await client.get_integration_info(config_entry_id)
# Notification Services Tools
elif name == "send_notification":
message = arguments["message"]
title = arguments.get("title")
target = arguments.get("target")
data = arguments.get("data")
result = await client.send_notification(message, title, target, data)
elif name == "get_notification_services":
result = await client.get_notification_services()
elif name == "dismiss_notification":
notification_id = arguments["notification_id"]
result = await client.dismiss_notification(notification_id)
# Entity Registry Management Tools
elif name == "get_entity_registry":
result = await client.get_entity_registry()
elif name == "update_entity_registry":
entity_id = arguments["entity_id"]
name_arg = arguments.get("name")
disabled_by = arguments.get("disabled_by")
area_id = arguments.get("area_id")
result = await client.update_entity_registry(entity_id, name_arg, disabled_by, area_id)
elif name == "enable_entity":
entity_id = arguments["entity_id"]
result = await client.enable_entity(entity_id)
elif name == "disable_entity":
entity_id = arguments["entity_id"]
result = await client.disable_entity(entity_id)
else:
result = {"error": f"Unknown tool: {name}"}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except Exception as e:
logger.error(f"Error in tool {name}: {e}")
return [TextContent(
type="text",
text=json.dumps({"error": str(e)})
)]
async def main():
"""Run the MCP server"""
# Import here to avoid issues if mcp package isn't available
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="homeassistant-mcp",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
asyncio.run(main())