naming changes

Author: akash329d

README.md

@@ -385,22 +385,22 @@ app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))
 
 For more information on mounting applications in Starlette, see the [Starlette documentation](https://www.starlette.io/routing/#submounting-routes).
 
-#### Message Queue Options
+#### Message Dispatch Options
 
-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:
+By default, the SSE server uses an in-memory message dispatch system for incoming POST messages. For production deployments or distributed scenarios, you can use Redis or implement your own message dispatch system that conforms to the `MessageDispatch` protocol:
 
 ```python
-# Using the built-in Redis message queue
+# Using the built-in Redis message dispatch
 from mcp.server.fastmcp import FastMCP
-from mcp.server.message_queue import RedisMessageQueue
+from mcp.server.message_queue import RedisMessageDispatch
 
-# Create a Redis message queue
-redis_queue = RedisMessageQueue(
+# Create a Redis message dispatch
+redis_dispatch = RedisMessageDispatch(
     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)
+# Pass the message dispatch instance to the server
+mcp = FastMCP("My App", message_queue=redis_dispatch)
 ```
 
 To use Redis, add the Redis dependency:

src/mcp/server/message_queue/base.py

@@ -12,8 +12,8 @@
 
 
 @runtime_checkable
-class MessageQueue(Protocol):
-    """Abstract interface for SSE messaging.
+class MessageDispatch(Protocol):
+    """Abstract interface for SSE message dispatching.
 
     This interface allows messages to be published to sessions and callbacks to be
     registered for message handling, enabling multiple servers to handle requests.
@@ -34,11 +34,11 @@ async def publish_message(
         ...
 
     @asynccontextmanager
-    async def active_for_request(self, session_id: UUID, callback: MessageCallback):
-        """Request-scoped context manager that ensures the listener is active.
+    async def subscribe(self, session_id: UUID, callback: MessageCallback):
+        """Request-scoped context manager that subscribes to messages for a session.
 
         Args:
-            session_id: The UUID of the session to activate
+            session_id: The UUID of the session to subscribe to
             callback: Async callback function to handle messages for this session
         """
         yield
@@ -55,49 +55,44 @@ async def session_exists(self, session_id: UUID) -> bool:
         ...
 
 
-class InMemoryMessageQueue:
-    """Default in-memory implementation of the MessageQueue interface.
+class InMemoryMessageDispatch:
+    """Default in-memory implementation of the MessageDispatch interface.
 
-    This implementation immediately calls registered callbacks when messages
-    are received.
+    This implementation immediately dispatches messages to registered callbacks when 
+    messages are received without any queuing behavior.
     """
 
     def __init__(self) -> None:
         self._callbacks: dict[UUID, MessageCallback] = {}
-        self._active_sessions: set[UUID] = set()
+        # We don't need a separate _active_sessions set since _callbacks already tracks this
 
     async def publish_message(
         self, session_id: UUID, message: types.JSONRPCMessage | Exception
     ) -> bool:
         """Publish a message for the specified session."""
-        if not await self.session_exists(session_id):
+        if session_id not in self._callbacks:
             logger.warning(f"Message received for unknown session {session_id}")
             return False
 
-        # Call the callback directly if registered
-        if session_id in self._callbacks:
-            await self._callbacks[session_id](message)
-            logger.debug(f"Called callback for session {session_id}")
-        else:
-            logger.warning(f"No callback registered for session {session_id}")
+        # Call the callback directly
+        await self._callbacks[session_id](message)
+        logger.debug(f"Called callback for session {session_id}")
 
         return True
 
     @asynccontextmanager
-    async def active_for_request(self, session_id: UUID, callback: MessageCallback):
-        """Request-scoped context manager that ensures the listener is active."""
-        self._active_sessions.add(session_id)
+    async def subscribe(self, session_id: UUID, callback: MessageCallback):
+        """Request-scoped context manager that subscribes to messages for a session."""
         self._callbacks[session_id] = callback
         logger.debug(f"Registered session {session_id} with callback")
 
         try:
             yield
         finally:
-            self._active_sessions.discard(session_id)
             if session_id in self._callbacks:
                 del self._callbacks[session_id]
             logger.debug(f"Unregistered session {session_id}")
 
     async def session_exists(self, session_id: UUID) -> bool:
         """Check if a session exists."""
-        return session_id in self._active_sessions
+        return session_id in self._callbacks

src/mcp/server/message_queue/redis.py

@@ -21,8 +21,8 @@
 logger = logging.getLogger(__name__)
 
 
-class RedisMessageQueue:
-    """Redis implementation of the MessageQueue interface using pubsub.
+class RedisMessageDispatch:
+    """Redis implementation of the MessageDispatch interface using pubsub.
 
     This implementation uses Redis pubsub for real-time message distribution across
     multiple servers handling the same sessions.
@@ -31,7 +31,7 @@ class RedisMessageQueue:
     def __init__(
         self, redis_url: str = "redis://localhost:6379/0", prefix: str = "mcp:pubsub:"
     ) -> None:
-        """Initialize Redis message queue.
+        """Initialize Redis message dispatch.
 
         Args:
             redis_url: Redis connection string
@@ -44,15 +44,15 @@ def __init__(
         self._callbacks: dict[UUID, MessageCallback] = {}
         # Ensures only one polling task runs at a time for message handling
         self._limiter = CapacityLimiter(1)
-        logger.debug(f"Initialized Redis message queue with URL: {redis_url}")
+        logger.debug(f"Initialized Redis message dispatch with URL: {redis_url}")
 
     def _session_channel(self, session_id: UUID) -> str:
         """Get the Redis channel for a session."""
         return f"{self._prefix}session:{session_id.hex}"
 
     @asynccontextmanager
-    async def active_for_request(self, session_id: UUID, callback: MessageCallback):
-        """Request-scoped context manager that ensures the listener task is running."""
+    async def subscribe(self, session_id: UUID, callback: MessageCallback):
+        """Request-scoped context manager that subscribes to messages for a session."""
         await self._redis.sadd(self._active_sessions_key, session_id.hex)
         self._callbacks[session_id] = callback
         channel = self._session_channel(session_id)
@@ -98,17 +98,23 @@ async def _listen_for_messages(self) -> None:
                 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']}"
-                            )
+                    if not isinstance(json_data, dict):
+                        logger.error(f"Received non-dict JSON data: {type(json_data)}")
+                        continue
+                        
+                    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:
+                        if session_id in self._callbacks:
+                            await self._callbacks[session_id](msg)
                         else:
-                            msg = types.JSONRPCMessage.model_validate_json(data)
-
-                    if msg and session_id in self._callbacks:
-                        await self._callbacks[session_id](msg)
+                            logger.warning(f"No callback registered for session {session_id}")
                 except Exception as e:
                     logger.error(f"Failed to process message: {e}")
 

src/mcp/server/sse.py

@@ -80,7 +80,7 @@ def __init__(
 
         super().__init__()
         self._endpoint = endpoint
-        self._message_queue = message_queue or InMemoryMessageQueue()
+        self._message_dispatch = message_queue or InMemoryMessageDispatch()
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
     @asynccontextmanager