changes

Author: akash329d

examples/fastmcp/unicode_example.py

@@ -4,8 +4,14 @@
 """
 
 from mcp.server.fastmcp import FastMCP
+from mcp.server.message_queue import RedisMessageQueue
 
-mcp = FastMCP()
+# Create a Redis message queue
+redis_queue = RedisMessageQueue(
+    redis_url="redis://localhost:6379/0", prefix="mcp:pubsub:"
+)
+
+mcp = FastMCP(message_queue=redis_queue)
 
 
 @mcp.tool(
@@ -61,4 +67,4 @@ def multilingual_hello() -> str:
 
 
 if __name__ == "__main__":
-    mcp.run()
+    mcp.run(transport="sse")

pyproject.toml

@@ -36,7 +36,8 @@ dependencies = [
 rich = ["rich>=13.9.4"]
 cli = ["typer>=0.12.4", "python-dotenv>=1.0.0"]
 ws = ["websockets>=15.0.1"]
-redis = ["redis>=5.0.0"]
+redis = ["redis>=5.2.1"]
+types-redis = ["types-redis>=4.6.0.20241004"]
 
 [project.scripts]
 mcp = "mcp.cli:app [cli]"

src/mcp/server/fastmcp/server.py

@@ -485,7 +485,6 @@ def sse_app(self) -> Starlette:
         """Return an instance of the SSE server app."""
         # 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

src/mcp/server/message_queue/redis.py

@@ -5,7 +5,7 @@
 from uuid import UUID
 
 import anyio
-from anyio import CapacityLimiter, from_thread
+from anyio import CapacityLimiter
 import mcp.types as types
 from mcp.server.message_queue.base import MessageCallback
 
@@ -74,8 +74,8 @@ async def _listen_for_messages(self) -> None:
         async with self._limiter:
             while True:
                 message: None | dict[str, Any] = await self._pubsub.get_message(  # type: ignore
-                    ignore_subscribe_messages=True
-                ) 
+                    ignore_subscribe_messages=True, timeout=None # type: ignore
+                )
                 if message is None:
                     continue
 
@@ -105,7 +105,7 @@ async def _listen_for_messages(self) -> None:
                             msg = types.JSONRPCMessage.model_validate_json(data)
 
                     if msg and session_id in self._callbacks:
-                        from_thread.run(self._callbacks[session_id], msg)
+                        await self._callbacks[session_id](msg)
                 except Exception as e:
                     logger.error(f"Failed to process message: {e}")