fix(redis): keep a coroutine redis_connect_func on async clients

redis-py awaits a redis_connect_func that is a coroutine function, so
dropping every connect func the async paths cannot convert took away an
auth path that worked.

litellm/_redis.py

@@ -577,10 +577,12 @@ def _init_async_redis_sentinel(redis_kwargs) -> async_redis.Redis:
 
 
 def _async_credential_provider(redis_connect_func: object | None) -> CredentialProvider | None:
-    """``redis_connect_func`` runs the AUTH exchange with the blocking client API, so on an
-    async connection its ``send_command``/``read_response`` calls return coroutines nobody
-    awaits and every connect fails. Async paths authenticate through a ``CredentialProvider``
-    instead, which redis-py consults per connection so the token stays fresh."""
+    """The Azure AD and GCP IAM connect funcs run their AUTH exchange with the blocking client
+    API, so on an async connection their ``send_command``/``read_response`` calls return
+    coroutines nobody awaits and every connect fails. Async paths authenticate through a
+    ``CredentialProvider`` instead, which redis-py consults per connection so the token stays
+    fresh. Any other ``redis_connect_func`` is left where it is, since redis-py awaits it
+    itself when it is a coroutine function."""
     gcp_service_account: Final = getattr(redis_connect_func, "_gcp_service_account", None)
     if gcp_service_account is not None:
         return GCPIAMCredentialProvider(gcp_service_account)
@@ -589,12 +591,6 @@ def _async_credential_provider(redis_connect_func: object | None) -> CredentialP
     if azure_credential is not None:
         return AzureADCredentialProvider(azure_credential, username=os.environ.get("REDIS_USERNAME") or None)
 
-    if redis_connect_func is not None:
-        verbose_logger.warning(
-            "REDIS: dropping redis_connect_func, which an async connection cannot run. "
-            "Configure Azure AD or GCP IAM auth so a credential provider handles the token instead."
-        )
-
     return None
 
 
@@ -625,11 +621,11 @@ def get_redis_async_client(
     **env_overrides,
 ) -> async_redis.Redis | async_redis.RedisCluster:
     redis_kwargs: Final = _get_redis_client_logic(**env_overrides)
-    credential_provider: Final = _async_credential_provider(redis_kwargs.pop("redis_connect_func", None))
+    credential_provider: Final = _async_credential_provider(redis_kwargs.get("redis_connect_func"))
     if credential_provider is not None:
         redis_kwargs["credential_provider"] = credential_provider
-        redis_kwargs.pop("username", None)
-        redis_kwargs.pop("password", None)
+        for superseded in ("redis_connect_func", "username", "password"):
+            redis_kwargs.pop(superseded, None)
 
     if "startup_nodes" in redis_kwargs:
         from redis.cluster import ClusterNode
@@ -693,11 +689,11 @@ def get_redis_connection_pool(
     **env_overrides,
 ) -> async_redis.BlockingConnectionPool | None:
     redis_kwargs: Final = _get_redis_client_logic(**env_overrides)
-    credential_provider: Final = _async_credential_provider(redis_kwargs.pop("redis_connect_func", None))
+    credential_provider: Final = _async_credential_provider(redis_kwargs.get("redis_connect_func"))
     if credential_provider is not None:
         redis_kwargs["credential_provider"] = credential_provider
-        redis_kwargs.pop("username", None)
-        redis_kwargs.pop("password", None)
+        for superseded in ("redis_connect_func", "username", "password"):
+            redis_kwargs.pop(superseded, None)
     verbose_logger.debug("get_redis_connection_pool: redis_kwargs", redis_kwargs)
 
     if "startup_nodes" in redis_kwargs:

tests/test_litellm/test_redis.py

@@ -929,8 +929,9 @@ GCP_IAM_CONNECT_FUNC = {"_gcp_service_account": "projects/-/serviceAccounts/sa@p
 def test_async_url_client_authenticates_through_credential_provider(markers, provider_cls):
     """A REDIS_URL config with Azure AD or GCP IAM must still reach the server with a credential.
 
-    The async client accepts redis_connect_func as a kwarg but never calls it, so the url
-    branch has to hand the connection a CredentialProvider or it authenticates with nothing.
+    The url branch forwards redis_connect_func straight to the async connection, which runs
+    its AUTH exchange with the blocking client API and dies, so the branch has to hand the
+    connection a CredentialProvider instead.
     """
     redis_kwargs = {
         "url": "rediss://redis-host:6380",
@@ -983,3 +984,24 @@ def test_async_url_client_drops_username_alongside_credential_provider():
     pool = client.connection_pool
     assert "username" not in pool.connection_kwargs
     pool.connection_class(**pool.connection_kwargs)
+
+
+@pytest.mark.parametrize("build_pool", [False, True], ids=["client", "pool"])
+def test_async_url_keeps_a_coroutine_connect_func(build_pool):
+    """redis-py awaits a coroutine redis_connect_func on an async connection, so one we cannot
+    turn into a credential provider has to be left where it is rather than dropped.
+    """
+
+    async def connect(connection):
+        return None
+
+    redis_kwargs = {
+        "url": "rediss://redis-host:6380",
+        "redis_connect_func": connect,
+    }
+
+    with patch("litellm._redis._get_redis_client_logic", return_value=redis_kwargs):
+        pool = get_redis_connection_pool() if build_pool else get_redis_async_client().connection_pool
+
+    assert pool.connection_kwargs["redis_connect_func"] is connect
+    assert "credential_provider" not in pool.connection_kwargs