hass-mcp-plus

Instructions

Implementation Reference

• app/server.py:116-149 (handler)

  Main handler function for the entity_action tool. Validates the action (on/off/toggle), maps it to the appropriate Home Assistant service name, extracts the domain from the entity_id, and calls the hass_call_service helper to execute the service call. Decorated with @mcp.tool() for registration and @async_handler for logging.

  @mcp.tool()
  @async_handler("entity_action")
  async def entity_action(entity_id: str, action: str, params: Optional[Dict[str, Any]] = None) -> dict:
      """
      Perform an action on a Home Assistant entity (on, off, toggle).
  
      Args:
          entity_id: Entity ID to control (e.g. 'light.living_room')
          action: 'on', 'off', or 'toggle'
          params: Additional service parameters (e.g. {"brightness": 255, "temperature": 22.5})
  
      Domain-specific params:
          Lights: brightness (0-255), color_temp, rgb_color, transition, effect
          Covers: position (0-100), tilt_position
          Climate: temperature, target_temp_high, target_temp_low, hvac_mode
          Media players: source, volume_level (0-1)
  
      Examples: entity_action("light.living_room", "on", {"brightness": 255})
                entity_action("switch.garden_lights", "off")
      """
      if action not in ["on", "off", "toggle"]:
          return {"error": f"Invalid action: {action}. Valid actions are 'on', 'off', 'toggle'"}
      
      # Map action to service name
      service = action if action == "toggle" else f"turn_{action}"
      
      # Extract the domain from the entity_id
      domain = entity_id.split(".")[0]
      
      # Prepare service data
      data = {"entity_id": entity_id, **(params or {})}
      
      logger.info(f"Performing action '{action}' on entity: {entity_id} with params: {sanitize_for_logging(params)}")
      return await hass_call_service(domain, service, data)

• app/hass/services.py:18-45 (helper)

  Core helper function that executes Home Assistant service calls. Validates the domain and service identifiers, rate-limits requests, and makes the HTTP POST request to the Home Assistant API endpoint. Returns the service call result or an empty dict if the response is an empty list.

  @handle_api_errors
  async def call_service(domain: str, service: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """Call a Home Assistant service"""
      validate_ha_identifier(domain, "domain")
      validate_ha_identifier(service, "service")
  
      if data is None:
          data = {}
      else:
          validate_service_payload(data)
  
      await _rate_limiter.acquire()
      client = await get_client()
      response = await client.post(
          f"{HA_URL}/api/services/{safe_url_path_segment(domain)}/{safe_url_path_segment(service)}",
          headers=get_ha_headers(),
          json=data
      )
      response.raise_for_status()
  
      result = response.json()
  
      # Handle list responses from Home Assistant
      # Service calls return lists (empty or with changed entity data)
      if isinstance(result, list):
          result = {"result": result} if result else {}
  
      return result

• app/server.py:116-116 (registration)

  The @mcp.tool() decorator registers the entity_action function as an MCP tool, making it available for invocation through the Model Context Protocol server.

  @mcp.tool()

• app/hass/validation.py:156-187 (schema)

  Validation function for service call payloads. Ensures the data parameter is JSON-serializable and does not exceed the maximum allowed payload size (MAX_SERVICE_PAYLOAD_BYTES), preventing resource exhaustion attacks.

  def validate_service_payload(
      data: Optional[dict],
      max_bytes: int = MAX_SERVICE_PAYLOAD_BYTES,
  ) -> Optional[dict]:
      """
      Validate service call payload size.
  
      Args:
          data: The service call data payload
          max_bytes: Maximum allowed size in bytes
  
      Returns:
          The validated data
  
      Raises:
          ValidationError: If payload exceeds size limit or is not serializable
      """
      if data is None:
          return data
  
      try:
          payload_size = len(json.dumps(data))
      except (TypeError, ValueError):
          raise ValidationError("Service data must be JSON-serializable")
  
      if payload_size > max_bytes:
          raise ValidationError(
              f"Service data payload too large: {payload_size:,} bytes "
              f"(maximum: {max_bytes:,} bytes)"
          )
  
      return data