Add model-immediate-response support to run_task helper

Add MODEL_IMMEDIATE_RESPONSE_KEY constant and model_immediate_response
parameter to the run_task() helper, allowing servers to provide an
immediate response string in CreateTaskResult._meta while tasks execute
in the background.

Per MCP spec, this is an optional field that hosts can use to pass
interim feedback to models rather than blocking on task completion.

Author: maxisbey

src/mcp/shared/experimental/tasks/__init__.py

@@ -20,6 +20,7 @@
 
 from mcp.shared.experimental.tasks.context import TaskContext
 from mcp.shared.experimental.tasks.helpers import (
+    MODEL_IMMEDIATE_RESPONSE_KEY,
     cancel_task,
     create_task_state,
     generate_task_id,
@@ -49,6 +50,7 @@
     "InMemoryTaskMessageQueue",
     "QueuedMessage",
     "RELATED_TASK_METADATA_KEY",
+    "MODEL_IMMEDIATE_RESPONSE_KEY",
     "run_task",
     "task_execution",
     "is_terminal",

src/mcp/shared/experimental/tasks/helpers.py

@@ -5,7 +5,7 @@
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 from uuid import uuid4
 
 from anyio.abc import TaskGroup
@@ -27,6 +27,11 @@
 if TYPE_CHECKING:
     from mcp.server.session import ServerSession
 
+# Metadata key for model-immediate-response (per MCP spec)
+# Servers MAY include this in CreateTaskResult._meta to provide an immediate
+# response string while the task executes in the background.
+MODEL_IMMEDIATE_RESPONSE_KEY = "io.modelcontextprotocol/model-immediate-response"
+
 
 def is_terminal(status: TaskStatus) -> bool:
     """
@@ -194,6 +199,7 @@ async def run_task(
     *,
     session: "ServerSession | None" = None,
     task_id: str | None = None,
+    model_immediate_response: str | None = None,
 ) -> tuple[CreateTaskResult, TaskContext]:
     """
     Create a task and spawn work to execute it.
@@ -209,6 +215,10 @@ async def run_task(
         work: Async function that does the actual work
         session: Optional session for sending notifications
         task_id: Optional task ID (generated if not provided)
+        model_immediate_response: Optional string to include in _meta as
+            io.modelcontextprotocol/model-immediate-response. This allows
+            hosts to pass an immediate response to the model while the
+            task executes in the background.
 
     Returns:
         Tuple of (CreateTaskResult to return to client, TaskContext for cancellation)
@@ -225,6 +235,7 @@ async def handle_tool(name: str, args: dict):
                         ctx.experimental.task_metadata,
                         lambda ctx: do_long_work(ctx, args),
                         session=ctx.session,
+                        model_immediate_response="Processing started, this may take a while.",
                     )
                     # Optionally store task_ctx for cancellation handling
                     return result
@@ -248,4 +259,9 @@ async def execute() -> None:
     # Spawn the work in the task group
     task_group.start_soon(execute)
 
-    return CreateTaskResult(task=task), ctx
+    # Build _meta if model_immediate_response is provided
+    meta: dict[str, Any] | None = None
+    if model_immediate_response is not None:
+        meta = {MODEL_IMMEDIATE_RESPONSE_KEY: model_immediate_response}
+
+    return CreateTaskResult(task=task, **{"_meta": meta} if meta else {}), ctx

tests/experimental/tasks/server/test_context.py

@@ -6,6 +6,7 @@
 import pytest
 
 from mcp.shared.experimental.tasks import (
+    MODEL_IMMEDIATE_RESPONSE_KEY,
     InMemoryTaskStore,
     TaskContext,
     create_task_state,
@@ -477,3 +478,57 @@ async def work_that_completes_after_cancel(ctx: TaskContext) -> CallToolResult:
         assert task.status == "cancelled"
 
     store.cleanup()
+
+
+@pytest.mark.anyio
+async def test_run_task_with_model_immediate_response() -> None:
+    """Test run_task includes model_immediate_response in _meta when provided."""
+    store = InMemoryTaskStore()
+
+    async def work(ctx: TaskContext) -> CallToolResult:
+        return CallToolResult(content=[TextContent(type="text", text="Done")])
+
+    immediate_msg = "Processing your request, please wait..."
+
+    async with anyio.create_task_group() as tg:
+        result, _ = await run_task(
+            tg,
+            store,
+            TaskMetadata(ttl=60000),
+            work,
+            model_immediate_response=immediate_msg,
+        )
+
+        # Result should have _meta with model-immediate-response
+        assert result.meta is not None
+        assert MODEL_IMMEDIATE_RESPONSE_KEY in result.meta
+        assert result.meta[MODEL_IMMEDIATE_RESPONSE_KEY] == immediate_msg
+
+        # Verify serialization uses _meta alias
+        serialized = result.model_dump(by_alias=True)
+        assert "_meta" in serialized
+        assert serialized["_meta"][MODEL_IMMEDIATE_RESPONSE_KEY] == immediate_msg
+
+    store.cleanup()
+
+
+@pytest.mark.anyio
+async def test_run_task_without_model_immediate_response() -> None:
+    """Test run_task has no _meta when model_immediate_response is not provided."""
+    store = InMemoryTaskStore()
+
+    async def work(ctx: TaskContext) -> CallToolResult:
+        return CallToolResult(content=[TextContent(type="text", text="Done")])
+
+    async with anyio.create_task_group() as tg:
+        result, _ = await run_task(
+            tg,
+            store,
+            TaskMetadata(ttl=60000),
+            work,
+        )
+
+        # Result should not have _meta
+        assert result.meta is None
+
+    store.cleanup()

tests/experimental/tasks/test_spec_compliance.py

@@ -383,7 +383,37 @@ def test_model_immediate_response_in_meta(self) -> None:
         Receiver MAY include io.modelcontextprotocol/model-immediate-response
         in _meta to provide immediate response while task executes.
         """
-        pytest.skip("TODO")
+        from mcp.shared.experimental.tasks import MODEL_IMMEDIATE_RESPONSE_KEY
+        from mcp.types import CreateTaskResult, Task
+
+        # Verify the constant has the correct value per spec
+        assert MODEL_IMMEDIATE_RESPONSE_KEY == "io.modelcontextprotocol/model-immediate-response"
+
+        # CreateTaskResult can include model-immediate-response in _meta
+        task = Task(
+            taskId="test-123",
+            status="working",
+            createdAt=TEST_DATETIME,
+            lastUpdatedAt=TEST_DATETIME,
+            ttl=60000,
+        )
+        immediate_msg = "Task started, processing your request..."
+        # Note: Must use _meta= (alias) not meta= due to Pydantic alias handling
+        result = CreateTaskResult(
+            task=task,
+            **{"_meta": {MODEL_IMMEDIATE_RESPONSE_KEY: immediate_msg}},
+        )
+
+        # Verify the metadata is present and correct
+        assert result.meta is not None
+        assert MODEL_IMMEDIATE_RESPONSE_KEY in result.meta
+        assert result.meta[MODEL_IMMEDIATE_RESPONSE_KEY] == immediate_msg
+
+        # Verify it serializes correctly with _meta alias
+        serialized = result.model_dump(by_alias=True)
+        assert "_meta" in serialized
+        assert MODEL_IMMEDIATE_RESPONSE_KEY in serialized["_meta"]
+        assert serialized["_meta"][MODEL_IMMEDIATE_RESPONSE_KEY] == immediate_msg
 
 
 # --- Getting Task Status (tasks/get) ---