Replace magic strings with constants for task status and metadata

- Add TASK_STATUS_* constants to types.py for task status values
- Use RELATED_TASK_METADATA_KEY constant instead of hardcoded string
- Use RelatedTaskMetadata model instead of raw dict with 'taskId'
- Update all task status usages to use constants

Author: maxisbey

src/mcp/server/experimental/task_context.py

@@ -19,6 +19,8 @@
 from mcp.shared.experimental.tasks.resolver import Resolver
 from mcp.shared.experimental.tasks.store import TaskStore
 from mcp.types import (
+    TASK_STATUS_INPUT_REQUIRED,
+    TASK_STATUS_WORKING,
     ClientCapabilities,
     CreateMessageResult,
     ElicitationCapability,
@@ -236,7 +238,7 @@ async def elicit(
             raise RuntimeError("handler is required for elicit(). Pass handler= to ServerTaskContext.")
 
         # Update status to input_required
-        await self._store.update_task(self.task_id, status="input_required")
+        await self._store.update_task(self.task_id, status=TASK_STATUS_INPUT_REQUIRED)
 
         # Build the request using session's helper
         request = self._session._build_elicit_request(  # pyright: ignore[reportPrivateUsage]
@@ -262,10 +264,10 @@ async def elicit(
         try:
             # Wait for response (routed back via TaskResultHandler)
             response_data = await resolver.wait()
-            await self._store.update_task(self.task_id, status="working")
+            await self._store.update_task(self.task_id, status=TASK_STATUS_WORKING)
             return ElicitResult.model_validate(response_data)
         except anyio.get_cancelled_exc_class():
-            await self._store.update_task(self.task_id, status="working")
+            await self._store.update_task(self.task_id, status=TASK_STATUS_WORKING)
             raise
 
     async def create_message(
@@ -313,7 +315,7 @@ async def create_message(
             raise RuntimeError("handler is required for create_message(). Pass handler= to ServerTaskContext.")
 
         # Update status to input_required
-        await self._store.update_task(self.task_id, status="input_required")
+        await self._store.update_task(self.task_id, status=TASK_STATUS_INPUT_REQUIRED)
 
         # Build the request using session's helper
         request = self._session._build_create_message_request(  # pyright: ignore[reportPrivateUsage]
@@ -345,8 +347,8 @@ async def create_message(
         try:
             # Wait for response (routed back via TaskResultHandler)
             response_data = await resolver.wait()
-            await self._store.update_task(self.task_id, status="working")
+            await self._store.update_task(self.task_id, status=TASK_STATUS_WORKING)
             return CreateMessageResult.model_validate(response_data)
         except anyio.get_cancelled_exc_class():
-            await self._store.update_task(self.task_id, status="working")
+            await self._store.update_task(self.task_id, status=TASK_STATUS_WORKING)
             raise

src/mcp/server/experimental/task_result_handler.py

@@ -17,7 +17,7 @@
 
 from mcp.server.session import ServerSession
 from mcp.shared.exceptions import McpError
-from mcp.shared.experimental.tasks.helpers import is_terminal
+from mcp.shared.experimental.tasks.helpers import RELATED_TASK_METADATA_KEY, is_terminal
 from mcp.shared.experimental.tasks.message_queue import TaskMessageQueue
 from mcp.shared.experimental.tasks.resolver import Resolver
 from mcp.shared.experimental.tasks.store import TaskStore
@@ -28,6 +28,7 @@
     GetTaskPayloadRequest,
     GetTaskPayloadResult,
     JSONRPCMessage,
+    RelatedTaskMetadata,
     RequestId,
 )
 
@@ -126,9 +127,9 @@ async def handle(
                 result = await self._store.get_result(task_id)
                 # GetTaskPayloadResult is a Result with extra="allow"
                 # The stored result contains the actual payload data
-                # Per spec: tasks/result MUST include _meta.io.modelcontextprotocol/related-task
-                # with taskId, as the result structure itself does not contain the task ID
-                related_task_meta: dict[str, Any] = {"io.modelcontextprotocol/related-task": {"taskId": task_id}}
+                # Per spec: tasks/result MUST include _meta with related-task metadata
+                related_task = RelatedTaskMetadata(taskId=task_id)
+                related_task_meta: dict[str, Any] = {RELATED_TASK_METADATA_KEY: related_task.model_dump(by_alias=True)}
                 if result is not None:
                     # Copy result fields and add required metadata
                     result_data = result.model_dump(by_alias=True)

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

@@ -7,7 +7,7 @@
 """
 
 from mcp.shared.experimental.tasks.store import TaskStore
-from mcp.types import Result, Task
+from mcp.types import TASK_STATUS_COMPLETED, TASK_STATUS_FAILED, Result, Task
 
 
 class TaskContext:
@@ -84,7 +84,7 @@ async def complete(self, result: Result) -> None:
         await self._store.store_result(self.task_id, result)
         self._task = await self._store.update_task(
             self.task_id,
-            status="completed",
+            status=TASK_STATUS_COMPLETED,
         )
 
     async def fail(self, error: str) -> None:
@@ -96,6 +96,6 @@ async def fail(self, error: str) -> None:
         """
         self._task = await self._store.update_task(
             self.task_id,
-            status="failed",
+            status=TASK_STATUS_FAILED,
             status_message=error,
         )

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

@@ -15,6 +15,10 @@
 from mcp.shared.experimental.tasks.store import TaskStore
 from mcp.types import (
     INVALID_PARAMS,
+    TASK_STATUS_CANCELLED,
+    TASK_STATUS_COMPLETED,
+    TASK_STATUS_FAILED,
+    TASK_STATUS_WORKING,
     CancelTaskResult,
     ErrorData,
     Task,
@@ -43,7 +47,7 @@ def is_terminal(status: TaskStatus) -> bool:
     Returns:
         True if the status is terminal (completed, failed, or cancelled)
     """
-    return status in ("completed", "failed", "cancelled")
+    return status in (TASK_STATUS_COMPLETED, TASK_STATUS_FAILED, TASK_STATUS_CANCELLED)
 
 
 async def cancel_task(
@@ -94,7 +98,7 @@ async def handle_cancel(request: CancelTaskRequest) -> CancelTaskResult:
         )
 
     # Update task to cancelled status
-    cancelled_task = await store.update_task(task_id, status="cancelled")
+    cancelled_task = await store.update_task(task_id, status=TASK_STATUS_CANCELLED)
     return CancelTaskResult(**cancelled_task.model_dump())
 
 
@@ -122,7 +126,7 @@ def create_task_state(
     now = datetime.now(timezone.utc)
     return Task(
         taskId=task_id or generate_task_id(),
-        status="working",
+        status=TASK_STATUS_WORKING,
         createdAt=now,
         lastUpdatedAt=now,
         ttl=metadata.ttl,

src/mcp/types.py

@@ -519,6 +519,13 @@ class ServerCapabilities(BaseModel):
 
 TaskStatus = Literal["working", "input_required", "completed", "failed", "cancelled"]
 
+# Task status constants
+TASK_STATUS_WORKING: Final[Literal["working"]] = "working"
+TASK_STATUS_INPUT_REQUIRED: Final[Literal["input_required"]] = "input_required"
+TASK_STATUS_COMPLETED: Final[Literal["completed"]] = "completed"
+TASK_STATUS_FAILED: Final[Literal["failed"]] = "failed"
+TASK_STATUS_CANCELLED: Final[Literal["cancelled"]] = "cancelled"
+
 
 class RelatedTaskMetadata(BaseModel):
     """