Move relay logic to elicitation.py, fix related-task metadata key

Moves relay_elicitation() into elicitation.py so it can reuse
handle_task_input() for the Redis push instead of duplicating that logic.
notifications.py just detects the trigger and calls it.

Also fixes the related-task metadata key from modelcontextprotocol.io/ to
io.modelcontextprotocol/ to match the current spec:
https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/tasks

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/fastmcp/server/tasks/__init__.py

@@ -5,7 +5,11 @@ This module implements protocol-level background task execution for MCP servers.
 
 from fastmcp.server.tasks.capabilities import get_task_capabilities
 from fastmcp.server.tasks.config import TaskConfig, TaskMeta, TaskMode
-from fastmcp.server.tasks.elicitation import elicit_for_task, handle_task_input
+from fastmcp.server.tasks.elicitation import (
+    elicit_for_task,
+    handle_task_input,
+    relay_elicitation,
+)
 from fastmcp.server.tasks.keys import (
     build_task_key,
     get_client_task_id_from_key,
@@ -29,5 +33,6 @@ __all__ = [
     "handle_task_input",
     "parse_task_key",
     "push_notification",
+    "relay_elicitation",
     "stop_subscriber",
 ]

src/fastmcp/server/tasks/elicitation.py

@@ -134,7 +134,7 @@ async def elicit_for_task(
             "ttl": ELICIT_TTL_SECONDS * 1000,
         },
         "_meta": {
-            "modelcontextprotocol.io/related-task": {
+            "io.modelcontextprotocol/related-task": {
                 "taskId": task_id,
                 "status": "input_required",
                 "statusMessage": message,
@@ -231,6 +231,62 @@ async def elicit_for_task(
     return mcp.types.ElicitResult(action="cancel", content=None)
 
 
+async def relay_elicitation(
+    session: ServerSession,
+    session_id: str,
+    task_id: str,
+    elicitation: dict[str, Any],
+    fastmcp: FastMCP,
+) -> None:
+    """Relay elicitation from a background task worker to the client.
+
+    Called by the notification subscriber when it detects an input_required
+    notification with elicitation metadata. Sends a standard elicitation/create
+    request to the client session, then uses handle_task_input() to push the
+    response to Redis so the blocked worker can resume.
+
+    Args:
+        session: MCP ServerSession
+        session_id: Session identifier
+        task_id: Background task ID
+        elicitation: Elicitation metadata (message, requestedSchema)
+        fastmcp: FastMCP server instance
+    """
+    try:
+        result = await session.elicit(
+            message=elicitation["message"],
+            requestedSchema=elicitation["requestedSchema"],
+        )
+        await handle_task_input(
+            task_id=task_id,
+            session_id=session_id,
+            action=result.action,
+            content=result.content,
+            fastmcp=fastmcp,
+        )
+        logger.debug(
+            "Relayed elicitation response for task %s (action=%s)",
+            task_id,
+            result.action,
+        )
+    except Exception as e:
+        logger.warning("Failed to relay elicitation for task %s: %s", task_id, e)
+        # Push a cancel response so the worker's BLPOP doesn't block forever
+        success = await handle_task_input(
+            task_id=task_id,
+            session_id=session_id,
+            action="cancel",
+            content=None,
+            fastmcp=fastmcp,
+        )
+        if not success:
+            logger.warning(
+                "Failed to push cancel response for task %s "
+                "(worker may block until TTL)",
+                task_id,
+            )
+
+
 async def handle_task_input(
     task_id: str,
     session_id: str,

src/fastmcp/server/tasks/handlers.py

@@ -127,7 +127,7 @@ async def submit_to_docket(
                 "pollInterval": poll_interval_ms,
             },
             "_meta": {
-                "modelcontextprotocol.io/related-task": {
+                "io.modelcontextprotocol/related-task": {
                     "taskId": server_task_id,
                 }
             },
@@ -173,7 +173,7 @@ async def submit_to_docket(
     )
 
     try:
-        await ensure_subscriber_running(session_id, ctx.session, docket)
+        await ensure_subscriber_running(session_id, ctx.session, docket, ctx.fastmcp)
 
         # Register cleanup callback on session exit (once per session)
         # This ensures subscriber is stopped when the session disconnects

src/fastmcp/server/tasks/notifications.py

@@ -31,6 +31,8 @@ if TYPE_CHECKING:
     from docket import Docket
     from mcp.server.session import ServerSession
 
+    from fastmcp.server.server import FastMCP
+
 logger = logging.getLogger(__name__)
 
 # Redis key patterns
@@ -75,6 +77,7 @@ async def notification_subscriber_loop(
     session_id: str,
     session: ServerSession,
     docket: Docket,
+    fastmcp: FastMCP,
 ) -> None:
     """Subscribe to notification queue and forward to session.
 
@@ -90,6 +93,7 @@ async def notification_subscriber_loop(
         session_id: Session identifier to subscribe to
         session: MCP ServerSession for sending notifications
         docket: Docket instance for Redis access
+        fastmcp: FastMCP server instance (for elicitation relay)
     """
     queue_key = docket.key(NOTIFICATION_QUEUE_KEY.format(session_id=session_id))
     active_key = docket.key(NOTIFICATION_ACTIVE_KEY.format(session_id=session_id))
@@ -118,7 +122,7 @@ async def notification_subscriber_loop(
                 try:
                     # Reconstruct and send MCP notification
                     await _send_mcp_notification(
-                        session, notification_dict, session_id, docket
+                        session, notification_dict, session_id, docket, fastmcp
                     )
                     logger.debug(
                         "Delivered notification to session %s (attempt %d)",
@@ -163,6 +167,7 @@ async def _send_mcp_notification(
     notification_dict: dict[str, Any],
     session_id: str,
     docket: Docket,
+    fastmcp: FastMCP,
 ) -> None:
     """Reconstruct MCP notification from dict and send to session.
 
@@ -174,7 +179,8 @@ async def _send_mcp_notification(
         session: MCP ServerSession
         notification_dict: Notification as dict (method, params, _meta)
         session_id: Session identifier (for elicitation relay)
-        docket: Docket instance (for elicitation relay)
+        docket: Docket instance (for notification delivery)
+        fastmcp: FastMCP server instance (for elicitation relay)
     """
     method = notification_dict.get("method", "notifications/tasks/status")
     if method != "notifications/tasks/status":
@@ -196,7 +202,7 @@ async def _send_mcp_notification(
     params = notification_dict.get("params", {})
     if params.get("status") == "input_required":
         meta = notification_dict.get("_meta", {})
-        related_task = meta.get("modelcontextprotocol.io/related-task", {})
+        related_task = meta.get("io.modelcontextprotocol/related-task", {})
         elicitation = related_task.get("elicitation")
         if elicitation:
             task_id = params.get("taskId")
@@ -205,95 +211,16 @@ async def _send_mcp_notification(
                     "input_required notification missing taskId, skipping relay"
                 )
                 return
+            from fastmcp.server.tasks.elicitation import relay_elicitation
+
             task = asyncio.create_task(
-                _relay_elicitation(session, session_id, task_id, elicitation, docket),
+                relay_elicitation(session, session_id, task_id, elicitation, fastmcp),
                 name=f"elicitation-relay-{task_id[:8]}",
             )
             _background_tasks.add(task)
             task.add_done_callback(_background_tasks.discard)
 
 
-async def _relay_elicitation(
-    session: ServerSession,
-    session_id: str,
-    task_id: str,
-    elicitation: dict[str, Any],
-    docket: Docket,
-) -> None:
-    """Relay elicitation from a background task worker to the client.
-
-    Sends a standard elicitation/create request to the client session, then
-    pushes the response to Redis so the blocked worker can resume.
-
-    Args:
-        session: MCP ServerSession
-        session_id: Session identifier
-        task_id: Background task ID
-        elicitation: Elicitation metadata (message, requestedSchema)
-        docket: Docket instance for Redis access
-    """
-    from fastmcp.server.tasks.elicitation import (
-        ELICIT_RESPONSE_KEY,
-        ELICIT_STATUS_KEY,
-        ELICIT_TTL_SECONDS,
-    )
-
-    try:
-        result = await session.elicit(
-            message=elicitation["message"],
-            requestedSchema=elicitation["requestedSchema"],
-        )
-
-        response_key = docket.key(
-            ELICIT_RESPONSE_KEY.format(session_id=session_id, task_id=task_id)
-        )
-        status_key = docket.key(
-            ELICIT_STATUS_KEY.format(session_id=session_id, task_id=task_id)
-        )
-
-        response = {
-            "action": result.action,
-            "content": result.content,
-        }
-
-        async with docket.redis() as redis:
-            await redis.lpush(  # type: ignore[invalid-await]
-                response_key, json.dumps(response)
-            )
-            await redis.expire(response_key, ELICIT_TTL_SECONDS)
-            await redis.set(status_key, "responded", ex=ELICIT_TTL_SECONDS)
-
-        logger.debug(
-            "Relayed elicitation response for task %s (action=%s)",
-            task_id,
-            result.action,
-        )
-    except Exception as e:
-        logger.warning("Failed to relay elicitation for task %s: %s", task_id, e)
-        # Push a cancel response so the worker's BLPOP doesn't block forever
-        try:
-            response_key = docket.key(
-                ELICIT_RESPONSE_KEY.format(session_id=session_id, task_id=task_id)
-            )
-            status_key = docket.key(
-                ELICIT_STATUS_KEY.format(session_id=session_id, task_id=task_id)
-            )
-            cancel = {"action": "cancel", "content": None}
-            async with docket.redis() as redis:
-                await redis.lpush(  # type: ignore[invalid-await]
-                    response_key, json.dumps(cancel)
-                )
-                await redis.expire(response_key, ELICIT_TTL_SECONDS)
-                await redis.set(status_key, "responded", ex=ELICIT_TTL_SECONDS)
-        except Exception as cancel_error:
-            logger.warning(
-                "Failed to push cancel response for task %s "
-                "(worker may block until TTL): %s",
-                task_id,
-                cancel_error,
-            )
-
-
 # =============================================================================
 # Subscriber Management
 # =============================================================================
@@ -312,6 +239,7 @@ async def ensure_subscriber_running(
     session_id: str,
     session: ServerSession,
     docket: Docket,
+    fastmcp: FastMCP,
 ) -> None:
     """Start notification subscriber if not already running (idempotent).
 
@@ -322,6 +250,7 @@ async def ensure_subscriber_running(
         session_id: Session identifier
         session: MCP ServerSession
         docket: Docket instance
+        fastmcp: FastMCP server instance (for elicitation relay)
     """
     # Check if subscriber already running for this session
     if session_id in _active_subscribers:
@@ -339,7 +268,7 @@ async def ensure_subscriber_running(
 
     # Start new subscriber task
     task = asyncio.create_task(
-        notification_subscriber_loop(session_id, session, docket),
+        notification_subscriber_loop(session_id, session, docket, fastmcp),
         name=f"notification-subscriber-{session_id[:8]}",
     )
     _active_subscribers[session_id] = (task, weakref.ref(session))

src/fastmcp/server/tasks/requests.py

@@ -300,7 +300,7 @@ async def tasks_result_handler(server: FastMCP, params: dict[str, Any]) -> Any:
                 content=[mcp.types.TextContent(type="text", text=str(error))],
                 isError=True,
                 _meta={  # type: ignore[call-arg]  # _meta is Pydantic alias for meta field
-                    "modelcontextprotocol.io/related-task": {
+                    "io.modelcontextprotocol/related-task": {
                         "taskId": client_task_id,
                     }
                 },
@@ -342,7 +342,7 @@ async def tasks_result_handler(server: FastMCP, params: dict[str, Any]) -> Any:
 
         # Build related-task metadata
         related_task_meta = {
-            "modelcontextprotocol.io/related-task": {
+            "io.modelcontextprotocol/related-task": {
                 "taskId": client_task_id,
             }
         }

tests/server/tasks/test_notifications.py

@@ -85,7 +85,7 @@ class TestNotificationIntegration:
             for candidate in reversed(candidates):
                 candidate_meta = getattr(candidate.root, "_meta", None)
                 related_task = (
-                    candidate_meta.get("modelcontextprotocol.io/related-task")
+                    candidate_meta.get("io.modelcontextprotocol/related-task")
                     if isinstance(candidate_meta, dict)
                     else None
                 )
@@ -100,7 +100,7 @@ class TestNotificationIntegration:
             task_meta = getattr(notification.root, "_meta", None)
             assert isinstance(task_meta, dict)
 
-            related_task = task_meta.get("modelcontextprotocol.io/related-task")
+            related_task = task_meta.get("io.modelcontextprotocol/related-task")
             assert isinstance(related_task, dict)
             assert related_task.get("taskId") == task.task_id
             assert related_task.get("status") == "input_required"

tests/server/tasks/test_task_metadata.py

@@ -2,7 +2,7 @@
 Tests for SEP-1686 related-task metadata in protocol responses.
 
 Per the spec, all task-related responses MUST include
-modelcontextprotocol.io/related-task in _meta.
+io.modelcontextprotocol/related-task in _meta.
 """
 
 import pytest
@@ -24,7 +24,7 @@ async def metadata_server():
 
 
 async def test_tasks_get_includes_related_task_metadata(metadata_server: FastMCP):
-    """tasks/get response includes modelcontextprotocol.io/related-task in _meta."""
+    """tasks/get response includes io.modelcontextprotocol/related-task in _meta."""
     async with Client(metadata_server) as client:
         # Submit a task
         task = await client.call_tool("test_tool", {"value": 5}, task=True)
@@ -40,7 +40,7 @@ async def test_tasks_get_includes_related_task_metadata(metadata_server: FastMCP
 
 
 async def test_tasks_result_includes_related_task_metadata(metadata_server: FastMCP):
-    """tasks/result response includes modelcontextprotocol.io/related-task in _meta."""
+    """tasks/result response includes io.modelcontextprotocol/related-task in _meta."""
     async with Client(metadata_server) as client:
         # Submit and complete a task
         task = await client.call_tool("test_tool", {"value": 7}, task=True)
@@ -53,7 +53,7 @@ async def test_tasks_result_includes_related_task_metadata(metadata_server: Fast
 
 
 async def test_tasks_list_includes_related_task_metadata(metadata_server: FastMCP):
-    """tasks/list response includes modelcontextprotocol.io/related-task in _meta."""
+    """tasks/list response includes io.modelcontextprotocol/related-task in _meta."""
     async with Client(metadata_server) as client:
         # List tasks via client (which uses protocol properly)
         result = await client.list_tasks()