|
| 1 | +""" |
| 2 | +Category management tools for Home Assistant. |
| 3 | +
|
| 4 | +This module provides tools for listing, creating, updating, and deleting |
| 5 | +Home Assistant categories. Categories are domain-scoped organizational groups |
| 6 | +(e.g., for automations, scripts, scenes, helpers) introduced in Home Assistant 2024.4. |
| 7 | +
|
| 8 | +To assign categories to entities, use ha_set_entity(categories=...). |
| 9 | +""" |
| 10 | + |
| 11 | +import logging |
| 12 | +from typing import Annotated, Any |
| 13 | + |
| 14 | +from fastmcp.exceptions import ToolError |
| 15 | +from pydantic import Field |
| 16 | + |
| 17 | +from ..errors import ErrorCode, create_error_response |
| 18 | +from .helpers import exception_to_structured_error, log_tool_usage, raise_tool_error |
| 19 | + |
| 20 | +logger = logging.getLogger(__name__) |
| 21 | + |
| 22 | + |
| 23 | +def register_category_tools(mcp: Any, client: Any, **kwargs: Any) -> None: |
| 24 | + """Register Home Assistant category management tools.""" |
| 25 | + |
| 26 | + @mcp.tool( |
| 27 | + annotations={ |
| 28 | + "idempotentHint": True, |
| 29 | + "readOnlyHint": True, |
| 30 | + "tags": ["category"], |
| 31 | + "title": "Get Category", |
| 32 | + } |
| 33 | + ) |
| 34 | + @log_tool_usage |
| 35 | + async def ha_config_get_category( |
| 36 | + scope: Annotated[ |
| 37 | + str, |
| 38 | + Field( |
| 39 | + description="Domain scope for categories (e.g., 'automation', 'script', 'scene', 'helpers').", |
| 40 | + ), |
| 41 | + ], |
| 42 | + category_id: Annotated[ |
| 43 | + str | None, |
| 44 | + Field( |
| 45 | + description="ID of the category to retrieve. If omitted, lists all categories for the scope.", |
| 46 | + default=None, |
| 47 | + ), |
| 48 | + ] = None, |
| 49 | + ) -> dict[str, Any]: |
| 50 | + """ |
| 51 | + Get category info - list all categories for a scope or get a specific one by ID. |
| 52 | +
|
| 53 | + Without a category_id: Lists all Home Assistant categories for the given scope. |
| 54 | + With a category_id: Returns configuration for that specific category. |
| 55 | +
|
| 56 | + Categories are domain-scoped organizational groups for automations, scripts, scenes, and helpers. |
| 57 | +
|
| 58 | + CATEGORY PROPERTIES: |
| 59 | + - ID (category_id), Name |
| 60 | + - Icon (optional) |
| 61 | +
|
| 62 | + EXAMPLES: |
| 63 | + - List automation categories: ha_config_get_category("automation") |
| 64 | + - List script categories: ha_config_get_category("script") |
| 65 | + - List helper categories: ha_config_get_category("helpers") |
| 66 | + - Get specific category: ha_config_get_category("automation", category_id="my_category_id") |
| 67 | +
|
| 68 | + Use ha_config_set_category() to create or update categories. |
| 69 | + Use ha_set_entity(categories={"automation": "category_id"}) to assign categories to entities. |
| 70 | + """ |
| 71 | + try: |
| 72 | + message: dict[str, Any] = { |
| 73 | + "type": "config/category_registry/list", |
| 74 | + "scope": scope, |
| 75 | + } |
| 76 | + |
| 77 | + result = await client.send_websocket_message(message) |
| 78 | + |
| 79 | + if not result.get("success"): |
| 80 | + raise_tool_error( |
| 81 | + create_error_response( |
| 82 | + ErrorCode.SERVICE_CALL_FAILED, |
| 83 | + result.get("error", "Failed to get categories"), |
| 84 | + context={"scope": scope, "category_id": category_id}, |
| 85 | + ) |
| 86 | + ) |
| 87 | + |
| 88 | + categories = result.get("result", []) |
| 89 | + |
| 90 | + # List mode - return all categories |
| 91 | + if category_id is None: |
| 92 | + return { |
| 93 | + "success": True, |
| 94 | + "count": len(categories), |
| 95 | + "categories": categories, |
| 96 | + "scope": scope, |
| 97 | + "message": f"Found {len(categories)} category(ies) for scope '{scope}'", |
| 98 | + } |
| 99 | + |
| 100 | + # Get mode - find specific category |
| 101 | + category = next( |
| 102 | + (cat for cat in categories if cat.get("category_id") == category_id), |
| 103 | + None, |
| 104 | + ) |
| 105 | + |
| 106 | + if category: |
| 107 | + return { |
| 108 | + "success": True, |
| 109 | + "category_id": category_id, |
| 110 | + "category": category, |
| 111 | + "scope": scope, |
| 112 | + "message": f"Found category: {category.get('name', category_id)}", |
| 113 | + } |
| 114 | + else: |
| 115 | + available_ids = [cat.get("category_id") for cat in categories[:10]] |
| 116 | + raise_tool_error( |
| 117 | + create_error_response( |
| 118 | + ErrorCode.ENTITY_NOT_FOUND, |
| 119 | + f"Category not found: {category_id}", |
| 120 | + context={ |
| 121 | + "category_id": category_id, |
| 122 | + "scope": scope, |
| 123 | + "available_category_ids": available_ids, |
| 124 | + }, |
| 125 | + suggestions=[ |
| 126 | + f"Use ha_config_get_category('{scope}') without category_id to see all categories" |
| 127 | + ], |
| 128 | + ) |
| 129 | + ) |
| 130 | + |
| 131 | + except ToolError: |
| 132 | + raise |
| 133 | + except Exception as e: |
| 134 | + logger.error(f"Error getting categories: {e}") |
| 135 | + exception_to_structured_error( |
| 136 | + e, |
| 137 | + context={"scope": scope, "category_id": category_id}, |
| 138 | + suggestions=[ |
| 139 | + "Check Home Assistant connection", |
| 140 | + "Verify WebSocket connection is active", |
| 141 | + "Ensure scope is valid (e.g., 'automation', 'script', 'scene', 'helpers')", |
| 142 | + ], |
| 143 | + ) |
| 144 | + |
| 145 | + @mcp.tool( |
| 146 | + annotations={ |
| 147 | + "destructiveHint": True, |
| 148 | + "tags": ["category"], |
| 149 | + "title": "Create or Update Category", |
| 150 | + } |
| 151 | + ) |
| 152 | + @log_tool_usage |
| 153 | + async def ha_config_set_category( |
| 154 | + name: Annotated[str, Field(description="Display name for the category")], |
| 155 | + scope: Annotated[ |
| 156 | + str, |
| 157 | + Field( |
| 158 | + description="Domain scope for the category (e.g., 'automation', 'script', 'scene', 'helpers').", |
| 159 | + ), |
| 160 | + ], |
| 161 | + category_id: Annotated[ |
| 162 | + str | None, |
| 163 | + Field( |
| 164 | + description="Category ID for updates. If not provided, creates a new category.", |
| 165 | + default=None, |
| 166 | + ), |
| 167 | + ] = None, |
| 168 | + icon: Annotated[ |
| 169 | + str | None, |
| 170 | + Field( |
| 171 | + description="Material Design Icon (e.g., 'mdi:tag', 'mdi:label')", |
| 172 | + default=None, |
| 173 | + ), |
| 174 | + ] = None, |
| 175 | + ) -> dict[str, Any]: |
| 176 | + """ |
| 177 | + Create or update a Home Assistant category. |
| 178 | +
|
| 179 | + Creates a new category if category_id is not provided, or updates an existing category if category_id is provided. |
| 180 | +
|
| 181 | + Categories are domain-scoped organizational groups for automations, scripts, |
| 182 | + scenes, and helpers. Unlike labels (which are cross-domain), categories are |
| 183 | + specific to a single domain scope. |
| 184 | +
|
| 185 | + EXAMPLES: |
| 186 | + - Create automation category: ha_config_set_category("Lighting", scope="automation") |
| 187 | + - Create with icon: ha_config_set_category("Security", scope="automation", icon="mdi:shield") |
| 188 | + - Update category: ha_config_set_category("Updated Name", scope="automation", category_id="my_category_id") |
| 189 | +
|
| 190 | + After creating a category, use ha_set_entity(categories={"automation": "category_id"}) to assign it. |
| 191 | + """ |
| 192 | + try: |
| 193 | + # Determine if this is a create or update |
| 194 | + action = "update" if category_id else "create" |
| 195 | + |
| 196 | + message: dict[str, Any] = { |
| 197 | + "type": f"config/category_registry/{action}", |
| 198 | + "scope": scope, |
| 199 | + "name": name, |
| 200 | + } |
| 201 | + |
| 202 | + if action == "update": |
| 203 | + message["category_id"] = category_id |
| 204 | + |
| 205 | + # Add optional fields only if they are explicitly provided (not None) |
| 206 | + if icon is not None: |
| 207 | + message["icon"] = icon |
| 208 | + |
| 209 | + result = await client.send_websocket_message(message) |
| 210 | + |
| 211 | + if result.get("success"): |
| 212 | + category_data = result.get("result", {}) |
| 213 | + action_past = "created" if action == "create" else "updated" |
| 214 | + return { |
| 215 | + "success": True, |
| 216 | + "category_id": category_data.get("category_id"), |
| 217 | + "category_data": category_data, |
| 218 | + "scope": scope, |
| 219 | + "message": f"Successfully {action_past} category: {name}", |
| 220 | + } |
| 221 | + else: |
| 222 | + raise_tool_error( |
| 223 | + create_error_response( |
| 224 | + ErrorCode.SERVICE_CALL_FAILED, |
| 225 | + f"Failed to {action} category: {result.get('error', 'Unknown error')}", |
| 226 | + context={ |
| 227 | + "name": name, |
| 228 | + "scope": scope, |
| 229 | + "category_id": category_id, |
| 230 | + }, |
| 231 | + ) |
| 232 | + ) |
| 233 | + |
| 234 | + except ToolError: |
| 235 | + raise |
| 236 | + except Exception as e: |
| 237 | + logger.error(f"Error setting category {name!r}: {e}") |
| 238 | + exception_to_structured_error( |
| 239 | + e, |
| 240 | + context={"name": name, "scope": scope, "category_id": category_id}, |
| 241 | + suggestions=[ |
| 242 | + "Check Home Assistant connection", |
| 243 | + "Verify the category name is valid", |
| 244 | + "For updates, verify the category_id exists using ha_config_get_category()", |
| 245 | + ], |
| 246 | + ) |
| 247 | + |
| 248 | + @mcp.tool( |
| 249 | + annotations={ |
| 250 | + "destructiveHint": True, |
| 251 | + "idempotentHint": True, |
| 252 | + "tags": ["category"], |
| 253 | + "title": "Remove Category", |
| 254 | + } |
| 255 | + ) |
| 256 | + @log_tool_usage |
| 257 | + async def ha_config_remove_category( |
| 258 | + scope: Annotated[ |
| 259 | + str, |
| 260 | + Field( |
| 261 | + description="Domain scope for the category (e.g., 'automation', 'script', 'scene', 'helpers').", |
| 262 | + ), |
| 263 | + ], |
| 264 | + category_id: Annotated[ |
| 265 | + str, |
| 266 | + Field(description="ID of the category to delete"), |
| 267 | + ], |
| 268 | + ) -> dict[str, Any]: |
| 269 | + """ |
| 270 | + Delete a Home Assistant category. |
| 271 | +
|
| 272 | + Removes the category from the category registry for the given scope |
| 273 | + (e.g., 'automation', 'script', 'scene', 'helpers'). |
| 274 | + This will also remove the category assignment from all entities in that scope. |
| 275 | +
|
| 276 | + EXAMPLES: |
| 277 | + - Delete category: ha_config_remove_category("automation", "my_category_id") |
| 278 | +
|
| 279 | + Use ha_config_get_category() to find category IDs. |
| 280 | +
|
| 281 | + **WARNING:** Deleting a category will remove it from all assigned entities. |
| 282 | + This action cannot be undone. |
| 283 | + """ |
| 284 | + try: |
| 285 | + message: dict[str, Any] = { |
| 286 | + "type": "config/category_registry/delete", |
| 287 | + "scope": scope, |
| 288 | + "category_id": category_id, |
| 289 | + } |
| 290 | + |
| 291 | + result = await client.send_websocket_message(message) |
| 292 | + |
| 293 | + if result.get("success"): |
| 294 | + return { |
| 295 | + "success": True, |
| 296 | + "category_id": category_id, |
| 297 | + "scope": scope, |
| 298 | + "message": f"Successfully deleted category: {category_id}", |
| 299 | + } |
| 300 | + else: |
| 301 | + raise_tool_error( |
| 302 | + create_error_response( |
| 303 | + ErrorCode.SERVICE_CALL_FAILED, |
| 304 | + f"Failed to delete category: {result.get('error', 'Unknown error')}", |
| 305 | + context={"category_id": category_id, "scope": scope}, |
| 306 | + ) |
| 307 | + ) |
| 308 | + |
| 309 | + except ToolError: |
| 310 | + raise |
| 311 | + except Exception as e: |
| 312 | + logger.error(f"Error removing category {category_id!r}: {e}") |
| 313 | + exception_to_structured_error( |
| 314 | + e, |
| 315 | + context={"category_id": category_id, "scope": scope}, |
| 316 | + suggestions=[ |
| 317 | + "Check Home Assistant connection", |
| 318 | + "Verify the category_id exists using ha_config_get_category()", |
| 319 | + ], |
| 320 | + ) |
0 commit comments