Add server→client task-augmented elicitation and sampling support

This implements the bidirectional task-augmented request pattern where the
server can send task-augmented elicitation/sampling requests to the client,
and the client can defer processing by returning CreateTaskResult.

Key changes:

- Add ExperimentalServerSessionFeatures with get_task(), get_task_result(),
  poll_task(), elicit_as_task(), and create_message_as_task() methods for
  server→client task operations

- Add shared polling utility (poll_until_terminal) used by both client and
  server to avoid code duplication

- Add elicit_as_task() and create_message_as_task() to ServerTaskContext for
  use inside task-augmented tool calls

- Add capability checks for task-augmented elicitation/sampling in
  ServerSession.check_client_capability()

- Add comprehensive tests for all four elicitation scenarios:
  1. Normal tool call + normal elicitation
  2. Normal tool call + task-augmented elicitation
  3. Task-augmented tool call + normal elicitation
  4. Task-augmented tool call + task-augmented elicitation

The implementation correctly handles the complex bidirectional flow where
the server polls the client while the client's tasks/result call is still
blocking, waiting for the tool task to complete.

Author: maxisbey

src/mcp/client/experimental/tasks.py

@@ -27,10 +27,8 @@
 from collections.abc import AsyncIterator
 from typing import TYPE_CHECKING, Any, TypeVar
 
-import anyio
-
 import mcp.types as types
-from mcp.shared.experimental.tasks.helpers import is_terminal
+from mcp.shared.experimental.tasks.polling import poll_until_terminal
 
 if TYPE_CHECKING:
     from mcp.client.session import ClientSession
@@ -222,13 +220,5 @@ async def poll_task(self, task_id: str) -> AsyncIterator[types.GetTaskResult]:
             # Task is now terminal, get the result
             result = await session.experimental.get_task_result(task_id, CallToolResult)
         """
-        while True:
-            status = await self.get_task(task_id)
+        async for status in poll_until_terminal(self.get_task, task_id):
             yield status
-
-            if is_terminal(status.status):
-                break
-
-            # Respect server's pollInterval hint, default to 500ms if not specified
-            interval_ms = status.pollInterval if status.pollInterval is not None else 500
-            await anyio.sleep(interval_ms / 1000)

src/mcp/server/experimental/session_features.py

@@ -0,0 +1,194 @@
+"""
+Experimental server session features for server→client task operations.
+
+This module provides the server-side equivalent of ExperimentalClientFeatures,
+allowing the server to send task-augmented requests to the client and poll for results.
+
+WARNING: These APIs are experimental and may change without notice.
+"""
+
+from collections.abc import AsyncIterator
+from typing import TYPE_CHECKING, Any, TypeVar
+
+import mcp.types as types
+from mcp.shared.experimental.tasks.polling import poll_until_terminal
+
+if TYPE_CHECKING:
+    from mcp.server.session import ServerSession
+
+ResultT = TypeVar("ResultT", bound=types.Result)
+
+
+class ExperimentalServerSessionFeatures:
+    """
+    Experimental server session features for server→client task operations.
+
+    This provides the server-side equivalent of ExperimentalClientFeatures,
+    allowing the server to send task-augmented requests to the client and
+    poll for results.
+
+    WARNING: These APIs are experimental and may change without notice.
+
+    Access via session.experimental:
+        result = await session.experimental.elicit_as_task(...)
+    """
+
+    def __init__(self, session: "ServerSession") -> None:
+        self._session = session
+
+    async def get_task(self, task_id: str) -> types.GetTaskResult:
+        """
+        Send tasks/get to the client to get task status.
+
+        Args:
+            task_id: The task identifier
+
+        Returns:
+            GetTaskResult containing the task status
+        """
+        return await self._session.send_request(
+            types.ServerRequest(types.GetTaskRequest(params=types.GetTaskRequestParams(taskId=task_id))),
+            types.GetTaskResult,
+        )
+
+    async def get_task_result(
+        self,
+        task_id: str,
+        result_type: type[ResultT],
+    ) -> ResultT:
+        """
+        Send tasks/result to the client to retrieve the final result.
+
+        Args:
+            task_id: The task identifier
+            result_type: The expected result type
+
+        Returns:
+            The task result, validated against result_type
+        """
+        return await self._session.send_request(
+            types.ServerRequest(types.GetTaskPayloadRequest(params=types.GetTaskPayloadRequestParams(taskId=task_id))),
+            result_type,
+        )
+
+    async def poll_task(self, task_id: str) -> AsyncIterator[types.GetTaskResult]:
+        """
+        Poll a client task until it reaches terminal status.
+
+        Yields GetTaskResult for each poll, allowing the caller to react to
+        status changes. Exits when task reaches a terminal status.
+
+        Respects the pollInterval hint from the client.
+
+        Args:
+            task_id: The task identifier
+
+        Yields:
+            GetTaskResult for each poll
+        """
+        async for status in poll_until_terminal(self.get_task, task_id):
+            yield status
+
+    async def elicit_as_task(
+        self,
+        message: str,
+        requestedSchema: types.ElicitRequestedSchema,
+        *,
+        ttl: int = 60000,
+    ) -> types.ElicitResult:
+        """
+        Send a task-augmented elicitation to the client and poll until complete.
+
+        The client will create a local task, process the elicitation asynchronously,
+        and return the result when ready. This method handles the full flow:
+        1. Send elicitation with task field
+        2. Receive CreateTaskResult from client
+        3. Poll client's task until terminal
+        4. Retrieve and return the final ElicitResult
+
+        Args:
+            message: The message to present to the user
+            requestedSchema: Schema defining the expected response
+            ttl: Task time-to-live in milliseconds
+
+        Returns:
+            The client's elicitation response
+        """
+        create_result = await self._session.send_request(
+            types.ServerRequest(
+                types.ElicitRequest(
+                    params=types.ElicitRequestFormParams(
+                        message=message,
+                        requestedSchema=requestedSchema,
+                        task=types.TaskMetadata(ttl=ttl),
+                    )
+                )
+            ),
+            types.CreateTaskResult,
+        )
+
+        task_id = create_result.task.taskId
+
+        async for _ in self.poll_task(task_id):
+            pass
+
+        return await self.get_task_result(task_id, types.ElicitResult)
+
+    async def create_message_as_task(
+        self,
+        messages: list[types.SamplingMessage],
+        *,
+        max_tokens: int,
+        ttl: int = 60000,
+        system_prompt: str | None = None,
+        include_context: types.IncludeContext | None = None,
+        temperature: float | None = None,
+        stop_sequences: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+        model_preferences: types.ModelPreferences | None = None,
+    ) -> types.CreateMessageResult:
+        """
+        Send a task-augmented sampling request and poll until complete.
+
+        The client will create a local task, process the sampling request
+        asynchronously, and return the result when ready.
+
+        Args:
+            messages: The conversation messages for sampling
+            max_tokens: Maximum tokens in the response
+            ttl: Task time-to-live in milliseconds
+            system_prompt: Optional system prompt
+            include_context: Context inclusion strategy
+            temperature: Sampling temperature
+            stop_sequences: Stop sequences
+            metadata: Additional metadata
+            model_preferences: Model selection preferences
+
+        Returns:
+            The sampling result from the client
+        """
+        create_result = await self._session.send_request(
+            types.ServerRequest(
+                types.CreateMessageRequest(
+                    params=types.CreateMessageRequestParams(
+                        messages=messages,
+                        maxTokens=max_tokens,
+                        systemPrompt=system_prompt,
+                        includeContext=include_context,
+                        temperature=temperature,
+                        stopSequences=stop_sequences,
+                        metadata=metadata,
+                        modelPreferences=model_preferences,
+                        task=types.TaskMetadata(ttl=ttl),
+                    )
+                )
+            ),
+            types.CreateTaskResult,
+        )
+
+        task_id = create_result.task.taskId
+
+        async for _ in self.poll_task(task_id):
+            pass
+
+        return await self.get_task_result(task_id, types.CreateMessageResult)