get_entities_by_area

Instructions

Implementation Reference

• app/server.py:377-421 (handler)

  The main handler for the 'get_entities_by_area' MCP tool. It fetches all entities via get_entities() with optional domain filtering, then filters them locally by matching the area name (case-insensitive). Returns a dict with matched area name, entity count, and entity list.

  @mcp.tool()
  @async_handler("get_entities_by_area")
  async def get_entities_by_area(
      area: str,
      domain: Optional[str] = None,
      lean: bool = True,
  ) -> Dict[str, Any]:
      """
      Get all entities assigned to a specific Home Assistant area (room).
  
      Area lookup is case-insensitive and matches the area's name as configured
      in Home Assistant (e.g., "Kitchen", "Living Room"). Entities inherit their
      area from their parent device when no area is set directly, matching HA's
      own resolution behavior.
  
      Args:
          area: Name of the area to filter by (case-insensitive)
          domain: Optional domain to further filter results (e.g., 'light')
          lean: If True (default), returns token-efficient entity records
  
      Returns:
          A dictionary containing:
          - area: The matched area name (as canonicalized by HA)
          - count: Number of matching entities
          - entities: List of entity records with their state and area
  
      Examples:
          get_entities_by_area(area="Kitchen") - everything in the kitchen
          get_entities_by_area(area="Living Room", domain="light") - lights only
      """
      logger.info(f"Listing entities in area: {area} (domain={domain})")
      entities = await get_entities(domain=domain, limit=10_000, lean=lean)
  
      needle = area.strip().lower()
      matches = [
          e for e in entities
          if (e.get("area") or "").lower() == needle
      ]
      canonical_area = matches[0]["area"] if matches else area
      return {
          "area": canonical_area,
          "count": len(matches),
          "entities": matches,
      }

• app/server.py:377-378 (registration)

  Registration of the tool via @mcp.tool() decorator on the get_entities_by_area function.

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

• app/hass.py:255-355 (helper)

  The get_entities() helper function called by the tool handler to fetch entities from HA. It retrieves all entity states from /api/states, enriches them with area data via get_all_areas(), applies domain/search filters, limits, and field formatting.

  @handle_api_errors
  async def get_entities(
      domain: Optional[str] = None, 
      search_query: Optional[str] = None, 
      limit: int = 100,
      fields: Optional[List[str]] = None,
      lean: bool = True
  ) -> List[Dict[str, Any]]:
      """
      Get a list of all entities from Home Assistant with optional filtering and search
      
      Args:
          domain: Optional domain to filter entities by (e.g., 'light', 'switch')
          search_query: Optional case-insensitive search term to filter by entity_id, friendly_name or other attributes
          limit: Maximum number of entities to return (default: 100)
          fields: Optional list of specific fields to include in each entity
          lean: If True (default), returns token-efficient versions with minimal fields
      
      Returns:
          List of entity dictionaries, optionally filtered by domain and search terms,
          and optionally limited to specific fields
      """
      # Get all entities directly
      client = await get_client()
      response = await client.get(f"{HA_URL}/api/states", headers=get_ha_headers())
      response.raise_for_status()
      entities = response.json()
  
      # Enrich each entity with area data. One bulk template call covers
      # the whole list regardless of size.
      areas = await get_all_areas(client)
      for entity in entities:
          entity["area"] = areas.get(entity["entity_id"])
  
      # Filter by domain if specified
      if domain:
          entities = [entity for entity in entities if entity["entity_id"].startswith(f"{domain}.")]
      
      # Search if query is provided
      if search_query and search_query.strip():
          search_term = search_query.lower().strip()
          filtered_entities = []
          
          for entity in entities:
              # Search in entity_id
              if search_term in entity["entity_id"].lower():
                  filtered_entities.append(entity)
                  continue
                  
              # Search in friendly_name
              friendly_name = entity.get("attributes", {}).get("friendly_name", "").lower()
              if friendly_name and search_term in friendly_name:
                  filtered_entities.append(entity)
                  continue
                  
              # Search in other common attributes (state, area_id, etc.)
              if search_term in entity.get("state", "").lower():
                  filtered_entities.append(entity)
                  continue
                  
              # Search in other attributes
              for attr_name, attr_value in entity.get("attributes", {}).items():
                  # Check if attribute value can be converted to string
                  if isinstance(attr_value, (str, int, float, bool)):
                      if search_term in str(attr_value).lower():
                          filtered_entities.append(entity)
                          break
          
          entities = filtered_entities
      
      # Apply the limit
      if limit > 0 and len(entities) > limit:
          entities = entities[:limit]
      
      # Apply field filtering if requested
      if fields:
          # Use explicit field list when provided
          return [filter_fields(entity, fields) for entity in entities]
      elif lean:
          # Apply domain-specific lean fields to each entity
          result = []
          for entity in entities:
              # Get the entity's domain
              entity_domain = entity["entity_id"].split('.')[0]
              
              # Start with basic lean fields
              lean_fields = DEFAULT_LEAN_FIELDS.copy()
              
              # Add domain-specific important attributes
              if entity_domain in DOMAIN_IMPORTANT_ATTRIBUTES:
                  for attr in DOMAIN_IMPORTANT_ATTRIBUTES[entity_domain]:
                      lean_fields.append(f"attr.{attr}")
              
              # Filter and add to result
              result.append(filter_fields(entity, lean_fields))
          
          return result
      else:
          # Return full entities
          return entities

• app/areas.py:119-134 (helper)

  Area resolution helper used by get_entities() to enrich entities with their area name. Uses a Jinja template via HA's /api/template endpoint to resolve entity->device->area mappings, cached with a 5-minute TTL.

  # Module-level singleton. Tests can reset via .invalidate() / monkeypatch.
  _cache = AreaCache()
  
  
  async def get_area(client: httpx.AsyncClient, entity_id: str) -> Optional[str]:
      return await _cache.get(client, entity_id)
  
  
  async def get_all_areas(client: httpx.AsyncClient) -> Dict[str, Optional[str]]:
      return await _cache.get_all(client)
  
  
  def invalidate_cache() -> None:
      _cache.invalidate()

• app/server.py:379-383 (schema)

  Tool input schema defined by function parameters: area (str, required), domain (Optional[str]), lean (bool, default True). Return type is Dict[str, Any] with 'area', 'count', and 'entities' keys.

  async def get_entities_by_area(
      area: str,
      domain: Optional[str] = None,
      lean: bool = True,
  ) -> Dict[str, Any]: