Python: Allow RedisHistoryProvider to borrow a Redis client

python/packages/redis/AGENTS.md

@@ -6,7 +6,9 @@ Redis-based storage for agent threads and context.
 
 - **`RedisHistoryProvider`** - Persistent chat history provider using scoped Redis keys. Scoped mode requires an
   application ID and non-empty session ID, and can additionally isolate tenants and agents. Use explicit legacy mode
-  only while deliberately migrating historical keys; scoped mode never accesses them automatically.
+  only while deliberately migrating historical keys; scoped mode never accesses them automatically. It can construct
+  an owned client from a URL or credential provider, or borrow a standalone async Redis client configured with
+  `decode_responses=True`; borrowed clients remain caller-owned.
 - **`RedisContextProvider`** - Context provider with Redis-backed retrieval
 - **`RedisSettings`** - TypedDict connection settings for vector stores, resolved with core `load_settings` from
   explicit URL overrides, an optional .env file, or `REDIS_URL`. URLs use `SecretString` to mask credentials.

python/packages/redis/README.md

@@ -63,7 +63,15 @@ settings loader. Connection precedence is an explicit `redis_url`, then
 `redis://localhost:6379` as the default. Settings mask credential-bearing URLs
 with `SecretString`; use `rediss://` when your server requires TLS.
 
-You may supply a standalone `redis.asyncio.Redis` client using
+`RedisHistoryProvider` can create its own client from a URL or credential
+provider, or borrow a standalone `redis.asyncio.Redis` client configured with
+`decode_responses=True`. A borrowed client lets the application configure
+connection policy such as bounded timeouts and health checks. It remains
+caller-owned and must not be combined with `redis_url` or `credential_provider`.
+Retry behavior also remains caller-controlled; transcript appends are not
+idempotent if a client replays a write after an ambiguous connection loss.
+
+For vector storage, you may supply a standalone `redis.asyncio.Redis` client using
 `decode_responses=False` and RESP2. Both URL-created and supplied clients must use
 strict UTF-8 encoding (`encoding="utf-8"`, `encoding_errors="strict"`, the defaults)
 so Unicode keys and string fields round-trip without lossy conversions.
@@ -97,6 +105,25 @@ history_provider = RedisHistoryProvider(
 )
 ```
 
+To bound transcript I/O with a caller-configured client:
+
+```python
+from redis.asyncio import Redis
+
+redis_client = Redis.from_url(
+    "redis://localhost:6379",
+    decode_responses=True,
+    socket_connect_timeout=3,
+    socket_timeout=5,
+)
+history_provider = RedisHistoryProvider(
+    redis_client=redis_client,
+    application_id="support-app",
+)
+```
+
+The application remains responsible for closing `redis_client`.
+
 Scoped mode rejects missing application or session identifiers rather than
 placing unrelated conversations under a shared fallback key. Identifiers are
 encoded independently, so they do not need to be globally unique across

python/packages/redis/agent_framework_redis/_history_provider.py

@@ -46,6 +46,7 @@ class RedisHistoryProvider(HistoryProvider):
 
     Stores conversation history in Redis Lists, with each session isolated by a
     key scoped to the application, optional tenant and agent, and provider source.
+    Caller-provided Redis clients are borrowed.
     """
 
     DEFAULT_SOURCE_ID: ClassVar[str] = "redis_memory"
@@ -67,6 +68,7 @@ def __init__(
         ssl: bool = True,
         username: str | None = None,
         *,
+        redis_client: redis.Redis | None = None,
         key_prefix: str = "chat_messages",
         tenant_id: str | None = None,
         application_id: str | None = None,
@@ -84,13 +86,16 @@ def __init__(
         Args:
             source_id: Unique identifier for this provider instance.
             redis_url: Redis connection URL (e.g., "redis://localhost:6379").
-                Mutually exclusive with credential_provider.
+                Mutually exclusive with credential_provider and redis_client.
             credential_provider: Redis credential provider for Azure AD authentication.
-                Requires host parameter. Mutually exclusive with redis_url.
+                Requires host parameter. Mutually exclusive with redis_url and redis_client.
             host: Redis host name. Required when using credential_provider.
             port: Redis port number. Defaults to 6380 (Azure Redis SSL port).
             ssl: Enable SSL/TLS connection. Defaults to True.
             username: Redis username.
+            redis_client: Borrowed standalone async Redis client configured with
+                ``decode_responses=True``. The caller owns its lifetime. Mutually exclusive
+                with redis_url and credential_provider.
             key_prefix: Base prefix for Redis keys. Scoped mode appends independently encoded
                 tenant, application, agent, provider source, and session segments.
                 Defaults to 'chat_messages'.
@@ -114,9 +119,11 @@ def __init__(
             store_context_from: If set, only store context from these source_ids.
 
         Raises:
-            ValueError: If neither redis_url nor credential_provider is provided.
-            ValueError: If both redis_url and credential_provider are provided.
+            TypeError: If redis_client is not a standalone async Redis client.
+            ValueError: If no Redis connection source is provided.
+            ValueError: If more than one Redis connection source is provided.
             ValueError: If credential_provider is used without host parameter.
+            ValueError: If redis_client does not use decoded string responses.
             ValueError: If max_messages is negative.
             ValueError: If key_format or its scoped identifiers are invalid.
         """
@@ -129,12 +136,19 @@ def __init__(
             store_context_from=store_context_from,
         )
 
-        if redis_url is None and credential_provider is None:
-            raise ValueError("Either redis_url or credential_provider must be provided")
-        if redis_url is not None and credential_provider is not None:
-            raise ValueError("redis_url and credential_provider are mutually exclusive")
+        connection_sources = sum(source is not None for source in (redis_client, redis_url, credential_provider))
+        if connection_sources == 0:
+            raise ValueError("Either redis_client, redis_url, or credential_provider must be provided")
+        if connection_sources > 1:
+            raise ValueError("redis_client, redis_url, and credential_provider are mutually exclusive")
         if credential_provider is not None and host is None:
             raise ValueError("host is required when using credential_provider")
+        if redis_client is not None:
+            if not isinstance(redis_client, redis.Redis):
+                raise TypeError("Redis history providers require a standalone redis.asyncio.Redis client")
+            connection_kwargs = cast(dict[str, Any], redis_client.connection_pool.connection_kwargs)  # pyright: ignore[reportUnknownMemberType]
+            if not connection_kwargs.get("decode_responses", False):
+                raise ValueError("Redis history providers require decode_responses=True")
         if max_messages is not None and max_messages < 0:
             raise ValueError("max_messages must be None (unlimited) or a non-negative integer")
         if key_format not in ("scoped", "legacy"):
@@ -163,8 +177,11 @@ def __init__(
         self.key_format = key_format
         self.max_messages = max_messages
         self.redis_url = redis_url
+        self._owns_client = redis_client is None
 
-        if credential_provider is not None and host is not None:
+        if redis_client is not None:
+            self._redis_client = redis_client
+        elif credential_provider is not None and host is not None:
             self._redis_client = redis.Redis(
                 host=host,
                 port=port,
@@ -313,8 +330,9 @@ async def clear(self, session_id: str | None) -> None:
         await self._redis_client.delete(self._redis_key(session_id))
 
     async def aclose(self) -> None:
-        """Close the Redis connection."""
-        await self._redis_client.aclose()
+        """Close the Redis connection when this provider owns it."""
+        if self._owns_client:
+            await self._redis_client.aclose()
 
 
 __all__ = ["RedisHistoryProvider"]

python/packages/redis/tests/test_providers.py

@@ -11,6 +11,7 @@
 import pytest
 from agent_framework import AgentResponse, Message
 from agent_framework._sessions import AgentSession, SessionContext
+from redis.asyncio import Redis
 
 from agent_framework_redis._context_provider import RedisContextProvider
 from agent_framework_redis._feature_usage import FeatureIndex
@@ -64,6 +65,7 @@ def mock_redis_client():
     client.llen = AsyncMock(return_value=0)
     client.ltrim = AsyncMock()
     client.delete = AsyncMock()
+    client.aclose = AsyncMock()
 
     mock_pipeline = AsyncMock()
     mock_pipeline.rpush = AsyncMock()
@@ -383,7 +385,7 @@ def test_custom_params(self, mock_redis_client: MagicMock):
         assert provider.store_inputs is False
 
     def test_no_redis_url_or_credential_raises(self):
-        with pytest.raises(ValueError, match="Either redis_url or credential_provider must be provided"):
+        with pytest.raises(ValueError, match="Either redis_client, redis_url, or credential_provider"):
             RedisHistoryProvider("mem")
 
     def test_both_url_and_credential_raises(self):
@@ -396,6 +398,46 @@ def test_both_url_and_credential_raises(self):
                 host="myhost",
             )
 
+    def test_borrowed_client_is_used_directly(self):
+        borrowed = Redis(decode_responses=True)
+
+        provider = RedisHistoryProvider("mem", redis_client=borrowed, application_id="test-app")
+
+        assert provider._redis_client is borrowed
+        assert provider._owns_client is False
+
+    @pytest.mark.parametrize(
+        "connection_kwargs",
+        [
+            {"redis_url": "redis://localhost:6379"},
+            {"credential_provider": MagicMock(), "host": "myhost"},
+        ],
+    )
+    def test_borrowed_client_rejects_other_connection_sources(self, connection_kwargs: dict[str, Any]):
+        borrowed = Redis(decode_responses=True)
+
+        with pytest.raises(ValueError, match="mutually exclusive"):
+            RedisHistoryProvider(
+                "mem",
+                redis_client=borrowed,
+                application_id="test-app",
+                **connection_kwargs,
+            )
+
+    def test_borrowed_client_must_be_async_redis(self):
+        with pytest.raises(TypeError, match=r"redis\.asyncio\.Redis"):
+            RedisHistoryProvider(
+                "mem",
+                redis_client=cast(Any, MagicMock()),
+                application_id="test-app",
+            )
+
+    def test_borrowed_client_requires_decoded_responses(self):
+        borrowed = Redis(decode_responses=False)
+
+        with pytest.raises(ValueError, match="decode_responses=True"):
+            RedisHistoryProvider("mem", redis_client=borrowed, application_id="test-app")
+
     def test_credential_provider_without_host_raises(self):
         mock_cred = MagicMock()
         with pytest.raises(ValueError, match="host is required"):
@@ -424,6 +466,29 @@ def test_credential_provider_with_host(self):
         assert provider.redis_url is None
 
 
+class TestRedisHistoryProviderClose:
+    async def test_owned_client_is_closed(self, mock_redis_client: MagicMock):
+        with patch(
+            "agent_framework_redis._history_provider.redis.from_url",
+            return_value=mock_redis_client,
+        ):
+            provider = RedisHistoryProvider("mem", redis_url="redis://localhost:6379", application_id="test-app")
+
+        await provider.aclose()
+
+        mock_redis_client.aclose.assert_awaited_once()
+
+    async def test_borrowed_client_is_not_closed(self):
+        borrowed = Redis(decode_responses=True)
+        with patch.object(borrowed, "aclose", new_callable=AsyncMock) as close:
+            provider = RedisHistoryProvider("mem", redis_client=borrowed, application_id="test-app")
+
+            await provider.aclose()
+
+            close.assert_not_awaited()
+        await borrowed.aclose()
+
+
 class TestRedisHistoryProviderRedisKey:
     def test_scoped_key_is_deterministic_and_encodes_opaque_components(self, mock_redis_client: MagicMock):
         with patch("agent_framework_redis._history_provider.redis.from_url") as mock_from_url: