format / lint

Author: akash329d

src/mcp/server/fastmcp/server.py

@@ -33,6 +33,7 @@
 from mcp.server.lowlevel.server import LifespanResultT
 from mcp.server.lowlevel.server import Server as MCPServer
 from mcp.server.lowlevel.server import lifespan as default_lifespan
+from mcp.server.message_queue import MessageQueue
 from mcp.server.session import ServerSession, ServerSessionT
 from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
@@ -49,7 +50,6 @@
 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__)
 
@@ -78,7 +78,9 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     message_path: str = "/messages/"
 
     # SSE message queue settings
-    message_queue: MessageQueue | None = Field(None, description="Custom message queue instance")
+    message_queue: MessageQueue | None = Field(
+        None, description="Custom message queue instance"
+    )
 
     # resource settings
     warn_on_duplicate_resources: bool = True
@@ -488,6 +490,7 @@ def sse_app(self) -> Starlette:
         # 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 default in-memory message queue")
 

src/mcp/server/message_queue/base.py

@@ -1,7 +1,8 @@
 import logging
-from typing import Protocol, runtime_checkable, Callable, Awaitable
-from uuid import UUID
+from collections.abc import Awaitable, Callable
 from contextlib import asynccontextmanager
+from typing import Protocol, runtime_checkable
+from uuid import UUID
 
 import mcp.types as types
 
@@ -35,7 +36,7 @@ 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.
-        
+
         Args:
             session_id: The UUID of the session to activate
             callback: Async callback function to handle messages for this session
@@ -57,7 +58,8 @@ async def session_exists(self, session_id: UUID) -> bool:
 class InMemoryMessageQueue:
     """Default in-memory implementation of the MessageQueue interface.
 
-    This implementation immediately calls registered callbacks when messages are received.
+    This implementation immediately calls registered callbacks when messages
+    are received.
     """
 
     def __init__(self) -> None:
@@ -78,7 +80,7 @@ async def publish_message(
             logger.debug(f"Called callback for session {session_id}")
         else:
             logger.warning(f"No callback registered for session {session_id}")
-            
+
         return True
 
     @asynccontextmanager
@@ -87,7 +89,7 @@ async def active_for_request(self, session_id: UUID, callback: MessageCallback):
         self._active_sessions.add(session_id)
         self._callbacks[session_id] = callback
         logger.debug(f"Registered session {session_id} with callback")
-        
+
         try:
             yield
         finally:

src/mcp/server/message_queue/redis.py

@@ -6,10 +6,10 @@
 
 import anyio
 from anyio import CapacityLimiter
+
 import mcp.types as types
 from mcp.server.message_queue.base import MessageCallback
 
-
 try:
     import redis.asyncio as redis
 except ImportError:
@@ -56,7 +56,7 @@ async def active_for_request(self, session_id: UUID, callback: MessageCallback):
         self._callbacks[session_id] = callback
         channel = self._session_channel(session_id)
         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:
             tg.start_soon(self._listen_for_messages)
@@ -74,11 +74,12 @@ 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, timeout=None # type: ignore
+                    ignore_subscribe_messages=True,
+                    timeout=None,  # type: ignore
                 )
                 if message is None:
                     continue
-                    
+
                 # Extract session ID from channel name
                 channel: str = cast(str, message["channel"])
                 if not channel.startswith(self._prefix):

src/mcp/server/sse.py

@@ -101,12 +101,12 @@ async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
 
         session_id = uuid4()
         session_uri = f"{quote(self._endpoint)}?session_id={session_id.hex}"
-        
+
         async def message_callback(message: types.JSONRPCMessage | Exception) -> None:
             """Callback that receives messages from the message queue"""
             logger.debug(f"Got message from queue for session {session_id}")
             await read_stream_writer.send(message)
-        
+
         logger.debug(f"Created new session with ID: {session_id}")
 
         sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[
@@ -137,7 +137,9 @@ async def sse_writer():
             logger.debug("Starting SSE response task")
             tg.start_soon(response, scope, receive, send)
 
-            async with self._message_queue.active_for_request(session_id, message_callback):
+            async with self._message_queue.active_for_request(
+                session_id, message_callback
+            ):
                 try:
                     logger.debug("Yielding read and write streams")
                     yield (read_stream, write_stream)