diff --git a/.github/actions/run-tests/action.yml b/.github/actions/run-tests/action.yml
index 9a07e71972..4e058dc64f 100644
--- a/.github/actions/run-tests/action.yml
+++ b/.github/actions/run-tests/action.yml
@@ -78,7 +78,22 @@ runs:
         echo "::endgroup::"
 
         echo "::group::Starting Redis servers"
-        redis_major_version=$(echo "$REDIS_VERSION" | grep -oP '^\d+')
+        # Check if REDIS_VERSION is in the custom map
+        mapped_version=""
+        if [[ -n "${REDIS_VERSION_CUSTOM_MAP:-}" ]]; then
+          for mapping in $REDIS_VERSION_CUSTOM_MAP; do
+            tag="${mapping%%:*}"
+            version="${mapping##*:}"
+            if [[ "$REDIS_VERSION" == "$tag" ]]; then
+              mapped_version="$version"
+              echo "Found custom mapping: $REDIS_VERSION -> $mapped_version"
+              break
+            fi
+          done
+        fi
+        # Use mapped version if found, otherwise use REDIS_VERSION
+        version_to_parse="${mapped_version:-$REDIS_VERSION}"
+        redis_major_version=$(echo "$version_to_parse" | grep -oP '^\d+')
         echo "REDIS_MAJOR_VERSION=${redis_major_version}" >> $GITHUB_ENV
 
         if (( redis_major_version < 8 )); then
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 78ba412164..43a7929ebc 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -31,6 +31,7 @@ env:
   # for example after 8.2.1 is published, 8.2 image contains 8.2.1 content
   CURRENT_CLIENT_LIBS_TEST_STACK_IMAGE_TAG: '8.4.0'
   CURRENT_REDIS_VERSION: '8.4.0'
+  REDIS_VERSION_CUSTOM_MAP: 'custom-21651605017-debian-amd64:8.6'
 
 jobs:
   dependency-audit:
@@ -76,7 +77,7 @@ jobs:
       max-parallel: 15
       fail-fast: false
       matrix:
-        redis-version: ['8.6-rc1-21356658603-debian-amd64', '${{ needs.redis_version.outputs.CURRENT }}', '8.2', '8.0.2' ,'7.4.4', '7.2.9']
+        redis-version: ['custom-21651605017-debian-amd64', '${{ needs.redis_version.outputs.CURRENT }}', '8.2', '8.0.2' ,'7.4.4', '7.2.9']
         python-version: ['3.10', '3.14']
         parser-backend: ['plain']
         event-loop: ['asyncio']
diff --git a/redis/_parsers/helpers.py b/redis/_parsers/helpers.py
index b6c3feb877..26bcd0e0d5 100644
--- a/redis/_parsers/helpers.py
+++ b/redis/_parsers/helpers.py
@@ -880,6 +880,7 @@ def string_keys_to_dict(key_string, callback):
         map(lambda ll: (float(ll[0]), float(ll[1])) if ll is not None else None, r)
     ),
     "HGETALL": lambda r: r and pairs_to_dict(r) or {},
+    "HOTKEYS GET": lambda r: [pairs_to_dict(m) for m in r],
     "MEMORY STATS": parse_memory_stats,
     "MODULE LIST": lambda r: [pairs_to_dict(m) for m in r],
     "RESET": str_if_bytes,
diff --git a/redis/commands/cluster.py b/redis/commands/cluster.py
index 568f4b4914..dea73763da 100644
--- a/redis/commands/cluster.py
+++ b/redis/commands/cluster.py
@@ -3,6 +3,7 @@
     TYPE_CHECKING,
     Any,
     AsyncIterator,
+    Awaitable,
     Dict,
     Iterable,
     Iterator,
@@ -38,6 +39,7 @@
     AsyncScriptCommands,
     DataAccessCommands,
     FunctionCommands,
+    HotkeysMetricsTypes,
     ManagementCommands,
     ModuleCommands,
     PubSubCommands,
@@ -827,6 +829,54 @@ def client_tracking_off(
             target_nodes=target_nodes,
         )
 
+    def hotkeys_start(
+        self,
+        metrics: List[HotkeysMetricsTypes],
+        count: Optional[int] = None,
+        duration: Optional[int] = None,
+        sample_ratio: Optional[int] = None,
+        slots: Optional[List[int]] = None,
+        **kwargs,
+    ) -> Union[str, bytes]:
+        """
+        Cluster client does not support hotkeys command. Please use the non-cluster client.
+
+        For more information see https://redis.io/commands/hotkeys-start
+        """
+        raise NotImplementedError(
+            "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
+        )
+
+    def hotkeys_stop(self, **kwargs) -> Union[str, bytes]:
+        """
+        Cluster client does not support hotkeys command. Please use the non-cluster client.
+
+        For more information see https://redis.io/commands/hotkeys-stop
+        """
+        raise NotImplementedError(
+            "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
+        )
+
+    def hotkeys_reset(self, **kwargs) -> Union[str, bytes]:
+        """
+        Cluster client does not support hotkeys command. Please use the non-cluster client.
+
+        For more information see https://redis.io/commands/hotkeys-reset
+        """
+        raise NotImplementedError(
+            "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
+        )
+
+    def hotkeys_get(self, **kwargs) -> list[dict[Union[str, bytes], Any]]:
+        """
+        Cluster client does not support hotkeys command. Please use the non-cluster client.
+
+        For more information see https://redis.io/commands/hotkeys-get
+        """
+        raise NotImplementedError(
+            "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
+        )
+
 
 class AsyncClusterManagementCommands(
     ClusterManagementCommands, AsyncManagementCommands
@@ -924,6 +974,56 @@ async def client_tracking_off(
             target_nodes=target_nodes,
         )
 
+    async def hotkeys_start(
+        self,
+        metrics: List[HotkeysMetricsTypes],
+        count: Optional[int] = None,
+        duration: Optional[int] = None,
+        sample_ratio: Optional[int] = None,
+        slots: Optional[List[int]] = None,
+        **kwargs,
+    ) -> Awaitable[Union[str, bytes]]:
+        """
+        Cluster client does not support hotkeys command. Please use the non-cluster client.
+
+        For more information see https://redis.io/commands/hotkeys-start
+        """
+        raise NotImplementedError(
+            "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
+        )
+
+    async def hotkeys_stop(self, **kwargs) -> Awaitable[Union[str, bytes]]:
+        """
+        Cluster client does not support hotkeys command. Please use the non-cluster client.
+
+        For more information see https://redis.io/commands/hotkeys-stop
+        """
+        raise NotImplementedError(
+            "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
+        )
+
+    async def hotkeys_reset(self, **kwargs) -> Awaitable[Union[str, bytes]]:
+        """
+        Cluster client does not support hotkeys command. Please use the non-cluster client.
+
+        For more information see https://redis.io/commands/hotkeys-reset
+        """
+        raise NotImplementedError(
+            "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
+        )
+
+    async def hotkeys_get(
+        self, **kwargs
+    ) -> Awaitable[list[dict[Union[str, bytes], Any]]]:
+        """
+        Cluster client does not support hotkeys command. Please use the non-cluster client.
+
+        For more information see https://redis.io/commands/hotkeys-get
+        """
+        raise NotImplementedError(
+            "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
+        )
+
 
 class ClusterDataAccessCommands(DataAccessCommands):
     """
diff --git a/redis/commands/core.py b/redis/commands/core.py
index fc5f276342..1b5f8ea027 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -405,6 +405,11 @@ def acl_whoami(self, **kwargs) -> ResponseT:
 AsyncACLCommands = ACLCommands
 
 
+class HotkeysMetricsTypes(Enum):
+    CPU = "CPU"
+    NET = "NET"
+
+
 class ManagementCommands(CommandsProtocol):
     """
     Redis management commands
@@ -1406,6 +1411,94 @@ def failover(self):
             "FAILOVER is intentionally not implemented in the client."
         )
 
+    def hotkeys_start(
+        self,
+        metrics: List[HotkeysMetricsTypes],
+        count: Optional[int] = None,
+        duration: Optional[int] = None,
+        sample_ratio: Optional[int] = None,
+        slots: Optional[List[int]] = None,
+        **kwargs,
+    ) -> Union[Awaitable[Union[str, bytes]], Union[str, bytes]]:
+        """
+        Start collecting hotkeys data.
+        Returns an error if there is an ongoing collection session.
+
+        Args:
+            count: The number of keys to collect in each criteria (CPU and network consumption)
+            metrics: List of metrics to track. Supported values: [HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET]
+            duration: Automatically stop the collection after `duration` seconds
+            sample_ratio: Commands are sampled with probability 1/ratio (1 means no sampling)
+            slots: Only track keys on the specified hash slots
+
+        For more information, see https://redis.io/commands/hotkeys-start
+        """
+        args: List[Union[str, int]] = ["HOTKEYS", "START"]
+
+        # Add METRICS
+        args.extend(["METRICS", len(metrics)])
+        args.extend([str(m.value) for m in metrics])
+
+        # Add COUNT
+        if count is not None:
+            args.extend(["COUNT", count])
+
+        # Add optional DURATION
+        if duration is not None:
+            args.extend(["DURATION", duration])
+
+        # Add optional SAMPLE ratio
+        if sample_ratio is not None:
+            args.extend(["SAMPLE", sample_ratio])
+
+        # Add optional SLOTS
+        if slots is not None:
+            args.append("SLOTS")
+            args.append(len(slots))
+            args.extend(slots)
+
+        return self.execute_command(*args, **kwargs)
+
+    def hotkeys_stop(
+        self, **kwargs
+    ) -> Union[Awaitable[Union[str, bytes]], Union[str, bytes]]:
+        """
+        Stop the ongoing hotkeys collection session (if any).
+        The results of the last collection session are kept for consumption with HOTKEYS GET.
+
+        For more information, see https://redis.io/commands/hotkeys-stop
+        """
+        return self.execute_command("HOTKEYS STOP", **kwargs)
+
+    def hotkeys_reset(
+        self, **kwargs
+    ) -> Union[Awaitable[Union[str, bytes]], Union[str, bytes]]:
+        """
+        Discard the last hotkeys collection session results (in order to save memory).
+        Error if there is an ongoing collection session.
+
+        For more information, see https://redis.io/commands/hotkeys-reset
+        """
+        return self.execute_command("HOTKEYS RESET", **kwargs)
+
+    def hotkeys_get(
+        self, **kwargs
+    ) -> Union[
+        Awaitable[list[dict[Union[str, bytes], Any]]],
+        list[dict[Union[str, bytes], Any]],
+    ]:
+        """
+        Retrieve the result of the ongoing collection session (if any),
+        or the last collection session (if any).
+
+        HOTKEYS GET response is wrapped in an array for aggregation support.
+        Each node returns a single-element array, allowing multiple node
+        responses to be concatenated by DMC or other aggregators.
+
+        For more information, see https://redis.io/commands/hotkeys-get
+        """
+        return self.execute_command("HOTKEYS GET", **kwargs)
+
 
 class AsyncManagementCommands(ManagementCommands):
     async def command_info(self, **kwargs) -> None:
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index a0c5fb5fc1..cca82f2b93 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -26,6 +26,7 @@
     LoadBalancingStrategy,
     get_node_name,
 )
+from redis.commands.core import HotkeysMetricsTypes
 from redis.crc import REDIS_CLUSTER_HASH_SLOTS, key_slot
 from redis.exceptions import (
     AskError,
@@ -2465,6 +2466,19 @@ async def test_acl_log(
 
         await user_client.aclose()
 
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_cluster(self, r: RedisCluster) -> None:
+        """Test all HOTKEYS commands in cluster are raising an error"""
+
+        with pytest.raises(NotImplementedError):
+            await r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+        with pytest.raises(NotImplementedError):
+            await r.hotkeys_get()
+        with pytest.raises(NotImplementedError):
+            await r.hotkeys_reset()
+        with pytest.raises(NotImplementedError):
+            await r.hotkeys_stop()
+
 
 class TestNodesManager:
     """
diff --git a/tests/test_asyncio/test_commands.py b/tests/test_asyncio/test_commands.py
index c7ec18e60b..f75948e561 100644
--- a/tests/test_asyncio/test_commands.py
+++ b/tests/test_asyncio/test_commands.py
@@ -21,7 +21,7 @@
     parse_info,
 )
 from redis.client import EMPTY_RESPONSE, NEVER_DECODE
-from redis.commands.core import DataPersistOptions
+from redis.commands.core import DataPersistOptions, HotkeysMetricsTypes
 from redis.commands.json.path import Path
 from redis.commands.search.field import TextField
 from redis.commands.search.query import Query
@@ -810,6 +810,312 @@ async def test_time(self, r: redis.Redis):
         assert isinstance(t[0], int)
         assert isinstance(t[1], int)
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_start_basic(self, r: redis.Redis):
+        """Test basic HOTKEYS START command with CPU metric"""
+        # Reset any previous session
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start collection with CPU metric
+        result = await r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_start_with_all_metrics(self, r: redis.Redis):
+        """Test HOTKEYS START with both CPU and NET metrics"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        result = await r.hotkeys_start(
+            count=5, metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET]
+        )
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_start_with_duration(self, r: redis.Redis):
+        """Test HOTKEYS START with duration parameter"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        result = await r.hotkeys_start(
+            count=10, metrics=[HotkeysMetricsTypes.CPU], duration=60
+        )
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_start_with_sample_ratio(self, r: redis.Redis):
+        """Test HOTKEYS START with sample ratio"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        result = await r.hotkeys_start(
+            count=10, metrics=[HotkeysMetricsTypes.CPU], sample_ratio=10
+        )
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_start_with_slots_fail_on_non_cluster_setup(
+        self, r: redis.Redis
+    ):
+        """Test HOTKEYS START with specific hash slots"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+        with pytest.raises(Exception):
+            await r.hotkeys_start(
+                count=10, metrics=[HotkeysMetricsTypes.CPU], slots=[0, 100, 200]
+            )
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_start_with_all_parameters(self, r: redis.Redis):
+        """Test HOTKEYS START with all optional parameters"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        result = await r.hotkeys_start(
+            count=5,
+            metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET],
+            duration=30,
+            sample_ratio=5,
+        )
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_stop(self, r: redis.Redis):
+        """Test HOTKEYS STOP command"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session
+        await r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+
+        # Stop the session
+        result = await r.hotkeys_stop()
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_reset(self, r: redis.Redis):
+        """Test HOTKEYS RESET command"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a session and generate some data
+        await r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+
+        # Perform some operations to generate hotkeys data
+        for i in range(5):
+            await r.set(f"resetkey{i}", f"value{i}")
+            await r.get(f"resetkey{i}")
+
+        # Stop the session
+        await r.hotkeys_stop()
+
+        # Get results before reset - should have data
+        result_before = await r.hotkeys_get()
+        assert isinstance(result_before, list)
+        for res_elem in result_before:
+            assert isinstance(res_elem, dict)
+            assert b"tracking-active" in res_elem
+
+        # Reset the results
+        result = await r.hotkeys_reset()
+        assert result == b"OK"
+
+        # Try to get results after reset - should fail or return empty
+        try:
+            result_after = await r.hotkeys_get()
+            # If it doesn't fail, verify the data is cleared
+            # The response might indicate no session exists
+            assert result_after != result_before
+        except Exception:
+            # Expected - no session exists after reset
+            pass
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_get_ongoing_session(self, r: redis.Redis):
+        """Test HOTKEYS GET during an ongoing collection session"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session
+        await r.hotkeys_start(
+            count=10, metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET]
+        )
+
+        # Perform some operations to generate hotkeys data
+        for i in range(10):
+            await r.set(f"key{i}", f"value{i}")
+            await r.get(f"key{i}")
+
+        # Get the results
+        result = await r.hotkeys_get()
+
+        # Verify the response structure
+        assert isinstance(result, list)
+        for res_elem in result:
+            assert isinstance(res_elem, dict)
+            # Check tracking-active is 1 (ongoing session)
+            assert res_elem[b"tracking-active"] == 1
+
+        # Stop the session
+        await r.hotkeys_stop()
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_get_terminated_session(self, r: redis.Redis):
+        """Test HOTKEYS GET after stopping a collection session"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session
+        await r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+
+        # Perform some operations
+        for i in range(5):
+            await r.set(f"testkey{i}", f"testvalue{i}")
+
+        # Stop the session
+        await r.hotkeys_stop()
+
+        # Get the results
+        result = await r.hotkeys_get()
+
+        # Verify the response structure
+        assert isinstance(result, list)
+        for res_elem in result:
+            assert isinstance(res_elem, dict)
+            # Check tracking-active is 0 (terminated session)
+            assert res_elem[b"tracking-active"] == 0
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_get_all_fields(self, r: redis.Redis):
+        """Test HOTKEYS GET returns all documented fields"""
+        try:
+            await r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session with all parameters
+        await r.hotkeys_start(
+            count=5,
+            metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET],
+            sample_ratio=1,
+        )
+
+        # Perform operations to generate data
+        for i in range(20):
+            await r.set("anyprefix:{3}:key", f"value{i}")
+            await r.get("anyprefix:{3}:key")
+            await r.set("anyprefix:{1}:key", f"value{i}")
+            await r.get("anyprefix:{1}:key")
+
+        # Stop the session
+        await r.hotkeys_stop()
+
+        # Get the results
+        result = await r.hotkeys_get()
+
+        # Verify all documented fields are present
+        expected_fields = [
+            b"tracking-active",
+            b"sample-ratio",
+            b"selected-slots",
+            b"net-bytes-all-commands-all-slots",
+            b"collection-start-time-unix-ms",
+            b"collection-duration-ms",
+            b"total-cpu-time-user-ms",
+            b"total-cpu-time-sys-ms",
+            b"total-net-bytes",
+            b"by-cpu-time-us",
+            b"by-net-bytes",
+        ]
+
+        for elem in result:
+            for field in expected_fields:
+                assert field in elem, (
+                    f"Field '{field}' is missing from HOTKEYS GET response"
+                )
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    async def test_hotkeys_get_all_fields_decoded(self, decoded_r: redis.Redis):
+        """Test HOTKEYS GET returns all documented fields"""
+        try:
+            await decoded_r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session with all parameters
+        await decoded_r.hotkeys_start(
+            count=5,
+            metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET],
+            sample_ratio=1,
+        )
+
+        # Perform operations to generate data
+        for i in range(20):
+            await decoded_r.set("anyprefix:{3}:key", f"value{i}")
+            await decoded_r.get(f"anyprefix:{3}:key")
+            await decoded_r.set("anyprefix:{1}:key", f"value{i}")
+            await decoded_r.get(f"anyprefix:{1}:key")
+
+        # Stop the session
+        await decoded_r.hotkeys_stop()
+
+        # Get the results
+        result = await decoded_r.hotkeys_get()
+
+        # Verify all documented fields are present
+        expected_fields = [
+            "tracking-active",
+            "sample-ratio",
+            "selected-slots",
+            "net-bytes-all-commands-all-slots",
+            "collection-start-time-unix-ms",
+            "collection-duration-ms",
+            "total-cpu-time-user-ms",
+            "total-cpu-time-sys-ms",
+            "total-net-bytes",
+            "by-cpu-time-us",
+            "by-net-bytes",
+        ]
+
+        for elem in result:
+            for field in expected_fields:
+                assert field in elem, (
+                    f"Field '{field}' is missing from HOTKEYS GET response"
+                )
+
     async def test_never_decode_option(self, r: redis.Redis):
         opts = {NEVER_DECODE: []}
         await r.delete("a")
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index 36b8ef65a6..d23cf7aaa6 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -29,6 +29,7 @@
     RedisCluster,
     get_node_name,
 )
+from redis.commands.core import HotkeysMetricsTypes
 from redis.connection import BlockingConnectionPool, Connection, ConnectionPool
 from redis.crc import key_slot
 from redis.exceptions import (
@@ -2608,6 +2609,19 @@ def try_delete_libs(self, r, *lib_names):
             except Exception:
                 pass
 
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_cluster(self, r):
+        """Test all HOTKEYS commands in cluster mode are raising an error"""
+
+        with pytest.raises(NotImplementedError):
+            r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+        with pytest.raises(NotImplementedError):
+            r.hotkeys_get()
+        with pytest.raises(NotImplementedError):
+            r.hotkeys_reset()
+        with pytest.raises(NotImplementedError):
+            r.hotkeys_stop()
+
 
 @pytest.mark.onlycluster
 class TestNodesManager:
diff --git a/tests/test_commands.py b/tests/test_commands.py
index e63c9a1e81..8967b6dd71 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -19,7 +19,7 @@
     parse_info,
 )
 from redis.client import EMPTY_RESPONSE, NEVER_DECODE
-from redis.commands.core import DataPersistOptions
+from redis.commands.core import DataPersistOptions, HotkeysMetricsTypes
 from redis.commands.json.path import Path
 from redis.commands.search.field import TextField
 from redis.commands.search.query import Query
@@ -1229,6 +1229,314 @@ def test_time(self, r):
         assert isinstance(t[0], int)
         assert isinstance(t[1], int)
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_start_basic(self, r):
+        """Test basic HOTKEYS START command with CPU metric"""
+        # Reset any previous session
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start collection with CPU metric
+        result = r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_start_with_all_metrics(self, r):
+        """Test HOTKEYS START with both CPU and NET metrics"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        result = r.hotkeys_start(
+            count=5, metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET]
+        )
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_start_with_duration(self, r):
+        """Test HOTKEYS START with duration parameter"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        result = r.hotkeys_start(
+            count=10, metrics=[HotkeysMetricsTypes.CPU], duration=60
+        )
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_start_with_sample_ratio(self, r):
+        """Test HOTKEYS START with sample ratio"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        result = r.hotkeys_start(
+            count=10, metrics=[HotkeysMetricsTypes.CPU], sample_ratio=10
+        )
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_start_fail__on_noncluster_setup_with_slots(self, r):
+        """Test HOTKEYS START with specific hash slots"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # slots is not supported argument for non-cluster setups
+        with pytest.raises(Exception):
+            r.hotkeys_start(
+                count=10, metrics=[HotkeysMetricsTypes.CPU], slots=[0, 100, 200]
+            )
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_start_with_all_parameters(self, r):
+        """Test HOTKEYS START with all optional parameters"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        result = r.hotkeys_start(
+            count=5,
+            metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET],
+            duration=30,
+            sample_ratio=5,
+        )
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_stop(self, r):
+        """Test HOTKEYS STOP command"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session
+        r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+
+        # Stop the session
+        result = r.hotkeys_stop()
+        assert result == b"OK"
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_reset(self, r):
+        """Test HOTKEYS RESET command"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a session and generate some data
+        r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+
+        # Perform some operations to generate hotkeys data
+        for i in range(5):
+            r.set(f"resetkey{i}", f"value{i}")
+            r.get(f"resetkey{i}")
+
+        # Stop the session
+        r.hotkeys_stop()
+
+        # Get results before reset - should have data
+        result_before = r.hotkeys_get()
+        assert isinstance(result_before, list)
+        for res_elem in result_before:
+            assert isinstance(res_elem, dict)
+            assert b"tracking-active" in res_elem
+
+        # Reset the results
+        result = r.hotkeys_reset()
+        assert result == b"OK"
+
+        # Try to get results after reset - should fail or return empty
+        try:
+            result_after = r.hotkeys_get()
+            # If it doesn't fail, verify the data is cleared
+            # The response might indicate no session exists
+            assert result_after != result_before
+        except Exception:
+            # Expected - no session exists after reset
+            pass
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_get_ongoing_session(self, r):
+        """Test HOTKEYS GET during an ongoing collection session"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session
+        r.hotkeys_start(
+            count=10, metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET]
+        )
+
+        # Perform some operations to generate hotkeys data
+        for i in range(10):
+            r.set(f"key{i}", f"value{i}")
+            r.get(f"key{i}")
+
+        # Get the results
+        result = r.hotkeys_get()
+
+        # Verify the response structure
+        assert isinstance(result, list)
+        for res_elem in result:
+            assert isinstance(res_elem, dict)
+            # Check tracking-active is 1 (ongoing session)
+            assert res_elem[b"tracking-active"] == 1
+
+        # Stop the session
+        r.hotkeys_stop()
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_get_terminated_session(self, r):
+        """Test HOTKEYS GET after stopping a collection session"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session
+        r.hotkeys_start(count=10, metrics=[HotkeysMetricsTypes.CPU])
+
+        # Perform some operations
+        for i in range(5):
+            r.set(f"testkey{i}", f"testvalue{i}")
+
+        # Stop the session
+        r.hotkeys_stop()
+
+        # Get the results
+        result = r.hotkeys_get()
+
+        # Verify the response structure
+        assert isinstance(result, list)
+        for res_elem in result:
+            assert isinstance(res_elem, dict)
+
+            # Check tracking-active is 0 (terminated session)
+            assert res_elem[b"tracking-active"] == 0
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_get_all_fields(self, r):
+        """Test HOTKEYS GET returns all documented fields"""
+        try:
+            r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session with all parameters
+        r.hotkeys_start(
+            count=5,
+            metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET],
+            sample_ratio=1,
+        )
+
+        # Perform operations to generate data
+        for i in range(20):
+            r.set("anyprefix:{3}:key", f"value{i}")
+            r.get(f"anyprefix:{3}:key")
+            r.set("anyprefix:{1}:key", f"value{i}")
+            r.get(f"anyprefix:{1}:key")
+
+        # Stop the session
+        r.hotkeys_stop()
+
+        # Get the results
+        result = r.hotkeys_get()
+        assert isinstance(result, list)
+
+        # Verify all documented fields are present
+        expected_fields = [
+            b"tracking-active",
+            b"sample-ratio",
+            b"selected-slots",
+            b"net-bytes-all-commands-all-slots",
+            b"collection-start-time-unix-ms",
+            b"collection-duration-ms",
+            b"total-cpu-time-user-ms",
+            b"total-cpu-time-sys-ms",
+            b"total-net-bytes",
+            b"by-cpu-time-us",
+            b"by-net-bytes",
+        ]
+
+        for res_elem in result:
+            for field in expected_fields:
+                assert field in res_elem, (
+                    f"Field '{field}' is missing from HOTKEYS GET response"
+                )
+
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.5.240")
+    def test_hotkeys_get_all_fields_decoded(self, decoded_r: redis.Redis):
+        """Test HOTKEYS GET returns all documented fields"""
+        try:
+            decoded_r.hotkeys_stop()
+        except Exception:
+            pass
+
+        # Start a collection session with all parameters
+        decoded_r.hotkeys_start(
+            count=5,
+            metrics=[HotkeysMetricsTypes.CPU, HotkeysMetricsTypes.NET],
+            sample_ratio=1,
+        )
+
+        # Perform operations to generate data
+        for i in range(20):
+            decoded_r.set("anyprefix:{3}:key", f"value{i}")
+            decoded_r.get(f"anyprefix:{3}:key")
+            decoded_r.set("anyprefix:{1}:key", f"value{i}")
+            decoded_r.get(f"anyprefix:{1}:key")
+
+        # Stop the session
+        decoded_r.hotkeys_stop()
+
+        # Get the results
+        result = decoded_r.hotkeys_get()
+
+        # Verify all documented fields are present
+        expected_fields = [
+            "tracking-active",
+            "sample-ratio",
+            "selected-slots",
+            "net-bytes-all-commands-all-slots",
+            "collection-start-time-unix-ms",
+            "collection-duration-ms",
+            "total-cpu-time-user-ms",
+            "total-cpu-time-sys-ms",
+            "total-net-bytes",
+            "by-cpu-time-us",
+            "by-net-bytes",
+        ]
+
+        for elem in result:
+            for field in expected_fields:
+                assert field in elem, (
+                    f"Field '{field}' is missing from HOTKEYS GET response"
+                )
+
     @skip_if_redis_enterprise()
     def test_bgsave(self, r):
         assert r.bgsave()