Add poll_task() method and update examples to use spec-compliant polling

The MCP Tasks spec requires clients to poll tasks/get watching for status
changes, then call tasks/result when status becomes input_required to
receive elicitation/sampling requests.

- Add poll_task() async iterator to ExperimentalClientFeatures that yields
  status on each poll and respects the server's pollInterval hint
- Update simple-task-client to use poll_task() instead of manual loop
- Update simple-task-interactive-client to poll first, then call
  tasks/result on input_required per the spec pattern

Author: maxisbey

examples/clients/simple-task-client/mcp_simple_task_client/main.py

@@ -28,18 +28,14 @@ async def run(url: str) -> None:
             task_id = result.task.taskId
             print(f"Task created: {task_id}")
 
-            # Poll until done
-            while True:
-                status = await session.experimental.get_task(task_id)
+            # Poll until done (respects server's pollInterval hint)
+            async for status in session.experimental.poll_task(task_id):
                 print(f"  Status: {status.status} - {status.statusMessage or ''}")
 
-                if status.status == "completed":
-                    break
-                elif status.status in ("failed", "cancelled"):
-                    print(f"Task ended with status: {status.status}")
-                    return
-
-                await asyncio.sleep(0.5)
+            # Check final status
+            if status.status != "completed":
+                print(f"Task ended with status: {status.status}")
+                return
 
             # Get the result
             task_result = await session.experimental.get_task_result(task_id, CallToolResult)

examples/clients/simple-task-interactive-client/mcp_simple_task_interactive_client/main.py

@@ -1,4 +1,10 @@
-"""Simple interactive task client demonstrating elicitation and sampling responses."""
+"""Simple interactive task client demonstrating elicitation and sampling responses.
+
+This example demonstrates the spec-compliant polling pattern:
+1. Poll tasks/get watching for status changes
+2. On input_required, call tasks/result to receive elicitation/sampling requests
+3. Continue until terminal status, then retrieve final result
+"""
 
 import asyncio
 from typing import Any
@@ -88,8 +94,17 @@ async def run(url: str) -> None:
             task_id = result.task.taskId
             print(f"Task created: {task_id}")
 
-            # get_task_result() delivers elicitation requests and blocks until complete
-            final = await session.experimental.get_task_result(task_id, CallToolResult)
+            # Poll until terminal, calling tasks/result on input_required
+            async for status in session.experimental.poll_task(task_id):
+                print(f"[Poll] Status: {status.status}")
+                if status.status == "input_required":
+                    # Server needs input - tasks/result delivers the elicitation request
+                    final = await session.experimental.get_task_result(task_id, CallToolResult)
+                    break
+            else:
+                # poll_task exited due to terminal status
+                final = await session.experimental.get_task_result(task_id, CallToolResult)
+
             print(f"Result: {get_text(final)}")
 
             # Demo 2: Sampling (write_haiku)
@@ -100,8 +115,15 @@ async def run(url: str) -> None:
             task_id = result.task.taskId
             print(f"Task created: {task_id}")
 
-            # get_task_result() delivers sampling requests and blocks until complete
-            final = await session.experimental.get_task_result(task_id, CallToolResult)
+            # Poll until terminal, calling tasks/result on input_required
+            async for status in session.experimental.poll_task(task_id):
+                print(f"[Poll] Status: {status.status}")
+                if status.status == "input_required":
+                    final = await session.experimental.get_task_result(task_id, CallToolResult)
+                    break
+            else:
+                final = await session.experimental.get_task_result(task_id, CallToolResult)
+
             print(f"Result:\n{get_text(final)}")
 
 

src/mcp/client/experimental/tasks.py

@@ -24,9 +24,13 @@
     await session.experimental.cancel_task(task_id)
 """
 
+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
 
 if TYPE_CHECKING:
     from mcp.client.session import ClientSession
@@ -191,3 +195,40 @@ async def cancel_task(self, task_id: str) -> types.CancelTaskResult:
             ),
             types.CancelTaskResult,
         )
+
+    async def poll_task(self, task_id: str) -> AsyncIterator[types.GetTaskResult]:
+        """
+        Poll a task until it reaches a terminal status.
+
+        Yields GetTaskResult for each poll, allowing the caller to react to
+        status changes (e.g., handle input_required). Exits when task reaches
+        a terminal status (completed, failed, cancelled).
+
+        Respects the pollInterval hint from the server.
+
+        Args:
+            task_id: The task identifier
+
+        Yields:
+            GetTaskResult for each poll
+
+        Example:
+            async for status in session.experimental.poll_task(task_id):
+                print(f"Status: {status.status}")
+                if status.status == "input_required":
+                    # Handle elicitation request via tasks/result
+                    pass
+
+            # 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)
+            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)

tests/experimental/tasks/client/test_poll_task.py

@@ -0,0 +1,121 @@
+"""Tests for poll_task async iterator."""
+
+from collections.abc import Callable, Coroutine
+from datetime import datetime, timezone
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from mcp.client.experimental.tasks import ExperimentalClientFeatures
+from mcp.types import GetTaskResult, TaskStatus
+
+
+def make_task_result(
+    status: TaskStatus = "working",
+    poll_interval: int = 0,
+    task_id: str = "test-task",
+    status_message: str | None = None,
+) -> GetTaskResult:
+    """Create GetTaskResult with sensible defaults."""
+    now = datetime.now(timezone.utc)
+    return GetTaskResult(
+        taskId=task_id,
+        status=status,
+        statusMessage=status_message,
+        createdAt=now,
+        lastUpdatedAt=now,
+        ttl=60000,
+        pollInterval=poll_interval,
+    )
+
+
+def make_status_sequence(
+    *statuses: TaskStatus,
+    task_id: str = "test-task",
+) -> Callable[[str], Coroutine[Any, Any, GetTaskResult]]:
+    """Create mock get_task that returns statuses in sequence."""
+    status_iter = iter(statuses)
+
+    async def mock_get_task(tid: str) -> GetTaskResult:
+        return make_task_result(status=next(status_iter), task_id=tid)
+
+    return mock_get_task
+
+
+@pytest.fixture
+def mock_session() -> AsyncMock:
+    return AsyncMock()
+
+
+@pytest.fixture
+def features(mock_session: AsyncMock) -> ExperimentalClientFeatures:
+    return ExperimentalClientFeatures(mock_session)
+
+
+@pytest.mark.anyio
+async def test_poll_task_yields_until_completed(features: ExperimentalClientFeatures) -> None:
+    """poll_task yields each status until terminal."""
+    features.get_task = make_status_sequence("working", "working", "completed")  # type: ignore[method-assign]
+
+    statuses = [s.status async for s in features.poll_task("test-task")]
+
+    assert statuses == ["working", "working", "completed"]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("terminal_status", ["completed", "failed", "cancelled"])
+async def test_poll_task_exits_on_terminal(features: ExperimentalClientFeatures, terminal_status: TaskStatus) -> None:
+    """poll_task exits immediately when task is already terminal."""
+    features.get_task = make_status_sequence(terminal_status)  # type: ignore[method-assign]
+
+    statuses = [s.status async for s in features.poll_task("test-task")]
+
+    assert statuses == [terminal_status]
+
+
+@pytest.mark.anyio
+async def test_poll_task_continues_through_input_required(features: ExperimentalClientFeatures) -> None:
+    """poll_task yields input_required and continues (non-terminal)."""
+    features.get_task = make_status_sequence("working", "input_required", "working", "completed")  # type: ignore[method-assign]
+
+    statuses = [s.status async for s in features.poll_task("test-task")]
+
+    assert statuses == ["working", "input_required", "working", "completed"]
+
+
+@pytest.mark.anyio
+async def test_poll_task_passes_task_id(features: ExperimentalClientFeatures) -> None:
+    """poll_task passes correct task_id to get_task."""
+    received_ids: list[str] = []
+
+    async def mock_get_task(task_id: str) -> GetTaskResult:
+        received_ids.append(task_id)
+        return make_task_result(status="completed", task_id=task_id)
+
+    features.get_task = mock_get_task  # type: ignore[method-assign]
+
+    _ = [s async for s in features.poll_task("my-task-123")]
+
+    assert received_ids == ["my-task-123"]
+
+
+@pytest.mark.anyio
+async def test_poll_task_yields_full_result(features: ExperimentalClientFeatures) -> None:
+    """poll_task yields complete GetTaskResult objects."""
+
+    async def mock_get_task(task_id: str) -> GetTaskResult:
+        return make_task_result(
+            status="completed",
+            task_id=task_id,
+            status_message="All done!",
+        )
+
+    features.get_task = mock_get_task  # type: ignore[method-assign]
+
+    results = [r async for r in features.poll_task("test-task")]
+
+    assert len(results) == 1
+    assert results[0].status == "completed"
+    assert results[0].statusMessage == "All done!"
+    assert results[0].taskId == "test-task"