Support redis-py 8

.github/workflows/test.yml

@@ -69,7 +69,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
-        redis-py-version: ["6.x", "7.x"]
+        redis-py-version: ["6.x", "7.x", "8.x"]
         redis-image: ["redis:8.2", "redis:8.4", "redis:latest"]
     steps:
       - name: Check out repository
@@ -106,6 +106,7 @@ jobs:
           case "$REDIS_PY_VERSION" in
             "6.x") spec="redis>=6,<7" ;;
             "7.x") spec="redis>=7,<8" ;;
+            "8.x") spec="redis>=8,<9" ;;
             *)
               echo "::error title=Unhandled redis-py-version::Matrix value '${REDIS_PY_VERSION}' has no install rule -- add a case branch in .github/workflows/test.yml."
               exit 1

pyproject.toml

@@ -25,7 +25,7 @@ classifiers = [
 dependencies = [
     "numpy>=1.26.0,<3",
     "pyyaml>=5.4,<7.0",
-    "redis>=5.0,<8.0",
+    "redis>=6.3.0,<9.0",
     "pydantic>=2,<3",
     "tenacity>=8.2.2",
     "ml-dtypes>=0.4.0,<1.0.0",

redisvl/extensions/cache/embeddings/embeddings.py

@@ -1,5 +1,6 @@
 """Embeddings cache implementation for RedisVL."""
 
+from collections.abc import Mapping
 from typing import Any, Iterable
 
 from redisvl.extensions.cache.base import BaseCache
@@ -111,7 +112,9 @@ def _prepare_entry_data(
         )
         return key, entry.to_dict()
 
-    def _process_cache_data(self, data: dict[str, Any] | None) -> dict[str, Any] | None:
+    def _process_cache_data(
+        self, data: Mapping[bytes | str, Any] | None
+    ) -> dict[str, Any] | None:
         """Process Redis hash data into a cache entry response.
 
         Args:
@@ -123,7 +126,7 @@ def _process_cache_data(self, data: dict[str, Any] | None) -> dict[str, Any] | N
         if not data:
             return None
 
-        cache_hit = CacheEntry(**convert_bytes(data))
+        cache_hit = CacheEntry(**convert_bytes(dict(data)))
         return cache_hit.model_dump(exclude_none=True)
 
     def _should_warn_for_async_only(self) -> bool:

redisvl/index/index.py

@@ -1799,7 +1799,8 @@ def batch_search(
 
                 for j, query_results in enumerate(results):
                     _built_query = batch_built_queries[j]
-                    parsed_result = search._parse_search(  # type: ignore
+                    parsed_result = search._parse_results(  # type: ignore
+                        "FT.SEARCH",
                         query_results,
                         query=_built_query,
                         duration=duration,
@@ -2998,7 +2999,8 @@ async def batch_search(
 
                 for j, query_results in enumerate(results):
                     _built_query = batch_built_queries[j]
-                    parsed_result = search._parse_search(  # type: ignore
+                    parsed_result = search._parse_results(  # type: ignore
+                        "FT.SEARCH",
                         query_results,
                         query=_built_query,
                         duration=duration,

redisvl/index/storage.py

@@ -1,12 +1,13 @@
-from collections.abc import Collection
-from typing import Any, Callable, Iterable
+from collections.abc import Collection, Mapping
+from typing import Any, Callable, Iterable, cast
 
 from pydantic import BaseModel, ValidationError
 from redis import __version__ as redis_version
 
 # Add imports for Pipeline types
 from redis.asyncio.client import Pipeline as AsyncPipeline
 from redis.asyncio.cluster import ClusterPipeline as AsyncClusterPipeline
+from redis.typing import EncodableT
 
 # Redis 5.x compatibility (6 fixed the import path)
 if redis_version.startswith("5"):
@@ -611,7 +612,10 @@ def _set(client: RedisClientOrPipeline, key: str, obj: dict[str, Any]):
             key (str): The key under which to store the hash.
             obj (Dict[str, Any]): The hash to store in Redis.
         """
-        client.hset(name=key, mapping=obj)
+        client.hset(
+            name=key,
+            mapping=cast(Mapping[EncodableT, EncodableT], obj),
+        )
 
     @staticmethod
     async def _aset(client: AsyncRedisClientOrPipeline, key: str, obj: dict[str, Any]):

redisvl/redis/connection.py

@@ -222,11 +222,16 @@ def convert_index_info_to_schema(index_info: dict[str, Any]) -> dict[str, Any]:
         Dict[str, Any]: Schema dictionary suitable for ``IndexSchema.from_dict()``.
     """
     index_name = index_info["index_name"]
-    prefixes = index_info["index_definition"][3]
+    index_definition = index_info["index_definition"]
+    if isinstance(index_definition, dict):
+        prefixes = index_definition["prefixes"]
+        storage_type = index_definition["key_type"].lower()
+    else:
+        prefixes = index_definition[3]
+        storage_type = index_definition[1].lower()
     # Normalize single-element prefix lists to string for backward compatibility
     if isinstance(prefixes, list) and len(prefixes) == 1:
         prefixes = prefixes[0]
-    storage_type = index_info["index_definition"][1].lower()
 
     # Parse stopwords if present in FT.INFO output
     # stopwords_list is only present when explicitly set (STOPWORDS 0 or custom list)
@@ -252,35 +257,42 @@ def parse_vector_attrs(attrs):
         # - Redis 7.x+: [... "VECTOR", "ALGORITHM", "FLAT", "TYPE", "FLOAT32", "DIM", "3", ...]
         #   Position 6+: all key-value pairs
 
-        # Check if we have any attributes beyond the type declaration
-        if len(attrs) <= 6:
-            # Redis 6.2.6-v9 or similar: no vector params in FT.INFO
-            # Return None to signal we can't parse this field properly
-            return None
-
-        vector_attrs = {}
-        start_pos = 6
-
-        # Detect format: if position 6 looks like an algorithm value (not a key),
-        # we're dealing with the older format
-        if len(attrs) > 6:
-            pos6_str = str(attrs[6]).upper()
-            # Check if position 6 is an algorithm value (FLAT, HNSW) vs a key (ALGORITHM, TYPE, DIM)
-            if pos6_str in ("FLAT", "HNSW"):
-                # Old format (Redis 6.2.x): position 6 is algorithm value, position 7 is param count
-                # Store the algorithm
-                vector_attrs["algorithm"] = pos6_str
-                # Skip to position 8 where key-value pairs start
-                start_pos = 8
+        if isinstance(attrs, dict):
+            vector_attrs = {
+                str(key).lower(): value
+                for key, value in attrs.items()
+                if key not in {"identifier", "attribute", "type", "flags"}
+            }
+        else:
+            # Check if we have any attributes beyond the type declaration
+            if len(attrs) <= 6:
+                # Redis 6.2.6-v9 or similar: no vector params in FT.INFO
+                # Return None to signal we can't parse this field properly
+                return None
+
+            vector_attrs = {}
+            start_pos = 6
+
+            # Detect format: if position 6 looks like an algorithm value (not a key),
+            # we're dealing with the older format
+            if len(attrs) > 6:
+                pos6_str = str(attrs[6]).upper()
+                # Check if position 6 is an algorithm value (FLAT, HNSW) vs a key (ALGORITHM, TYPE, DIM)
+                if pos6_str in ("FLAT", "HNSW"):
+                    # Old format (Redis 6.2.x): position 6 is algorithm value, position 7 is param count
+                    # Store the algorithm
+                    vector_attrs["algorithm"] = pos6_str
+                    # Skip to position 8 where key-value pairs start
+                    start_pos = 8
 
-        try:
-            for i in range(start_pos, len(attrs), 2):
-                if i + 1 < len(attrs):
-                    key = str(attrs[i]).lower()
-                    vector_attrs[key] = attrs[i + 1]
-        except (IndexError, TypeError, ValueError):
-            # Silently continue - we'll validate required fields below
-            pass
+            try:
+                for i in range(start_pos, len(attrs), 2):
+                    if i + 1 < len(attrs):
+                        key = str(attrs[i]).lower()
+                        vector_attrs[key] = attrs[i + 1]
+            except (IndexError, TypeError, ValueError):
+                # Silently continue - we'll validate required fields below
+                pass
 
         # Normalize to expected field names
         normalized = {}
@@ -402,7 +414,6 @@ def parse_attrs(attrs, field_type=None):
         # 'SORTABLE', 'NOSTEM' don't have corresponding values.
         # Their presence indicates boolean True
         # TODO 'WITHSUFFIXTRIE' is another boolean attr, but is not returned by ft.info
-        original = attrs.copy()
         parsed_attrs = {}
 
         # Handle all boolean attributes first, regardless of position
@@ -415,6 +426,24 @@ def parse_attrs(attrs, field_type=None):
             "NOINDEX": "no_index",
         }
 
+        if isinstance(attrs, dict):
+            flags = attrs.get("flags", []).copy()
+            for redis_attr, python_attr in boolean_attrs.items():
+                if redis_attr in flags:
+                    parsed_attrs[python_attr] = True
+            if "UNF" in flags and field_type == "TEXT":
+                parsed_attrs["unf"] = True
+            parsed_attrs.update(
+                {
+                    str(key).lower(): value
+                    for key, value in attrs.items()
+                    if key not in {"identifier", "attribute", "type", "flags"}
+                }
+            )
+            return parsed_attrs
+
+        original = attrs.copy()
+
         # Special handling for UNF:
         # - For NUMERIC fields, Redis always adds UNF when SORTABLE is present
         # - For TEXT fields, UNF is only present when explicitly set
@@ -442,20 +471,28 @@ def parse_attrs(attrs, field_type=None):
 
     for field_attrs in index_fields:
         # parse field info
-        name = field_attrs[1] if storage_type == "hash" else field_attrs[3]
-        field = {"name": name, "type": field_attrs[5].lower()}
-        if storage_type == "json":
-            field["path"] = field_attrs[1]
+        if isinstance(field_attrs, dict):
+            name = field_attrs["attribute"]
+            field_type = field_attrs["type"]
+            field = {"name": name, "type": field_type.lower()}
+            if storage_type == "json":
+                field["path"] = field_attrs["identifier"]
+        else:
+            name = field_attrs[1] if storage_type == "hash" else field_attrs[3]
+            field_type = field_attrs[5]
+            field = {"name": name, "type": field_type.lower()}
+            if storage_type == "json":
+                field["path"] = field_attrs[1]
         # parse field attrs
-        if field_attrs[5] == "VECTOR":
+        if field_type == "VECTOR":
             attrs = parse_vector_attrs(field_attrs)
             if attrs is None:
                 # Vector field attributes cannot be parsed on this Redis version
                 # Skip this field - it cannot be properly reconstructed
                 continue
             field["attrs"] = attrs
         else:
-            field["attrs"] = parse_attrs(field_attrs, field_type=field_attrs[5])
+            field["attrs"] = parse_attrs(field_attrs, field_type=field_type)
         # append field
         schema_fields.append(field)
 
@@ -533,6 +570,9 @@ def get_redis_connection(
                 variable is not set.
         """
         url = redis_url or get_address_from_env()
+        # redis-py 8 defaults to RESP3, which changes raw Search command reply
+        # shapes. Keep RedisVL's existing RESP2 behavior unless requested.
+        kwargs.setdefault("protocol", 2)
         client: SyncRedisClient
         if url.startswith("redis+sentinel"):
             client = RedisConnectionFactory._redis_sentinel_client(url, Redis, **kwargs)
@@ -580,6 +620,8 @@ async def _get_aredis_connection(
         """
         _deprecated_url = kwargs.pop("url", None)
         url = _deprecated_url or redis_url or get_address_from_env()
+        # Keep sync and async clients on the same backward-compatible default.
+        kwargs.setdefault("protocol", 2)
 
         client: AsyncRedisClient
         if url.startswith("redis+sentinel"):
@@ -640,6 +682,7 @@ def get_async_redis_connection(
         )
         _deprecated_url = kwargs.pop("url", None)
         url = _deprecated_url or redis_url or get_address_from_env()
+        kwargs.setdefault("protocol", 2)
 
         if url.startswith("redis+sentinel"):
             return RedisConnectionFactory._redis_sentinel_client(
@@ -665,6 +708,7 @@ def get_redis_cluster_connection(
     ) -> RedisCluster:
         """Creates and returns a synchronous Redis client for a Redis cluster."""
         url = redis_url or get_address_from_env()
+        kwargs.setdefault("protocol", 2)
         return RedisCluster.from_url(url, **kwargs)
 
     @staticmethod
@@ -674,6 +718,7 @@ def get_async_redis_cluster_connection(
     ) -> AsyncRedisCluster:
         """Creates and returns an asynchronous Redis client for a Redis cluster."""
         url = redis_url or get_address_from_env()
+        kwargs.setdefault("protocol", 2)
         # Strip 'cluster' parameter as AsyncRedisCluster doesn't accept it
         cleaned_url, cleaned_kwargs = _strip_cluster_from_url_and_kwargs(url, **kwargs)
         return AsyncRedisCluster.from_url(cleaned_url, **cleaned_kwargs)

tests/integration/test_async_search_index.py

@@ -542,6 +542,49 @@ async def test_batch_search(async_index):
     assert results[1].docs[0]["id"] == "rvl:2"
 
 
+@pytest.mark.asyncio
+async def test_default_client_from_existing_and_batch_search(
+    redis_url, redis_test_name
+):
+    """User-provided async redis-py clients support introspection and batch search."""
+    name = redis_test_name("async_resp3_index")
+    prefix = f"{name}:"
+    client = AsyncRedis.from_url(redis_url)
+    index = AsyncSearchIndex.from_dict(
+        {
+            "index": {"name": name, "prefix": prefix},
+            "fields": [
+                {"name": "test", "type": "tag"},
+                {
+                    "name": "embedding",
+                    "type": "vector",
+                    "attrs": {
+                        "dims": 3,
+                        "distance_metric": "cosine",
+                        "algorithm": "flat",
+                        "datatype": "float32",
+                    },
+                },
+            ],
+        },
+        redis_client=client,
+    )
+
+    try:
+        await index.create()
+        await index.load([{"id": "1", "test": "foo"}], id_field="id")
+
+        reopened = await AsyncSearchIndex.from_existing(name, redis_client=client)
+        results = await reopened.batch_search(["@test:{foo}"])
+
+        assert reopened.schema == index.schema
+        assert results[0].total == 1
+        assert results[0].docs[0]["id"] == f"{prefix}1"
+    finally:
+        await index.delete(drop=True)
+        await client.aclose()
+
+
 @pytest.mark.parametrize(
     "queries",
     [

tests/integration/test_mcp/test_transport.py

@@ -249,16 +249,30 @@ async def test_server_read_only_mode_hides_upsert_over_http(
 
 async def _post_mcp(port: int, headers: dict) -> httpx.Response:
     """POST a minimal payload to /mcp with custom headers; return the response."""
-    async with httpx.AsyncClient(timeout=5.0) as client:
-        return await client.post(
-            f"http://127.0.0.1:{port}/mcp",
-            headers={
-                "content-type": "application/json",
-                "accept": "application/json, text/event-stream",
-                **headers,
-            },
-            json={"jsonrpc": "2.0", "id": 1, "method": "ping"},
-        )
+    reader, writer = await asyncio.open_connection("127.0.0.1", port)
+    payload = b'{"jsonrpc":"2.0","id":1,"method":"ping"}'
+    request_headers = {
+        "host": f"127.0.0.1:{port}",
+        "content-type": "application/json",
+        "accept": "application/json, text/event-stream",
+        "content-length": str(len(payload)),
+        "connection": "close",
+        **headers,
+    }
+    request = "POST /mcp HTTP/1.1\r\n" + "".join(
+        f"{name}: {value}\r\n" for name, value in request_headers.items()
+    )
+    writer.write(request.encode("ascii") + b"\r\n" + payload)
+    await writer.drain()
+
+    status_line = await reader.readline()
+    status_code = int(status_line.split()[1])
+    while await reader.readline() not in (b"\r\n", b""):
+        pass
+    body = await reader.read()
+    writer.close()
+    await writer.wait_closed()
+    return httpx.Response(status_code, content=body)
 
 
 @pytest.mark.asyncio