hass-mcp-plus

Instructions

Implementation Reference

• app/server.py:286-300 (handler)

  The MCP tool handler for 'domain_summary'. This async function is decorated with @mcp.tool() for registration and @async_handler for logging. It accepts 'domain' and 'example_limit' parameters and delegates to the summarize_domain helper function for the actual implementation logic.

  @mcp.tool()
  @async_handler("domain_summary")
  async def domain_summary(domain: str, example_limit: int = 3) -> Dict[str, Any]:
      """
      Get a summary of entities in a domain (counts, state distribution, examples).
  
      Args:
          domain: Domain to summarize (e.g. 'light', 'switch', 'sensor')
          example_limit: Max examples per state (default: 3)
  
      Examples: domain_summary("light")
                domain_summary("climate", example_limit=5)
      """
      logger.info(f"Getting domain summary for: {domain}")
      return await summarize_domain(domain, example_limit)

• app/hass/summary.py:17-83 (helper)

  The core business logic for domain_summary. This async function fetches entities for the specified domain, counts states, collects examples, and generates a summary including total count, state distribution, example entities, and common attributes. Decorated with @handle_api_errors for error handling.

  async def summarize_domain(domain: str, example_limit: int = 3) -> Dict[str, Any]:
      """
      Generate a summary of entities in a domain
  
      Args:
          domain: The domain to summarize (e.g., 'light', 'switch')
          example_limit: Maximum number of examples to include for each state
  
      Returns:
          Dictionary with summary information
      """
      validate_ha_identifier(domain, "domain")
  
      entities = await get_entities(domain=domain)
  
      # Check if we got an error response
      if isinstance(entities, dict) and "error" in entities:
          return entities  # Just pass through the error
  
      try:
          # Initialize summary data
          total_count = len(entities)
          state_counts = {}
          state_examples = {}
          attributes_summary = {}
  
          # Process entities to build the summary
          for entity in entities:
              state = entity.get("state", "unknown")
  
              # Count states
              if state not in state_counts:
                  state_counts[state] = 0
                  state_examples[state] = []
              state_counts[state] += 1
  
              # Add examples (up to the limit)
              if len(state_examples[state]) < example_limit:
                  example = {
                      "entity_id": entity["entity_id"],
                      "friendly_name": entity.get("attributes", {}).get("friendly_name", entity["entity_id"])
                  }
                  state_examples[state].append(example)
  
              # Collect attribute keys for summary
              for attr_key in entity.get("attributes", {}):
                  if attr_key not in attributes_summary:
                      attributes_summary[attr_key] = 0
                  attributes_summary[attr_key] += 1
  
          # Create the summary
          summary = {
              "domain": domain,
              "total_count": total_count,
              "state_distribution": state_counts,
              "examples": state_examples,
              "common_attributes": sorted(
                  [(k, v) for k, v in attributes_summary.items()],
                  key=lambda x: x[1],
                  reverse=True
              )[:10]  # Top 10 most common attributes
          }
  
          return summary
      except Exception as e:
          logger.error("Error generating domain summary: %s", e)
          return {"error": "Error generating domain summary"}

• app/hass/validation.py:59-80 (schema)

  Input validation function that validates the 'domain' parameter using HA_IDENTIFIER_PATTERN. Ensures the domain is a non-empty string matching the pattern '^[a-z][a-z0-9_]*$' (lowercase letters, digits, underscores). Raises ValidationError if invalid.

  def validate_ha_identifier(value: str, name: str = "identifier") -> str:
      """
      Validate a Home Assistant identifier (domain, service name, etc.).
  
      Args:
          value: The identifier to validate
          name: Human-readable name for error messages (e.g., 'domain', 'service')
  
      Returns:
          The validated identifier
  
      Raises:
          ValidationError: If the identifier format is invalid
      """
      if not value or not isinstance(value, str):
          raise ValidationError(f"{name.capitalize()} must be a non-empty string")
  
      if not HA_IDENTIFIER_PATTERN.match(value):
          raise ValidationError(
              f"Invalid {name}: '{value}'. "
              f"Must contain only lowercase letters, numbers, and underscores."
          )

• app/server.py:286-287 (registration)

  The @mcp.tool() decorator on the domain_summary function registers it as an MCP tool with the FastMCP server, making it available for tool calls.

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