Split ServerAsyncOperationManager into AsyncOperationStore and AsyncOperationBroker components

Author: LucaButBoring

README.md

@@ -1661,6 +1661,40 @@ For more information on mounting applications in Starlette, see the [Starlette d
 
 ## Advanced Usage
 
+### 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.
+
+#### Operation Lifecycle
+
+Async operations follow this lifecycle:
+
+1. **Submitted** - Operation token generated and stored
+2. **Working** - Task begins execution  
+3. **Completed/Failed/Cancelled** - Operation reaches terminal state with results
+
+#### Custom Store and Broker
+
+```python
+from mcp.server.fastmcp import FastMCP
+from mcp.shared.async_operations import ServerAsyncOperationManager
+
+# 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
+)
+
+# Use with FastMCP
+mcp = FastMCP("My Server", async_operations=operation_manager)
+```
+
+For a complete SQLite-based implementation example, see [`examples/servers/sqlite-async-operations/`](examples/servers/sqlite-async-operations/).
+
 ### Low-Level Server
 
 For more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server, including lifecycle management through the lifespan API:

examples/clients/async-reconnect-client/mcp_async_reconnect_client/client.py

@@ -10,6 +10,8 @@ async def call_async_tool(session: ClientSession, token: str | None):
 
     if not token:
         result = await session.call_tool("fetch_website", arguments={"url": "https://modelcontextprotocol.io"})
+        if result.isError:
+            raise RuntimeError(f"Error calling tool: {result}")
         assert result.operation
         token = result.operation.token
         print(f"Operation started with token: {token}")

examples/servers/sqlite-async-operations/.gitignore

@@ -0,0 +1 @@
+*.db

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

@@ -0,0 +1,36 @@
+# SQLite Async Operations Example
+
+This example demonstrates how to implement custom async operations storage and task queuing using SQLite with the MCP Python SDK.
+
+## Architecture
+
+The example showcases the pluggable architecture of the async operations system:
+
+- `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
+- `FastMCP`: Configured with the custom async operations manager
+
+## Usage
+
+Install and run the server:
+
+```bash
+# Using stdio transport (default)
+# Run with default SQLite database
+uv run mcp-sqlite-async-operations
+
+# Run with custom database path
+uv run mcp-sqlite-async-operations --db-path /path/to/custom.db
+
+# Using streamable-http transport on custom port
+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`)
+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/__init__.py

@@ -0,0 +1 @@
+

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

@@ -0,0 +1,4 @@
+from .server import main
+
+if __name__ == "__main__":
+    main()