Hass-MCP

Instructions

Implementation Reference

• app/server.py:360-491 (handler)

  Implementation of the search_entities_tool handler. Decorated with @mcp.tool() for MCP registration and @async_handler for logging. Handles entity search queries, empty queries as all-entities listing, simplifies results with domain counts and key attributes per domain.

  @mcp.tool()
  @async_handler("search_entities_tool")
  async def search_entities_tool(query: str, limit: int = 20) -> Dict[str, Any]:
      """
      Search for entities matching a query string
      
      Args:
          query: The search query to match against entity IDs, names, and attributes.
                (Note: Does not support wildcards. To get all entities, leave this blank or use list_entities tool)
          limit: Maximum number of results to return (default: 20)
      
      Returns:
          A dictionary containing search results and metadata:
          - count: Total number of matching entities found
          - results: List of matching entities with essential information
          - domains: Map of domains with counts (e.g. {"light": 3, "sensor": 2})
          
      Examples:
          query="temperature" - find temperature entities
          query="living room", limit=10 - find living room entities
          query="", limit=500 - list all entity types
          
      """
      logger.info(f"Searching for entities matching: '{query}' with limit: {limit}")
      
      # Special case - treat "*" as empty query to just return entities without filtering
      if query == "*":
          query = ""
          logger.info("Converting '*' to empty query (retrieving all entities up to limit)")
      
      # Handle empty query as a special case to just return entities up to the limit
      if not query or not query.strip():
          logger.info(f"Empty query - retrieving up to {limit} entities without filtering")
          entities = await get_entities(limit=limit, lean=True)
          
          # Check if there was an error
          if isinstance(entities, dict) and "error" in entities:
              return {"error": entities["error"], "count": 0, "results": [], "domains": {}}
          
          # No query, but we'll return a structured result anyway
          domains_count = {}
          simplified_entities = []
          
          for entity in entities:
              domain = entity["entity_id"].split(".")[0]
              
              # Count domains
              if domain not in domains_count:
                  domains_count[domain] = 0
              domains_count[domain] += 1
              
              # Create simplified entity representation
              simplified_entity = {
                  "entity_id": entity["entity_id"],
                  "state": entity["state"],
                  "domain": domain,
                  "friendly_name": entity.get("attributes", {}).get("friendly_name", entity["entity_id"])
              }
              
              # Add key attributes based on domain
              attributes = entity.get("attributes", {})
              
              # Include domain-specific important attributes
              if domain == "light" and "brightness" in attributes:
                  simplified_entity["brightness"] = attributes["brightness"]
              elif domain == "sensor" and "unit_of_measurement" in attributes:
                  simplified_entity["unit"] = attributes["unit_of_measurement"]
              elif domain == "climate" and "temperature" in attributes:
                  simplified_entity["temperature"] = attributes["temperature"]
              elif domain == "media_player" and "media_title" in attributes:
                  simplified_entity["media_title"] = attributes["media_title"]
              
              simplified_entities.append(simplified_entity)
          
          # Return structured response for empty query
          return {
              "count": len(simplified_entities),
              "results": simplified_entities,
              "domains": domains_count,
              "query": "all entities (no filtering)"
          }
      
      # Normal search with non-empty query
      entities = await get_entities(search_query=query, limit=limit, lean=True)
      
      # Check if there was an error
      if isinstance(entities, dict) and "error" in entities:
          return {"error": entities["error"], "count": 0, "results": [], "domains": {}}
      
      # Prepare the results
      domains_count = {}
      simplified_entities = []
      
      for entity in entities:
          domain = entity["entity_id"].split(".")[0]
          
          # Count domains
          if domain not in domains_count:
              domains_count[domain] = 0
          domains_count[domain] += 1
          
          # Create simplified entity representation
          simplified_entity = {
              "entity_id": entity["entity_id"],
              "state": entity["state"],
              "domain": domain,
              "friendly_name": entity.get("attributes", {}).get("friendly_name", entity["entity_id"])
          }
          
          # Add key attributes based on domain
          attributes = entity.get("attributes", {})
          
          # Include domain-specific important attributes
          if domain == "light" and "brightness" in attributes:
              simplified_entity["brightness"] = attributes["brightness"]
          elif domain == "sensor" and "unit_of_measurement" in attributes:
              simplified_entity["unit"] = attributes["unit_of_measurement"]
          elif domain == "climate" and "temperature" in attributes:
              simplified_entity["temperature"] = attributes["temperature"]
          elif domain == "media_player" and "media_title" in attributes:
              simplified_entity["media_title"] = attributes["media_title"]
          
          simplified_entities.append(simplified_entity)
      
      # Return structured response
      return {
          "count": len(simplified_entities),
          "results": simplified_entities,
          "domains": domains_count,
          "query": query
      }

• app/server.py:362-382 (schema)

  Type hints and comprehensive docstring defining input schema (query: str, limit: int=20) and output schema (Dict[str, Any] with specific keys: count, results, domains, query or error).

  async def search_entities_tool(query: str, limit: int = 20) -> Dict[str, Any]:
      """
      Search for entities matching a query string
      
      Args:
          query: The search query to match against entity IDs, names, and attributes.
                (Note: Does not support wildcards. To get all entities, leave this blank or use list_entities tool)
          limit: Maximum number of results to return (default: 20)
      
      Returns:
          A dictionary containing search results and metadata:
          - count: Total number of matching entities found
          - results: List of matching entities with essential information
          - domains: Map of domains with counts (e.g. {"light": 3, "sensor": 2})
          
      Examples:
          query="temperature" - find temperature entities
          query="living room", limit=10 - find living room entities
          query="", limit=500 - list all entity types
          
      """

• app/server.py:360-361 (registration)

  MCP tool registration via @mcp.tool() decorator and named via @async_handler("search_entities_tool").

  @mcp.tool()
  @async_handler("search_entities_tool")