Add lastUpdatedAt field and related-task metadata for spec conformance

This addresses two critical spec compliance gaps:

1. Add `lastUpdatedAt` field to Task model
   - Required by spec: ISO 8601 timestamp updated on every status change
   - Added to Task model in types.py
   - Initialized alongside createdAt in create_task_state()
   - Updated in InMemoryTaskStore.update_task() on any change
   - Included in all Task responses and notifications

2. Add related-task metadata to tasks/result response
   - Per spec: tasks/result MUST include _meta with
     io.modelcontextprotocol/related-task containing the taskId
   - Required because result structure doesn't contain task ID
   - Merges with any existing _meta from stored result

Author: maxisbey

examples/servers/simple-task-interactive/mcp_simple_task_interactive/server.py

@@ -187,6 +187,7 @@ async def handle_get_task(request: types.GetTaskRequest) -> types.GetTaskResult:
         status=task.status,
         statusMessage=task.statusMessage,
         createdAt=task.createdAt,
+        lastUpdatedAt=task.lastUpdatedAt,
         ttl=task.ttl,
         pollInterval=task.pollInterval,
     )

examples/servers/simple-task/mcp_simple_task/server.py

@@ -90,6 +90,7 @@ async def handle_get_task(request: types.GetTaskRequest) -> types.GetTaskResult:
         status=task.status,
         statusMessage=task.statusMessage,
         createdAt=task.createdAt,
+        lastUpdatedAt=task.lastUpdatedAt,
         ttl=task.ttl,
         pollInterval=task.pollInterval,
     )

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

@@ -132,6 +132,7 @@ async def _send_notification(self) -> None:
                         status=self._task.status,
                         statusMessage=self._task.statusMessage,
                         createdAt=self._task.createdAt,
+                        lastUpdatedAt=self._task.lastUpdatedAt,
                         ttl=self._task.ttl,
                         pollInterval=self._task.pollInterval,
                     )

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

@@ -54,10 +54,12 @@ def create_task_state(
     Returns:
         A new Task in "working" status
     """
+    now = datetime.now(timezone.utc)
     return Task(
         taskId=task_id or generate_task_id(),
         status="working",
-        createdAt=datetime.now(timezone.utc),
+        createdAt=now,
+        lastUpdatedAt=now,
         ttl=metadata.ttl,
         pollInterval=500,  # Default 500ms poll interval
     )

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

@@ -122,6 +122,9 @@ async def update_task(
         if status_message is not None:
             stored.task.statusMessage = status_message
 
+        # Update lastUpdatedAt on any change
+        stored.task.lastUpdatedAt = datetime.now(timezone.utc)
+
         # If task is now terminal and has TTL, reset expiry timer
         if status is not None and is_terminal(status) and stored.task.ttl is not None:
             stored.expires_at = self._calculate_expiry(stored.task.ttl)

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

@@ -128,10 +128,17 @@ 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}}
                 if result is not None:
-                    # Copy result fields into GetTaskPayloadResult
-                    return GetTaskPayloadResult.model_validate(result.model_dump(by_alias=True))
-                return GetTaskPayloadResult()
+                    # Copy result fields and add required metadata
+                    result_data = result.model_dump(by_alias=True)
+                    # Merge with existing _meta if present
+                    existing_meta: dict[str, Any] = result_data.get("_meta") or {}
+                    result_data["_meta"] = {**existing_meta, **related_task_meta}
+                    return GetTaskPayloadResult.model_validate(result_data)
+                return GetTaskPayloadResult.model_validate({"_meta": related_task_meta})
 
             # Wait for task update (status change or new messages)
             await self._wait_for_task_update(task_id)

src/mcp/types.py

@@ -492,6 +492,9 @@ class Task(BaseModel):
     createdAt: datetime  # Pydantic will enforce ISO 8601 and re-serialize as a string later
     """ISO 8601 timestamp when the task was created."""
 
+    lastUpdatedAt: datetime
+    """ISO 8601 timestamp when the task was last updated."""
+
     ttl: Annotated[int, Field(strict=True)] | None
     """Actual retention duration from creation in milliseconds, null for unlimited."""
 

tests/experimental/tasks/client/test_handlers.py

@@ -76,6 +76,7 @@ async def get_task_handler(
                 status=task.status,
                 statusMessage=task.statusMessage,
                 createdAt=task.createdAt,
+                lastUpdatedAt=task.lastUpdatedAt,
                 ttl=task.ttl,
                 pollInterval=task.pollInterval,
             )
@@ -327,6 +328,7 @@ async def cancel_task_handler(
                 taskId=updated.taskId,
                 status=updated.status,
                 createdAt=updated.createdAt,
+                lastUpdatedAt=updated.lastUpdatedAt,
                 ttl=updated.ttl,
             )
 
@@ -450,6 +452,7 @@ async def get_task_handler(
                 status=task.status,
                 statusMessage=task.statusMessage,
                 createdAt=task.createdAt,
+                lastUpdatedAt=task.lastUpdatedAt,
                 ttl=task.ttl,
                 pollInterval=task.pollInterval,
             )

tests/experimental/tasks/client/test_tasks.py

@@ -96,6 +96,7 @@ async def handle_get_task(request: GetTaskRequest) -> GetTaskResult:
             status=task.status,
             statusMessage=task.statusMessage,
             createdAt=task.createdAt,
+            lastUpdatedAt=task.lastUpdatedAt,
             ttl=task.ttl,
             pollInterval=task.pollInterval,
         )
@@ -419,6 +420,7 @@ async def handle_get_task(request: GetTaskRequest) -> GetTaskResult:
             status=task.status,
             statusMessage=task.statusMessage,
             createdAt=task.createdAt,
+            lastUpdatedAt=task.lastUpdatedAt,
             ttl=task.ttl,
             pollInterval=task.pollInterval,
         )
@@ -437,6 +439,7 @@ async def handle_cancel_task(request: CancelTaskRequest) -> CancelTaskResult:
             taskId=updated_task.taskId,
             status=updated_task.status,
             createdAt=updated_task.createdAt,
+            lastUpdatedAt=updated_task.lastUpdatedAt,
             ttl=updated_task.ttl,
         )
 

tests/experimental/tasks/server/test_elicitation_flow.py

@@ -169,6 +169,7 @@ async def handle_get_task(request: GetTaskRequest) -> GetTaskResult:
             status=task.status,
             statusMessage=task.statusMessage,
             createdAt=task.createdAt,
+            lastUpdatedAt=task.lastUpdatedAt,
             ttl=task.ttl,
             pollInterval=task.pollInterval,
         )