Clean up LRO code and update docs

Author: LucaButBoring

README.md

@@ -1663,7 +1663,7 @@ For more information on mounting applications in Starlette, see the [Starlette d
 
 ### Persistent Async Operations
 
-For production deployments, you may want async operations to survive server restarts. The `ServerAsyncOperationManager` uses pluggable `AsyncOperationStore` and `AsyncOperationBroker` components to handle operation persistence and task queuing.
+For production deployments, you may want async operations to survive server restarts. The `ServerAsyncOperationManager` uses pluggable `OperationEventQueue`, `AsyncOperationStore`, and `AsyncOperationBroker` components to handle operation persistence and task queuing.
 
 #### Operation Lifecycle
 
@@ -1679,14 +1679,20 @@ Async operations follow this lifecycle:
 from mcp.server.fastmcp import FastMCP
 from mcp.shared.async_operations import ServerAsyncOperationManager
 
+# Create custom event queues
+custom_request_queue = MyAsyncOperationEventQueue()
+custom_response_queue = MyAsyncOperationEventQueue()
+
 # Create custom store and broker implementations
 custom_store = MyAsyncOperationStore()
 custom_broker = MyAsyncOperationBroker()
 
 # Create operation manager with custom components
 operation_manager = ServerAsyncOperationManager(
     store=custom_store,
-    broker=custom_broker
+    broker=custom_broker,
+    operation_request_queue=custom_request_queue,
+    operation_response_queue=custom_response_queue,
 )
 
 # Use with FastMCP

examples/servers/sqlite-async-operations/README.md

@@ -6,6 +6,7 @@ This example demonstrates how to implement custom async operations storage and t
 
 The example showcases the pluggable architecture of the async operations system:
 
+- `SQLiteOperationEventQueue`: Custom event queue that manages operation messages for disconnected clients
 - `SQLiteAsyncOperationStore`: Custom implementation that persists operations to SQLite
 - `SQLiteAsyncOperationBroker`: Custom implementation that persists pending tasks to SQLite
 - `ServerAsyncOperationManager`: Uses both custom store and broker for full persistence
@@ -30,7 +31,7 @@ uv run mcp-sqlite-async-operations --transport streamable-http --port 8000
 ## Testing Persistent Async Operations
 
 1. Start the server
-2. Call one of the async tools (`long_computation` or `fetch_data`)
+2. Call the async tool (`fetch_data`)
 3. **Restart the server while the operation is running**
 4. The operation will automatically resume and complete
 5. Use the operation token to check status and retrieve results

examples/servers/sqlite-async-operations/mcp_sqlite_async_operations/server.py

@@ -216,24 +216,25 @@ async def cleanup_expired(self) -> int:
 class SQLiteOperationEventQueue(OperationEventQueue):
     """SQLite-based implementation of OperationEventQueue for operation-specific event delivery."""
 
-    def __init__(self, db_path: str = "async_operations.db"):
+    def __init__(self, db_path: str = "async_operations.db", table_name: str = "operation_events"):
         self.db_path = db_path
+        self.table_name = table_name
         self._init_db()
 
     def _init_db(self):
         """Initialize the SQLite database for operation event queuing."""
         with sqlite3.connect(self.db_path) as conn:
-            conn.execute("""
-                CREATE TABLE IF NOT EXISTS operation_events (
+            conn.execute(f"""
+                CREATE TABLE IF NOT EXISTS {self.table_name} (
                     id INTEGER PRIMARY KEY AUTOINCREMENT,
                     operation_token TEXT NOT NULL,
                     message TEXT NOT NULL,
                     created_at REAL NOT NULL
                 )
             """)
-            conn.execute("""
+            conn.execute(f"""
                 CREATE INDEX IF NOT EXISTS idx_operation_events_token_created 
-                ON operation_events(operation_token, created_at)
+                ON {self.table_name}(operation_token, created_at)
             """)
             conn.commit()
 
@@ -244,8 +245,8 @@ async def enqueue_event(self, operation_token: str, message: types.JSONRPCMessag
 
         with sqlite3.connect(self.db_path) as conn:
             conn.execute(
-                """
-                INSERT INTO operation_events (operation_token, message, created_at)
+                f"""
+                INSERT INTO {self.table_name} (operation_token, message, created_at)
                 VALUES (?, ?, ?)
             """,
                 (operation_token, message_json, created_at),
@@ -259,8 +260,8 @@ async def dequeue_events(self, operation_token: str) -> list[types.JSONRPCMessag
 
             # Get all events for this operation token
             cursor = conn.execute(
-                """
-                SELECT id, message FROM operation_events 
+                f"""
+                SELECT id, message FROM {self.table_name} 
                 WHERE operation_token = ?
                 ORDER BY created_at
             """,
@@ -279,7 +280,7 @@ async def dequeue_events(self, operation_token: str) -> list[types.JSONRPCMessag
             # Delete the dequeued events
             if event_ids:
                 placeholders = ",".join("?" * len(event_ids))
-                conn.execute(f"DELETE FROM operation_events WHERE id IN ({placeholders})", event_ids)
+                conn.execute(f"DELETE FROM {self.table_name} WHERE id IN ({placeholders})", event_ids)
                 conn.commit()
 
             return events
@@ -419,13 +420,18 @@ class UserPreferences(BaseModel):
 def main(port: int, transport: str, db_path: str):
     """Run the SQLite async operations example server."""
     # Create components with specified database path
-    operation_event_queue = SQLiteOperationEventQueue(db_path)
+    operation_request_queue = SQLiteOperationEventQueue(db_path, "operation_requests")
+    operation_response_queue = SQLiteOperationEventQueue(db_path, "operation_responses")
     broker = SQLiteAsyncOperationBroker(db_path)
     store = SQLiteAsyncOperationStore(db_path)
-    manager = ServerAsyncOperationManager(store=store, broker=broker, operation_request_queue=operation_event_queue)
+    manager = ServerAsyncOperationManager(
+        store=store,
+        broker=broker,
+        operation_request_queue=operation_request_queue,
+        operation_response_queue=operation_response_queue,
+    )
     mcp = FastMCP(
         "SQLite Async Operations Demo",
-        operation_event_queue=operation_event_queue,
         async_operations=manager,
     )
 

src/mcp/server/fastmcp/server.py

@@ -45,7 +45,6 @@
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
-from mcp.shared.async_operations import OperationEventQueue
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import (
     AnyFunction,
@@ -148,7 +147,6 @@ def __init__(  # noqa: PLR0913
         event_store: EventStore | None = None,
         *,
         async_operations: ServerAsyncOperationManager | None = None,
-        operation_event_queue: OperationEventQueue | None = None,
         tools: list[Tool] | None = None,
         debug: bool = False,
         log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
@@ -168,7 +166,7 @@ def __init__(  # noqa: PLR0913
         auth: AuthSettings | None = None,
         transport_security: TransportSecuritySettings | None = None,
     ):
-        from mcp.shared.async_operations import InMemoryOperationEventQueue, ServerAsyncOperationManager
+        from mcp.shared.async_operations import ServerAsyncOperationManager
 
         self.settings = Settings(
             debug=debug,
@@ -190,21 +188,14 @@ def __init__(  # noqa: PLR0913
             transport_security=transport_security,
         )
 
-        self._operation_event_queue = operation_event_queue or InMemoryOperationEventQueue()
-        self._operation_response_queue = InMemoryOperationEventQueue()
-        self._async_operations = async_operations or ServerAsyncOperationManager(
-            operation_request_queue=self._operation_event_queue,
-            operation_response_queue=self._operation_response_queue,
-        )
+        self._async_operations = async_operations or ServerAsyncOperationManager()
 
         self._mcp_server = MCPServer(
             name=name or "FastMCP",
             instructions=instructions,
             website_url=website_url,
             icons=icons,
             async_operations=self._async_operations,
-            operation_request_queue=self._operation_event_queue,
-            operation_response_queue=self._operation_response_queue,
             # TODO(Marcelo): It seems there's a type mismatch between the lifespan type from an FastMCP and Server.
             # We need to create a Lifespan type that is a generic on the server type, like Starlette does.
             lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan),  # type: ignore
@@ -228,7 +219,6 @@ def __init__(  # noqa: PLR0913
         if auth_server_provider and not token_verifier:
             self._token_verifier = ProviderTokenVerifier(auth_server_provider)
         self._event_store = event_store
-        self._operation_event_queue = operation_event_queue
         self._custom_starlette_routes: list[Route] = []
         self.dependencies = self.settings.dependencies
         self._session_manager: StreamableHTTPSessionManager | None = None

src/mcp/server/lowlevel/server.py

@@ -95,7 +95,7 @@ async def main():
 from mcp.types import NEXT_PROTOCOL_VERSION, Operation, RequestId
 
 if TYPE_CHECKING:
-    from mcp.shared.async_operations import OperationEventQueue, ServerAsyncOperationManager
+    from mcp.shared.async_operations import ServerAsyncOperationManager
 
 logger = logging.getLogger(__name__)
 
@@ -145,8 +145,6 @@ def __init__(
         website_url: str | None = None,
         icons: list[types.Icon] | None = None,
         async_operations: ServerAsyncOperationManager | None = None,
-        operation_request_queue: OperationEventQueue | None = None,
-        operation_response_queue: OperationEventQueue | None = None,
         lifespan: Callable[
             [Server[LifespanResultT, RequestT]],
             AbstractAsyncContextManager[LifespanResultT],
@@ -160,10 +158,7 @@ def __init__(
         self.website_url = website_url
         self.icons = icons
         self.lifespan = lifespan
-        self.async_operations = async_operations or ServerAsyncOperationManager(
-            operation_request_queue=operation_request_queue,
-            operation_response_queue=operation_response_queue,
-        )
+        self.async_operations = async_operations or ServerAsyncOperationManager()
         self.async_operations.set_handler(self._execute_tool_async)
         # Track request ID to operation token mapping for cancellation
         self._request_to_operation: dict[RequestId, str] = {}