|
1 | | -#!/usr/bin/env python3 |
2 | | -"""Home Assistant MCP Server Add-on startup script.""" |
3 | | - |
4 | | -import json |
5 | | -import os |
6 | | -import secrets |
7 | | -import sys |
8 | | -from pathlib import Path |
9 | | - |
10 | | - |
11 | | -def log_info(message: str) -> None: |
12 | | - """Log info message.""" |
13 | | - print(f"[INFO] {message}", flush=True) |
14 | | - |
15 | | - |
16 | | -def log_error(message: str) -> None: |
17 | | - """Log error message.""" |
18 | | - print(f"[ERROR] {message}", file=sys.stderr, flush=True) |
19 | | - |
20 | | - |
21 | | -def generate_secret_path() -> str: |
22 | | - """Generate a secure random path with 128-bit entropy. |
23 | | -
|
24 | | - Format: /private_<22-char-urlsafe-token> |
25 | | - Example: /private_zctpwlX7ZkIAr7oqdfLPxw |
26 | | - """ |
27 | | - return "/private_" + secrets.token_urlsafe(16) |
28 | | - |
29 | | - |
30 | | -def get_or_create_secret_path(data_dir: Path, custom_path: str = "") -> str: |
31 | | - """Get existing secret path or create a new one. |
32 | | -
|
33 | | - Args: |
34 | | - data_dir: Path to the /data directory |
35 | | - custom_path: Optional custom path from config (overrides auto-generated) |
36 | | -
|
37 | | - Returns: |
38 | | - The secret path to use |
39 | | - """ |
40 | | - secret_file = data_dir / "secret_path.txt" |
41 | | - |
42 | | - # If custom path is provided, use it and update the stored path |
43 | | - if custom_path and custom_path.strip(): |
44 | | - path = custom_path.strip() |
45 | | - if not path.startswith("/"): |
46 | | - path = "/" + path |
47 | | - log_info(f"Using custom secret path from configuration") |
48 | | - # Update stored path for consistency |
49 | | - secret_file.write_text(path) |
50 | | - return path |
51 | | - |
52 | | - # Check if we have a stored secret path |
53 | | - if secret_file.exists(): |
54 | | - try: |
55 | | - stored_path = secret_file.read_text().strip() |
56 | | - if stored_path: |
57 | | - log_info(f"Using existing auto-generated secret path") |
58 | | - return stored_path |
59 | | - except Exception as e: |
60 | | - log_error(f"Failed to read stored secret path: {e}") |
61 | | - |
62 | | - # Generate new secret path |
63 | | - new_path = generate_secret_path() |
64 | | - log_info("Generated new secret path with 128-bit entropy") |
65 | | - try: |
66 | | - data_dir.mkdir(parents=True, exist_ok=True) |
67 | | - secret_file.write_text(new_path) |
68 | | - return new_path |
69 | | - except Exception as e: |
70 | | - log_error(f"Failed to save secret path: {e}") |
71 | | - # Return the path anyway - it will work for this session |
72 | | - return new_path |
73 | | - |
74 | | - |
75 | | -def main() -> int: |
76 | | - """Start the Home Assistant MCP Server.""" |
77 | | - log_info("Starting Home Assistant MCP Server...") |
78 | | - |
79 | | - # Read configuration from Supervisor |
80 | | - config_file = Path("/data/options.json") |
81 | | - data_dir = Path("/data") |
82 | | - backup_hint = "normal" # default |
83 | | - custom_secret_path = "" # default |
84 | | - |
85 | | - if config_file.exists(): |
86 | | - try: |
87 | | - with open(config_file) as f: |
88 | | - config = json.load(f) |
89 | | - backup_hint = config.get("backup_hint", "normal") |
90 | | - custom_secret_path = config.get("secret_path", "") |
91 | | - except Exception as e: |
92 | | - log_error(f"Failed to read config: {e}, using defaults") |
93 | | - |
94 | | - # Generate or retrieve secret path |
95 | | - secret_path = get_or_create_secret_path(data_dir, custom_secret_path) |
96 | | - |
97 | | - log_info(f"Backup hint mode: {backup_hint}") |
98 | | - |
99 | | - # Set up environment for ha-mcp |
100 | | - os.environ["HOMEASSISTANT_URL"] = "http://supervisor/core" |
101 | | - os.environ["BACKUP_HINT"] = backup_hint |
102 | | - |
103 | | - # Validate Supervisor token |
104 | | - supervisor_token = os.environ.get("SUPERVISOR_TOKEN") |
105 | | - if not supervisor_token: |
106 | | - log_error("SUPERVISOR_TOKEN not found! Cannot authenticate.") |
107 | | - return 1 |
108 | | - |
109 | | - os.environ["HOMEASSISTANT_TOKEN"] = supervisor_token |
110 | | - |
111 | | - log_info(f"Home Assistant URL: {os.environ['HOMEASSISTANT_URL']}") |
112 | | - log_info("Authentication configured via Supervisor token") |
113 | | - |
114 | | - # Fixed port (internal container port) |
115 | | - port = 9583 |
116 | | - |
117 | | - log_info("") |
118 | | - log_info("=" * 80) |
119 | | - log_info(f"🔐 MCP Server URL: http://<home-assistant-ip>:9583{secret_path}") |
120 | | - log_info("") |
121 | | - log_info(f" Secret Path: {secret_path}") |
122 | | - log_info("") |
123 | | - log_info(" ⚠️ IMPORTANT: Copy this exact URL - the secret path is required!") |
124 | | - log_info(" 💡 This path is auto-generated and persisted to /data/secret_path.txt") |
125 | | - log_info("=" * 80) |
126 | | - log_info("") |
127 | | - |
128 | | - # Import and run MCP server directly |
129 | | - try: |
130 | | - log_info("Importing ha_mcp module...") |
131 | | - from ha_mcp.__main__ import mcp |
132 | | - |
133 | | - log_info("Starting MCP server...") |
134 | | - mcp.run( |
135 | | - transport="streamable-http", |
136 | | - host="0.0.0.0", |
137 | | - port=port, |
138 | | - path=secret_path, |
139 | | - log_level="info", |
140 | | - ) |
141 | | - except Exception as e: |
142 | | - log_error(f"Failed to start MCP server: {e}") |
143 | | - import traceback |
144 | | - |
145 | | - traceback.print_exc() |
146 | | - return 1 |
147 | | - |
148 | | - return 0 |
149 | | - |
150 | | - |
151 | | -if __name__ == "__main__": |
152 | | - sys.exit(main()) |
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Home Assistant MCP Server Add-on startup script.""" |
| 3 | + |
| 4 | +import json |
| 5 | +import os |
| 6 | +import secrets |
| 7 | +import sys |
| 8 | +from datetime import datetime |
| 9 | +from pathlib import Path |
| 10 | + |
| 11 | + |
| 12 | +def _log_with_timestamp(level: str, message: str, stream=None) -> None: |
| 13 | + """Log a message with a timestamp.""" |
| 14 | + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") |
| 15 | + print(f"{now} [{level}] {message}", file=stream, flush=True) |
| 16 | + |
| 17 | + |
| 18 | +def log_info(message: str) -> None: |
| 19 | + """Log info message.""" |
| 20 | + _log_with_timestamp("INFO", message) |
| 21 | + |
| 22 | + |
| 23 | +def log_error(message: str) -> None: |
| 24 | + """Log error message.""" |
| 25 | + _log_with_timestamp("ERROR", message, sys.stderr) |
| 26 | + |
| 27 | + |
| 28 | +def generate_secret_path() -> str: |
| 29 | + """Generate a secure random path with 128-bit entropy. |
| 30 | +
|
| 31 | + Format: /private_<22-char-urlsafe-token> |
| 32 | + Example: /private_zctpwlX7ZkIAr7oqdfLPxw |
| 33 | + """ |
| 34 | + return "/private_" + secrets.token_urlsafe(16) |
| 35 | + |
| 36 | + |
| 37 | +def get_or_create_secret_path(data_dir: Path, custom_path: str = "") -> str: |
| 38 | + """Get existing secret path or create a new one. |
| 39 | +
|
| 40 | + Args: |
| 41 | + data_dir: Path to the /data directory |
| 42 | + custom_path: Optional custom path from config (overrides auto-generated) |
| 43 | +
|
| 44 | + Returns: |
| 45 | + The secret path to use |
| 46 | + """ |
| 47 | + secret_file = data_dir / "secret_path.txt" |
| 48 | + |
| 49 | + # If custom path is provided, use it and update the stored path |
| 50 | + if custom_path and custom_path.strip(): |
| 51 | + path = custom_path.strip() |
| 52 | + if not path.startswith("/"): |
| 53 | + path = "/" + path |
| 54 | + log_info("Using custom secret path from configuration") |
| 55 | + # Update stored path for consistency |
| 56 | + secret_file.write_text(path) |
| 57 | + return path |
| 58 | + |
| 59 | + # Check if we have a stored secret path |
| 60 | + if secret_file.exists(): |
| 61 | + try: |
| 62 | + stored_path = secret_file.read_text().strip() |
| 63 | + if stored_path: |
| 64 | + log_info("Using existing auto-generated secret path") |
| 65 | + return stored_path |
| 66 | + except Exception as e: |
| 67 | + log_error(f"Failed to read stored secret path: {e}") |
| 68 | + |
| 69 | + # Generate new secret path |
| 70 | + new_path = generate_secret_path() |
| 71 | + log_info("Generated new secret path with 128-bit entropy") |
| 72 | + try: |
| 73 | + data_dir.mkdir(parents=True, exist_ok=True) |
| 74 | + secret_file.write_text(new_path) |
| 75 | + return new_path |
| 76 | + except Exception as e: |
| 77 | + log_error(f"Failed to save secret path: {e}") |
| 78 | + # Return the path anyway - it will work for this session |
| 79 | + return new_path |
| 80 | + |
| 81 | + |
| 82 | +def main() -> int: |
| 83 | + """Start the Home Assistant MCP Server.""" |
| 84 | + log_info("Starting Home Assistant MCP Server...") |
| 85 | + |
| 86 | + # Read configuration from Supervisor |
| 87 | + config_file = Path("/data/options.json") |
| 88 | + data_dir = Path("/data") |
| 89 | + backup_hint = "normal" # default |
| 90 | + custom_secret_path = "" # default |
| 91 | + |
| 92 | + if config_file.exists(): |
| 93 | + try: |
| 94 | + with open(config_file) as f: |
| 95 | + config = json.load(f) |
| 96 | + backup_hint = config.get("backup_hint", "normal") |
| 97 | + custom_secret_path = config.get("secret_path", "") |
| 98 | + except Exception as e: |
| 99 | + log_error(f"Failed to read config: {e}, using defaults") |
| 100 | + |
| 101 | + # Generate or retrieve secret path |
| 102 | + secret_path = get_or_create_secret_path(data_dir, custom_secret_path) |
| 103 | + |
| 104 | + log_info(f"Backup hint mode: {backup_hint}") |
| 105 | + |
| 106 | + # Set up environment for ha-mcp |
| 107 | + os.environ["HOMEASSISTANT_URL"] = "http://supervisor/core" |
| 108 | + os.environ["BACKUP_HINT"] = backup_hint |
| 109 | + |
| 110 | + # Validate Supervisor token |
| 111 | + supervisor_token = os.environ.get("SUPERVISOR_TOKEN") |
| 112 | + if not supervisor_token: |
| 113 | + log_error("SUPERVISOR_TOKEN not found! Cannot authenticate.") |
| 114 | + return 1 |
| 115 | + |
| 116 | + os.environ["HOMEASSISTANT_TOKEN"] = supervisor_token |
| 117 | + |
| 118 | + log_info(f"Home Assistant URL: {os.environ['HOMEASSISTANT_URL']}") |
| 119 | + log_info("Authentication configured via Supervisor token") |
| 120 | + |
| 121 | + # Fixed port (internal container port) |
| 122 | + port = 9583 |
| 123 | + |
| 124 | + log_info("") |
| 125 | + log_info("=" * 80) |
| 126 | + log_info(f"🔐 MCP Server URL: http://<home-assistant-ip>:9583{secret_path}") |
| 127 | + log_info("") |
| 128 | + log_info(f" Secret Path: {secret_path}") |
| 129 | + log_info("") |
| 130 | + log_info(" ⚠️ IMPORTANT: Copy this exact URL - the secret path is required!") |
| 131 | + log_info(" 💡 This path is auto-generated and persisted to /data/secret_path.txt") |
| 132 | + log_info("=" * 80) |
| 133 | + log_info("") |
| 134 | + |
| 135 | + # Import and run MCP server directly |
| 136 | + try: |
| 137 | + log_info("Importing ha_mcp module...") |
| 138 | + from ha_mcp.__main__ import mcp, _get_timestamped_uvicorn_log_config |
| 139 | + |
| 140 | + log_info("Starting MCP server...") |
| 141 | + mcp.run( |
| 142 | + transport="streamable-http", |
| 143 | + host="0.0.0.0", |
| 144 | + port=port, |
| 145 | + path=secret_path, |
| 146 | + log_level="info", |
| 147 | + uvicorn_config={"log_config": _get_timestamped_uvicorn_log_config()}, |
| 148 | + ) |
| 149 | + except Exception as e: |
| 150 | + log_error(f"Failed to start MCP server: {e}") |
| 151 | + import traceback |
| 152 | + |
| 153 | + traceback.print_exc() |
| 154 | + return 1 |
| 155 | + |
| 156 | + return 0 |
| 157 | + |
| 158 | + |
| 159 | +if __name__ == "__main__": |
| 160 | + sys.exit(main()) |
0 commit comments