diff --git a/docs/connections.rst b/docs/connections.rst
index b547a7659e..9538c0234c 100644
--- a/docs/connections.rst
+++ b/docs/connections.rst
@@ -95,6 +95,11 @@ Connection
 .. autoclass:: redis.connection.Connection
     :members:
 
+For deployments that use rotating credentials, ``max_connection_lifetime``
+can be set in seconds. A connection that reaches this age is refreshed the
+next time it is checked out from the pool, allowing the configured credential
+provider to obtain fresh credentials without interrupting an active command.
+
 Connection (Async)
 ==================
 .. autoclass:: redis.asyncio.connection.Connection
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 9cd0d082a4..d93668681d 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -561,6 +561,8 @@ class AbstractConnection(AsyncMaintNotificationsAbstractConnection):
         "retry_on_timeout",
         "retry_on_error",
         "health_check_interval",
+        "max_connection_lifetime",
+        "_connection_created_at",
         "next_health_check",
         "last_active_at",
         "encoder",
@@ -597,6 +599,7 @@ def __init__(
         parser_class: Type[BaseParser] = DefaultParser,
         socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
         health_check_interval: float = 0,
+        max_connection_lifetime: float | None = None,
         client_name: str | None = None,
         lib_name: str | object | None = SENTINEL,
         lib_version: str | object | None = SENTINEL,
@@ -627,6 +630,11 @@ def __init__(
 
         Parameters
         ----------
+        max_connection_lifetime : float, optional
+            Maximum time in seconds to reuse a connected socket. When the limit
+            is reached, the next connection attempt reconnects the socket so
+            credential providers can refresh their credentials. ``None``
+            disables the limit.
         driver_info : DriverInfo, optional
             Driver metadata for CLIENT SETINFO. If provided, lib_name and lib_version
             are ignored. If not provided, a DriverInfo will be created from lib_name
@@ -679,6 +687,10 @@ def __init__(
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
+        if max_connection_lifetime is not None and max_connection_lifetime <= 0:
+            raise ValueError("max_connection_lifetime must be positive")
+        self.max_connection_lifetime = max_connection_lifetime
+        self._connection_created_at: float | None = None
         self.next_health_check: float = -1
         self.encoder = encoder_class(encoding, encoding_errors, decode_responses)
         self.redis_connect_func = redis_connect_func
@@ -818,11 +830,21 @@ async def connect(self):
             with_failure_count=True,
         )
 
+    def is_connection_expired(self) -> bool:
+        return (
+            self.max_connection_lifetime is not None
+            and self._connection_created_at is not None
+            and time.monotonic() - self._connection_created_at
+            >= self.max_connection_lifetime
+        )
+
     async def connect_check_health(
         self, check_health: bool = True, retry_socket_connect: bool = True
     ):
         if self.is_connected:
-            return
+            if not self.is_connection_expired():
+                return
+            await self.disconnect()
         # Track actual retry attempts for error reporting
         actual_retry_attempts = 0
 
@@ -895,6 +917,9 @@ def failure_callback(error, failure_count):
             if task and inspect.isawaitable(task):
                 await task
 
+        if self.max_connection_lifetime is not None:
+            self._connection_created_at = time.monotonic()
+
     def mark_for_reconnect(self):
         self._should_reconnect = True
 
@@ -1052,6 +1077,7 @@ async def disconnect(
         health_check_failed: bool = False,
     ) -> None:
         """Disconnects from the Redis server"""
+        self._connection_created_at = None
         # On Python 3.13+, asyncio.timeout() raises RuntimeError when called
         # outside a running Task (e.g. during GC finalization or event-loop
         # callbacks).  In that context we fall back to a synchronous close.
@@ -1717,6 +1743,7 @@ def parse_ssl_verify_flags(value):
         "db": int,
         "socket_timeout": float,
         "socket_connect_timeout": float,
+        "max_connection_lifetime": float,
         "socket_read_size": int,
         "socket_keepalive": to_bool,
         "retry_on_timeout": to_bool,
@@ -1740,6 +1767,7 @@ class ConnectKwargs(TypedDict, total=False):
     host: str
     port: int
     db: int
+    max_connection_lifetime: float
     path: str
 
 
diff --git a/redis/asyncio/sentinel.py b/redis/asyncio/sentinel.py
index 51f54aa670..bac82d6aec 100644
--- a/redis/asyncio/sentinel.py
+++ b/redis/asyncio/sentinel.py
@@ -51,7 +51,9 @@ async def connect_to(self, address):
 
     async def _connect_retry(self):
         if self._reader:
-            return  # already connected
+            if not self.is_connection_expired():
+                return  # already connected
+            await self.disconnect()
         if self.connection_pool.is_master:
             await self.connect_to(await self.connection_pool.get_master_address())
         else:
diff --git a/redis/connection.py b/redis/connection.py
index bd2e141240..0494246fef 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -817,6 +817,7 @@ def __init__(
         parser_class=DefaultParser,
         socket_read_size: int = DEFAULT_SOCKET_READ_SIZE,
         health_check_interval: int = 0,
+        max_connection_lifetime: Optional[float] = None,
         client_name: Optional[str] = None,
         lib_name: Union[Optional[str], object] = SENTINEL,
         lib_version: Union[Optional[str], object] = SENTINEL,
@@ -850,6 +851,11 @@ def __init__(
         `retry` to a valid `Retry` object.
         To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
 
+        `max_connection_lifetime` limits how long a connected socket can be
+        reused. When the limit is reached, the next connection attempt closes
+        the socket and reconnects, allowing credential providers to refresh
+        their credentials. `None` disables the limit.
+
         Parameters
         ----------
         driver_info : DriverInfo, optional
@@ -907,6 +913,10 @@ def __init__(
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
+        if max_connection_lifetime is not None and max_connection_lifetime <= 0:
+            raise ValueError("max_connection_lifetime must be positive")
+        self.max_connection_lifetime = max_connection_lifetime
+        self._connection_created_at: Optional[float] = None
         self.next_health_check = 0
         self.redis_connect_func = redis_connect_func
         self.encoder = Encoder(encoding, encoding_errors, decode_responses)
@@ -1025,11 +1035,21 @@ def connect(self):
             lambda error: self.disconnect(error),
         )
 
+    def is_connection_expired(self) -> bool:
+        return (
+            self.max_connection_lifetime is not None
+            and self._connection_created_at is not None
+            and time.monotonic() - self._connection_created_at
+            >= self.max_connection_lifetime
+        )
+
     def connect_check_health(
         self, check_health: bool = True, retry_socket_connect: bool = True
     ):
         if self._sock:
-            return
+            if not self.is_connection_expired():
+                return
+            self.disconnect()
         # Track actual retry attempts for error reporting
         actual_retry_attempts = [0]
 
@@ -1091,6 +1111,9 @@ def failure_callback(error, failure_count):
             if callback:
                 callback(self)
 
+        if self.max_connection_lifetime is not None:
+            self._connection_created_at = time.monotonic()
+
     @abstractmethod
     def _connect(self):
         pass
@@ -1227,6 +1250,7 @@ def disconnect(self, *args, **kwargs):
 
         conn_sock = self._sock
         self._sock = None
+        self._connection_created_at = None
         # reset the reconnect flag
         self.reset_should_reconnect()
 
@@ -2259,6 +2283,7 @@ def parse_ssl_verify_flags(value):
     "db": int,
     "socket_timeout": float,
     "socket_connect_timeout": float,
+    "max_connection_lifetime": float,
     "socket_read_size": int,
     "socket_keepalive": to_bool,
     "retry_on_timeout": to_bool,
diff --git a/redis/sentinel.py b/redis/sentinel.py
index 2a34f65b92..5b32ca8dc4 100644
--- a/redis/sentinel.py
+++ b/redis/sentinel.py
@@ -51,7 +51,9 @@ def connect_to(self, address):
 
     def _connect_retry(self):
         if self._sock:
-            return  # already connected
+            if not self.is_connection_expired():
+                return  # already connected
+            self.disconnect()
         if self.connection_pool.is_master:
             self.connect_to(self.connection_pool.get_master_address())
         else:
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index c56b5d9dd4..03ff6b5dfd 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -325,6 +325,63 @@ async def get_conn():
     await r.aclose()
 
 
+@pytest.mark.asyncio
+async def test_reconnects_expired_connection():
+    conn = Connection(max_connection_lifetime=10)
+    conn._reader = mock.Mock()
+    conn._writer = mock.Mock()
+    conn._connection_created_at = 0
+
+    async def disconnect():
+        conn._reader = None
+        conn._writer = None
+
+    async def connect():
+        conn._reader = mock.Mock()
+        conn._writer = mock.Mock()
+
+    conn.disconnect = mock.AsyncMock(side_effect=disconnect)
+    conn._connect = mock.AsyncMock(side_effect=connect)
+    conn.on_connect_check_health = mock.AsyncMock()
+
+    with patch("redis.asyncio.connection.time.monotonic", return_value=11):
+        await conn.connect_check_health()
+
+    conn.disconnect.assert_awaited_once_with()
+    conn._connect.assert_awaited_once_with()
+    assert conn._connection_created_at == 11
+    conn._reader = None
+    conn._writer = None
+
+
+@pytest.mark.asyncio
+async def test_reuses_connection_before_lifetime_expires():
+    conn = Connection(max_connection_lifetime=10)
+    conn._reader = mock.Mock()
+    conn._writer = mock.Mock()
+    conn._connection_created_at = 0
+    conn.disconnect = mock.AsyncMock()
+    conn._connect = mock.AsyncMock()
+
+    with patch("redis.asyncio.connection.time.monotonic", return_value=9):
+        await conn.connect_check_health()
+
+    conn.disconnect.assert_not_awaited()
+    conn._connect.assert_not_awaited()
+    conn._reader = None
+    conn._writer = None
+
+
+@pytest.mark.asyncio
+async def test_parse_url_connection_lifetime():
+    assert (
+        parse_url("redis://localhost?max_connection_lifetime=60")[
+            "max_connection_lifetime"
+        ]
+        == 60.0
+    )
+
+
 @skip_if_server_version_lt("4.0.0")
 @pytest.mark.redismod
 @pytest.mark.onlynoncluster
diff --git a/tests/test_connection.py b/tests/test_connection.py
index 244bf3b54f..1d5b8693f1 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -543,6 +543,39 @@ def test_redis_client_default_driver_info():
 
 @pytest.mark.fixed_client
 class TestConnection:
+    @pytest.mark.parametrize("max_connection_lifetime", [0, -1])
+    def test_rejects_non_positive_connection_lifetime(self, max_connection_lifetime):
+        with pytest.raises(ValueError, match="max_connection_lifetime"):
+            Connection(max_connection_lifetime=max_connection_lifetime)
+
+    def test_reconnects_expired_connection(self):
+        conn = Connection(max_connection_lifetime=10)
+        conn._sock = mock.Mock()
+        conn._connection_created_at = 0
+        conn.disconnect = mock.Mock(side_effect=lambda: setattr(conn, "_sock", None))
+        conn._connect = mock.Mock(return_value=mock.Mock())
+        conn.on_connect_check_health = mock.Mock()
+
+        with patch("redis.connection.time.monotonic", return_value=11):
+            conn.connect_check_health()
+
+        conn.disconnect.assert_called_once_with()
+        conn._connect.assert_called_once_with()
+        assert conn._connection_created_at == 11
+
+    def test_reuses_connection_before_lifetime_expires(self):
+        conn = Connection(max_connection_lifetime=10)
+        conn._sock = mock.Mock()
+        conn._connection_created_at = 0
+        conn.disconnect = mock.Mock()
+        conn._connect = mock.Mock()
+
+        with patch("redis.connection.time.monotonic", return_value=9):
+            conn.connect_check_health()
+
+        conn.disconnect.assert_not_called()
+        conn._connect.assert_not_called()
+
     def test_disconnect(self):
         conn = Connection()
         mock_sock = mock.Mock()
@@ -662,6 +695,14 @@ def test_connect_timeout_error_without_retry(self):
         assert conn._connect.call_count == 1
         self.clear(conn)
 
+    def test_parse_url_connection_lifetime(self):
+        assert (
+            parse_url("redis://localhost?max_connection_lifetime=60")[
+                "max_connection_lifetime"
+            ]
+            == 60.0
+        )
+
 
 @pytest.mark.onlynoncluster
 @pytest.mark.parametrize(