diff --git a/src/fastmcp/server/tasks/__init__.py b/src/fastmcp/server/tasks/__init__.py index 20dd733a8..008332db5 100644 --- a/src/fastmcp/server/tasks/__init__.py +++ b/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", ] diff --git a/src/fastmcp/server/tasks/elicitation.py b/src/fastmcp/server/tasks/elicitation.py index 1a85da9f8..cb148cfc7 100644 --- a/src/fastmcp/server/tasks/elicitation.py +++ b/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, diff --git a/src/fastmcp/server/tasks/handlers.py b/src/fastmcp/server/tasks/handlers.py index 494fce87f..fa8ba3ce4 100644 --- a/src/fastmcp/server/tasks/handlers.py +++ b/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 diff --git a/src/fastmcp/server/tasks/notifications.py b/src/fastmcp/server/tasks/notifications.py index 01f049dae..67417bd62 100644 --- a/src/fastmcp/server/tasks/notifications.py +++ b/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)) diff --git a/src/fastmcp/server/tasks/requests.py b/src/fastmcp/server/tasks/requests.py index 61286d831..fae63c08d 100644 --- a/src/fastmcp/server/tasks/requests.py +++ b/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, } } diff --git a/tests/server/tasks/test_notifications.py b/tests/server/tasks/test_notifications.py index 34b169bb5..f0e144fd4 100644 --- a/tests/server/tasks/test_notifications.py +++ b/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" diff --git a/tests/server/tasks/test_task_metadata.py b/tests/server/tasks/test_task_metadata.py index c603ff6a6..32ce2b849 100644 --- a/tests/server/tasks/test_task_metadata.py +++ b/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()