Address review feedback

- Remove set_task_result_handler wrapper method and its test
- Remove internal comment block for request builders
- Mark add_response_router as experimental in docstring
- Add experimental tasks documentation link to README
- Add comment explaining why experimental field uses Any type
  (circular import: mcp.server.__init__ -> fastmcp -> context)

Author: maxisbey

README.md

@@ -2512,6 +2512,7 @@ MCP servers declare capabilities during initialization:
 ## Documentation
 
 - [API Reference](https://modelcontextprotocol.github.io/python-sdk/api/)
+- [Experimental Features (Tasks)](https://modelcontextprotocol.github.io/python-sdk/experimental/tasks/)
 - [Model Context Protocol documentation](https://modelcontextprotocol.io)
 - [Model Context Protocol specification](https://modelcontextprotocol.io/specification/latest)
 - [Officially supported servers](https://github.com/modelcontextprotocol/servers)

src/mcp/server/session.py

@@ -52,7 +52,6 @@ async def handle_list_prompts(ctx: RequestContext) -> list[types.Prompt]:
 from mcp.shared.experimental.tasks.capabilities import check_tasks_capability
 from mcp.shared.experimental.tasks.helpers import RELATED_TASK_METADATA_KEY
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
-from mcp.shared.response_router import ResponseRouter
 from mcp.shared.session import (
     BaseSession,
     RequestResponder,
@@ -157,28 +156,6 @@ def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
 
         return True
 
-    def set_task_result_handler(self, handler: ResponseRouter) -> None:
-        """
-        Set a response router for task-augmented requests.
-
-        This enables response routing for task-augmented requests. When a
-        ServerTaskContext enqueues an elicitation request, the response will be
-        routed back through this handler.
-
-        The handler is automatically registered as a response router.
-
-        Args:
-            handler: The ResponseRouter (typically TaskResultHandler) to use
-
-        Example:
-            from mcp.server.experimental.task_result_handler import TaskResultHandler
-            task_store = InMemoryTaskStore()
-            message_queue = InMemoryTaskMessageQueue()
-            handler = TaskResultHandler(task_store, message_queue)
-            session.set_task_result_handler(handler)
-        """
-        self.add_response_router(handler)
-
     async def _receive_loop(self) -> None:
         async with self._incoming_message_stream_writer:
             await super()._receive_loop()
@@ -483,14 +460,6 @@ async def send_elicit_complete(
             related_request_id,
         )
 
-    # =========================================================================
-    # Request builders for task queueing (internal use)
-    # =========================================================================
-    #
-    # These methods build JSON-RPC requests without sending them. They are used
-    # by TaskContext to construct requests that will be queued instead of sent
-    # directly, avoiding code duplication between ServerSession and TaskContext.
-
     def _build_elicit_form_request(
         self,
         message: str,

src/mcp/shared/context.py

@@ -21,5 +21,9 @@ class RequestContext(Generic[SessionT, LifespanContextT, RequestT]):
     meta: RequestParams.Meta | None
     session: SessionT
     lifespan_context: LifespanContextT
-    experimental: Any = field(default=None)  # Set to Experimental instance by Server
+    # NOTE: This is typed as Any to avoid circular imports. The actual type is
+    # mcp.server.experimental.request_context.Experimental, but importing it here
+    # triggers mcp.server.__init__ -> fastmcp -> tools -> back to this module.
+    # The Server sets this to an Experimental instance at runtime.
+    experimental: Any = field(default=None)
     request: RequestT | None = None

src/mcp/shared/session.py

@@ -211,6 +211,8 @@ def add_response_router(self, router: ResponseRouter) -> None:
         response stream mechanism. This is used by TaskResultHandler to route
         responses for queued task requests back to their resolvers.
 
+        WARNING: This is an experimental API that may change without notice.
+
         Args:
             router: A ResponseRouter implementation
         """

tests/experimental/tasks/client/test_tasks.py

@@ -64,7 +64,6 @@ async def list_tools():
     async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent] | CreateTaskResult:
         ctx = server.request_context
         app = ctx.lifespan_context
-
         if ctx.experimental.is_task:
             task_metadata = ctx.experimental.task_metadata
             assert task_metadata is not None
@@ -174,7 +173,6 @@ async def list_tools():
     async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent] | CreateTaskResult:
         ctx = server.request_context
         app = ctx.lifespan_context
-
         if ctx.experimental.is_task:
             task_metadata = ctx.experimental.task_metadata
             assert task_metadata is not None
@@ -283,7 +281,6 @@ async def list_tools():
     async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent] | CreateTaskResult:
         ctx = server.request_context
         app = ctx.lifespan_context
-
         if ctx.experimental.is_task:
             task_metadata = ctx.experimental.task_metadata
             assert task_metadata is not None
@@ -381,7 +378,6 @@ async def list_tools():
     async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent] | CreateTaskResult:
         ctx = server.request_context
         app = ctx.lifespan_context
-
         if ctx.experimental.is_task:
             task_metadata = ctx.experimental.task_metadata
             assert task_metadata is not None

tests/experimental/tasks/server/test_integration.py

@@ -93,7 +93,6 @@ async def list_tools():
     async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent] | CreateTaskResult:
         ctx = server.request_context
         app = ctx.lifespan_context
-
         if name == "process_data" and ctx.experimental.is_task:
             # 1. Create task in store
             task_metadata = ctx.experimental.task_metadata
@@ -254,7 +253,6 @@ async def list_tools():
     async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent] | CreateTaskResult:
         ctx = server.request_context
         app = ctx.lifespan_context
-
         if name == "failing_task" and ctx.experimental.is_task:
             task_metadata = ctx.experimental.task_metadata
             assert task_metadata is not None

tests/experimental/tasks/server/test_server.py

@@ -8,13 +8,10 @@
 
 from mcp.client.session import ClientSession
 from mcp.server import Server
-from mcp.server.experimental.task_result_handler import TaskResultHandler
 from mcp.server.lowlevel import NotificationOptions
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
 from mcp.shared.exceptions import McpError
-from mcp.shared.experimental.tasks.in_memory_task_store import InMemoryTaskStore
-from mcp.shared.experimental.tasks.message_queue import InMemoryTaskMessageQueue
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.response_router import ResponseRouter
 from mcp.shared.session import RequestResponder
@@ -559,38 +556,6 @@ async def run_server() -> None:
             tg.cancel_scope.cancel()
 
 
-@pytest.mark.anyio
-async def test_set_task_result_handler() -> None:
-    """Test that set_task_result_handler adds the handler as a response router."""
-    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[SessionMessage](10)
-    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[SessionMessage](10)
-
-    store = InMemoryTaskStore()
-    queue = InMemoryTaskMessageQueue()
-    handler = TaskResultHandler(store, queue)
-
-    try:
-        async with ServerSession(
-            client_to_server_receive,
-            server_to_client_send,
-            InitializationOptions(
-                server_name="test-server",
-                server_version="1.0.0",
-                capabilities=ServerCapabilities(),
-            ),
-        ) as server_session:
-            # Use set_task_result_handler (the method we're testing)
-            server_session.set_task_result_handler(handler)
-
-            # Verify handler was added as a response router
-            assert handler in server_session._response_routers
-    finally:  # pragma: no cover
-        await server_to_client_send.aclose()
-        await server_to_client_receive.aclose()
-        await client_to_server_send.aclose()
-        await client_to_server_receive.aclose()
-
-
 @pytest.mark.anyio
 async def test_build_elicit_form_request() -> None:
     """Test that _build_elicit_form_request builds a proper elicitation request."""