#!/usr/bin/env python3
"""
Enhanced Home Assistant MCP Server
Provides Claude with read/write access to Home Assistant + automation management
"""
import sys
import traceback
import asyncio
import os
from pathlib import Path
from typing import Any
# Debug logging - will appear in Claude Desktop logs
print("Enhanced MCP Server starting...", file=sys.stderr)
# Try to import dependencies with error handling
try:
print("Importing httpx...", file=sys.stderr)
import httpx
print("Importing python-dotenv...", file=sys.stderr)
from dotenv import load_dotenv
print("Importing MCP SDK...", file=sys.stderr)
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
print("All imports successful!", file=sys.stderr)
except ImportError as e:
print(f"Import error: {e}", file=sys.stderr)
print("Make sure all dependencies are installed:", file=sys.stderr)
print(" pip install mcp httpx python-dotenv", file=sys.stderr)
traceback.print_exc(file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"Unexpected error during imports: {e}", file=sys.stderr)
traceback.print_exc(file=sys.stderr)
sys.exit(1)
# Load environment variables from .env file
try:
env_path = Path(__file__).parent / ".env"
print(f"Loading environment from: {env_path}", file=sys.stderr)
if env_path.exists():
load_dotenv(env_path)
print("Environment file loaded successfully", file=sys.stderr)
else:
print(f"Warning: .env file not found at {env_path}", file=sys.stderr)
except Exception as e:
print(f"Error loading .env file: {e}", file=sys.stderr)
traceback.print_exc(file=sys.stderr)
# Configuration
HA_URL = os.getenv("HA_URL", "http://192.168.3.10:8123")
HA_TOKEN = os.getenv("HA_TOKEN")
print(f"HA_URL: {HA_URL}", file=sys.stderr)
print(f"HA_TOKEN present: {bool(HA_TOKEN)}", file=sys.stderr)
if not HA_TOKEN:
print("ERROR: HA_TOKEN environment variable not set!", file=sys.stderr)
print("Please create a .env file with:", file=sys.stderr)
print(" HA_URL=http://192.168.3.10:8123", file=sys.stderr)
print(" HA_TOKEN=your_long_lived_token_here", file=sys.stderr)
sys.exit(1)
# Create server instance
print("Creating MCP server instance...", file=sys.stderr)
app = Server("homeassistant-enhanced")
print("Server instance created", file=sys.stderr)
# Helper function for API calls
async def ha_api_call(method: str, endpoint: str, data: dict = None) -> dict:
"""Make authenticated API call to Home Assistant"""
headers = {
"Authorization": f"Bearer {HA_TOKEN}",
"Content-Type": "application/json"
}
try:
async with httpx.AsyncClient(timeout=30.0) as client:
if method == "GET":
response = await client.get(f"{HA_URL}{endpoint}", headers=headers)
elif method == "POST":
response = await client.post(
f"{HA_URL}{endpoint}",
headers=headers,
json=data or {}
)
elif method == "DELETE":
response = await client.delete(f"{HA_URL}{endpoint}", headers=headers)
else:
raise ValueError(f"Unsupported method: {method}")
response.raise_for_status()
# Some endpoints return empty responses
if response.text:
return response.json()
return {}
except httpx.HTTPStatusError as e:
error_msg = f"HTTP {e.response.status_code}: {e.response.text}"
print(f"API Error: {error_msg}", file=sys.stderr)
raise
except httpx.RequestError as e:
error_msg = f"Request failed: {str(e)}"
print(f"API Error: {error_msg}", file=sys.stderr)
raise
except Exception as e:
print(f"Unexpected API error: {str(e)}", file=sys.stderr)
raise
# Define tools
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List all available tools"""
print("list_tools() called", file=sys.stderr)
return [
Tool(
name="get_state",
description="Get the current state and attributes of any Home Assistant entity",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "The entity ID (e.g., 'light.office', 'sensor.temperature')"
}
},
"required": ["entity_id"]
}
),
Tool(
name="list_entities",
description="List all entities, optionally filtered by domain (light, sensor, etc.)",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Optional domain filter (e.g., 'light', 'sensor', 'automation')"
}
}
}
),
Tool(
name="call_service",
description="Call any Home Assistant service to control devices",
inputSchema={
"type": "object",
"properties": {
"domain": {
"type": "string",
"description": "Service domain (e.g., 'light', 'climate', 'switch')"
},
"service": {
"type": "string",
"description": "Service name (e.g., 'turn_on', 'turn_off', 'set_temperature')"
},
"entity_id": {
"type": "string",
"description": "Target entity ID"
},
"data": {
"type": "object",
"description": "Additional service data (e.g., brightness, temperature)",
"additionalProperties": True
}
},
"required": ["domain", "service", "entity_id"]
}
),
Tool(
name="trigger_automation",
description="Manually trigger a Home Assistant automation",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Automation entity ID (e.g., 'automation.office_lights_consolidated')"
}
},
"required": ["entity_id"]
}
),
Tool(
name="get_history",
description="Get historical state changes for an entity",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Entity ID to get history for"
},
"hours": {
"type": "number",
"description": "Number of hours of history to retrieve (default: 24)",
"default": 24
}
},
"required": ["entity_id"]
}
),
# ============ NEW AUTOMATION MANAGEMENT TOOLS ============
Tool(
name="create_automation",
description="Create a new Home Assistant automation from YAML configuration",
inputSchema={
"type": "object",
"properties": {
"automation_config": {
"type": "object",
"description": "Complete automation configuration (must include 'id', 'alias', 'trigger', 'action')",
"additionalProperties": True
}
},
"required": ["automation_config"]
}
),
Tool(
name="update_automation",
description="Update an existing Home Assistant automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "The automation ID (not entity_id, just the ID from config)"
},
"automation_config": {
"type": "object",
"description": "Updated automation configuration",
"additionalProperties": True
}
},
"required": ["automation_id", "automation_config"]
}
),
Tool(
name="delete_automation",
description="Delete a Home Assistant automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "The automation ID to delete (not entity_id, just the ID from config)"
}
},
"required": ["automation_id"]
}
),
Tool(
name="get_automation_config",
description="Get the full YAML configuration of an automation",
inputSchema={
"type": "object",
"properties": {
"automation_id": {
"type": "string",
"description": "The automation ID (not entity_id, just the ID from config)"
}
},
"required": ["automation_id"]
}
),
Tool(
name="list_automation_configs",
description="List all automation configurations (not just states, but full YAML configs)",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="reload_automations",
description="Reload all automations after making changes",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="enable_automation",
description="Enable a disabled automation",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Automation entity ID (e.g., 'automation.office_lights')"
}
},
"required": ["entity_id"]
}
),
Tool(
name="disable_automation",
description="Disable an automation",
inputSchema={
"type": "object",
"properties": {
"entity_id": {
"type": "string",
"description": "Automation entity ID (e.g., 'automation.office_lights')"
}
},
"required": ["entity_id"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""Handle tool execution"""
print(f"Tool called: {name} with arguments: {arguments}", file=sys.stderr)
try:
# ============ ORIGINAL TOOLS ============
if name == "get_state":
entity_id = arguments["entity_id"]
print(f"Getting state for: {entity_id}", file=sys.stderr)
result = await ha_api_call("GET", f"/api/states/{entity_id}")
response = f"""Entity: {result['entity_id']}
State: {result['state']}
Last Changed: {result['last_changed']}
Last Updated: {result['last_updated']}
Attributes:
"""
for key, value in result.get('attributes', {}).items():
response += f" {key}: {value}\n"
print(f"get_state successful for {entity_id}", file=sys.stderr)
return [TextContent(type="text", text=response)]
elif name == "list_entities":
domain = arguments.get("domain")
print(f"Listing entities, domain filter: {domain}", file=sys.stderr)
result = await ha_api_call("GET", "/api/states")
if domain:
result = [e for e in result if e['entity_id'].startswith(f"{domain}.")]
response = f"Found {len(result)} entities:\n\n"
for entity in result[:50]: # Limit to first 50
response += f"{entity['entity_id']}: {entity['state']}\n"
if len(result) > 50:
response += f"\n... and {len(result) - 50} more"
print(f"list_entities successful, returned {len(result)} entities", file=sys.stderr)
return [TextContent(type="text", text=response)]
elif name == "call_service":
domain = arguments["domain"]
service = arguments["service"]
entity_id = arguments["entity_id"]
data = arguments.get("data", {})
print(f"Calling service: {domain}.{service} on {entity_id}", file=sys.stderr)
payload = {
"entity_id": entity_id,
**data
}
result = await ha_api_call("POST", f"/api/services/{domain}/{service}", payload)
print(f"Service call successful", file=sys.stderr)
return [TextContent(
type="text",
text=f"Successfully called {domain}.{service} on {entity_id}"
)]
elif name == "trigger_automation":
entity_id = arguments["entity_id"]
print(f"Triggering automation: {entity_id}", file=sys.stderr)
result = await ha_api_call(
"POST",
"/api/services/automation/trigger",
{"entity_id": entity_id}
)
print(f"Automation triggered successfully", file=sys.stderr)
return [TextContent(
type="text",
text=f"Successfully triggered {entity_id}"
)]
elif name == "get_history":
entity_id = arguments["entity_id"]
hours = arguments.get("hours", 24)
print(f"Getting history for {entity_id}, last {hours} hours", file=sys.stderr)
from datetime import datetime, timedelta
end_time = datetime.now()
start_time = end_time - timedelta(hours=hours)
result = await ha_api_call(
"GET",
f"/api/history/period/{start_time.isoformat()}?filter_entity_id={entity_id}"
)
if not result or not result[0]:
return [TextContent(
type="text",
text=f"No history found for {entity_id} in the last {hours} hours"
)]
history = result[0]
response = f"History for {entity_id} (last {hours} hours):\n\n"
for entry in history[-10:]: # Last 10 state changes
response += f"{entry['last_changed']}: {entry['state']}\n"
print(f"History retrieved successfully", file=sys.stderr)
return [TextContent(type="text", text=response)]
# ============ NEW AUTOMATION MANAGEMENT TOOLS ============
elif name == "create_automation":
automation_config = arguments["automation_config"]
automation_id = automation_config.get("id")
if not automation_id:
return [TextContent(
type="text",
text="Error: automation_config must include an 'id' field"
)]
print(f"Creating automation: {automation_id}", file=sys.stderr)
result = await ha_api_call(
"POST",
f"/api/config/automation/config/{automation_id}",
automation_config
)
print(f"Automation created successfully", file=sys.stderr)
return [TextContent(
type="text",
text=f"✓ Successfully created automation '{automation_config.get('alias', automation_id)}' (ID: {automation_id})"
)]
elif name == "update_automation":
automation_id = arguments["automation_id"]
automation_config = arguments["automation_config"]
print(f"Updating automation: {automation_id}", file=sys.stderr)
result = await ha_api_call(
"POST",
f"/api/config/automation/config/{automation_id}",
automation_config
)
print(f"Automation updated successfully", file=sys.stderr)
return [TextContent(
type="text",
text=f"✓ Successfully updated automation '{automation_config.get('alias', automation_id)}' (ID: {automation_id})"
)]
elif name == "delete_automation":
automation_id = arguments["automation_id"]
print(f"Deleting automation: {automation_id}", file=sys.stderr)
result = await ha_api_call(
"DELETE",
f"/api/config/automation/config/{automation_id}"
)
print(f"Automation deleted successfully", file=sys.stderr)
return [TextContent(
type="text",
text=f"✓ Successfully deleted automation (ID: {automation_id})"
)]
elif name == "get_automation_config":
automation_id = arguments["automation_id"]
print(f"Getting config for automation: {automation_id}", file=sys.stderr)
result = await ha_api_call(
"GET",
f"/api/config/automation/config/{automation_id}"
)
import json
config_json = json.dumps(result, indent=2)
print(f"Automation config retrieved successfully", file=sys.stderr)
return [TextContent(
type="text",
text=f"Configuration for automation {automation_id}:\n\n{config_json}"
)]
elif name == "list_automation_configs":
print(f"Listing all automation configurations", file=sys.stderr)
result = await ha_api_call("GET", "/api/config/automation/config")
response = f"Found {len(result)} automation configurations:\n\n"
for config in result:
automation_id = config.get('id', 'unknown')
alias = config.get('alias', 'No alias')
response += f"- {alias} (ID: {automation_id})\n"
print(f"Automation configs listed successfully", file=sys.stderr)
return [TextContent(type="text", text=response)]
elif name == "reload_automations":
print(f"Reloading all automations", file=sys.stderr)
result = await ha_api_call(
"POST",
"/api/services/automation/reload",
{}
)
print(f"Automations reloaded successfully", file=sys.stderr)
return [TextContent(
type="text",
text="✓ Successfully reloaded all automations"
)]
elif name == "enable_automation":
entity_id = arguments["entity_id"]
print(f"Enabling automation: {entity_id}", file=sys.stderr)
result = await ha_api_call(
"POST",
"/api/services/automation/turn_on",
{"entity_id": entity_id}
)
print(f"Automation enabled successfully", file=sys.stderr)
return [TextContent(
type="text",
text=f"✓ Successfully enabled {entity_id}"
)]
elif name == "disable_automation":
entity_id = arguments["entity_id"]
print(f"Disabling automation: {entity_id}", file=sys.stderr)
result = await ha_api_call(
"POST",
"/api/services/automation/turn_off",
{"entity_id": entity_id}
)
print(f"Automation disabled successfully", file=sys.stderr)
return [TextContent(
type="text",
text=f"✓ Successfully disabled {entity_id}"
)]
else:
error_msg = f"Unknown tool: {name}"
print(error_msg, file=sys.stderr)
return [TextContent(type="text", text=error_msg)]
except httpx.HTTPStatusError as e:
error_msg = f"Home Assistant API error: {e.response.status_code} - {e.response.text}"
print(error_msg, file=sys.stderr)
return [TextContent(type="text", text=error_msg)]
except Exception as e:
error_msg = f"Error executing tool {name}: {str(e)}"
print(error_msg, file=sys.stderr)
traceback.print_exc(file=sys.stderr)
return [TextContent(type="text", text=error_msg)]
async def main():
"""Run the MCP server"""
print("Starting main() function...", file=sys.stderr)
try:
# Test connection to Home Assistant first
print("Testing connection to Home Assistant...", file=sys.stderr)
test_response = await ha_api_call("GET", "/api/")
print(f"Connection test successful: {test_response.get('message', 'OK')}", file=sys.stderr)
print("Starting MCP stdio server...", file=sys.stderr)
async with stdio_server() as (read_stream, write_stream):
print("Enhanced MCP server running with automation management, waiting for requests...", file=sys.stderr)
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
except Exception as e:
print(f"Fatal error in main(): {e}", file=sys.stderr)
traceback.print_exc(file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
print("Script executing as main...", file=sys.stderr)
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Server stopped by user", file=sys.stderr)
except Exception as e:
print(f"Fatal error: {e}", file=sys.stderr)
traceback.print_exc(file=sys.stderr)
sys.exit(1)
