fixes

Author: akash329d

README.md

@@ -397,22 +397,23 @@ For more information on mounting applications in Starlette, see the [Starlette d
 By default, the SSE server uses an in-memory message queue for incoming POST messages. For production deployments or distributed scenarios, you can use Redis:
 
 ```python
+# Using the built-in Redis message queue
 from mcp.server.fastmcp import FastMCP
+from mcp.server.message_queue import RedisMessageQueue
 
-mcp = FastMCP(
-    "My App",
-    settings={
-        "message_queue": "redis",
-        "redis_url": "redis://localhost:6379/0",
-        "redis_prefix": "mcp:queue:",
-    },
+# Create a Redis message queue
+redis_queue = RedisMessageQueue(
+    redis_url="redis://localhost:6379/0", prefix="mcp:pubsub:"
 )
+
+# Pass the message queue instance to the server
+mcp = FastMCP("My App", message_queue=redis_queue)
 ```
 
 To use Redis, add the Redis dependency:
 
 ```bash
-uv add "mcp[redis]"
+uv add redis types-redis
 ```
 
 ## Examples

pyproject.toml

@@ -30,8 +30,6 @@ dependencies = [
     "sse-starlette>=1.6.1",
     "pydantic-settings>=2.5.2",
     "uvicorn>=0.23.1",
-    "redis==5.2.1",
-    "types-redis==4.6.0.20241004",
 ]
 
 [project.optional-dependencies]

src/mcp/server/fastmcp/server.py

@@ -49,6 +49,7 @@
 from mcp.types import Resource as MCPResource
 from mcp.types import ResourceTemplate as MCPResourceTemplate
 from mcp.types import Tool as MCPTool
+from mcp.server.message_queue import MessageQueue
 
 logger = get_logger(__name__)
 
@@ -77,9 +78,7 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     message_path: str = "/messages/"
 
     # SSE message queue settings
-    message_queue: Literal["memory", "redis"] = "memory"
-    redis_url: str = "redis://localhost:6379/0"
-    redis_prefix: str = "mcp:queue:"
+    message_queue: MessageQueue | None = Field(None, description="Custom message queue instance")
 
     # resource settings
     warn_on_duplicate_resources: bool = True
@@ -484,25 +483,14 @@ async def run_sse_async(self) -> None:
 
     def sse_app(self) -> Starlette:
         """Return an instance of the SSE server app."""
-        message_queue = None
-        if self.settings.message_queue == "redis":
-            try:
-                from mcp.server.message_queue import RedisMessageQueue
-
-                message_queue = RedisMessageQueue(
-                    redis_url=self.settings.redis_url, prefix=self.settings.redis_prefix
-                )
-                logger.info(f"Using Redis message queue at {self.settings.redis_url}")
-            except ImportError:
-                logger.error(
-                    "Redis message queue requested but 'redis' package not installed. "
-                )
-                raise
-        else:
+        # Use a custom provided message queue if available
+        message_queue = self.settings.message_queue
+        
+        # If no message queue is provided, create an in-memory queue as default
+        if message_queue is None:
             from mcp.server.message_queue import InMemoryMessageQueue
-
             message_queue = InMemoryMessageQueue()
-            logger.info("Using in-memory message queue")
+            logger.info("Using default in-memory message queue")
 
         sse = SseServerTransport(
             self.settings.message_path, message_queue=message_queue

src/mcp/server/message_queue/redis.py

@@ -1,14 +1,13 @@
 import json
 import logging
+from contextlib import asynccontextmanager
+from typing import Any, cast
 from uuid import UUID
 
 import anyio
-from anyio import CapacityLimiter
+from anyio import CapacityLimiter, from_thread
 import mcp.types as types
 from mcp.server.message_queue.base import MessageCallback
-from typing import Any, cast
-from anyio import from_thread
-from contextlib import asynccontextmanager
 
 
 try:
@@ -38,8 +37,8 @@ def __init__(
             redis_url: Redis connection string
             prefix: Key prefix for Redis channels to avoid collisions
         """
-        self._redis = redis.from_url(redis_url, decode_responses=True) # type: ignore
-        self._pubsub = self._redis.pubsub(ignore_subscribe_messages=True) # type: ignore
+        self._redis = redis.from_url(redis_url, decode_responses=True)  # type: ignore
+        self._pubsub = self._redis.pubsub(ignore_subscribe_messages=True)  # type: ignore
         self._prefix = prefix
         self._active_sessions_key = f"{prefix}active_sessions"
         self._callbacks: dict[UUID, MessageCallback] = {}
@@ -53,11 +52,10 @@ def _session_channel(self, session_id: UUID) -> str:
     @asynccontextmanager
     async def active_for_request(self, session_id: UUID, callback: MessageCallback):
         """Request-scoped context manager that ensures the listener task is running."""
-
         await self._redis.sadd(self._active_sessions_key, session_id.hex)
         self._callbacks[session_id] = callback
         channel = self._session_channel(session_id)
-        await self._pubsub.subscribe(channel) # type: ignore
+        await self._pubsub.subscribe(channel)  # type: ignore
 
         logger.debug(f"Registered session {session_id} in Redis with callback")
         async with anyio.create_task_group() as tg:
@@ -66,7 +64,7 @@ async def active_for_request(self, session_id: UUID, callback: MessageCallback):
                 yield
             finally:
                 tg.cancel_scope.cancel()
-                await self._pubsub.unsubscribe(channel) # type: ignore
+                await self._pubsub.unsubscribe(channel)  # type: ignore
                 await self._redis.srem(self._active_sessions_key, session_id.hex)
                 del self._callbacks[session_id]
                 logger.debug(f"Unregistered session {session_id} from Redis")
@@ -75,40 +73,41 @@ async def _listen_for_messages(self) -> None:
         """Background task that listens for messages on subscribed channels."""
         async with self._limiter:
             while True:
-                message: None | dict[str, Any] = await self._pubsub.get_message( # type: ignore
+                message: None | dict[str, Any] = await self._pubsub.get_message(  # type: ignore
                     ignore_subscribe_messages=True
                 ) 
-                if message is not None:
-                    # Extract session ID from channel name
-                    channel: str = cast(str, message["channel"])
-                    if not channel.startswith(self._prefix):
-                        continue
-
-                    session_hex = channel.split(":")[-1]
-                    try:
-                        session_id = UUID(hex=session_hex)
-                    except ValueError:
-                        logger.error(f"Invalid session channel: {channel}")
-                        continue
-
-                    # Deserialize the message
-                    data: str = cast(str, message["data"])
-                    msg: None | types.JSONRPCMessage | Exception = None
-                    try:
-                        json_data = json.loads(data)
-                        if isinstance(json_data, dict):
-                            json_dict: dict[str, Any] = json_data
-                            if json_dict.get("_exception", False):
-                                msg = Exception(
-                                    f"{json_dict['type']}: {json_dict['message']}"
-                                )
-                            else:
-                                msg = types.JSONRPCMessage.model_validate_json(data)
-
-                        if msg and session_id in self._callbacks:
-                            from_thread.run(self._callbacks[session_id], msg)
-                    except Exception as e:
-                        logger.error(f"Failed to process message: {e}")
+                if message is None:
+                    continue
+                    
+                # Extract session ID from channel name
+                channel: str = cast(str, message["channel"])
+                if not channel.startswith(self._prefix):
+                    continue
+
+                session_hex = channel.split(":")[-1]
+                try:
+                    session_id = UUID(hex=session_hex)
+                except ValueError:
+                    logger.error(f"Invalid session channel: {channel}")
+                    continue
+
+                data: str = cast(str, message["data"])
+                msg: None | types.JSONRPCMessage | Exception = None
+                try:
+                    json_data = json.loads(data)
+                    if isinstance(json_data, dict):
+                        json_dict: dict[str, Any] = json_data
+                        if json_dict.get("_exception", False):
+                            msg = Exception(
+                                f"{json_dict['type']}: {json_dict['message']}"
+                            )
+                        else:
+                            msg = types.JSONRPCMessage.model_validate_json(data)
+
+                    if msg and session_id in self._callbacks:
+                        from_thread.run(self._callbacks[session_id], msg)
+                except Exception as e:
+                    logger.error(f"Failed to process message: {e}")
 
     async def publish_message(
         self, session_id: UUID, message: types.JSONRPCMessage | Exception
@@ -136,8 +135,6 @@ async def publish_message(
 
     async def session_exists(self, session_id: UUID) -> bool:
         """Check if a session exists."""
-        # Explicitly annotate the result as bool to help the type checker
-        result = bool(
+        return bool(
             await self._redis.sismember(self._active_sessions_key, session_id.hex)  # type: ignore[attr-defined]
         )
-        return result