diff --git a/docs/python-sdk/fastmcp-server-context.mdx b/docs/python-sdk/fastmcp-server-context.mdx index 19d67e5d6..2873100b7 100644 --- a/docs/python-sdk/fastmcp-server-context.mdx +++ b/docs/python-sdk/fastmcp-server-context.mdx @@ -189,12 +189,16 @@ report_progress(self, progress: float, total: float | None = None, message: str Report progress for the current operation. +Works in both foreground (MCP progress notifications) and background +(Docket task execution) contexts. + **Args:** - `progress`: Current progress value e.g. 24 - `total`: Optional total value e.g. 100 +- `message`: Optional status message describing current progress -#### `list_resources` +#### `list_resources` ```python list_resources(self) -> list[SDKResource] @@ -206,7 +210,7 @@ List all available resources from the server. - List of Resource objects available on the server -#### `list_prompts` +#### `list_prompts` ```python list_prompts(self) -> list[SDKPrompt] @@ -218,7 +222,7 @@ List all available prompts from the server. - List of Prompt objects available on the server -#### `get_prompt` +#### `get_prompt` ```python get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> GetPromptResult @@ -234,7 +238,7 @@ Get a prompt by name with optional arguments. - The prompt result -#### `read_resource` +#### `read_resource` ```python read_resource(self, uri: str | AnyUrl) -> ResourceResult @@ -249,7 +253,7 @@ Read a resource by URI. - ResourceResult with contents -#### `log` +#### `log` ```python log(self, message: str, level: LoggingLevel | None = None, logger_name: str | None = None, extra: Mapping[str, Any] | None = None) -> None @@ -267,7 +271,7 @@ Messages sent to Clients are also logged to the `fastmcp.server.context.to_clien - `extra`: Optional mapping for additional arguments -#### `transport` +#### `transport` ```python transport(self) -> TransportType | None @@ -279,7 +283,7 @@ Returns the transport type used to run this server: "stdio", "sse", or "streamable-http". Returns None if called outside of a server context. -#### `client_supports_extension` +#### `client_supports_extension` ```python client_supports_extension(self, extension_id: str) -> bool @@ -304,7 +308,7 @@ Example:: return "text-only client" -#### `client_id` +#### `client_id` ```python client_id(self) -> str | None @@ -313,7 +317,7 @@ client_id(self) -> str | None Get the client ID if available. -#### `request_id` +#### `request_id` ```python request_id(self) -> str @@ -324,7 +328,7 @@ Get the unique ID for this request. Raises RuntimeError if MCP request context is not available. -#### `session_id` +#### `session_id` ```python session_id(self) -> str @@ -341,7 +345,7 @@ the same client session. - for other transports. -#### `session` +#### `session` ```python session(self) -> ServerSession @@ -355,7 +359,7 @@ In background task mode: Returns the session stored at Context creation. Raises RuntimeError if no session is available. -#### `debug` +#### `debug` ```python debug(self, message: str, logger_name: str | None = None, extra: Mapping[str, Any] | None = None) -> None @@ -366,7 +370,7 @@ Send a `DEBUG`-level message to the connected MCP Client. Messages sent to Clients are also logged to the `fastmcp.server.context.to_client` logger with a level of `DEBUG`. -#### `info` +#### `info` ```python info(self, message: str, logger_name: str | None = None, extra: Mapping[str, Any] | None = None) -> None @@ -377,7 +381,7 @@ Send a `INFO`-level message to the connected MCP Client. Messages sent to Clients are also logged to the `fastmcp.server.context.to_client` logger with a level of `DEBUG`. -#### `warning` +#### `warning` ```python warning(self, message: str, logger_name: str | None = None, extra: Mapping[str, Any] | None = None) -> None @@ -388,7 +392,7 @@ Send a `WARNING`-level message to the connected MCP Client. Messages sent to Clients are also logged to the `fastmcp.server.context.to_client` logger with a level of `DEBUG`. -#### `error` +#### `error` ```python error(self, message: str, logger_name: str | None = None, extra: Mapping[str, Any] | None = None) -> None @@ -399,7 +403,7 @@ Send a `ERROR`-level message to the connected MCP Client. Messages sent to Clients are also logged to the `fastmcp.server.context.to_client` logger with a level of `DEBUG`. -#### `list_roots` +#### `list_roots` ```python list_roots(self) -> list[Root] @@ -408,7 +412,7 @@ list_roots(self) -> list[Root] List the roots available to the server, as indicated by the client. -#### `send_notification` +#### `send_notification` ```python send_notification(self, notification: mcp.types.ServerNotificationType) -> None @@ -420,7 +424,7 @@ Send a notification to the client immediately. - `notification`: An MCP notification instance (e.g., ToolListChangedNotification()) -#### `close_sse_stream` +#### `close_sse_stream` ```python close_sse_stream(self) -> None @@ -438,7 +442,7 @@ Instead of holding a connection open for minutes, you can periodically close and let the client reconnect. -#### `sample_step` +#### `sample_step` ```python sample_step(self, messages: str | Sequence[str | SamplingMessage]) -> SampleStep @@ -481,7 +485,7 @@ regardless of this setting. - - .text: The text content (if any) -#### `sample` +#### `sample` ```python sample(self, messages: str | Sequence[str | SamplingMessage]) -> SamplingResult[ResultT] @@ -490,7 +494,7 @@ sample(self, messages: str | Sequence[str | SamplingMessage]) -> SamplingResult[ Overload: With result_type, returns SamplingResult[ResultT]. -#### `sample` +#### `sample` ```python sample(self, messages: str | Sequence[str | SamplingMessage]) -> SamplingResult[str] @@ -499,7 +503,7 @@ sample(self, messages: str | Sequence[str | SamplingMessage]) -> SamplingResult[ Overload: Without result_type, returns SamplingResult[str]. -#### `sample` +#### `sample` ```python sample(self, messages: str | Sequence[str | SamplingMessage]) -> SamplingResult[ResultT] | SamplingResult[str] @@ -547,43 +551,43 @@ regardless of this setting. - - .history: All messages exchanged during sampling -#### `elicit` +#### `elicit` ```python elicit(self, message: str, response_type: None) -> AcceptedElicitation[dict[str, Any]] | DeclinedElicitation | CancelledElicitation ``` -#### `elicit` +#### `elicit` ```python elicit(self, message: str, response_type: type[T]) -> AcceptedElicitation[T] | DeclinedElicitation | CancelledElicitation ``` -#### `elicit` +#### `elicit` ```python elicit(self, message: str, response_type: list[str]) -> AcceptedElicitation[str] | DeclinedElicitation | CancelledElicitation ``` -#### `elicit` +#### `elicit` ```python elicit(self, message: str, response_type: dict[str, dict[str, str]]) -> AcceptedElicitation[str] | DeclinedElicitation | CancelledElicitation ``` -#### `elicit` +#### `elicit` ```python elicit(self, message: str, response_type: list[list[str]]) -> AcceptedElicitation[list[str]] | DeclinedElicitation | CancelledElicitation ``` -#### `elicit` +#### `elicit` ```python elicit(self, message: str, response_type: list[dict[str, dict[str, str]]]) -> AcceptedElicitation[list[str]] | DeclinedElicitation | CancelledElicitation ``` -#### `elicit` +#### `elicit` ```python elicit(self, message: str, response_type: type[T] | list[str] | dict[str, dict[str, str]] | list[list[str]] | list[dict[str, dict[str, str]]] | None = None) -> AcceptedElicitation[T] | AcceptedElicitation[dict[str, Any]] | AcceptedElicitation[str] | AcceptedElicitation[list[str]] | DeclinedElicitation | CancelledElicitation @@ -612,7 +616,7 @@ type or dataclass or BaseModel. If it is a primitive type, an object schema with a single "value" field will be generated. -#### `set_state` +#### `set_state` ```python set_state(self, key: str, value: Any) -> None @@ -625,7 +629,7 @@ The key is automatically prefixed with the session identifier. State expires after 1 day to prevent unbounded memory growth. -#### `get_state` +#### `get_state` ```python get_state(self, key: str) -> Any @@ -636,7 +640,7 @@ Get a value from the session-scoped state store. Returns None if the key is not found. -#### `delete_state` +#### `delete_state` ```python delete_state(self, key: str) -> None @@ -645,7 +649,7 @@ delete_state(self, key: str) -> None Delete a value from the session-scoped state store. -#### `enable_components` +#### `enable_components` ```python enable_components(self) -> None @@ -669,7 +673,7 @@ ResourceListChangedNotification, and PromptListChangedNotification. - `match_all`: If True, matches all components regardless of other criteria. -#### `disable_components` +#### `disable_components` ```python disable_components(self) -> None @@ -693,7 +697,7 @@ ResourceListChangedNotification, and PromptListChangedNotification. - `match_all`: If True, matches all components regardless of other criteria. -#### `reset_visibility` +#### `reset_visibility` ```python reset_visibility(self) -> None diff --git a/docs/python-sdk/fastmcp-server-tasks-elicitation.mdx b/docs/python-sdk/fastmcp-server-tasks-elicitation.mdx index 1b06baa8e..3bdd697ef 100644 --- a/docs/python-sdk/fastmcp-server-tasks-elicitation.mdx +++ b/docs/python-sdk/fastmcp-server-tasks-elicitation.mdx @@ -13,7 +13,7 @@ in Docket workers. Unlike regular MCP requests, background tasks don't have an active request context, so elicitation requires special handling: 1. Set task status to "input_required" via Redis -2. Send notifications/tasks/updated with elicitation metadata +2. Send notifications/tasks/status with elicitation metadata 3. Wait for client to send input via tasks/sendInput 4. Resume task execution with the provided input @@ -26,7 +26,7 @@ internal APIs for background task coordination. ### `elicit_for_task` ```python -elicit_for_task(task_id: str, session: ServerSession, message: str, schema: dict[str, Any], fastmcp: FastMCP) -> mcp.types.ElicitResult +elicit_for_task(task_id: str, session: ServerSession | None, message: str, schema: dict[str, Any], fastmcp: FastMCP) -> mcp.types.ElicitResult ``` @@ -50,7 +50,29 @@ in a Docket worker context where there's no active MCP request. - `McpError`: If the elicitation request fails -### `handle_task_input` +### `relay_elicitation` + +```python +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 + + +### `handle_task_input` ```python handle_task_input(task_id: str, session_id: str, action: str, content: dict[str, Any] | None, fastmcp: FastMCP) -> bool diff --git a/docs/python-sdk/fastmcp-server-tasks-handlers.mdx b/docs/python-sdk/fastmcp-server-tasks-handlers.mdx index bee4b0b93..94e094174 100644 --- a/docs/python-sdk/fastmcp-server-tasks-handlers.mdx +++ b/docs/python-sdk/fastmcp-server-tasks-handlers.mdx @@ -13,7 +13,7 @@ Handles queuing tool/prompt/resource executions to Docket as background tasks. ## Functions -### `submit_to_docket` +### `submit_to_docket` ```python submit_to_docket(task_type: Literal['tool', 'resource', 'template', 'prompt'], key: str, component: Tool | Resource | ResourceTemplate | Prompt, arguments: dict[str, Any] | None = None, task_meta: TaskMeta | None = None) -> mcp.types.CreateTaskResult diff --git a/docs/python-sdk/fastmcp-server-tasks-notifications.mdx b/docs/python-sdk/fastmcp-server-tasks-notifications.mdx new file mode 100644 index 000000000..6652d7600 --- /dev/null +++ b/docs/python-sdk/fastmcp-server-tasks-notifications.mdx @@ -0,0 +1,113 @@ +--- +title: notifications +sidebarTitle: notifications +--- + +# `fastmcp.server.tasks.notifications` + + +Distributed notification queue for background task events (SEP-1686). + +Enables distributed Docket workers to send MCP notifications to clients +without holding session references. Workers push to a Redis queue, +the MCP server process subscribes and forwards to the client's session. + +Pattern: Fire-and-forward with retry +- One queue per session_id +- LPUSH/BRPOP for reliable ordered delivery +- Retry up to 3 times on delivery failure, then discard +- TTL-based expiration for stale messages + +Note: Docket's execution.subscribe() handles task state/progress events via +Redis Pub/Sub. This module handles elicitation-specific notifications that +require reliable delivery (input_required prompts, cancel signals). + + +## Functions + +### `push_notification` + +```python +push_notification(session_id: str, notification: dict[str, Any], docket: Docket) -> None +``` + + +Push notification to session's queue (called from Docket worker). + +Used for elicitation-specific notifications (input_required, cancel) +that need reliable delivery across distributed processes. + +**Args:** +- `session_id`: Target session's identifier +- `notification`: MCP notification dict (method, params, _meta) +- `docket`: Docket instance for Redis access + + +### `notification_subscriber_loop` + +```python +notification_subscriber_loop(session_id: str, session: ServerSession, docket: Docket, fastmcp: FastMCP) -> None +``` + + +Subscribe to notification queue and forward to session. + +Runs in the MCP server process. Bridges distributed workers to clients. + +This loop: +1. Maintains a heartbeat (active subscriber marker for debugging) +2. Blocks on BRPOP waiting for notifications +3. Forwards notifications to the client's session +4. Retries failed deliveries, then discards (no dead-letter queue) + +**Args:** +- `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) + + +### `ensure_subscriber_running` + +```python +ensure_subscriber_running(session_id: str, session: ServerSession, docket: Docket, fastmcp: FastMCP) -> None +``` + + +Start notification subscriber if not already running (idempotent). + +Subscriber is created on first task submission and cleaned up on disconnect. +Safe to call multiple times for the same session. + +**Args:** +- `session_id`: Session identifier +- `session`: MCP ServerSession +- `docket`: Docket instance +- `fastmcp`: FastMCP server instance (for elicitation relay) + + +### `stop_subscriber` + +```python +stop_subscriber(session_id: str) -> None +``` + + +Stop notification subscriber for a session. + +Called when session disconnects. Pending messages remain in queue +for delivery if client reconnects (with TTL expiration). + +**Args:** +- `session_id`: Session identifier + + +### `get_subscriber_count` + +```python +get_subscriber_count() -> int +``` + + +Get number of active subscribers (for monitoring). + diff --git a/examples/task_elicitation.py b/examples/task_elicitation.py new file mode 100644 index 000000000..51f7b046a --- /dev/null +++ b/examples/task_elicitation.py @@ -0,0 +1,82 @@ +""" +Background task elicitation demo. + +A background task (Docket) that pauses mid-execution to ask the user a +question, waits for the answer, then resumes and finishes. + +Works with both in-memory and Redis backends: + + # In-memory (single process, no Redis needed) + FASTMCP_DOCKET_URL=memory:// uv run python examples/task_elicitation.py + + # Redis (distributed, needs a worker running separately) + # Terminal 1: docker compose -f examples/tasks/docker-compose.yml up -d + # Terminal 2: FASTMCP_DOCKET_URL=redis://localhost:24242/0 \ + # uv run fastmcp tasks worker examples/task_elicitation.py + # Terminal 3: FASTMCP_DOCKET_URL=redis://localhost:24242/0 \ + # uv run python examples/task_elicitation.py + +Requires the `docket` extra (included in dev dependencies). +""" + +import asyncio +from dataclasses import dataclass + +from mcp.types import TextContent + +from fastmcp import Context, FastMCP +from fastmcp.client import Client +from fastmcp.server.elicitation import AcceptedElicitation + +mcp = FastMCP("Task Elicitation Demo") + + +@dataclass +class DinnerPrefs: + cuisine: str + vegetarian: bool + + +@mcp.tool(task=True) +async def plan_dinner(ctx: Context) -> str: + """Plan a dinner menu, asking the user what they're in the mood for.""" + + await ctx.report_progress(0, 2, "Asking what you'd like...") + + result = await ctx.elicit( + "What kind of dinner are you in the mood for?", + response_type=DinnerPrefs, + ) + + if not isinstance(result, AcceptedElicitation): + return "Dinner cancelled!" + + prefs = result.data + await ctx.report_progress(1, 2, "Planning your menu...") + await asyncio.sleep(1) + await ctx.report_progress(2, 2, "Done!") + + veg = "vegetarian " if prefs.vegetarian else "" + return f"Tonight's menu: a lovely {veg}{prefs.cuisine} dinner!" + + +async def handle_elicitation(message, response_type, params, context): + """Handle elicitation requests from background tasks.""" + print(f" Server asks: {message}") + print(" Responding with: cuisine=Thai, vegetarian=True") + return DinnerPrefs(cuisine="Thai", vegetarian=True) + + +async def main(): + async with Client(mcp, elicitation_handler=handle_elicitation) as client: + print("Starting background task...") + task = await client.call_tool("plan_dinner", {}, task=True) + print(f" task_id = {task.task_id}\n") + + result = await task.result() + assert isinstance(result.content[0], TextContent) + print(f"\nResult: {result.content[0].text}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/fastmcp/server/context.py b/src/fastmcp/server/context.py index a39817e58..d65e87c43 100644 --- a/src/fastmcp/server/context.py +++ b/src/fastmcp/server/context.py @@ -1133,7 +1133,7 @@ class Context: return await elicit_for_task( task_id=self._task_id, # type: ignore[arg-type] - session=self.session, + session=self._session, message=message, schema=schema, fastmcp=self.fastmcp, 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 299382df6..cb148cfc7 100644 --- a/src/fastmcp/server/tasks/elicitation.py +++ b/src/fastmcp/server/tasks/elicitation.py @@ -41,7 +41,7 @@ ELICIT_TTL_SECONDS = 3600 async def elicit_for_task( task_id: str, - session: ServerSession, + session: ServerSession | None, message: str, schema: dict[str, Any], fastmcp: FastMCP, @@ -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 2c65e31e1..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)) @@ -117,7 +121,9 @@ async def notification_subscriber_loop( try: # Reconstruct and send MCP notification - await _send_mcp_notification(session, notification_dict) + await _send_mcp_notification( + session, notification_dict, session_id, docket, fastmcp + ) logger.debug( "Delivered notification to session %s (attempt %d)", session_id, @@ -159,12 +165,22 @@ async def notification_subscriber_loop( async def _send_mcp_notification( session: ServerSession, notification_dict: dict[str, Any], + session_id: str, + docket: Docket, + fastmcp: FastMCP, ) -> None: """Reconstruct MCP notification from dict and send to session. + For input_required notifications with elicitation metadata, also sends + a standard elicitation/create request to the client and relays the + response back to the worker via Redis. + Args: session: MCP ServerSession notification_dict: Notification as dict (method, params, _meta) + session_id: Session identifier (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": @@ -181,11 +197,37 @@ async def _send_mcp_notification( await session.send_notification(server_notification) + # If this is an input_required notification with elicitation metadata, + # relay the elicitation to the client via standard elicitation/create + params = notification_dict.get("params", {}) + if params.get("status") == "input_required": + meta = notification_dict.get("_meta", {}) + related_task = meta.get("io.modelcontextprotocol/related-task", {}) + elicitation = related_task.get("elicitation") + if elicitation: + task_id = params.get("taskId") + if not task_id: + logger.warning( + "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, fastmcp), + name=f"elicitation-relay-{task_id[:8]}", + ) + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) + # ============================================================================= # Subscriber Management # ============================================================================= +# Strong references to fire-and-forget relay tasks (prevent GC mid-flight) +_background_tasks: set[asyncio.Task[None]] = set() + # Registry of active subscribers per session (prevents duplicates) # Uses weakref to session to detect disconnects _active_subscribers: dict[ @@ -197,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). @@ -207,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: @@ -224,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_context_background_task.py b/tests/server/tasks/test_context_background_task.py index 219a7ae21..2b5a1efa9 100644 --- a/tests/server/tasks/test_context_background_task.py +++ b/tests/server/tasks/test_context_background_task.py @@ -13,6 +13,7 @@ from mcp import ServerSession from fastmcp import FastMCP from fastmcp.client import Client +from fastmcp.client.elicitation import ElicitResult from fastmcp.server.context import Context from fastmcp.server.elicitation import AcceptedElicitation, DeclinedElicitation from fastmcp.server.tasks.elicitation import handle_task_input @@ -227,67 +228,31 @@ class TestBackgroundTaskIntegration: assert captured["is_background"] is True async def test_elicit_accept_flow(self): - """E2E: tool elicits input, client accepts, tool receives value. - - Flow: - 1. Tool calls ctx.elicit("name?", str) — blocks waiting for input - 2. Client polls handle_task_input(action="accept", content={"value":"Bob"}) - 3. Tool resumes with AcceptedElicitation(data="Bob") - """ + """E2E: tool elicits input, client accepts via elicitation_handler.""" mcp = FastMCP("elicit-accept-test") - elicit_started = asyncio.Event() - captured: dict[str, str | None] = {"task_id": None, "session_id": None} @mcp.tool(task=True) async def ask_name(ctx: Context) -> str: - captured["task_id"] = ctx.task_id - captured["session_id"] = ctx.session_id - elicit_started.set() - result = await ctx.elicit("What is your name?", str) if isinstance(result, AcceptedElicitation): return f"Hello, {result.data}!" return "No name provided" - async with Client(mcp) as client: + async def handler(message, response_type, params, ctx): + return ElicitResult(action="accept", content={"value": "Bob"}) + + async with Client(mcp, elicitation_handler=handler) as client: task = await client.call_tool("ask_name", {}, task=True) - await asyncio.wait_for(elicit_started.wait(), timeout=5.0) - - assert captured["task_id"] is not None - assert captured["session_id"] is not None - - # Poll until the "waiting" status is stored in Redis - success = False - for _ in range(40): - success = await handle_task_input( - task_id=captured["task_id"], - session_id=captured["session_id"], - action="accept", - content={"value": "Bob"}, - fastmcp=mcp, - ) - if success: - break - await asyncio.sleep(0.05) - - assert success is True, "handle_task_input should succeed within 2s" - await task.wait(timeout=10.0) result = await task.result() assert result.data == "Hello, Bob!" async def test_elicit_decline_flow(self): - """E2E: tool elicits input, client declines, tool gets DeclinedElicitation.""" + """E2E: tool elicits input, client declines via elicitation_handler.""" mcp = FastMCP("elicit-decline-test") - elicit_started = asyncio.Event() - captured: dict[str, str | None] = {"task_id": None, "session_id": None} @mcp.tool(task=True) async def optional_input(ctx: Context) -> str: - captured["task_id"] = ctx.task_id - captured["session_id"] = ctx.session_id - elicit_started.set() - result = await ctx.elicit("Want to provide a name?", str) if isinstance(result, DeclinedElicitation): return "User declined" @@ -295,34 +260,17 @@ class TestBackgroundTaskIntegration: return f"Got: {result.data}" return "Cancelled" - async with Client(mcp) as client: + async def handler(message, response_type, params, ctx): + return ElicitResult(action="decline") + + async with Client(mcp, elicitation_handler=handler) as client: task = await client.call_tool("optional_input", {}, task=True) - await asyncio.wait_for(elicit_started.wait(), timeout=5.0) - - assert captured["task_id"] is not None - assert captured["session_id"] is not None - - success = False - for _ in range(40): - success = await handle_task_input( - task_id=captured["task_id"], - session_id=captured["session_id"], - action="decline", - content=None, - fastmcp=mcp, - ) - if success: - break - await asyncio.sleep(0.05) - - assert success is True - await task.wait(timeout=10.0) result = await task.result() assert result.data == "User declined" async def test_elicit_with_pydantic_model(self): - """E2E: tool elicits structured Pydantic input, data round-trips correctly.""" + """E2E: tool elicits structured Pydantic input via elicitation_handler.""" from pydantic import BaseModel class UserInfo(BaseModel): @@ -330,43 +278,20 @@ class TestBackgroundTaskIntegration: age: int mcp = FastMCP("elicit-pydantic-test") - elicit_started = asyncio.Event() - captured: dict[str, str | None] = {"task_id": None, "session_id": None} @mcp.tool(task=True) async def get_user_info(ctx: Context) -> str: - captured["task_id"] = ctx.task_id - captured["session_id"] = ctx.session_id - elicit_started.set() - result = await ctx.elicit("Provide user info", UserInfo) if isinstance(result, AcceptedElicitation): assert isinstance(result.data, UserInfo) return f"{result.data.name} is {result.data.age}" return "No info" - async with Client(mcp) as client: + async def handler(message, response_type, params, ctx): + return ElicitResult(action="accept", content={"name": "Alice", "age": 30}) + + async with Client(mcp, elicitation_handler=handler) as client: task = await client.call_tool("get_user_info", {}, task=True) - await asyncio.wait_for(elicit_started.wait(), timeout=5.0) - - assert captured["task_id"] is not None - assert captured["session_id"] is not None - - success = False - for _ in range(40): - success = await handle_task_input( - task_id=captured["task_id"], - session_id=captured["session_id"], - action="accept", - content={"name": "Alice", "age": 30}, - fastmcp=mcp, - ) - if success: - break - await asyncio.sleep(0.05) - - assert success is True - await task.wait(timeout=10.0) result = await task.result() assert result.data == "Alice is 30" diff --git a/tests/server/tasks/test_notifications.py b/tests/server/tasks/test_notifications.py index 07b2b7f3b..f0e144fd4 100644 --- a/tests/server/tasks/test_notifications.py +++ b/tests/server/tasks/test_notifications.py @@ -7,14 +7,14 @@ No mocking of Redis, sessions, or Docket internals. import asyncio -import mcp.types +import mcp.types as mcp_types from fastmcp import FastMCP from fastmcp.client import Client +from fastmcp.client.elicitation import ElicitResult from fastmcp.client.messages import MessageHandler from fastmcp.server.context import Context from fastmcp.server.elicitation import AcceptedElicitation -from fastmcp.server.tasks.elicitation import handle_task_input from fastmcp.server.tasks.notifications import ( get_subscriber_count, ) @@ -25,12 +25,12 @@ class NotificationCaptureHandler(MessageHandler): def __init__(self) -> None: super().__init__() - self.notifications: list[mcp.types.ServerNotification] = [] + self.notifications: list[mcp_types.ServerNotification] = [] - async def on_notification(self, message: mcp.types.ServerNotification) -> None: + async def on_notification(self, message: mcp_types.ServerNotification) -> None: self.notifications.append(message) - def for_method(self, method: str) -> list[mcp.types.ServerNotification]: + def for_method(self, method: str) -> list[mcp_types.ServerNotification]: return [ notification for notification in self.notifications @@ -41,66 +41,68 @@ class NotificationCaptureHandler(MessageHandler): class TestNotificationIntegration: """Integration tests for the notification queue using real Docket memory backend. - The elicitation flow implicitly validates the full notification pipeline: - 1. Tool calls ctx.elicit() → stores request in Redis → pushes notification - 2. Subscriber picks up notification → sends MCP notification to client - 3. Client calls handle_task_input() → LPUSH response → BLPOP wakes tool + The elicitation flow validates the full notification pipeline: + 1. Tool calls ctx.elicit() -> stores request in Redis -> pushes notification + 2. Subscriber picks up notification -> sends MCP notification to client + 3. Subscriber relays elicitation/create to client -> handler responds + 4. Relay pushes response to Redis -> BLPOP wakes tool """ async def test_notification_delivered_during_elicitation(self): - """Full E2E: notification queue delivers input_required metadata to client.""" + """Full E2E: notification queue delivers input_required metadata to client. + + The elicitation relay handles the response via the client's + elicitation_handler. We verify both the notification metadata + structure and the end-to-end elicitation flow. + """ mcp = FastMCP("notification-test") notification_handler = NotificationCaptureHandler() - elicit_started = asyncio.Event() - captured: dict[str, str | None] = {"task_id": None, "session_id": None} @mcp.tool(task=True) async def elicit_tool(ctx: Context) -> str: - captured["task_id"] = ctx.task_id - captured["session_id"] = ctx.session_id - elicit_started.set() - result = await ctx.elicit("Enter value", str) if isinstance(result, AcceptedElicitation): return f"got: {result.data}" return "no value" - async with Client(mcp, message_handler=notification_handler) as client: + async def elicitation_handler(message, response_type, params, ctx): + return ElicitResult(action="accept", content={"value": "hello"}) + + async with Client( + mcp, + message_handler=notification_handler, + elicitation_handler=elicitation_handler, + ) as client: task = await client.call_tool("elicit_tool", {}, task=True) - await asyncio.wait_for(elicit_started.wait(), timeout=5.0) - assert captured["task_id"] is not None - assert captured["session_id"] is not None + await task.wait(timeout=10.0) + result = await task.result() + assert result.data == "got: hello" - notification: mcp.types.ServerNotification | None = None - for _ in range(40): - candidates = notification_handler.for_method( - "notifications/tasks/status" + # Verify the input_required notification was delivered with metadata + notification: mcp_types.ServerNotification | None = None + candidates = notification_handler.for_method("notifications/tasks/status") + for candidate in reversed(candidates): + candidate_meta = getattr(candidate.root, "_meta", None) + related_task = ( + candidate_meta.get("io.modelcontextprotocol/related-task") + if isinstance(candidate_meta, dict) + else None ) - for candidate in reversed(candidates): - candidate_meta = getattr(candidate.root, "_meta", None) - related_task = ( - candidate_meta.get("modelcontextprotocol.io/related-task") - if isinstance(candidate_meta, dict) - else None - ) - if ( - isinstance(related_task, dict) - and related_task.get("status") == "input_required" - ): - notification = candidate - break - if notification is not None: + if ( + isinstance(related_task, dict) + and related_task.get("status") == "input_required" + ): + notification = candidate break - await asyncio.sleep(0.05) assert notification is not None, "expected notifications/tasks/status" 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") == captured["task_id"] + assert related_task.get("taskId") == task.task_id assert related_task.get("status") == "input_required" elicitation = related_task.get("elicitation") @@ -109,25 +111,6 @@ class TestNotificationIntegration: assert isinstance(elicitation.get("requestId"), str) assert isinstance(elicitation.get("requestedSchema"), dict) - success = False - for _ in range(40): - success = await handle_task_input( - task_id=captured["task_id"], - session_id=captured["session_id"], - action="accept", - content={"value": "hello"}, - fastmcp=mcp, - ) - if success: - break - await asyncio.sleep(0.05) - - assert success is True - - await task.wait(timeout=10.0) - result = await task.result() - assert result.data == "got: hello" - async def test_subscriber_started_and_cleaned_up(self): """Subscriber starts during background task and stops when client disconnects.""" mcp = FastMCP("subscriber-test") diff --git a/tests/server/tasks/test_task_elicitation_relay.py b/tests/server/tasks/test_task_elicitation_relay.py new file mode 100644 index 000000000..42362edd2 --- /dev/null +++ b/tests/server/tasks/test_task_elicitation_relay.py @@ -0,0 +1,191 @@ +"""Tests for background task elicitation relay (notifications.py). + +The relay bridges distributed background tasks to clients via the standard +MCP elicitation/create protocol. When a worker calls ctx.elicit(), the +notification subscriber detects the input_required notification and sends +an elicitation/create request to the client session. The client's +elicitation_handler fires, and the relay pushes the response to Redis +for the blocked worker. + +These tests use Client(mcp) with the real memory:// Docket backend. +""" + +import asyncio +from dataclasses import dataclass + +from pydantic import BaseModel + +from fastmcp import FastMCP +from fastmcp.client import Client +from fastmcp.client.elicitation import ElicitResult +from fastmcp.server.context import Context +from fastmcp.server.elicitation import ( + AcceptedElicitation, + CancelledElicitation, + DeclinedElicitation, +) + + +class TestElicitationRelay: + """E2E tests for elicitation flowing through the standard MCP protocol.""" + + async def test_accept_via_elicitation_handler(self): + """Tool elicits, client handler accepts, tool gets the value.""" + mcp = FastMCP("relay-accept") + + @mcp.tool(task=True) + async def ask_name(ctx: Context) -> str: + result = await ctx.elicit("What is your name?", str) + if isinstance(result, AcceptedElicitation): + return f"Hello, {result.data}!" + return "No name" + + async def handler(message, response_type, params, ctx): + assert message == "What is your name?" + return ElicitResult(action="accept", content={"value": "Alice"}) + + async with Client(mcp, elicitation_handler=handler) as client: + task = await client.call_tool("ask_name", {}, task=True) + result = await task.result() + assert result.data == "Hello, Alice!" + + async def test_decline_via_elicitation_handler(self): + """Tool elicits, client handler declines, tool gets DeclinedElicitation.""" + mcp = FastMCP("relay-decline") + + @mcp.tool(task=True) + async def optional_input(ctx: Context) -> str: + result = await ctx.elicit("Provide a name?", str) + if isinstance(result, DeclinedElicitation): + return "User declined" + if isinstance(result, AcceptedElicitation): + return f"Got: {result.data}" + return "Cancelled" + + async def handler(message, response_type, params, ctx): + return ElicitResult(action="decline") + + async with Client(mcp, elicitation_handler=handler) as client: + task = await client.call_tool("optional_input", {}, task=True) + result = await task.result() + assert result.data == "User declined" + + async def test_cancel_via_elicitation_handler(self): + """Tool elicits, client handler cancels, tool gets CancelledElicitation.""" + mcp = FastMCP("relay-cancel") + + @mcp.tool(task=True) + async def cancellable(ctx: Context) -> str: + result = await ctx.elicit("Input?", str) + if isinstance(result, CancelledElicitation): + return "Cancelled" + return "Not cancelled" + + async def handler(message, response_type, params, ctx): + return ElicitResult(action="cancel") + + async with Client(mcp, elicitation_handler=handler) as client: + task = await client.call_tool("cancellable", {}, task=True) + result = await task.result() + assert result.data == "Cancelled" + + async def test_dataclass_round_trips_through_relay(self): + """Structured dataclass type round-trips through the relay.""" + mcp = FastMCP("relay-dataclass") + + @dataclass + class UserInfo: + name: str + age: int + + @mcp.tool(task=True) + async def get_user(ctx: Context) -> str: + result = await ctx.elicit("Provide user info", UserInfo) + if isinstance(result, AcceptedElicitation): + assert isinstance(result.data, UserInfo) + return f"{result.data.name} is {result.data.age}" + return "No info" + + async def handler(message, response_type, params, ctx): + return ElicitResult(action="accept", content={"name": "Bob", "age": 30}) + + async with Client(mcp, elicitation_handler=handler) as client: + task = await client.call_tool("get_user", {}, task=True) + result = await task.result() + assert result.data == "Bob is 30" + + async def test_pydantic_model_round_trips_through_relay(self): + """Structured Pydantic model round-trips through the relay.""" + mcp = FastMCP("relay-pydantic") + + class Config(BaseModel): + host: str + port: int + + @mcp.tool(task=True) + async def get_config(ctx: Context) -> str: + result = await ctx.elicit("Server config?", Config) + if isinstance(result, AcceptedElicitation): + assert isinstance(result.data, Config) + return f"{result.data.host}:{result.data.port}" + return "No config" + + async def handler(message, response_type, params, ctx): + return ElicitResult( + action="accept", content={"host": "localhost", "port": 8080} + ) + + async with Client(mcp, elicitation_handler=handler) as client: + task = await client.call_tool("get_config", {}, task=True) + result = await task.result() + assert result.data == "localhost:8080" + + async def test_multiple_sequential_elicitations(self): + """Tool calls ctx.elicit() twice, both go through the relay.""" + mcp = FastMCP("relay-multi") + + @mcp.tool(task=True) + async def two_questions(ctx: Context) -> str: + r1 = await ctx.elicit("First name?", str) + r2 = await ctx.elicit("Last name?", str) + if isinstance(r1, AcceptedElicitation) and isinstance( + r2, AcceptedElicitation + ): + return f"{r1.data} {r2.data}" + return "Incomplete" + + call_count = 0 + + async def handler(message, response_type, params, ctx): + nonlocal call_count + call_count += 1 + if call_count == 1: + assert message == "First name?" + return ElicitResult(action="accept", content={"value": "Jane"}) + else: + assert message == "Last name?" + return ElicitResult(action="accept", content={"value": "Doe"}) + + async with Client(mcp, elicitation_handler=handler) as client: + task = await client.call_tool("two_questions", {}, task=True) + result = await task.result() + assert result.data == "Jane Doe" + assert call_count == 2 + + async def test_no_elicitation_handler_returns_cancel(self): + """Without an elicitation_handler, the relay fails and task gets cancel.""" + mcp = FastMCP("relay-no-handler") + + @mcp.tool(task=True) + async def needs_input(ctx: Context) -> str: + result = await ctx.elicit("Input?", str) + if isinstance(result, CancelledElicitation): + return "Cancelled as expected" + if isinstance(result, AcceptedElicitation): + return f"Got: {result.data}" + return "Other" + + async with Client(mcp) as client: + task = await client.call_tool("needs_input", {}, task=True) + result = await asyncio.wait_for(task.result(), timeout=15.0) + assert result.data == "Cancelled as expected" 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()