diff --git a/redis/_parsers/base.py b/redis/_parsers/base.py
index d5e4add661..77ae0bca7d 100644
--- a/redis/_parsers/base.py
+++ b/redis/_parsers/base.py
@@ -1,3 +1,4 @@
+import logging
 import sys
 from abc import ABC
 from asyncio import IncompleteReadError, StreamReader, TimeoutError
@@ -56,6 +57,8 @@
     "Client sent AUTH, but no password is set": AuthenticationError,
 }
 
+logger = logging.getLogger(__name__)
+
 
 class BaseParser(ABC):
     EXCEPTION_CLASSES = {
@@ -199,28 +202,41 @@ def handle_push_response(self, response, **kwargs):
             *_MOVING_MESSAGE,
         ):
             return self.pubsub_push_handler_func(response)
-        if msg_type in _INVALIDATION_MESSAGE and self.invalidation_push_handler_func:
-            return self.invalidation_push_handler_func(response)
-        if msg_type in _MOVING_MESSAGE and self.node_moving_push_handler_func:
-            host, port = response[2].decode().split(":")
-            ttl = response[1]
-            id = 1  # Hardcoded value until the notification starts including the id
-            notification = NodeMovingEvent(id, host, port, ttl)
-            return self.node_moving_push_handler_func(notification)
-        if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func:
-            if msg_type in _MIGRATING_MESSAGE:
-                ttl = response[1]
-                id = 2  # Hardcoded value until the notification starts including the id
-                notification = NodeMigratingEvent(id, ttl)
-            elif msg_type in _MIGRATED_MESSAGE:
-                id = 3  # Hardcoded value until the notification starts including the id
-                notification = NodeMigratedEvent(id)
-            else:
+
+        try:
+            if (
+                msg_type in _INVALIDATION_MESSAGE
+                and self.invalidation_push_handler_func
+            ):
+                return self.invalidation_push_handler_func(response)
+            if msg_type in _MOVING_MESSAGE and self.node_moving_push_handler_func:
+                # Expected message format is: MOVING <seq_number> <time> <endpoint>
+                id = response[1]
+                ttl = response[2]
+                host, port = response[3].decode().split(":")
+                notification = NodeMovingEvent(id, host, port, ttl)
+                return self.node_moving_push_handler_func(notification)
+
+            if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func:
                 notification = None
-            if notification is not None:
-                return self.maintenance_push_handler_func(notification)
-            else:
-                return None
+
+                if msg_type in _MIGRATING_MESSAGE:
+                    # Expected message format is: MIGRATING <seq_number> <time> <shard_id-s>
+                    id = response[1]
+                    ttl = response[2]
+                    notification = NodeMigratingEvent(id, ttl)
+                elif msg_type in _MIGRATED_MESSAGE:
+                    id = response[1]
+                    notification = NodeMigratedEvent(id)
+
+                if notification is not None:
+                    return self.maintenance_push_handler_func(notification)
+        except Exception as e:
+            logger.error(
+                "Error handling {} message ({}): {}".format(msg_type, response, e)
+            )
+
+        return None
 
     def set_pubsub_push_handler(self, pubsub_push_handler_func):
         self.pubsub_push_handler_func = pubsub_push_handler_func
@@ -249,6 +265,7 @@ async def handle_pubsub_push_response(self, response):
 
     async def handle_push_response(self, response, **kwargs):
         """Handle push responses asynchronously"""
+
         msg_type = response[0]
         if msg_type not in (
             *_INVALIDATION_MESSAGE,
@@ -256,24 +273,40 @@ async def handle_push_response(self, response, **kwargs):
             *_MOVING_MESSAGE,
         ):
             return await self.pubsub_push_handler_func(response)
-        if msg_type in _INVALIDATION_MESSAGE and self.invalidation_push_handler_func:
-            return await self.invalidation_push_handler_func(response)
-        if msg_type in _MOVING_MESSAGE and self.node_moving_push_handler_func:
-            # push notification from enterprise cluster for node moving
-            host, port = response[2].split(":")
-            ttl = response[1]
-            id = 1  # Hardcoded value for async parser
-            notification = NodeMovingEvent(id, host, port, ttl)
-            return await self.node_moving_push_handler_func(notification)
-        if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func:
-            if msg_type in _MIGRATING_MESSAGE:
-                ttl = response[1]
-                id = 2  # Hardcoded value for async parser
-                notification = NodeMigratingEvent(id, ttl)
-            elif msg_type in _MIGRATED_MESSAGE:
-                id = 3  # Hardcoded value for async parser
-                notification = NodeMigratedEvent(id)
-            return await self.maintenance_push_handler_func(notification)
+
+        try:
+            if (
+                msg_type in _INVALIDATION_MESSAGE
+                and self.invalidation_push_handler_func
+            ):
+                return await self.invalidation_push_handler_func(response)
+            if msg_type in _MOVING_MESSAGE and self.node_moving_push_handler_func:
+                # push notification from enterprise cluster for node moving
+                id = response[1]
+                ttl = response[2]
+                host, port = response[3].split(":")
+                notification = NodeMovingEvent(id, host, port, ttl)
+                return await self.node_moving_push_handler_func(notification)
+
+            if msg_type in _MAINTENANCE_MESSAGES and self.maintenance_push_handler_func:
+                notification = None
+
+                if msg_type in _MIGRATING_MESSAGE:
+                    id = response[1]
+                    ttl = response[2]
+                    notification = NodeMigratingEvent(id, ttl)
+                elif msg_type in _MIGRATED_MESSAGE:
+                    id = response[1]
+                    notification = NodeMigratedEvent(id)
+
+                if notification is not None:
+                    return await self.maintenance_push_handler_func(notification)
+        except Exception as e:
+            logger.error(
+                "Error handling {} message ({}): {}".format(msg_type, response, e)
+            )
+
+        return None
 
     def set_pubsub_push_handler(self, pubsub_push_handler_func):
         """Set the pubsub push handler function"""
diff --git a/redis/connection.py b/redis/connection.py
index 3f1b54e26c..948ff6ec68 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -264,6 +264,13 @@ def should_reconnect(self):
         """
         pass
 
+    @abstractmethod
+    def get_resolved_ip(self):
+        """
+        Get resolved ip address for the connection.
+        """
+        pass
+
     @abstractmethod
     def update_current_socket_timeout(self, relax_timeout: Optional[float] = None):
         """
@@ -400,28 +407,15 @@ def __init__(
             parser_class = _RESP3Parser
         self.set_parser(parser_class)
 
-        if maintenance_events_config and maintenance_events_config.enabled:
-            if maintenance_events_pool_handler:
-                self._parser.set_node_moving_push_handler(
-                    maintenance_events_pool_handler.handle_event
-                )
-            self._maintenance_event_connection_handler = (
-                MaintenanceEventConnectionHandler(self, maintenance_events_config)
-            )
-            self._parser.set_maintenance_push_handler(
-                self._maintenance_event_connection_handler.handle_event
-            )
+        self.maintenance_events_config = maintenance_events_config
 
-            self.orig_host_address = (
-                orig_host_address if orig_host_address else self.host
-            )
-            self.orig_socket_timeout = (
-                orig_socket_timeout if orig_socket_timeout else self.socket_timeout
-            )
-            self.orig_socket_connect_timeout = (
-                orig_socket_connect_timeout
-                if orig_socket_connect_timeout
-                else self.socket_connect_timeout
+        # Set up maintenance events if enabled
+        if maintenance_events_config and maintenance_events_config.enabled:
+            self._enable_maintenance_events(
+                maintenance_events_pool_handler,
+                orig_host_address,
+                orig_socket_timeout,
+                orig_socket_connect_timeout,
             )
         self._should_reconnect = False
         self.maintenance_state = maintenance_state
@@ -481,6 +475,45 @@ def set_parser(self, parser_class):
         """
         self._parser = parser_class(socket_read_size=self._socket_read_size)
 
+    def _enable_maintenance_events(
+        self,
+        maintenance_events_pool_handler=None,
+        orig_host_address=None,
+        orig_socket_timeout=None,
+        orig_socket_connect_timeout=None,
+    ):
+        """Enable maintenance events by setting up handlers and storing original connection parameters."""
+        if (
+            not self.maintenance_events_config
+            or not self.maintenance_events_config.enabled
+        ):
+            return
+
+        # Set up pool handler if available
+        if maintenance_events_pool_handler:
+            self._parser.set_node_moving_push_handler(
+                maintenance_events_pool_handler.handle_event
+            )
+
+        # Set up connection handler
+        self._maintenance_event_connection_handler = MaintenanceEventConnectionHandler(
+            self, self.maintenance_events_config
+        )
+        self._parser.set_maintenance_push_handler(
+            self._maintenance_event_connection_handler.handle_event
+        )
+
+        # Store original connection parameters
+        self.orig_host_address = orig_host_address if orig_host_address else self.host
+        self.orig_socket_timeout = (
+            orig_socket_timeout if orig_socket_timeout else self.socket_timeout
+        )
+        self.orig_socket_connect_timeout = (
+            orig_socket_connect_timeout
+            if orig_socket_connect_timeout
+            else self.socket_connect_timeout
+        )
+
     def set_maintenance_event_pool_handler(
         self, maintenance_event_pool_handler: MaintenanceEventPoolHandler
     ):
@@ -623,6 +656,37 @@ def on_connect_check_health(self, check_health: bool = True):
             ):
                 raise ConnectionError("Invalid RESP version")
 
+        # Send maintenance notifications handshake if RESP3 is active and maintenance events are enabled
+        if (
+            self.protocol not in [2, "2"]
+            and self.maintenance_events_config
+            and self.maintenance_events_config.enabled
+            and hasattr(self, "_maintenance_event_connection_handler")
+        ):
+            try:
+                endpoint_type = self.maintenance_events_config.get_endpoint_type(
+                    self.host, self
+                )
+                self.send_command(
+                    "CLIENT",
+                    "MAINT_NOTIFICATIONS",
+                    "ON",
+                    "moving-endpoint-type",
+                    endpoint_type,
+                    check_health=check_health,
+                )
+                response = self.read_response()
+                if str_if_bytes(response) != "OK":
+                    raise ConnectionError(
+                        "The server doesn't support maintenance notifications"
+                    )
+            except Exception as e:
+                # Log warning but don't fail the connection
+                import logging
+
+                logger = logging.getLogger(__name__)
+                logger.warning(f"Failed to enable maintenance notifications: {e}")
+
         # if a client_name is given, set it
         if self.client_name:
             self.send_command(
@@ -859,6 +923,56 @@ def re_auth(self):
             self.read_response()
             self._re_auth_token = None
 
+    def get_resolved_ip(self) -> Optional[str]:
+        """
+        Extract the resolved IP address from an
+        established connection or resolve it from the host.
+
+        First tries to get the actual IP from the socket (most accurate),
+        then falls back to DNS resolution if needed.
+
+        Args:
+            connection: The connection object to extract the IP from
+
+        Returns:
+            str: The resolved IP address, or None if it cannot be determined
+        """
+
+        # Method 1: Try to get the actual IP from the established socket connection
+        # This is most accurate as it shows the exact IP being used
+        try:
+            if self._sock is not None:
+                peer_addr = self._sock.getpeername()
+                if peer_addr and len(peer_addr) >= 1:
+                    # For TCP sockets, peer_addr is typically (host, port) tuple
+                    # Return just the host part
+                    return peer_addr[0]
+        except (AttributeError, OSError):
+            # Socket might not be connected or getpeername() might fail
+            pass
+
+        # Method 2: Fallback to DNS resolution of the host
+        # This is less accurate but works when socket is not available
+        try:
+            host = getattr(self, "host", "localhost")
+            port = getattr(self, "port", 6379)
+            if host:
+                # Use getaddrinfo to resolve the hostname to IP
+                # This mimics what the connection would do during _connect()
+                addr_info = socket.getaddrinfo(
+                    host, port, socket.AF_UNSPEC, socket.SOCK_STREAM
+                )
+                if addr_info:
+                    # Return the IP from the first result
+                    # addr_info[0] is (family, socktype, proto, canonname, sockaddr)
+                    # sockaddr[0] is the IP address
+                    return addr_info[0][4][0]
+        except (AttributeError, OSError, socket.gaierror):
+            # DNS resolution might fail
+            pass
+
+        return None
+
     @property
     def maintenance_state(self) -> MaintenanceState:
         return self._maintenance_state
diff --git a/redis/maintenance_events.py b/redis/maintenance_events.py
index f99ad37397..2c533d372a 100644
--- a/redis/maintenance_events.py
+++ b/redis/maintenance_events.py
@@ -1,5 +1,7 @@
 import enum
+import ipaddress
 import logging
+import re
 import threading
 import time
 from abc import ABC, abstractmethod
@@ -15,6 +17,27 @@ class MaintenanceState(enum.Enum):
     FAILING_OVER = "failing_over"
 
 
+class EndpointType:
+    """Constants for valid endpoint types used in CLIENT MAINT_NOTIFICATIONS command."""
+
+    INTERNAL_IP = "internal-ip"
+    INTERNAL_FQDN = "internal-fqdn"
+    EXTERNAL_IP = "external-ip"
+    EXTERNAL_FQDN = "external-fqdn"
+    NONE = "none"
+
+    @classmethod
+    def get_valid_types(cls):
+        """Return a set of all valid endpoint types."""
+        return {
+            cls.INTERNAL_IP,
+            cls.INTERNAL_FQDN,
+            cls.EXTERNAL_IP,
+            cls.EXTERNAL_FQDN,
+            cls.NONE,
+        }
+
+
 if TYPE_CHECKING:
     from redis.connection import (
         BlockingConnectionPool,
@@ -361,6 +384,100 @@ def __hash__(self) -> int:
         return hash((self.__class__, self.id))
 
 
+def _is_private_fqdn(host: str) -> bool:
+    """
+    Determine if an FQDN is likely to be internal/private.
+
+    This uses heuristics based on RFC 952 and RFC 1123 standards:
+    - .local domains (RFC 6762 - Multicast DNS)
+    - .internal domains (common internal convention)
+    - Single-label hostnames (no dots)
+    - Common internal TLDs
+
+    Args:
+        host (str): The FQDN to check
+
+    Returns:
+        bool: True if the FQDN appears to be internal/private
+    """
+    host_lower = host.lower().rstrip(".")
+
+    # Single-label hostnames (no dots) are typically internal
+    if "." not in host_lower:
+        return True
+
+    # Common internal/private domain patterns
+    internal_patterns = [
+        r"\.local$",  # mDNS/Bonjour domains
+        r"\.internal$",  # Common internal convention
+        r"\.corp$",  # Corporate domains
+        r"\.lan$",  # Local area network
+        r"\.intranet$",  # Intranet domains
+        r"\.private$",  # Private domains
+    ]
+
+    for pattern in internal_patterns:
+        if re.search(pattern, host_lower):
+            return True
+
+    # If none of the internal patterns match, assume it's external
+    return False
+
+
+def _get_resolved_ip_from_connection(
+    connection: "ConnectionInterface",
+) -> Optional[str]:
+    """
+    Extract the resolved IP address from an established connection.
+
+    First tries to get the actual IP from the socket (most accurate),
+    then falls back to DNS resolution if needed.
+
+    Args:
+        connection: The connection object to extract the IP from
+
+    Returns:
+        str: The resolved IP address, or None if it cannot be determined
+    """
+    import socket
+
+    # Method 1: Try to get the actual IP from the established socket connection
+    # This is most accurate as it shows the exact IP being used
+    try:
+        sock = getattr(connection, "_sock", None)
+        if sock is not None:
+            peer_addr = sock.getpeername()
+            if peer_addr and len(peer_addr) >= 1:
+                # For TCP sockets, peer_addr is typically (host, port) tuple
+                # Return just the host part
+                return peer_addr[0]
+    except (AttributeError, OSError):
+        # Socket might not be connected or getpeername() might fail
+        pass
+
+    # Method 2: Fallback to DNS resolution of the host
+    # This is less accurate but works when socket is not available
+    try:
+        host = getattr(connection, "host", None)
+        port = getattr(connection, "port", 6379)
+        if host:
+            # Use getaddrinfo to resolve the hostname to IP
+            # This mimics what the connection would do during _connect()
+            addr_info = socket.getaddrinfo(
+                host, port, socket.AF_UNSPEC, socket.SOCK_STREAM
+            )
+            if addr_info:
+                # Return the IP from the first result
+                # addr_info[0] is (family, socktype, proto, canonname, sockaddr)
+                # sockaddr[0] is the IP address
+                return addr_info[0][4][0]
+    except (AttributeError, OSError, socket.gaierror):
+        # DNS resolution might fail
+        pass
+
+    return None
+
+
 class MaintenanceEventsConfig:
     """
     Configuration class for maintenance events handling behaviour. Events are received through
@@ -373,9 +490,10 @@ class MaintenanceEventsConfig:
 
     def __init__(
         self,
-        enabled: bool = False,
+        enabled: bool = True,
         proactive_reconnect: bool = True,
         relax_timeout: Optional[Number] = 20,
+        endpoint_type: Optional[str] = None,
     ):
         """
         Initialize a new MaintenanceEventsConfig.
@@ -387,18 +505,39 @@ def __init__(
                 Defaults to True.
             relax_timeout (Number): The relax timeout to use for the connection during maintenance.
                 If -1 is provided - the relax timeout is disabled. Defaults to 20.
+            endpoint_type (Optional[str]): Override for the endpoint type to use in CLIENT MAINT_NOTIFICATIONS.
+                Must be one of: 'internal-ip', 'internal-fqdn', 'external-ip', 'external-fqdn', 'none'.
+                If None, the endpoint type will be automatically determined based on the host and TLS configuration.
+                Defaults to None.
 
+        Raises:
+            ValueError: If endpoint_type is provided but is not a valid endpoint type.
         """
         self.enabled = enabled
         self.relax_timeout = relax_timeout
         self.proactive_reconnect = proactive_reconnect
 
+        # Validate endpoint_type if provided
+        if (
+            endpoint_type is not None
+            and endpoint_type not in EndpointType.get_valid_types()
+        ):
+            valid_types = ", ".join(
+                f"'{t}'" for t in sorted(EndpointType.get_valid_types())
+            )
+            raise ValueError(
+                f"Invalid endpoint_type '{endpoint_type}'. Must be one of: {valid_types}"
+            )
+
+        self.endpoint_type = endpoint_type
+
     def __repr__(self) -> str:
         return (
             f"{self.__class__.__name__}("
             f"enabled={self.enabled}, "
             f"proactive_reconnect={self.proactive_reconnect}, "
             f"relax_timeout={self.relax_timeout}, "
+            f"endpoint_type={self.endpoint_type!r}"
             f")"
         )
 
@@ -413,6 +552,58 @@ def is_relax_timeouts_enabled(self) -> bool:
         """
         return self.relax_timeout != -1
 
+    def get_endpoint_type(self, host: str, connection: "ConnectionInterface") -> str:
+        """
+        Determine the appropriate endpoint type for CLIENT MAINT_NOTIFICATIONS command.
+
+        Logic:
+        1. If endpoint_type is explicitly set, use it
+        2. Otherwise, check the original host from connection.host:
+           - If host is an IP address, use it directly to determine internal-ip vs external-ip
+           - If host is an FQDN, get the resolved IP to determine internal-fqdn vs external-fqdn
+
+        Args:
+            host: User provided hostname to analyze
+            connection: The connection object to analyze for endpoint type determination
+
+        Returns:
+        """
+
+        # If endpoint_type is explicitly set, use it
+        if self.endpoint_type is not None:
+            return self.endpoint_type
+
+        # Check if the host is an IP address
+        try:
+            ip_addr = ipaddress.ip_address(host)
+            # Host is an IP address - use it directly
+            is_private = ip_addr.is_private
+            return EndpointType.INTERNAL_IP if is_private else EndpointType.EXTERNAL_IP
+        except ValueError:
+            # Host is an FQDN - need to check resolved IP to determine internal vs external
+            pass
+
+        # Host is an FQDN, get the resolved IP to determine if it's internal or external
+        resolved_ip = connection.get_resolved_ip()
+
+        if resolved_ip:
+            try:
+                ip_addr = ipaddress.ip_address(resolved_ip)
+                is_private = ip_addr.is_private
+                # Use FQDN types since the original host was an FQDN
+                return (
+                    EndpointType.INTERNAL_FQDN
+                    if is_private
+                    else EndpointType.EXTERNAL_FQDN
+                )
+            except ValueError:
+                # This shouldn't happen since we got the IP from the socket, but fallback
+                pass
+
+        # Final fallback: use heuristics on the FQDN itself
+        is_private = _is_private_fqdn(host)
+        return EndpointType.INTERNAL_FQDN if is_private else EndpointType.EXTERNAL_FQDN
+
 
 class MaintenanceEventPoolHandler:
     def __init__(
diff --git a/tests/test_maintenance_events.py b/tests/test_maintenance_events.py
index a59b834a4e..d52c395172 100644
--- a/tests/test_maintenance_events.py
+++ b/tests/test_maintenance_events.py
@@ -2,6 +2,7 @@
 from unittest.mock import Mock, patch, MagicMock
 import pytest
 
+from redis.connection import ConnectionInterface
 from redis.maintenance_events import (
     MaintenanceEvent,
     NodeMovingEvent,
@@ -13,6 +14,7 @@
     MaintenanceEventPoolHandler,
     MaintenanceEventConnectionHandler,
     MaintenanceState,
+    EndpointType,
 )
 
 
@@ -368,7 +370,7 @@ class TestMaintenanceEventsConfig:
     def test_init_defaults(self):
         """Test MaintenanceEventsConfig initialization with defaults."""
         config = MaintenanceEventsConfig()
-        assert config.enabled is False
+        assert config.enabled is True
         assert config.proactive_reconnect is True
         assert config.relax_timeout == 20
 
@@ -672,3 +674,180 @@ def test_handle_maintenance_completed_event_success(self):
         self.mock_connection.reset_tmp_settings.assert_called_once_with(
             reset_relax_timeout=True
         )
+
+
+class TestEndpointType:
+    """Test the EndpointType class functionality."""
+
+    def test_endpoint_type_constants(self):
+        """Test that the EndpointType constants are correct."""
+        assert EndpointType.INTERNAL_IP == "internal-ip"
+        assert EndpointType.INTERNAL_FQDN == "internal-fqdn"
+        assert EndpointType.EXTERNAL_IP == "external-ip"
+        assert EndpointType.EXTERNAL_FQDN == "external-fqdn"
+        assert EndpointType.NONE == "none"
+
+    def test_get_valid_types(self):
+        """Test that get_valid_types returns the expected set."""
+        valid_types = EndpointType.get_valid_types()
+        expected_types = {
+            "internal-ip",
+            "internal-fqdn",
+            "external-ip",
+            "external-fqdn",
+            "none",
+        }
+        assert valid_types == expected_types
+
+
+class TestMaintenanceEventsConfigEndpointType:
+    """Test MaintenanceEventsConfig endpoint type functionality."""
+
+    def setup_method(self):
+        """Set up common mock classes for all tests."""
+
+        class MockSocket:
+            def __init__(self, resolved_ip):
+                self.resolved_ip = resolved_ip
+
+            def getpeername(self):
+                return (self.resolved_ip, 6379)
+
+        class MockConnection(ConnectionInterface):
+            def __init__(self, host, resolved_ip=None, is_ssl=False):
+                self.host = host
+                self.port = 6379
+                self._sock = MockSocket(resolved_ip) if resolved_ip else None
+                self.__class__.__name__ = "SSLConnection" if is_ssl else "Connection"
+
+            def get_resolved_ip(self):
+                # Call the actual method from AbstractConnection
+                from redis.connection import AbstractConnection
+
+                return AbstractConnection.get_resolved_ip(self)  # type: ignore
+
+        self.MockSocket = MockSocket
+        self.MockConnection = MockConnection
+
+    def test_config_validation_valid_endpoint_types(self):
+        """Test that MaintenanceEventsConfig accepts valid endpoint types."""
+        for endpoint_type in EndpointType.get_valid_types():
+            config = MaintenanceEventsConfig(endpoint_type=endpoint_type)
+            assert config.endpoint_type == endpoint_type
+
+    def test_config_validation_invalid_endpoint_type(self):
+        """Test that MaintenanceEventsConfig raises ValueError for invalid endpoint type."""
+        with pytest.raises(ValueError, match="Invalid endpoint_type"):
+            MaintenanceEventsConfig(endpoint_type="invalid-type")
+
+    def test_config_validation_none_endpoint_type(self):
+        """Test that MaintenanceEventsConfig accepts None as endpoint type."""
+        config = MaintenanceEventsConfig(endpoint_type=None)
+        assert config.endpoint_type is None
+
+    def test_endpoint_type_detection_ip_addresses(self):
+        """Test endpoint type detection for IP addresses."""
+        config = MaintenanceEventsConfig()
+
+        # Test private IPv4 addresses
+        conn1 = self.MockConnection("192.168.1.1", resolved_ip="192.168.1.1")
+        assert (
+            config.get_endpoint_type("192.168.1.1", conn1) == EndpointType.INTERNAL_IP
+        )
+
+        # Test public IPv4 addresses
+        conn2 = self.MockConnection("8.8.8.8", resolved_ip="8.8.8.8")
+        assert config.get_endpoint_type("8.8.8.8", conn2) == EndpointType.EXTERNAL_IP
+
+        # Test IPv6 loopback
+        conn3 = self.MockConnection("::1")
+        assert config.get_endpoint_type("::1", conn3) == EndpointType.INTERNAL_IP
+
+        # Test IPv6 public address
+        conn4 = self.MockConnection("2001:4860:4860::8888")
+        assert (
+            config.get_endpoint_type("2001:4860:4860::8888", conn4)
+            == EndpointType.EXTERNAL_IP
+        )
+
+    def test_endpoint_type_detection_fqdn_with_resolved_ip(self):
+        """Test endpoint type detection for FQDNs with resolved IP addresses."""
+        config = MaintenanceEventsConfig()
+
+        # Test FQDN resolving to private IP
+        conn1 = self.MockConnection(
+            "redis.internal.company.com", resolved_ip="192.168.1.1"
+        )
+        assert (
+            config.get_endpoint_type("redis.internal.company.com", conn1)
+            == EndpointType.INTERNAL_FQDN
+        )
+
+        # Test FQDN resolving to public IP
+        conn2 = self.MockConnection("db123.redis.com", resolved_ip="8.8.8.8")
+        assert (
+            config.get_endpoint_type("db123.redis.com", conn2)
+            == EndpointType.EXTERNAL_FQDN
+        )
+
+        # Test internal FQDN resolving to public IP (should use resolved IP)
+        conn3 = self.MockConnection(
+            "redis.internal.company.com", resolved_ip="10.8.8.8"
+        )
+        assert (
+            config.get_endpoint_type("redis.internal.company.com", conn3)
+            == EndpointType.INTERNAL_FQDN
+        )
+
+        # Test FQDN with TLS
+        conn4 = self.MockConnection(
+            "redis.internal.company.com", resolved_ip="192.168.1.1", is_ssl=True
+        )
+        assert (
+            config.get_endpoint_type("redis.internal.company.com", conn4)
+            == EndpointType.INTERNAL_FQDN
+        )
+
+        conn5 = self.MockConnection(
+            "db123.redis.com", resolved_ip="8.8.8.8", is_ssl=True
+        )
+        assert (
+            config.get_endpoint_type("db123.redis.com", conn5)
+            == EndpointType.EXTERNAL_FQDN
+        )
+
+    def test_endpoint_type_detection_fqdn_heuristics(self):
+        """Test endpoint type detection using FQDN heuristics when no resolved IP is available."""
+        config = MaintenanceEventsConfig()
+
+        # Test localhost (should be internal)
+        conn1 = self.MockConnection("localhost")
+        assert (
+            config.get_endpoint_type("localhost", conn1) == EndpointType.INTERNAL_FQDN
+        )
+
+        # Test .local domain (should be internal)
+        conn2 = self.MockConnection("server.local")
+        assert (
+            config.get_endpoint_type("server.local", conn2)
+            == EndpointType.INTERNAL_FQDN
+        )
+
+        # Test public domain (should be external)
+        conn3 = self.MockConnection("example.com")
+        assert (
+            config.get_endpoint_type("example.com", conn3) == EndpointType.EXTERNAL_FQDN
+        )
+
+    def test_endpoint_type_override(self):
+        """Test that configured endpoint_type overrides detection."""
+
+        # Test with endpoint_type set to NONE
+        config = MaintenanceEventsConfig(endpoint_type=EndpointType.NONE)
+        conn = self.MockConnection("localhost")
+
+        assert config.get_endpoint_type("localhost", conn) == EndpointType.NONE
+
+        # Test with endpoint_type set to EXTERNAL_IP
+        config = MaintenanceEventsConfig(endpoint_type=EndpointType.EXTERNAL_IP)
+        assert config.get_endpoint_type("localhost", conn) == EndpointType.EXTERNAL_IP
diff --git a/tests/test_maintenance_events_handling.py b/tests/test_maintenance_events_handling.py
index b6fc2116c7..c00f261d24 100644
--- a/tests/test_maintenance_events_handling.py
+++ b/tests/test_maintenance_events_handling.py
@@ -63,35 +63,35 @@ def send(self, data):
             # Check if this is a key that should trigger a push message
             if b"key_receive_migrating_" in data or b"key_receive_migrating" in data:
                 # MIGRATING push message before SET key_receive_migrating_X response
-                # Format: >2\r\n$9\r\nMIGRATING\r\n:10\r\n (2 elements: MIGRATING, ttl)
-                migrating_push = ">2\r\n$9\r\nMIGRATING\r\n:10\r\n"
+                # Format: >3\r\n$9\r\nMIGRATING\r\n:1\r\n:10\r\n (3 elements: MIGRATING, id, ttl)
+                migrating_push = ">3\r\n$9\r\nMIGRATING\r\n:1\r\n:10\r\n"
                 response = migrating_push.encode() + response
             elif b"key_receive_migrated_" in data or b"key_receive_migrated" in data:
                 # MIGRATED push message before SET key_receive_migrated_X response
-                # Format: >1\r\n$8\r\nMIGRATED\r\n (1 element: MIGRATED)
-                migrated_push = ">1\r\n$8\r\nMIGRATED\r\n"
+                # Format: >2\r\n$8\r\nMIGRATED\r\n:1\r\n (2 elements: MIGRATED, id)
+                migrated_push = ">2\r\n$8\r\nMIGRATED\r\n:1\r\n"
                 response = migrated_push.encode() + response
             elif (
                 b"key_receive_failing_over_" in data
                 or b"key_receive_failing_over" in data
             ):
                 # FAILING_OVER push message before SET key_receive_failing_over_X response
-                # Format: >2\r\n$12\r\nFAILING_OVER\r\n:10\r\n (2 elements: FAILING_OVER, ttl)
-                failing_over_push = ">2\r\n$12\r\nFAILING_OVER\r\n:10\r\n"
+                # Format: >3\r\n$12\r\nFAILING_OVER\r\n:1\r\n:10\r\n (3 elements: FAILING_OVER, id, ttl)
+                failing_over_push = ">3\r\n$12\r\nFAILING_OVER\r\n:1\r\n:10\r\n"
                 response = failing_over_push.encode() + response
             elif (
                 b"key_receive_failed_over_" in data
                 or b"key_receive_failed_over" in data
             ):
                 # FAILED_OVER push message before SET key_receive_failed_over_X response
-                # Format: >1\r\n$11\r\nFAILED_OVER\r\n (1 element: FAILED_OVER)
-                failed_over_push = ">1\r\n$11\r\nFAILED_OVER\r\n"
+                # Format: >2\r\n$11\r\nFAILED_OVER\r\n:1\r\n (2 elements: FAILED_OVER, id)
+                failed_over_push = ">2\r\n$11\r\nFAILED_OVER\r\n:1\r\n"
                 response = failed_over_push.encode() + response
             elif b"key_receive_moving_" in data:
                 # MOVING push message before SET key_receive_moving_X response
-                # Format: >3\r\n$6\r\nMOVING\r\n:15\r\n+localhost:6379\r\n (3 elements: MOVING, ttl, host:port)
+                # Format: >4\r\n$6\r\nMOVING\r\n:1\r\n:1\r\n+1.2.3.4:6379\r\n (4 elements: MOVING, id, ttl, host:port)
                 # Note: Using + instead of $ to send as simple string instead of bulk string
-                moving_push = f">3\r\n$6\r\nMOVING\r\n:{MockSocket.MOVING_TIMEOUT}\r\n+{MockSocket.AFTER_MOVING_ADDRESS}\r\n"
+                moving_push = f">4\r\n$6\r\nMOVING\r\n:1\r\n:{MockSocket.MOVING_TIMEOUT}\r\n+{MockSocket.AFTER_MOVING_ADDRESS}\r\n"
                 response = moving_push.encode() + response
 
             self.pending_responses.append(response)