downgraded redis to 3.2.1

Author: cshinaver

.tool-versions

@@ -0,0 +1 @@
+python 3.10.16

pyproject.toml

@@ -30,14 +30,15 @@ dependencies = [
     "sse-starlette>=1.6.1",
     "pydantic-settings>=2.5.2",
     "uvicorn>=0.23.1; sys_platform != 'emscripten'",
-    "fakeredis==2.28.1",
+    "fakeredis==1.9.0",
+    "redis==3.2.1",
 ]
 
 [project.optional-dependencies]
 rich = ["rich>=13.9.4"]
 cli = ["typer>=0.12.4", "python-dotenv>=1.0.0"]
 ws = ["websockets>=15.0.1"]
-redis = ["redis>=5.2.1", "types-redis>=4.6.0.20241004"]
+redis = ["redis==3.2.1"]
 
 [project.scripts]
 mcp = "mcp.cli:app [cli]"
@@ -115,5 +116,7 @@ filterwarnings = [
     # This should be fixed on Uvicorn's side.
     "ignore::DeprecationWarning:websockets",
     "ignore:websockets.server.WebSocketServerProtocol is deprecated:DeprecationWarning",
-    "ignore:Returning str or bytes.*:DeprecationWarning:mcp.server.lowlevel"
+    "ignore:Returning str or bytes.*:DeprecationWarning:mcp.server.lowlevel",
+    # Ignore distutils deprecation from Redis 3.2.1
+    "ignore:The distutils package is deprecated.*:DeprecationWarning"
 ]

src/mcp/server/message_queue/redis.py

@@ -1,4 +1,5 @@
 import logging
+import threading
 from contextlib import asynccontextmanager
 from typing import Any, cast
 from uuid import UUID
@@ -12,7 +13,7 @@
 from mcp.server.message_queue.base import MessageCallback
 
 try:
-    import redis.asyncio as redis
+    import redis
 except ImportError:
     raise ImportError(
         "Redis support requires the 'redis' package. "
@@ -40,19 +41,45 @@ def __init__(
             prefix: Key prefix for Redis channels to avoid collisions
             session_ttl: TTL in seconds for session keys (default: 1 hour)
         """
-        self._redis = redis.from_url(redis_url, decode_responses=True)  # type: ignore
-        self._pubsub = self._redis.pubsub(ignore_subscribe_messages=True)  # type: ignore
+        # Parse Redis URL and create connection
+        if redis_url.startswith("redis://"):
+            host_port = redis_url.replace("redis://", "").split("/")[0]
+            if ":" in host_port:
+                host, port = host_port.split(":")
+                port = int(port)
+            else:
+                host, port = host_port, 6379
+            db = int(redis_url.split("/")[-1]) if "/" in redis_url else 0
+            self._redis = redis.StrictRedis(host=host, port=port, db=db, decode_responses=True)
+        else:
+            self._redis = redis.StrictRedis.from_url(redis_url, decode_responses=True)
+        
+        self._pubsub = self._redis.pubsub(ignore_subscribe_messages=True)
         self._prefix = prefix
         self._session_ttl = session_ttl
         # Maps session IDs to the callback and task group for that SSE session.
         self._session_state: dict[UUID, tuple[MessageCallback, TaskGroup]] = {}
+        # Thread for pubsub listening
+        self._pubsub_thread = None
+        # Lock for thread safety
+        self._lock = threading.RLock()
+        # Tracks current subscriptions
+        self._subscriptions: set[str] = set()
         # Ensures only one polling task runs at a time for message handling
         self._limiter = CapacityLimiter(1)
+        # Active sessions set key
+        self._active_sessions_key = f"{self._prefix}active_sessions"
         logger.debug(f"Redis message dispatch initialized: {redis_url}")
 
     async def close(self):
-        await self._pubsub.aclose() # type: ignore
-        await self._redis.aclose() # type: ignore
+        """Close Redis connections."""
+        # Stop pubsub thread if running
+        if self._pubsub_thread:
+            self._pubsub_thread.stop()
+        
+        # Clean up pubsub and connection
+        self._pubsub.close()
+        # Redis connection in 3.2.1 doesn't need explicit closing
 
     def _session_channel(self, session_id: UUID) -> str:
         """Get the Redis channel for a session."""
@@ -66,24 +93,55 @@ def _session_key(self, session_id: UUID) -> str:
     async def subscribe(self, session_id: UUID, callback: MessageCallback):
         """Request-scoped context manager that subscribes to messages for a session."""
         session_key = self._session_key(session_id)
-        await self._redis.setex(session_key, self._session_ttl, "1")  # type: ignore
+        
+        # Run Redis operations in anyio's run_sync to make blocking calls non-blocking
+        await anyio.to_thread.run_sync(
+            lambda: self._redis.setex(session_key, self._session_ttl, "1")
+        )
+        
+        # Add to active sessions set
+        await anyio.to_thread.run_sync(
+            lambda: self._redis.sadd(self._active_sessions_key, session_id.hex)
+        )
 
         channel = self._session_channel(session_id)
-        await self._pubsub.subscribe(channel)  # type: ignore
-
+        
+        # Use lock for thread safety
+        with self._lock:
+            # Subscribe to channel
+            await anyio.to_thread.run_sync(lambda: self._pubsub.subscribe(channel))
+            self._subscriptions.add(channel)
+        
         logger.debug(f"Subscribing to Redis channel for session {session_id}")
+        
         async with anyio.create_task_group() as tg:
             self._session_state[session_id] = (callback, tg)
-            tg.start_soon(self._listen_for_messages)
+            # Start message listener if not running
+            if not self._pubsub_thread:
+                tg.start_soon(self._listen_for_messages)
+            
             # Start heartbeat for this session
             tg.start_soon(self._session_heartbeat, session_id)
             try:
                 yield
             finally:
                 with anyio.CancelScope(shield=True):
                     tg.cancel_scope.cancel()
-                    await self._pubsub.unsubscribe(channel)  # type: ignore
-                    await self._redis.delete(session_key)  # type: ignore
+                    
+                    # Unsubscribe
+                    with self._lock:
+                        await anyio.to_thread.run_sync(
+                            lambda: self._pubsub.unsubscribe(channel)
+                        )
+                        self._subscriptions.discard(channel)
+                    
+                    # Delete session key and remove from active sessions
+                    await anyio.to_thread.run_sync(lambda: self._redis.delete(session_key))
+                    await anyio.to_thread.run_sync(
+                        lambda: self._redis.srem(self._active_sessions_key, session_id.hex)
+                    )
+                    
+                    # Clean up session state
                     del self._session_state[session_id]
                     logger.debug(f"Unsubscribed from Redis channel: {session_id}")
 
@@ -96,7 +154,10 @@ async def _session_heartbeat(self, session_id: UUID) -> None:
                 # Refresh TTL at half the TTL interval to avoid expiration
                 await anyio.sleep(self._session_ttl / 2)
                 with anyio.CancelScope(shield=True):
-                    await self._redis.expire(session_key, self._session_ttl)  # type: ignore
+                    # Run in thread to avoid blocking
+                    await anyio.to_thread.run_sync(
+                        lambda: self._redis.expire(session_key, self._session_ttl)
+                    )
             except anyio.get_cancelled_exc_class():
                 break
             except Exception as e:
@@ -123,33 +184,39 @@ async def _listen_for_messages(self) -> None:
         """Background task that listens for messages on subscribed channels."""
         async with self._limiter:
             while True:
+                # Check for cancellation
                 await lowlevel.checkpoint()
+                
+                # Use a shield to prevent cancellation during message processing
                 with CancelScope(shield=True):
-                    message: None | dict[str, Any] = await self._pubsub.get_message(  # type: ignore
-                        ignore_subscribe_messages=True,
-                        timeout=0.1,  # type: ignore
+                    # Get message with non-blocking call using thread
+                    message = await anyio.to_thread.run_sync(
+                        lambda: self._pubsub.get_message(
+                            ignore_subscribe_messages=True,
+                            timeout=0.1,
+                        )
                     )
+                    
                     if message is None:
+                        # No message available, sleep briefly and try again
+                        await anyio.sleep(0.01)
                         continue
 
                     channel: str = cast(str, message["channel"])
                     session_id = self._extract_session_id(channel)
                     if session_id is None:
-                        logger.debug(
-                            f"Ignoring message from non-MCP channel: {channel}"
-                        )
+                        logger.debug(f"Ignoring message from non-MCP channel: {channel}")
                         continue
 
                     data: str = cast(str, message["data"])
                     try:
                         if session_state := self._session_state.get(session_id):
+                            # Process message in task group
                             session_state[1].start_soon(
                                 self._handle_message, session_id, data
                             )
                         else:
-                            logger.warning(
-                                f"Message dropped: unknown session {session_id}"
-                            )
+                            logger.warning(f"Message dropped: unknown session {session_id}")
                     except Exception as e:
                         logger.error(f"Error processing message for {session_id}: {e}")
 
@@ -186,11 +253,15 @@ async def publish_message(
             data = message.model_dump_json()
 
         channel = self._session_channel(session_id)
-        await self._redis.publish(channel, data)  # type: ignore[attr-defined]
+        
+        # Run publish in thread to avoid blocking
+        await anyio.to_thread.run_sync(lambda: self._redis.publish(channel, data))
         logger.debug(f"Message published to Redis channel for session {session_id}")
         return True
 
     async def session_exists(self, session_id: UUID) -> bool:
         """Check if a session exists."""
         session_key = self._session_key(session_id)
-        return bool(await self._redis.exists(session_key))  # type: ignore
+        # Run exists command in thread to avoid blocking
+        exists = await anyio.to_thread.run_sync(lambda: self._redis.exists(session_key))
+        return bool(exists)

tests/server/message_queue/test_redis.py

@@ -9,7 +9,9 @@
 
 # Set up fakeredis for testing
 try:
-    from fakeredis import aioredis as fake_redis
+    import fakeredis
+    # Older fakeredis (v1.9.0) doesn't have aioredis module
+    fake_redis = fakeredis
 except ImportError:
     pytest.skip(
         "fakeredis is required for testing Redis functionality", allow_module_level=True
@@ -20,7 +22,7 @@
 async def redis_dispatch():
     """Create a Redis message dispatch with a fake Redis client."""
     # Mock the redis module entirely within RedisMessageDispatch
-    with patch("mcp.server.message_queue.redis.redis", fake_redis.FakeRedis):
+    with patch("mcp.server.message_queue.redis.redis.StrictRedis", fake_redis.FakeStrictRedis):
         from mcp.server.message_queue.redis import RedisMessageDispatch
 
         dispatch = RedisMessageDispatch(session_ttl=5)  # Shorter TTL for testing
@@ -53,7 +55,8 @@ async def test_session_ttl(redis_dispatch):
 
     async with redis_dispatch.subscribe(session_id, AsyncMock()):
         session_key = redis_dispatch._session_key(session_id)
-        ttl = await redis_dispatch._redis.ttl(session_key)  # type: ignore
+        # Updated for synchronous Redis client
+        ttl = await anyio.to_thread.run_sync(lambda: redis_dispatch._redis.ttl(session_key))
         assert ttl > 0
         assert ttl <= redis_dispatch._session_ttl
 
@@ -67,14 +70,14 @@ async def test_session_heartbeat(redis_dispatch):
         session_key = redis_dispatch._session_key(session_id)
 
         # Initial TTL
-        initial_ttl = await redis_dispatch._redis.ttl(session_key)  # type: ignore
+        initial_ttl = await anyio.to_thread.run_sync(lambda: redis_dispatch._redis.ttl(session_key))
         assert initial_ttl > 0
 
         # Wait for heartbeat to run
         await anyio.sleep(redis_dispatch._session_ttl / 2 + 0.5)
 
         # TTL should be refreshed
-        refreshed_ttl = await redis_dispatch._redis.ttl(session_key)  # type: ignore
+        refreshed_ttl = await anyio.to_thread.run_sync(lambda: redis_dispatch._redis.ttl(session_key))
         assert refreshed_ttl > 0
         assert refreshed_ttl <= redis_dispatch._session_ttl
 

tests/server/message_queue/test_redis_integration.py

@@ -22,7 +22,9 @@
 
 # Set up fakeredis for testing
 try:
-    from fakeredis import aioredis as fake_redis
+    import fakeredis
+    # Older fakeredis (v1.9.0) doesn't have aioredis module
+    fake_redis = fakeredis
 except ImportError:
     pytest.skip(
         "fakeredis is required for testing Redis functionality", allow_module_level=True
@@ -64,10 +66,10 @@ async def handle_call_tool(name: str, args: dict) -> list[TextContent]:
 def make_redis_server_app() -> Starlette:
     """Create test Starlette app with SSE transport and Redis message dispatch"""
     # Create a mock Redis instance
-    mock_redis = fake_redis.FakeRedis()
+    mock_redis = fake_redis.FakeStrictRedis(decode_responses=True)
 
     # Patch the redis module within RedisMessageDispatch
-    with patch("mcp.server.message_queue.redis.redis", mock_redis):
+    with patch("mcp.server.message_queue.redis.redis.StrictRedis", lambda *args, **kwargs: mock_redis):
         from mcp.server.message_queue.redis import RedisMessageDispatch
 
         # Create Redis message dispatch with mock redis
@@ -173,11 +175,11 @@ async def test_redis_integration_tool_call(server: None, server_url: str) -> Non
 async def test_redis_integration_session_lifecycle() -> None:
     """Test that sessions are properly added to and removed from Redis using direct Redis access"""
     # Create a fresh Redis instance with decode_responses=True to get str instead of bytes
-    mock_redis = fake_redis.FakeRedis(decode_responses=True)
+    mock_redis = fake_redis.FakeStrictRedis(decode_responses=True)
     active_sessions_key = "mcp:pubsub:active_sessions"
 
     # Mock Redis in RedisMessageDispatch
-    with patch("mcp.server.message_queue.redis.redis.from_url", return_value=mock_redis):
+    with patch("mcp.server.message_queue.redis.redis.StrictRedis", lambda *args, **kwargs: mock_redis):
         from mcp.server.message_queue.redis import RedisMessageDispatch
 
         # Create Redis message dispatch with our specific mock redis instance
@@ -197,7 +199,7 @@ async def mock_callback(message):
             await anyio.sleep(0.05)
 
             # Check that session was added to Redis
-            active_sessions = await mock_redis.smembers(active_sessions_key)
+            active_sessions = mock_redis.smembers(active_sessions_key)
             assert len(active_sessions) == 1
             assert list(active_sessions)[0] == session_id.hex
 
@@ -208,7 +210,7 @@ async def mock_callback(message):
         await anyio.sleep(0.05)
 
         # After context exit, verify the session was removed
-        final_sessions = await mock_redis.smembers(active_sessions_key)
+        final_sessions = mock_redis.smembers(active_sessions_key)
         assert len(final_sessions) == 0
         assert not await message_dispatch.session_exists(session_id)
 
@@ -217,10 +219,10 @@ async def mock_callback(message):
 async def test_redis_integration_message_publishing_direct() -> None:
     """Test message publishing through Redis channels using direct Redis access"""
     # Create a fresh Redis instance with decode_responses=True to get str instead of bytes
-    mock_redis = fake_redis.FakeRedis(decode_responses=True)
+    mock_redis = fake_redis.FakeStrictRedis(decode_responses=True)
 
     # Mock Redis in RedisMessageDispatch
-    with patch("mcp.server.message_queue.redis.redis.from_url", return_value=mock_redis):
+    with patch("mcp.server.message_queue.redis.redis.StrictRedis", lambda *args, **kwargs: mock_redis):
         from mcp.server.message_queue.redis import RedisMessageDispatch
         from mcp.types import JSONRPCMessage, JSONRPCRequest
 
@@ -252,8 +254,8 @@ async def message_callback(message):
             assert success
 
             # Give some time for the message to be processed
-            # Use a shorter sleep since we're in controlled test environment
-            await anyio.sleep(0.1)
+            # Use a longer sleep since we're with older Redis version
+            await anyio.sleep(0.5)
 
             # Verify that the message was received
             assert len(messages_received) > 0, "No messages were received through the callback"