From 7699deb99c31b40004cd4e10dafcd70e0848806a Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 15:00:08 -0400 Subject: [PATCH 1/9] Pass the MCP conformance suite's draft and pending scenarios Pin the suite, build out the fixture, and fix the protocol gaps it found. --- fastmcp_slim/fastmcp/prompts/base.py | 39 ++ .../fastmcp/server/mixins/mcp_operations.py | 28 +- fastmcp_slim/fastmcp/server/server.py | 19 + fastmcp_tasks/fastmcp_tasks/extension.py | 48 +- fastmcp_tasks/fastmcp_tasks/handlers.py | 55 +- fastmcp_tasks/fastmcp_tasks/input_loop.py | 10 + fastmcp_tasks/fastmcp_tasks/input_store.py | 60 ++- fastmcp_tasks/fastmcp_tasks/models.py | 11 +- tests/conformance/expected-failures.yml | 23 +- tests/conformance/server.py | 498 +++++++++++++++++- tests/conformance/test_conformance.py | 25 +- tests/server/test_mrtr_guards.py | 58 ++ tests/tasks/client/test_client_tool_tasks.py | 4 +- tests/tasks/server/test_extension.py | 19 +- tests/tasks/server/test_guard_reentrant.py | 78 +++ tests/tasks/server/test_task_config.py | 2 +- tests/test_exceptions.py | 57 ++ 17 files changed, 979 insertions(+), 55 deletions(-) diff --git a/fastmcp_slim/fastmcp/prompts/base.py b/fastmcp_slim/fastmcp/prompts/base.py index d23ab0e0f..b7e4fa3f6 100644 --- a/fastmcp_slim/fastmcp/prompts/base.py +++ b/fastmcp_slim/fastmcp/prompts/base.py @@ -10,6 +10,7 @@ import pydantic_core if TYPE_CHECKING: from fastmcp.prompts.function_prompt import FunctionPrompt +import mcp_types from mcp import GetPromptResult from mcp_types import ( AudioContent, @@ -188,6 +189,38 @@ class PromptResult(pydantic.BaseModel): ) +class InputRequiredPromptResult(PromptResult): + """The full result of a single multi-round-trip prompt leg (SEP-2322). + + `InputRequiredResult` is a result type, not a `tools/call` feature: any + request may resolve to one. When a prompt returns an `InputRequiredResult` + from its body to ask the client for input, that ask is the legitimate + result of this `prompts/get` — so FastMCP wraps it in this `PromptResult` + subclass, mirroring `InputRequiredToolResult`, and it flows through the + middleware chain as an ordinary return value. + + Invariant: the wrapped `InputRequiredResult` is never rendered as prompt + messages. `messages` is always empty; the wire handler (`_on_get_prompt`) + reads `.input_required` and returns it to the runner. + """ + + input_required: mcp_types.InputRequiredResult = Field( + description="The client-input request this leg resolved to (SEP-2322)" + ) + + def __init__(self, input_required: mcp_types.InputRequiredResult) -> None: + # Bypass PromptResult's message-normalizing __init__: an input-required + # leg carries no messages (see the invariant above), and + # `input_required` is a required field PromptResult.__init__ can't set. + pydantic.BaseModel.__init__( + self, + messages=[], + description=None, + meta=None, + input_required=input_required, + ) + + class Prompt(FastMCPComponent): """A prompt template that can be rendered with parameters.""" @@ -287,6 +320,12 @@ class Prompt(FastMCPComponent): if isinstance(raw_value, PromptResult): return raw_value + if isinstance(raw_value, mcp_types.InputRequiredResult): + # The prompt asked the client for input (SEP-2322). Wrap it so the + # ask travels the middleware chain as an ordinary result; the wire + # handler unwraps it. + return InputRequiredPromptResult(raw_value) + if isinstance(raw_value, str): return PromptResult(raw_value, description=self.description, meta=self.meta) diff --git a/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py b/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py index af85dd08c..642e28879 100644 --- a/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py +++ b/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py @@ -28,6 +28,7 @@ from fastmcp.exceptions import ( NotFoundError, to_mcp_error, ) +from fastmcp.prompts.base import InputRequiredPromptResult from fastmcp.server.completions import CompletionValues, normalize_completion from fastmcp.server.dependencies import bind_request_context, extract_version_spec from fastmcp.tools.base import InputRequiredToolResult, ToolResult @@ -306,8 +307,12 @@ class MCPOperationsMixin: try: result = await self.read_resource(str(uri), version=version) except (DisabledError, NotFoundError) as e: - raise to_mcp_error( - NotFoundError(f"Resource not found: {str(uri)!r}") + # SEP-2164: echo the requested URI in `data` so a client that + # pipelined several reads can tell which one is missing. + raise MCPError( + code=INVALID_PARAMS, + message=f"Resource not found: {str(uri)!r}", + data={"uri": str(uri)}, ) from e except FastMCPError as e: # Resource-visible errors (ResourceError, ValidationError, ...) @@ -326,7 +331,7 @@ class MCPOperationsMixin: self: FastMCP, ctx: ServerRequestContext, params: GetPromptRequestParams, - ) -> mcp_types.GetPromptResult: + ) -> mcp_types.GetPromptResult | mcp_types.InputRequiredResult: """Handle MCP 'prompts/get' requests.""" with bind_request_context(ctx): name = params.name @@ -351,6 +356,23 @@ class MCPOperationsMixin: # Masking already happened inside render_prompt. raise to_mcp_error(e) from e + if isinstance(result, InputRequiredPromptResult): + # The prompt requested client input (SEP-2322). As with tools, + # the multi-round-trip result type only exists at 2026-07-28, so + # name the era problem on an older connection rather than + # failing as a generic "invalid result". + if ctx.protocol_version not in MODERN_PROTOCOL_VERSIONS: + raise MCPError( + code=INVALID_PARAMS, + message=( + f"Prompt {name!r} returned an InputRequiredResult to " + "request client input, but the multi-round-trip result " + "type (SEP-2322) only exists at MCP 2026-07-28; this " + f"connection negotiated {ctx.protocol_version!r}." + ), + ) + return result.input_required + return result.to_mcp_prompt_result() async def _on_set_logging_level( diff --git a/fastmcp_slim/fastmcp/server/server.py b/fastmcp_slim/fastmcp/server/server.py index 15e03c5f2..51c0a476c 100644 --- a/fastmcp_slim/fastmcp/server/server.py +++ b/fastmcp_slim/fastmcp/server/server.py @@ -31,6 +31,7 @@ from mcp_types import ( CallToolRequestParams, ToolAnnotations, ) +from mcp_types.jsonrpc import MISSING_REQUIRED_CLIENT_CAPABILITY from pydantic import AnyUrl from pydantic import ValidationError as PydanticValidationError from starlette.routing import BaseRoute @@ -1484,6 +1485,24 @@ class FastMCP( ) raise except Exception as e: + # Most MCPErrors raised under a tool describe how the call + # went — a timeout, an upstream error a proxy forwarded — + # and are masked into an `isError` result like any other + # failure. A missing-client-capability error is different: + # it says the request cannot be serviced at all, and + # SEP-2575 requires it on the wire as -32021 (HTTP 400). + # Flattening it into a result would drop the code and tell + # the client the call had succeeded. + if ( + isinstance(e, MCPError) + and e.error.code == MISSING_REQUIRED_CLIENT_CAPABILITY + ): + logger.debug( + "Tool %r requires a client capability the client did " + "not declare", + name, + ) + raise logger.exception(f"Error calling tool {name!r}") # Handle actionable errors that should reach the LLM # even when masking is enabled diff --git a/fastmcp_tasks/fastmcp_tasks/extension.py b/fastmcp_tasks/fastmcp_tasks/extension.py index bc34daaa8..7ba4f123d 100644 --- a/fastmcp_tasks/fastmcp_tasks/extension.py +++ b/fastmcp_tasks/fastmcp_tasks/extension.py @@ -32,10 +32,12 @@ from typing import TYPE_CHECKING, Any from mcp.server.context import ServerRequestContext from mcp.shared.exceptions import MCPError +from mcp.shared.inbound import MCP_NAME_HEADER, decode_header_value +from mcp_types.jsonrpc import HEADER_MISMATCH from mcp_types.version import MODERN_PROTOCOL_VERSIONS from fastmcp.exceptions import NotFoundError -from fastmcp.server.dependencies import extract_version_spec +from fastmcp.server.dependencies import extract_version_spec, get_http_request from fastmcp.server.extensions import ( MethodBinding, ServerExtension, @@ -140,7 +142,7 @@ class TasksExtension(ServerExtension): """Reject a task method from a client that did not declare the extension. SEP-2663: a client issuing `tasks/get`/`tasks/update`/`tasks/cancel` - without the tasks capability in the request's `_meta` gets -32003. A + without the tasks capability in the request's `_meta` gets -32021. A client normally only holds a taskId because it declared the capability on the creating `tools/call`, but the method-level check is an explicit MUST, so enforce it here rather than assume. @@ -156,22 +158,56 @@ class TasksExtension(ServerExtension): data=missing_capability_error_data(), ) + def _require_matching_task_route(self, task_id: str) -> None: + """Reject a task method whose `Mcp-Name` header disagrees with its body. + + SEP-2243 mirrors a request's name-shaped field into `Mcp-Name` so + intermediaries can route without parsing the body, and requires servers + that read the body to check the two agree. SEP-2663 extends that to the + tasks namespace, where the name-shaped field is `taskId`. The core SDK's + pre-dispatch ladder only knows the base protocol's name-bearing methods, + so the extension enforces its own. + """ + try: + request = get_http_request() + except RuntimeError: + # Not an HTTP transport, so there are no routing headers to check. + return + header = request.headers.get(MCP_NAME_HEADER) + if header is None: + return + if decode_header_value(header) != task_id: + raise MCPError( + code=HEADER_MISMATCH, + message=( + f"{MCP_NAME_HEADER} header does not match the request body's " + "'taskId' parameter" + ), + ) + + def _check_task_request( + self, ctx: ServerRequestContext[Any, Any], task_id: str + ) -> None: + """Run both gates every `tasks/*` method shares.""" + self._require_tasks_capability(ctx) + self._require_matching_task_route(task_id) + async def _handle_get( self, ctx: ServerRequestContext[Any, Any], params: GetTaskParams ) -> GetTaskResult: - self._require_tasks_capability(ctx) + self._check_task_request(ctx, params.task_id) return await tasks_get(self.server, params.task_id) async def _handle_update( self, ctx: ServerRequestContext[Any, Any], params: UpdateTaskParams ) -> UpdateTaskResult: - self._require_tasks_capability(ctx) + self._check_task_request(ctx, params.task_id) return await tasks_update(self.server, params.task_id, params.input_responses) async def _handle_cancel( self, ctx: ServerRequestContext[Any, Any], params: CancelTaskParams ) -> CancelTaskResult: - self._require_tasks_capability(ctx) + self._check_task_request(ctx, params.task_id) return await tasks_cancel(self.server, params.task_id) async def intercept_tool_call( @@ -183,7 +219,7 @@ class TasksExtension(ServerExtension): """Decide whether to run this ``tools/call`` as a task. Consults the tool's ``TaskConfig`` mode and the client's per-request - opt-in: ``required`` always tasks (raising -32003 if the client did not + opt-in: ``required`` always tasks (raising -32021 if the client did not opt in), ``optional`` tasks only when the client opted in, ``forbidden`` never tasks. A non-task call passes straight through to the tool body. """ diff --git a/fastmcp_tasks/fastmcp_tasks/handlers.py b/fastmcp_tasks/fastmcp_tasks/handlers.py index b12aabf19..526416474 100644 --- a/fastmcp_tasks/fastmcp_tasks/handlers.py +++ b/fastmcp_tasks/fastmcp_tasks/handlers.py @@ -42,6 +42,7 @@ from fastmcp_tasks.input_store import ( acquire_update_lock, acquire_update_lock_blocking, clear_outstanding, + discard_outstanding, is_cancelled, load_current_leg, load_task_args, @@ -296,18 +297,26 @@ async def tasks_get(server: FastMCP, task_id: str) -> GetTaskResult: if execution.state == ExecutionState.FAILED: message = "Task failed" + error: dict[str, Any] = { + "code": mcp_types.INTERNAL_ERROR, + "message": message, + } try: await execution.get_result(timeout=timedelta(seconds=0)) # On a FAILED execution, get_result re-raises the exception the task # itself raised — an arbitrary user-defined type, so no narrower catch - # exists. Its message becomes the task's error payload. - except Exception as error: - message = str(error) - return build( - "failed", - status_message=message, - error={"code": mcp_types.INTERNAL_ERROR, "message": message}, - ) + # exists. Its message becomes the task's error payload; an MCPError + # already *is* a JSON-RPC error, so its code and data are preserved + # rather than flattened to an internal error. + except MCPError as protocol_error: + message = protocol_error.error.message + error = {"code": protocol_error.error.code, "message": message} + if protocol_error.error.data is not None: + error["data"] = protocol_error.error.data + except Exception as unexpected: + message = str(unexpected) + error = {"code": mcp_types.INTERNAL_ERROR, "message": message} + return build("failed", status_message=message, error=error) if execution.state == ExecutionState.CANCELLED: return build("cancelled") @@ -352,21 +361,37 @@ async def tasks_update( if await is_cancelled(docket, task_scope, task_id): return UpdateTaskResult() - translated = await translate_responses( + matched = await translate_responses( docket, task_scope, task_id, leg_number, input_responses ) - if translated is None: + if matched is None: # Nothing matched the current leg's outstanding requests: the leg was # already answered, or the keys are unknown. Idempotent no-op. return UpdateTaskResult() + translated, answered_keys = matched - # Store the answers for the next leg to read, then enqueue that leg. - # Ordering matters: the answers must be in Redis before the next leg's - # worker context loads them, and current_leg must not advance to an - # execution that is not yet durable — so enqueue (with its durable wait) - # precedes the pointer swap. + # Store the answers for the next leg to read. They accumulate: a client + # may answer a multi-request ask one update at a time. await store_input_responses(docket, task_scope, task_id, translated) + # Retire only what this update answered. While anything is still + # outstanding the task stays `input_required` and `tasks/get` surfaces + # the remaining keys — the leg re-enters only once every request has an + # answer (SEP-2663 partial fulfillment). + await discard_outstanding( + docket, task_scope, task_id, leg_number, answered_keys + ) + still_pending = await read_outstanding_inputs( + docket, task_scope, task_id, leg_number + ) + if still_pending: + return UpdateTaskResult() + + # Every request is answered, so enqueue the next leg. Ordering matters: + # the answers must be in Redis before the next leg's worker context + # loads them, and current_leg must not advance to an execution that is + # not yet durable — so enqueue (with its durable wait) precedes the + # pointer swap. component = await registered_component_for_key( server, parse_task_key(base_task_key)["component_identifier"] ) diff --git a/fastmcp_tasks/fastmcp_tasks/input_loop.py b/fastmcp_tasks/fastmcp_tasks/input_loop.py index 07145d5bf..f1f146c09 100644 --- a/fastmcp_tasks/fastmcp_tasks/input_loop.py +++ b/fastmcp_tasks/fastmcp_tasks/input_loop.py @@ -31,6 +31,7 @@ import logging from typing import TYPE_CHECKING, Any import mcp_types +from mcp.shared.exceptions import MCPError from fastmcp.exceptions import FastMCPError from fastmcp.tools.base import InputRequiredToolResult, ToolResult @@ -142,6 +143,15 @@ def reentrant_task_fn( result = await fn(*args, **kwargs) except FastMCPError as exc: return _error_result(tool_name, exc) + except MCPError: + # A protocol fault, not a tool error. SEP-2663 reserves `failed` + # for exactly this, so it must escape the wrapper: the Docket + # execution fails and `tasks/get` inlines the JSON-RPC error + # instead of reporting a completed task with an `isError` result. + logger.exception( + "background task tool %r raised a protocol error", tool_name + ) + raise except Exception as exc: logger.exception("background task tool %r raised", tool_name) return _error_result(tool_name, exc) diff --git a/fastmcp_tasks/fastmcp_tasks/input_store.py b/fastmcp_tasks/fastmcp_tasks/input_store.py index d13a6e82e..9dab545ea 100644 --- a/fastmcp_tasks/fastmcp_tasks/input_store.py +++ b/fastmcp_tasks/fastmcp_tasks/input_store.py @@ -37,6 +37,8 @@ import mcp_types from fastmcp_tasks.keys import task_redis_prefix if TYPE_CHECKING: + from collections.abc import Iterable + from docket import Docket logger = logging.getLogger(__name__) @@ -250,6 +252,11 @@ async def store_outstanding( await redis.hset(map_key, surfaced, tool_key) await redis.expire(requests_key, ttl_seconds) await redis.expire(map_key, ttl_seconds) + # The answers that drove this leg have been consumed by the body that + # just parked, so drop them: responses accumulate per leg (a client may + # answer a multi-request ask one key at a time), and a stale carry-over + # would make the next leg look already-answered. + await redis.delete(_input_responses_key(docket, task_scope, task_id)) if request_state is not None: await redis.set(state_key, request_state, ex=ttl_seconds) else: @@ -306,7 +313,7 @@ async def translate_responses( task_id: str, leg: int, responses: dict[str, Any], -) -> dict[str, mcp_types.Result] | None: +) -> tuple[dict[str, mcp_types.Result], list[str]] | None: """Translate a ``tasks/update`` payload into typed, tool-keyed responses. ``responses`` is keyed by the surfaced keys the client received for ``leg``. @@ -314,6 +321,10 @@ async def translate_responses( answer is validated into the result type its request maps to and re-keyed to the tool's own request key. Returns ``None`` when nothing matched, so the caller can treat a stale or empty update as an idempotent no-op. + + Returns the tool-keyed answers alongside the surfaced keys they came from, + so the caller can retire exactly the answered requests and leave the rest + outstanding. """ outstanding = await read_outstanding_inputs(docket, task_scope, task_id, leg) if not outstanding: @@ -321,6 +332,7 @@ async def translate_responses( mapping = await _read_outstanding_map(docket, task_scope, task_id, leg) translated: dict[str, mcp_types.Result] = {} + matched: list[str] = [] for surfaced_key, raw in responses.items(): payload = outstanding.get(surfaced_key) if payload is None: @@ -331,8 +343,11 @@ async def translate_responses( method = payload.get("method", "elicitation/create") result_type = result_type_for_method(method) translated[tool_key] = result_type.model_validate(raw) + matched.append(surfaced_key) - return translated or None + if not translated: + return None + return translated, matched async def store_input_responses( @@ -347,6 +362,11 @@ async def store_input_responses( The responses are stored typed-but-serialized (``{"type", "data"}``) so the next leg's context factory reconstructs real result objects keyed by the tool's own request keys. + + Answers merge into whatever the leg has already collected: a client may + answer a multi-request ask one `tasks/update` at a time, and the leg only + re-enters once every request has been answered. Callers hold the per-task + update lock, so the read-modify-write cannot interleave. """ stored = { tool_key: { @@ -355,12 +375,38 @@ async def store_input_responses( } for tool_key, result in translated.items() } + responses_key = _input_responses_key(docket, task_scope, task_id) async with docket.redis() as redis: - await redis.set( - _input_responses_key(docket, task_scope, task_id), - json.dumps(stored), - ex=ttl_seconds, - ) + existing_raw = _decode(await redis.get(responses_key)) + if existing_raw: + try: + existing = json.loads(existing_raw) + except json.JSONDecodeError: + existing = {} + if isinstance(existing, dict): + stored = {**existing, **stored} + await redis.set(responses_key, json.dumps(stored), ex=ttl_seconds) + + +async def discard_outstanding( + docket: Docket, + task_scope: str | None, + task_id: str, + leg: int, + surfaced_keys: Iterable[str], +) -> None: + """Drop just the surfaced keys an update answered, keeping the rest pending. + + Partial fulfillment (SEP-2663): a leg that asked several questions stays + ``input_required`` until all are answered, and each ``tasks/get`` in between + must surface only the still-unanswered keys. + """ + keys = list(surfaced_keys) + if not keys: + return + async with docket.redis() as redis: + await redis.hdel(_requests_key(docket, task_scope, task_id, leg), *keys) + await redis.hdel(_map_key(docket, task_scope, task_id, leg), *keys) async def clear_outstanding( diff --git a/fastmcp_tasks/fastmcp_tasks/models.py b/fastmcp_tasks/fastmcp_tasks/models.py index 761b7ebf4..a58fde3a1 100644 --- a/fastmcp_tasks/fastmcp_tasks/models.py +++ b/fastmcp_tasks/fastmcp_tasks/models.py @@ -22,6 +22,9 @@ from __future__ import annotations from typing import Any, Literal from mcp_types import RequestParams, Result +from mcp_types.jsonrpc import ( + MISSING_REQUIRED_CLIENT_CAPABILITY as _MISSING_REQUIRED_CLIENT_CAPABILITY, +) from pydantic import BaseModel, ConfigDict, Field __all__ = [ @@ -42,8 +45,10 @@ __all__ = [ #: JSON-RPC error code for "Missing Required Client Capability" (SEP-2663). A #: tool whose task mode is `required` returns this when the client did not opt -#: the tasks extension in for the request. -MISSING_REQUIRED_CLIENT_CAPABILITY = -32003 +#: the tasks extension in for the request, as do the `tasks/*` methods when the +#: client never negotiated the extension. Re-exported from the SDK so the code +#: tracks the protocol rather than an early draft's number. +MISSING_REQUIRED_CLIENT_CAPABILITY = _MISSING_REQUIRED_CLIENT_CAPABILITY TaskStatus = Literal["working", "input_required", "completed", "failed", "cancelled"] @@ -186,7 +191,7 @@ class CancelTaskRequest(BaseModel): def missing_capability_error_data() -> dict[str, Any]: - """Build the `data.requiredCapabilities` payload for a -32003 error. + """Build the `data.requiredCapabilities` payload for a -32021 error. A `required`-mode tool called without the client opting the tasks extension in for the request returns this so the client learns which capability to diff --git a/tests/conformance/expected-failures.yml b/tests/conformance/expected-failures.yml index 46b2081de..9e3ed9dde 100644 --- a/tests/conformance/expected-failures.yml +++ b/tests/conformance/expected-failures.yml @@ -1,6 +1,23 @@ +# Scenarios the conformance suite runs that FastMCP does not pass. +# +# This is a baseline, not a to-do list: every entry needs a reason, and anything +# that is merely unimplemented in the *fixture* belongs in server.py instead. +# The suite is run with `--suite all`, so draft and pending scenarios count too. + server: - - completion-complete - - server-sse-polling + # Resource subscriptions (resources/subscribe, resources/unsubscribe) are not + # implemented. The server correctly advertises `resources.subscribe: false`, + # but the suite calls the methods regardless of the declared capability. Both + # scenarios were removed in MCP 2026-07-28, the version FastMCP targets, so + # this affects handshake-era clients only. - resources-subscribe - resources-unsubscribe - - dns-rebinding-protection + + # SEP-2663 MRTR-to-tasks composition: a task-supporting guard tool is + # expected to gather its input over foreground multi-round-trip rounds and + # only mint the task on the final round. FastMCP instead creates the task up + # front and parks it at `input_required`, answered through `tasks/update` — + # the model the `tasks-mrtr-input` scenario exercises. Supporting both would + # need the tool to declare which one it wants, which is an unmade API + # decision rather than a bug. + - tasks-mrtr-composition diff --git a/tests/conformance/server.py b/tests/conformance/server.py index 9a6a97c28..ed91ee289 100644 --- a/tests/conformance/server.py +++ b/tests/conformance/server.py @@ -9,17 +9,33 @@ import base64 import json import sys from enum import Enum as PyEnum +from typing import Annotated import mcp_types -from mcp_types import EmbeddedResource, ImageContent, TextContent +import uvicorn +from mcp.shared.exceptions import MCPError +from mcp_types import ( + ClientCapabilities, + Completion, + EmbeddedResource, + ImageContent, + MissingRequiredClientCapabilityErrorData, + PromptReference, + TextContent, +) +from mcp_types.jsonrpc import MISSING_REQUIRED_CLIENT_CAPABILITY from pydantic import BaseModel, Field from fastmcp import FastMCP from fastmcp.exceptions import ToolError from fastmcp.prompts import Message +from fastmcp.server.completions import CompletionValues from fastmcp.server.context import Context +from fastmcp.server.event_store import EventStore from fastmcp.tools.function_tool import FunctionTool +from fastmcp.utilities.tasks import TaskConfig from fastmcp.utilities.types import Audio, Image +from fastmcp_tasks import TasksExtension # Minimal 1x1 red PNG for image tests (89 bytes) _1X1_PNG = base64.b64decode( @@ -47,6 +63,29 @@ _SILENT_WAV = ( server = FastMCP("conformance-test-server", dereference_schemas=False) +def require_client_capability(ctx: Context, capability: str) -> None: + """Raise `-32021` unless the client declared *capability* on this request. + + SEP-2575 makes capability negotiation per-request: the client repeats its + capabilities in each request's `_meta`, and a server that needs one the + client did not declare must answer with a + `MissingRequiredClientCapabilityError` whose `data.requiredCapabilities` is + a `ClientCapabilities` object keyed by the missing capability. + """ + client_params = ctx.session.client_params + declared = client_params.capabilities if client_params else None + if declared is not None and getattr(declared, capability, None) is not None: + return + data = MissingRequiredClientCapabilityErrorData( + required_capabilities=ClientCapabilities.model_validate({capability: {}}) + ) + raise MCPError( + code=MISSING_REQUIRED_CLIENT_CAPABILITY, + message=f"Client did not declare the required {capability!r} capability", + data=data.model_dump(by_alias=True, mode="json", exclude_none=True), + ) + + # --------------------------------------------------------------------------- # Tools # --------------------------------------------------------------------------- @@ -261,6 +300,7 @@ server.add_tool( "type": "object", "$defs": { "address": { + "$anchor": "address", "type": "object", "properties": { "street": {"type": "string"}, @@ -272,12 +312,423 @@ server.add_tool( "name": {"type": "string"}, "address": {"$ref": "#/$defs/address"}, }, + # SEP-2106 requires servers to pass composition and conditional + # keywords through to the client untouched. + "allOf": [ + { + "anyOf": [ + {"required": ["name"]}, + {"required": ["address"]}, + ] + } + ], + "if": {"required": ["address"]}, + "then": {"properties": {"name": {"minLength": 1}}}, + "else": {}, "additionalProperties": False, }, ) ) +@server.tool(name="test_reconnection") +async def test_reconnection(ctx: Context) -> str: + """Closes the POST stream mid-call so the client must resume (SEP-1699). + + The result is written after the stream is gone, so it can only reach the + client through the event store on reconnect. + """ + await ctx.report_progress(0, 100) + await ctx.close_sse_stream() + await asyncio.sleep(0.1) + return "Reconnection test complete." + + +@server.tool(name="test_custom_headers") +async def test_custom_headers( + message: Annotated[str, Field(json_schema_extra={"x-mcp-header": "Message"})], +) -> str: + """Mirrors an argument into an `Mcp-Param-Message` header (SEP-2243). + + The annotation is what makes the header recognized; the transport compares + the header against this argument before the tool ever runs. + """ + return f"Received message: {message}" + + +@server.tool(name="test_missing_capability") +async def test_missing_capability(ctx: Context) -> str: + """Requires the client to have declared the sampling capability (SEP-2575). + + A stateless server may not rely on a capability the client did not declare + in this request's `io.modelcontextprotocol/clientCapabilities` `_meta` + block, so an undeclared caller gets `-32021` rather than a tool result. + """ + require_client_capability(ctx, "sampling") + return "Client declared the sampling capability." + + +# --------------------------------------------------------------------------- +# Multi-round-trip input requests (SEP-2322) +# +# A guard component returns an `InputRequiredResult` naming what it needs; the +# client fulfils those requests and calls again, and the answers arrive on +# `ctx.input_responses` with any `ctx.request_state` echoed back. The framework +# seals and verifies `request_state`, so a tampered echo is rejected before a +# handler sees it. +# --------------------------------------------------------------------------- + + +def _elicit_request(message: str, field: str) -> mcp_types.ElicitRequest: + """A single-field form elicitation for *field*.""" + return mcp_types.ElicitRequest( + method="elicitation/create", + params=mcp_types.ElicitRequestFormParams( + message=message, + requested_schema={ + "type": "object", + "properties": {field: {"type": "string"}}, + "required": [field], + }, + ), + ) + + +def _sampling_request(text: str, max_tokens: int) -> mcp_types.CreateMessageRequest: + """A one-message sampling request.""" + return mcp_types.CreateMessageRequest( + method="sampling/createMessage", + params=mcp_types.CreateMessageRequestParams( + messages=[ + mcp_types.SamplingMessage( + role="user", + content=TextContent(type="text", text=text), + ) + ], + max_tokens=max_tokens, + ), + ) + + +def _elicited_field(responses: mcp_types.InputResponses, key: str, field: str) -> str: + """The accepted value of *field* from the elicitation answered under *key*.""" + answer = responses[key] + if not isinstance(answer, mcp_types.ElicitResult) or answer.content is None: + return "" + return str(answer.content.get(field, "")) + + +@server.tool(name="test_input_required_result_elicitation") +async def test_input_required_result_elicitation( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """Asks the client one elicitation question, then greets the answer. + + A retry whose `inputResponses` omit the key is re-asked rather than + errored: the answer is still missing, so the honest result is the same + request again. + """ + responses = ctx.input_responses + if responses is None or "user_name" not in responses: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={"user_name": _elicit_request("What is your name?", "name")}, + ) + return f"Hello, {_elicited_field(responses, 'user_name', 'name')}!" + + +@server.tool(name="test_input_required_result_sampling") +async def test_input_required_result_sampling( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """Asks the client to sample an answer, then echoes the sampled text.""" + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "capital_question": _sampling_request( + "What is the capital of France?", 100 + ) + }, + ) + answer = responses["capital_question"] + text = "" + if isinstance(answer, mcp_types.CreateMessageResult) and isinstance( + answer.content, TextContent + ): + text = answer.content.text + return f"Sampling result: {text}" + + +@server.tool(name="test_input_required_result_list_roots") +async def test_input_required_result_list_roots( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """Asks the client for its roots, then reports them back.""" + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "client_roots": mcp_types.ListRootsRequest(method="roots/list") + }, + ) + answer = responses["client_roots"] + roots = ( + [str(root.uri) for root in answer.roots] + if isinstance(answer, mcp_types.ListRootsResult) + else [] + ) + return f"Client roots: {', '.join(roots)}" + + +@server.tool(name="test_input_required_result_request_state") +async def test_input_required_result_request_state( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """Carries opaque state across the round trip and confirms it came back.""" + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "confirm": mcp_types.ElicitRequest( + method="elicitation/create", + params=mcp_types.ElicitRequestFormParams( + message="Please confirm", + requested_schema={ + "type": "object", + "properties": {"ok": {"type": "boolean"}}, + "required": ["ok"], + }, + ), + ) + }, + request_state="conformance-state-v1", + ) + if ctx.request_state != "conformance-state-v1": + raise ToolError("requestState was not echoed back intact") + return "state-ok: requestState round-tripped" + + +@server.tool(name="test_input_required_result_multiple_inputs") +async def test_input_required_result_multiple_inputs( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """Asks for elicitation, sampling, and roots in a single round.""" + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "user_name": _elicit_request("What is your name?", "name"), + "greeting": _sampling_request("Generate a greeting", 50), + "client_roots": mcp_types.ListRootsRequest(method="roots/list"), + }, + request_state="conformance-multi-v1", + ) + name = _elicited_field(responses, "user_name", "name") + return f"Collected {len(responses)} responses for {name}" + + +@server.tool(name="test_input_required_result_multi_round") +async def test_input_required_result_multi_round( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """Asks two dependent questions across three rounds.""" + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "step1": _elicit_request("Step 1: What is your name?", "name") + }, + request_state="round-1", + ) + if "step1" in responses: + name = _elicited_field(responses, "step1", "name") + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "step2": _elicit_request( + "Step 2: What is your favorite color?", "color" + ) + }, + request_state=f"round-2:{name}", + ) + color = _elicited_field(responses, "step2", "color") + name = (ctx.request_state or "round-2:").split(":", 1)[1] + return f"{name} likes {color}" + + +@server.tool(name="test_input_required_result_tampered_state") +async def test_input_required_result_tampered_state( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """Round-trips sealed state so a tampered echo is rejected by the framework.""" + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "confirm": _elicit_request("Please confirm", "confirmation") + }, + request_state="sealed-state-v1", + ) + return f"Accepted state: {ctx.request_state}" + + +@server.tool(name="test_input_required_result_capabilities") +async def test_input_required_result_capabilities( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """Asks only for the input methods this client declared it can answer.""" + responses = ctx.input_responses + if responses is not None: + return f"Collected {len(responses)} responses" + + client_params = ctx.session.client_params + declared = client_params.capabilities if client_params else None + requests: dict[str, mcp_types.InputRequest] = {} + if declared is not None and declared.sampling is not None: + requests["capital_question"] = _sampling_request( + "What is the capital of France?", 100 + ) + if declared is not None and declared.elicitation is not None: + requests["user_name"] = _elicit_request("What is your name?", "name") + if declared is not None and declared.roots is not None: + requests["client_roots"] = mcp_types.ListRootsRequest(method="roots/list") + if not requests: + return "Client declared no input capabilities" + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests=requests, + ) + + +# --------------------------------------------------------------------------- +# Background tasks (SEP-2663) +# +# The tasks extension is what turns `task=`-declared tools into background +# work; registering it also advertises `io.modelcontextprotocol/tasks` under +# `capabilities.extensions` and gates the `tasks/*` methods on negotiation. +# The in-memory Docket backend keeps the fixture to a single process. +# --------------------------------------------------------------------------- + +server.add_extension(TasksExtension(url="memory://")) + + +@server.tool(name="greet") +async def greet(name: str) -> str: + """A sync-only tool: never runs as a task.""" + return f"Hello, {name}!" + + +@server.tool(name="slow_compute", task=True) +async def slow_compute(seconds: float = 1.0, label: str = "") -> str: + """Sleeps for *seconds*, so a cancel can land while it is still running.""" + await asyncio.sleep(seconds) + return f"Computed {label} after {seconds} seconds" + + +@server.tool(name="failing_job", task=TaskConfig(mode="required")) +async def failing_job() -> str: + """Reports a tool execution error: `completed` with `result.isError`. + + Registered `required` so a client that never negotiated the extension gets + `-32021` rather than a synchronous run. + """ + await asyncio.sleep(1) + raise ToolError("This job intentionally fails for testing") + + +@server.tool(name="protocol_error_job", task=True) +async def protocol_error_job() -> str: + """Raises a protocol-level error: `failed` with an inlined `error`.""" + raise MCPError( + code=mcp_types.INTERNAL_ERROR, + message="Protocol-level failure for testing", + ) + + +@server.tool(name="confirm_delete", task=True) +async def confirm_delete( + filename: str, ctx: Context +) -> str | mcp_types.InputRequiredResult: + """Parks the task on one elicitation before doing the (pretend) deletion.""" + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "confirm": _elicit_request( + f"Confirm deletion of {filename}?", "confirmation" + ) + }, + ) + answer = _elicited_field(responses, "confirm", "confirmation") + return f"Deleted {filename}: {answer}" + + +@server.tool(name="multi_input", task=True) +async def multi_input(ctx: Context) -> str | mcp_types.InputRequiredResult: + """Parks the task on two elicitations at once, so they can be answered separately.""" + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "first": _elicit_request("First question?", "first"), + "second": _elicit_request("Second question?", "second"), + }, + ) + first = _elicited_field(responses, "first", "first") + second = _elicited_field(responses, "second", "second") + return f"Answers: {first}, {second}" + + +@server.tool(name="test_tool_with_task", task=TaskConfig(mode="required")) +async def test_tool_with_task(ctx: Context) -> str | mcp_types.InputRequiredResult: + """Gathers input over MRTR, then escalates the final round to a task. + + The composition is the point: round 1 is a plain `InputRequiredResult` + with no `taskId`, and the round that actually does the work becomes a + `CreateTaskResult` because the tool requires task execution. + """ + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={"user_name": _elicit_request("What is your name?", "name")}, + ) + return f"Task completed for {_elicited_field(responses, 'user_name', 'name')}" + + +# --------------------------------------------------------------------------- +# Completions +# --------------------------------------------------------------------------- + +_PROMPT_ARG_COMPLETIONS = ["paris", "park", "party"] + + +@server.completion +async def complete( + ref: mcp_types.PromptReference | mcp_types.ResourceTemplateReference, + argument: mcp_types.CompletionArgument, + context: mcp_types.CompletionContext | None, +) -> CompletionValues: + """Suggests values for `test_prompt_with_arguments` arguments.""" + if isinstance(ref, PromptReference) and ref.name == "test_prompt_with_arguments": + matches = [ + value + for value in _PROMPT_ARG_COMPLETIONS + if value.startswith(argument.value) + ] + return Completion(values=matches, total=len(matches), has_more=False) + return None + + # --------------------------------------------------------------------------- # Resources # --------------------------------------------------------------------------- @@ -372,6 +823,49 @@ async def test_prompt_with_image() -> list: ] +@server.prompt(name="test_input_required_result_prompt") +async def test_input_required_result_prompt( + ctx: Context, +) -> str | mcp_types.InputRequiredResult: + """A prompt that gathers its context by elicitation before rendering. + + `InputRequiredResult` is universal — it is a result type, not a tools/call + feature — so `prompts/get` can ask for input the same way a tool does. + """ + responses = ctx.input_responses + if responses is None: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "user_context": _elicit_request( + "What context should the prompt use?", "context" + ) + }, + ) + context_value = _elicited_field(responses, "user_context", "context") + return f"Prompt rendered with context: {context_value}" + + +MCP_PATH = "/mcp" + + +def build_app(): + """The ASGI app the conformance suite is run against. + + Shared by the pytest fixture and the `__main__` entry point so both exercise + the same configuration. The event store is what makes SSE resumption work, + which `test_reconnection` depends on; host/origin protection is a spec MUST + for a localhost server without TLS or auth. + """ + return server.http_app( + transport="streamable-http", + path=MCP_PATH, + host_origin_protection=True, + event_store=EventStore(), + retry_interval=100, + ) + + if __name__ == "__main__": port = int(sys.argv[1]) if len(sys.argv) > 1 else 8000 - server.run(transport="streamable-http", host="127.0.0.1", port=port) + uvicorn.run(build_app(), host="127.0.0.1", port=port, log_level="warning") diff --git a/tests/conformance/test_conformance.py b/tests/conformance/test_conformance.py index 78749f8bc..824c7b401 100644 --- a/tests/conformance/test_conformance.py +++ b/tests/conformance/test_conformance.py @@ -1,5 +1,14 @@ """Run the MCP conformance test suite against a FastMCP server. +The suite is pinned rather than tracking `@latest`: upstream adds scenarios for +draft SEPs, so an unpinned run turns CI red on somebody else's release rather +than on a change of ours. Bumping `CONFORMANCE_VERSION` is how new scenarios +arrive, and the diff shows what they cost. + +`--suite all` includes draft and pending scenarios, which is deliberate — most +of what FastMCP implements ahead of a spec release lives there. Anything that +does not pass is listed in `expected-failures.yml` with a reason. + Requires Node.js and npx to be available on PATH. Mark: pytest -m conformance """ @@ -17,7 +26,9 @@ import uvicorn CONFORMANCE_DIR = Path(__file__).parent EXPECTED_FAILURES = CONFORMANCE_DIR / "expected-failures.yml" HOST = "127.0.0.1" -MCP_PATH = "/mcp" + +#: Pinned version of `@modelcontextprotocol/conformance`. Bump deliberately. +CONFORMANCE_VERSION = "0.2.0-alpha.9" def _get_free_port() -> int: @@ -36,12 +47,10 @@ def _require_npx(): @pytest.fixture(scope="module") def conformance_server(_require_npx): """Start the conformance test server in a background thread.""" - from tests.conformance.server import server as mcp_server + from tests.conformance.server import MCP_PATH, build_app port = _get_free_port() - app = mcp_server.http_app(transport="streamable-http", path=MCP_PATH) - - config = uvicorn.Config(app, host=HOST, port=port, log_level="warning") + config = uvicorn.Config(build_app(), host=HOST, port=port, log_level="warning") uv_server = uvicorn.Server(config) thread = threading.Thread(target=uv_server.run, daemon=True) @@ -66,13 +75,13 @@ def conformance_server(_require_npx): @pytest.mark.conformance -@pytest.mark.timeout(120) +@pytest.mark.timeout(180) def test_mcp_conformance(conformance_server): """Run the full MCP conformance test suite against the server.""" cmd = [ "npx", "--yes", - "@modelcontextprotocol/conformance@latest", + f"@modelcontextprotocol/conformance@{CONFORMANCE_VERSION}", "server", "--url", conformance_server, @@ -83,7 +92,7 @@ def test_mcp_conformance(conformance_server): if EXPECTED_FAILURES.exists(): cmd.extend(["--expected-failures", str(EXPECTED_FAILURES)]) - result = subprocess.run(cmd, capture_output=True, text=True, timeout=90) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=150) # Print output for visibility in test results if result.stdout: diff --git a/tests/server/test_mrtr_guards.py b/tests/server/test_mrtr_guards.py index 8355bdebe..79c03a606 100644 --- a/tests/server/test_mrtr_guards.py +++ b/tests/server/test_mrtr_guards.py @@ -1215,3 +1215,61 @@ class TestHttpTransport: assert asked == ["Where would you like to fly?", "When to Paris?"] assert result.data == "Booked Paris on 2026-08-01" + + +class TestPromptGuard: + """`InputRequiredResult` is a result type, not a tools/call feature, so a + prompt can ask for input the same way a tool does (SEP-2322).""" + + @staticmethod + def _context_prompt_server() -> FastMCP: + mcp = FastMCP("prompt-guard") + + @mcp.prompt + async def summarize(ctx: Context) -> str | InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _ask( + _elicit("context", "What context?", "context"), + key="context", + request_state=None, + ) + return f"Summarizing with {_accepted(responses, 'context')['context']}" + + return mcp + + async def test_prompt_emits_input_required(self): + """The asking round reaches the wire as an InputRequiredResult.""" + async with Client(self._context_prompt_server(), mode="auto") as client: + result = await client.session.get_prompt( + "summarize", allow_input_required=True + ) + + assert isinstance(result, InputRequiredResult) + assert "context" in result.input_requests + + async def test_prompt_completes_with_responses(self): + """Answering the ask renders the prompt on the next round.""" + mcp = self._context_prompt_server() + async with Client(mcp, mode="auto") as client: + ask = await client.session.get_prompt( + "summarize", allow_input_required=True + ) + assert isinstance(ask, InputRequiredResult) + done = await client.session.get_prompt( + "summarize", + input_responses={ + "context": mcp_types.ElicitResult( + action="accept", content={"context": "quarterly report"} + ) + }, + ) + + assert done.messages[0].content.text == ("Summarizing with quarterly report") + + async def test_prompt_guard_rejected_on_handshake_era(self): + """The result type only exists at 2026-07-28, so an older connection + gets the era named rather than a generic invalid-result failure.""" + async with Client(self._context_prompt_server(), mode="legacy") as client: + with pytest.raises(MCPError, match="2026-07-28"): + await client.session.get_prompt("summarize") diff --git a/tests/tasks/client/test_client_tool_tasks.py b/tests/tasks/client/test_client_tool_tasks.py index 815adaeba..f809b34d0 100644 --- a/tests/tasks/client/test_client_tool_tasks.py +++ b/tests/tasks/client/test_client_tool_tasks.py @@ -135,9 +135,9 @@ async def test_tool_task_cancel(): assert final.status == "cancelled" -async def test_required_mode_without_optin_raises_32003(): +async def test_required_mode_without_optin_raises_32021(): """A legacy client never negotiates the tasks capability, so a required-mode - tool call is rejected with the -32003 missing-capability error.""" + tool call is rejected with the -32021 missing-capability error.""" mcp = FastMCP("required-test") mcp.add_extension(TasksExtension()) diff --git a/tests/tasks/server/test_extension.py b/tests/tasks/server/test_extension.py index 5260565a5..44dc7ef9c 100644 --- a/tests/tasks/server/test_extension.py +++ b/tests/tasks/server/test_extension.py @@ -1,7 +1,7 @@ """End-to-end tests for the SEP-2663 `TasksExtension` server adapter. Covers the decide-and-task interceptor (forbidden/optional/required modes and the --32003 missing-capability error), the tasks/get|update|cancel handlers, status +-32021 missing-capability error), the tasks/get|update|cancel handlers, status mapping, inlined completed results, argument-coercion parity, TTL, and capability advertisement. Server-side tasks are driven in-process via `task_helpers` because there is no client task-submission API until Phase 4. @@ -360,7 +360,7 @@ async def test_legacy_era_opt_in_is_ignored(): async def test_legacy_era_required_tool_raises_missing_capability(): - """`required` tools refuse legacy-era calls with -32003 even when opted in.""" + """`required` tools refuse legacy-era calls with -32021 even when opted in.""" mcp = _tasks_server() async with running_task_server(mcp): srctx = ServerRequestContext( @@ -377,7 +377,7 @@ async def test_legacy_era_required_tool_raises_missing_capability(): with bind_request_context(srctx): with pytest.raises(MCPError) as exc_info: await mcp.call_tool("must_task", {"n": 3}) - assert exc_info.value.error.code == -32003 + assert exc_info.value.error.code == -32021 # --------------------------------------------------------------------------- @@ -409,12 +409,21 @@ async def test_worker_hooks_survive_sibling_server_shutdown(): # --------------------------------------------------------------------------- -# Compliance: -32003 on task methods for non-declaring clients (SEP-2663) +# Compliance: -32021 on task methods for non-declaring clients (SEP-2663) # --------------------------------------------------------------------------- +def test_missing_capability_code_is_the_protocol_value(): + """The code must track the SDK, not an early SEP-2663 draft. + + It shipped hardcoded as -32003, which no client recognizes: SEP-2575 + assigns -32021 to MissingRequiredClientCapability. + """ + assert MISSING_REQUIRED_CLIENT_CAPABILITY == -32021 + + async def test_task_method_without_capability_raises_missing_capability(): - """tasks/get from a client that did not declare the extension gets -32003.""" + """tasks/get from a client that did not declare the extension gets -32021.""" mcp = _tasks_server() extension = cast(TasksExtension, mcp._extensions[TASKS_EXTENSION_ID]) # A request context with no tasks capability in its _meta. diff --git a/tests/tasks/server/test_guard_reentrant.py b/tests/tasks/server/test_guard_reentrant.py index 5adad9c32..ef57fee4d 100644 --- a/tests/tasks/server/test_guard_reentrant.py +++ b/tests/tasks/server/test_guard_reentrant.py @@ -14,6 +14,8 @@ from __future__ import annotations from typing import Any import mcp_types +from mcp.shared.exceptions import MCPError +from mcp_types import INTERNAL_ERROR from fastmcp import Context, FastMCP from fastmcp_tasks import TasksExtension @@ -244,3 +246,79 @@ async def test_state_only_guard_round_fails_clearly(): assert final.result is not None assert final.result["isError"] is True assert "state-only" in final.result["content"][0]["text"] + + +async def test_partial_update_keeps_task_parked_on_remaining_request(): + """SEP-2663 partial fulfillment: a leg that asked two questions stays + `input_required` until both are answered, and each `tasks/get` in between + surfaces only what is still outstanding.""" + mcp = FastMCP("partial") + mcp.add_extension(TasksExtension()) + + @mcp.tool(task=True) + async def two_questions(ctx: Context) -> str | mcp_types.InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _input_required( + { + "first": _elicit_request("First?"), + "second": _elicit_request("Second?"), + } + ) + return f"{_answer(responses, 'first')}+{_answer(responses, 'second')}" + + async with running_task_server(mcp): + created = await submit_task(mcp, "two_questions", {}) + parked = await wait_for_task( + mcp, created.task_id, target_states=frozenset({"input_required"}) + ) + assert parked.input_requests is not None + keys = sorted(parked.input_requests) + assert len(keys) == 2 + + answered, pending = keys[0], keys[1] + await update_task( + mcp, + created.task_id, + {answered: {"action": "accept", "content": {"value": "one"}}}, + ) + + still_parked = await get_task(mcp, created.task_id) + assert still_parked.status == "input_required" + assert still_parked.input_requests is not None + assert list(still_parked.input_requests) == [pending] + + # Answering the last one resumes the leg, which now sees both answers. + await update_task( + mcp, + created.task_id, + {pending: {"action": "accept", "content": {"value": "two"}}}, + ) + final = await wait_for_task(mcp, created.task_id) + + assert final.status == "completed" + assert final.result is not None + assert final.result["content"][0]["text"] == "one+two" + + +async def test_protocol_error_fails_the_task_with_inlined_error(): + """SEP-2663 reserves `failed` for protocol faults: an `MCPError` raised by + the body is inlined as a JSON-RPC error rather than reported as a completed + task carrying an `isError` result (which is what a `ToolError` produces).""" + mcp = FastMCP("protocol-fault") + mcp.add_extension(TasksExtension()) + + @mcp.tool(task=True) + async def explodes() -> str: + raise MCPError(code=INTERNAL_ERROR, message="protocol fault", data={"x": 1}) + + async with running_task_server(mcp): + created = await submit_task(mcp, "explodes", {}) + final = await wait_for_task(mcp, created.task_id) + + assert final.status == "failed" + assert final.result is None + assert final.error is not None + assert final.error["code"] == INTERNAL_ERROR + assert final.error["message"] == "protocol fault" + assert final.error["data"] == {"x": 1} diff --git a/tests/tasks/server/test_task_config.py b/tests/tasks/server/test_task_config.py index 42c5e7e98..f839debff 100644 --- a/tests/tasks/server/test_task_config.py +++ b/tests/tasks/server/test_task_config.py @@ -166,7 +166,7 @@ class TestToolModeEnforcement: return mcp async def test_required_mode_without_opt_in_raises(self): - """Required mode raises -32003 when called without a tasks opt-in.""" + """Required mode raises -32021 when called without a tasks opt-in.""" mcp = self._server() async with running_task_server(mcp): with pytest.raises(MCPError) as exc_info: diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py index d85ea0322..042274212 100644 --- a/tests/test_exceptions.py +++ b/tests/test_exceptions.py @@ -5,6 +5,7 @@ from __future__ import annotations import pytest from mcp import MCPError from mcp_types import INTERNAL_ERROR, INVALID_PARAMS +from mcp_types.jsonrpc import MISSING_REQUIRED_CLIENT_CAPABILITY from fastmcp import Client, FastMCP from fastmcp.exceptions import ( @@ -71,6 +72,20 @@ class TestWireErrorCodes: assert exc_info.value.error.code == INVALID_PARAMS assert "Resource not found" in exc_info.value.error.message + async def test_resource_not_found_echoes_uri_in_data(self): + """SEP-2164 SHOULD: the error names which URI was missing. + + A client that pipelined several reads cannot otherwise tell which one + failed from the message alone. + """ + mcp = FastMCP("test-server") + + async with Client(mcp) as client: + with pytest.raises(MCPError) as exc_info: + await client.read_resource_mcp("config://missing") + + assert exc_info.value.error.data == {"uri": "config://missing"} + async def test_prompt_not_found_uses_invalid_params(self): mcp = FastMCP("test-server") @@ -80,3 +95,45 @@ class TestWireErrorCodes: assert exc_info.value.error.code == INVALID_PARAMS assert "Unknown prompt" in exc_info.value.error.message + + +class TestMissingClientCapabilityFromTool: + """A tool's `-32021` must reach the wire, not become an `isError` result. + + SEP-2575 makes this error a statement about the *request* — the server + cannot service it at all — so flattening it into a tool result would drop + the code and tell the client the call succeeded. Every other `MCPError` + raised under a tool still masks into a result, since those describe how the + call went rather than whether it could run. + """ + + @staticmethod + def _server() -> FastMCP: + mcp = FastMCP("capability-test") + + @mcp.tool + async def needs_sampling() -> str: + raise MCPError( + code=MISSING_REQUIRED_CLIENT_CAPABILITY, + message="Client did not declare the required 'sampling' capability", + data={"requiredCapabilities": {"sampling": {}}}, + ) + + @mcp.tool + async def upstream_failed() -> str: + raise MCPError(code=INTERNAL_ERROR, message="upstream exploded") + + return mcp + + async def test_capability_error_propagates_as_protocol_error(self): + async with Client(self._server()) as client: + with pytest.raises(MCPError) as exc_info: + await client.call_tool("needs_sampling") + + assert exc_info.value.error.code == MISSING_REQUIRED_CLIENT_CAPABILITY + assert exc_info.value.error.data == {"requiredCapabilities": {"sampling": {}}} + + async def test_other_mcp_errors_still_become_tool_errors(self): + async with Client(self._server()) as client: + with pytest.raises(ToolError): + await client.call_tool("upstream_failed") From 96e12569b1cc19b15fafaada51dc098b83997cdc Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 15:36:40 -0400 Subject: [PATCH 2/9] Address review: durable re-entry, resource guards, deterministic test Keep the final outstanding input marker until the next task leg is durable, so a racing tasks/get cannot read a parked leg as complete. Let resources and resource templates return InputRequiredResult like tools and prompts. Identify parked requests by their question rather than sort order. --- fastmcp_slim/fastmcp/resources/base.py | 35 +++++++++ .../fastmcp/server/mixins/mcp_operations.py | 20 +++++- fastmcp_tasks/fastmcp_tasks/handlers.py | 26 ++++--- tests/server/test_mrtr_guards.py | 72 +++++++++++++++++++ tests/tasks/server/test_guard_reentrant.py | 51 ++++++++++++- 5 files changed, 191 insertions(+), 13 deletions(-) diff --git a/fastmcp_slim/fastmcp/resources/base.py b/fastmcp_slim/fastmcp/resources/base.py index 7210a8e63..30ed132ea 100644 --- a/fastmcp_slim/fastmcp/resources/base.py +++ b/fastmcp_slim/fastmcp/resources/base.py @@ -210,6 +210,35 @@ class ResourceResult(pydantic.BaseModel): ) +class InputRequiredResourceResult(ResourceResult): + """The full result of a single multi-round-trip resource read (SEP-2322). + + `InputRequiredResult` is a result type, not a `tools/call` feature: any + request may resolve to one. When a resource or resource template returns an + `InputRequiredResult` from its body to ask the client for input, that ask is + the legitimate result of this `resources/read` — so FastMCP wraps it in this + `ResourceResult` subclass, mirroring `InputRequiredToolResult` and + `InputRequiredPromptResult`, and it flows through the middleware chain as an + ordinary return value. + + Invariant: the wrapped `InputRequiredResult` is never serialized as resource + contents. `contents` is always empty; the wire handler (`_on_read_resource`) + reads `.input_required` and returns it to the runner. + """ + + input_required: mcp_types.InputRequiredResult = pydantic.Field( + description="The client-input request this read resolved to (SEP-2322)" + ) + + def __init__(self, input_required: mcp_types.InputRequiredResult) -> None: + # Bypass ResourceResult's content-normalizing __init__: an + # input-required read carries no contents (see the invariant above), and + # `input_required` is a required field ResourceResult.__init__ can't set. + pydantic.BaseModel.__init__( + self, contents=[], meta=None, input_required=input_required + ) + + def _public_content_meta(meta: dict[str, Any] | None) -> dict[str, Any] | None: """Strip FastMCP's internal bookkeeping out of component meta. @@ -259,6 +288,12 @@ def convert_raw_to_resource_result( if isinstance(raw_value, ResourceResult): return raw_value + if isinstance(raw_value, mcp_types.InputRequiredResult): + # The resource asked the client for input (SEP-2322). Wrap it so the + # ask travels the middleware chain as an ordinary result; the wire + # handler unwraps it. + return InputRequiredResourceResult(raw_value) + meta = _public_content_meta(meta) # For plain str/bytes returns, wrap in ResourceContent with the diff --git a/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py b/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py index 642e28879..5e96e37dc 100644 --- a/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py +++ b/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py @@ -29,6 +29,7 @@ from fastmcp.exceptions import ( to_mcp_error, ) from fastmcp.prompts.base import InputRequiredPromptResult +from fastmcp.resources.base import InputRequiredResourceResult from fastmcp.server.completions import CompletionValues, normalize_completion from fastmcp.server.dependencies import bind_request_context, extract_version_spec from fastmcp.tools.base import InputRequiredToolResult, ToolResult @@ -296,7 +297,7 @@ class MCPOperationsMixin: self: FastMCP, ctx: ServerRequestContext, params: ReadResourceRequestParams, - ) -> mcp_types.ReadResourceResult: + ) -> mcp_types.ReadResourceResult | mcp_types.InputRequiredResult: """Handle MCP 'resources/read' requests.""" with bind_request_context(ctx): uri = params.uri @@ -325,6 +326,23 @@ class MCPOperationsMixin: # already happened inside read_resource. raise to_mcp_error(e) from e + if isinstance(result, InputRequiredResourceResult): + # The resource requested client input (SEP-2322). As with tools + # and prompts, the multi-round-trip result type only exists at + # 2026-07-28, so name the era problem on an older connection + # rather than failing as a generic "invalid result". + if ctx.protocol_version not in MODERN_PROTOCOL_VERSIONS: + raise MCPError( + code=INVALID_PARAMS, + message=( + f"Resource {str(uri)!r} returned an InputRequiredResult " + "to request client input, but the multi-round-trip " + "result type (SEP-2322) only exists at MCP 2026-07-28; " + f"this connection negotiated {ctx.protocol_version!r}." + ), + ) + return result.input_required + return result.to_mcp_result(uri) async def _on_get_prompt( diff --git a/fastmcp_tasks/fastmcp_tasks/handlers.py b/fastmcp_tasks/fastmcp_tasks/handlers.py index 526416474..836441c98 100644 --- a/fastmcp_tasks/fastmcp_tasks/handlers.py +++ b/fastmcp_tasks/fastmcp_tasks/handlers.py @@ -374,17 +374,25 @@ async def tasks_update( # may answer a multi-request ask one update at a time. await store_input_responses(docket, task_scope, task_id, translated) - # Retire only what this update answered. While anything is still - # outstanding the task stays `input_required` and `tasks/get` surfaces - # the remaining keys — the leg re-enters only once every request has an - # answer (SEP-2663 partial fulfillment). - await discard_outstanding( - docket, task_scope, task_id, leg_number, answered_keys - ) - still_pending = await read_outstanding_inputs( + # A partial update retires only the keys it answered, leaving the task + # `input_required` with the rest surfaced; the leg re-enters only once + # every request has an answer (SEP-2663 partial fulfillment). + # + # The *last* answer deliberately leaves its marker in place. Outstanding + # requests are what make a completed-but-parked leg read as + # `input_required` rather than as a finished task, so retiring the final + # one before the next leg is durable would let a racing `tasks/get` + # report the task complete with a `None` result — and would strand it + # there for good if the enqueue below failed, since a retried update + # would no longer match any key. `clear_outstanding` runs after the + # pointer swap instead. + outstanding = await read_outstanding_inputs( docket, task_scope, task_id, leg_number ) - if still_pending: + if set(outstanding) - set(answered_keys): + await discard_outstanding( + docket, task_scope, task_id, leg_number, answered_keys + ) return UpdateTaskResult() # Every request is answered, so enqueue the next leg. Ordering matters: diff --git a/tests/server/test_mrtr_guards.py b/tests/server/test_mrtr_guards.py index 79c03a606..134eda8b4 100644 --- a/tests/server/test_mrtr_guards.py +++ b/tests/server/test_mrtr_guards.py @@ -1273,3 +1273,75 @@ class TestPromptGuard: async with Client(self._context_prompt_server(), mode="legacy") as client: with pytest.raises(MCPError, match="2026-07-28"): await client.session.get_prompt("summarize") + + +class TestResourceGuard: + """Resources and templates ask for input the same way tools and prompts do.""" + + @staticmethod + def _resource_server() -> FastMCP: + mcp = FastMCP("resource-guard") + + @mcp.resource("data://report") + async def report(ctx: Context) -> str | InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _ask( + _elicit("context", "Which quarter?", "context"), + key="context", + request_state=None, + ) + return f"Report for {_accepted(responses, 'context')['context']}" + + @mcp.resource("data://report/{section}") + async def section_report( + section: str, ctx: Context + ) -> str | InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _ask( + _elicit("context", f"Which quarter for {section}?", "context"), + key="context", + request_state=None, + ) + quarter = _accepted(responses, "context")["context"] + return f"{section} for {quarter}" + + return mcp + + async def test_resource_emits_input_required(self): + async with Client(self._resource_server(), mode="auto") as client: + result = await client.session.read_resource( + "data://report", allow_input_required=True + ) + + assert isinstance(result, InputRequiredResult) + assert "context" in result.input_requests + + async def test_resource_completes_with_responses(self): + async with Client(self._resource_server(), mode="auto") as client: + done = await client.session.read_resource( + "data://report", + input_responses={ + "context": mcp_types.ElicitResult( + action="accept", content={"context": "Q3"} + ) + }, + ) + + assert done.contents[0].text == "Report for Q3" + + async def test_resource_template_emits_input_required(self): + """Templates share the converter, so the ask survives there too.""" + async with Client(self._resource_server(), mode="auto") as client: + result = await client.session.read_resource( + "data://report/revenue", allow_input_required=True + ) + + assert isinstance(result, InputRequiredResult) + assert "context" in result.input_requests + + async def test_resource_guard_rejected_on_handshake_era(self): + async with Client(self._resource_server(), mode="legacy") as client: + with pytest.raises(MCPError, match="2026-07-28"): + await client.session.read_resource("data://report") diff --git a/tests/tasks/server/test_guard_reentrant.py b/tests/tasks/server/test_guard_reentrant.py index ef57fee4d..93123734e 100644 --- a/tests/tasks/server/test_guard_reentrant.py +++ b/tests/tasks/server/test_guard_reentrant.py @@ -60,6 +60,14 @@ def _input_required( ) +def _key_asking(input_requests: dict[str, Any], message: str) -> str: + """The surfaced key whose parked request asks *message*.""" + for key, payload in input_requests.items(): + if payload["params"]["message"] == message: + return key + raise AssertionError(f"no parked request asks {message!r}") + + async def _park_key(mcp: FastMCP, task_id: str) -> str: parked = await wait_for_task( mcp, task_id, target_states=frozenset({"input_required"}) @@ -273,10 +281,12 @@ async def test_partial_update_keeps_task_parked_on_remaining_request(): mcp, created.task_id, target_states=frozenset({"input_required"}) ) assert parked.input_requests is not None - keys = sorted(parked.input_requests) - assert len(keys) == 2 + assert len(parked.input_requests) == 2 - answered, pending = keys[0], keys[1] + # Surfaced keys are freshly minted per request, so they carry no order + # a test can rely on. Identify each by the question it asks. + answered = _key_asking(parked.input_requests, "First?") + pending = _key_asking(parked.input_requests, "Second?") await update_task( mcp, created.task_id, @@ -301,6 +311,41 @@ async def test_partial_update_keeps_task_parked_on_remaining_request(): assert final.result["content"][0]["text"] == "one+two" +async def test_final_answer_keeps_task_parked_until_next_leg_is_durable(): + """The last answer must not retire its outstanding marker early. + + Outstanding requests are what make a completed-but-parked leg read as + `input_required`. Discarding the final one before the next leg is enqueued + would let a `tasks/get` landing in that window see a finished execution with + no result and report the task complete. + """ + mcp = FastMCP("durable-reentry") + mcp.add_extension(TasksExtension()) + + @mcp.tool(task=True) + async def one_question(ctx: Context) -> str | mcp_types.InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _input_required({"only": _elicit_request("Only?")}) + return f"got {_answer(responses, 'only')}" + + async with running_task_server(mcp): + created = await submit_task(mcp, "one_question", {}) + key = await _park_key(mcp, created.task_id) + + await update_task( + mcp, + created.task_id, + {key: {"action": "accept", "content": {"value": "answer"}}}, + ) + final = await wait_for_task(mcp, created.task_id) + + # The task must land on the real result, never on a phantom completion. + assert final.status == "completed" + assert final.result is not None + assert final.result["content"][0]["text"] == "got answer" + + async def test_protocol_error_fails_the_task_with_inlined_error(): """SEP-2663 reserves `failed` for protocol faults: an `MCPError` raised by the body is inlined as a JSON-RPC error rather than reported as a completed From 7ca58583fd20eb2476c650332d2f70a7e12ff616 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 15:49:45 -0400 Subject: [PATCH 3/9] Document the guard pattern for prompts and resources --- docs/servers/elicitation.mdx | 53 ++++++++++++++++++++++++++++++++++++ docs/servers/prompts.mdx | 4 +++ docs/servers/resources.mdx | 4 +++ 3 files changed, 61 insertions(+) diff --git a/docs/servers/elicitation.mdx b/docs/servers/elicitation.mdx index 9b7b0c164..76bfea4d3 100644 --- a/docs/servers/elicitation.mdx +++ b/docs/servers/elicitation.mdx @@ -541,6 +541,59 @@ on handshake-era connections. If you need to support both eras, branch on `ctx.protocol_version`: return an `InputRequiredResult` on modern connections and fall back to [`ctx.elicit()`](#requesting-input-on-handshake-connections) on handshake-era ones. +### Prompts and resources + +`InputRequiredResult` is a **result type**, not a tools feature: any request can resolve to one. Prompts, resources, and resource templates ask for input exactly the way tools do — return an `InputRequiredResult`, read `ctx.input_responses` on the next round, and the client re-issues the same `prompts/get` or `resources/read` with the answer attached. + +This prompt gathers the context it needs before rendering: + +```python +from fastmcp import FastMCP, Context +from mcp_types import InputRequiredResult, ElicitRequest, ElicitRequestFormParams + +mcp = FastMCP("Reporting Server") + +ask_for_quarter = InputRequiredResult( + result_type="input_required", + input_requests={ + "quarter": ElicitRequest( + method="elicitation/create", + params=ElicitRequestFormParams( + message="Which quarter should the summary cover?", + requested_schema={ + "type": "object", + "properties": {"quarter": {"type": "string"}}, + "required": ["quarter"], + }, + ), + ) + }, +) + + +@mcp.prompt +async def summarize(ctx: Context) -> str | InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return ask_for_quarter + quarter = responses["quarter"].content["quarter"] + return f"Summarize the {quarter} results." +``` + +Resources and resource templates work the same way, with the URI standing in for the tool name: + +```python +@mcp.resource("report://summary") +async def report(ctx: Context) -> str | InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return ask_for_quarter + quarter = responses["quarter"].content["quarter"] + return f"Revenue report for {quarter}" +``` + +The same protocol requirement applies: returning an `InputRequiredResult` from a prompt or resource needs a 2026-07-28 connection, and FastMCP names the era mismatch if one arrives on an older one. Client-side, `read_resource` and `get_prompt` drive the loop the way `call_tool` does, so a configured elicitation handler answers all three without extra wiring. + ### Sampling and roots Elicitation is the most common request to carry this way, and **roots** requests work identically — the `input_requests` map holds them the same way, and each answer comes back in `ctx.input_responses` under its key (an `ElicitResult` or `ListRootsResult`). See [Client Roots](/clients/roots) for what a roots request contains. `fastmcp.Client` answers both from the handlers you already configured, so a guard tool that mixes them needs no extra client wiring. diff --git a/docs/servers/prompts.mdx b/docs/servers/prompts.mdx index 4e824dec0..ae7172522 100644 --- a/docs/servers/prompts.mdx +++ b/docs/servers/prompts.mdx @@ -438,6 +438,10 @@ Notifications are only sent when these operations occur within an active MCP req Clients can handle these notifications using a [message handler](/clients/notifications) to automatically refresh their prompt lists or update their interfaces. +## Requesting Input + +A prompt can ask the client for information before it renders. On an MCP 2026-07-28 connection, return an `InputRequiredResult` describing what you need; the client answers and re-issues the `prompts/get`, and your function runs again with the answer on `ctx.input_responses`. See [Elicitation](/servers/elicitation#prompts-and-resources) for the full pattern. + ## Server Behavior ### Duplicate Prompts diff --git a/docs/servers/resources.mdx b/docs/servers/resources.mdx index 7f45c7280..e0daf6e76 100644 --- a/docs/servers/resources.mdx +++ b/docs/servers/resources.mdx @@ -783,6 +783,10 @@ def get_data_by_id(id: str) -> dict: When `mask_error_details=True`, only error messages from `ResourceError` will include details, other exceptions will be converted to a generic message. +## Requesting Input + +A resource or resource template can ask the client for information before it produces content. On an MCP 2026-07-28 connection, return an `InputRequiredResult` describing what you need; the client answers and re-issues the `resources/read`, and your function runs again with the answer on `ctx.input_responses`. See [Elicitation](/servers/elicitation#prompts-and-resources) for the full pattern. + ## Server Behavior ### Duplicate Resources From 4ef7d16419bc8e65fd5fe7814fe1416f127d2dd4 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 16:06:17 -0400 Subject: [PATCH 4/9] Keep multi-round-trip asks out of the response cache Prompt and resource asks carry no content, so caching one stored an empty result and the client never saw the question. Bypass the cache on continuation legs and return asks unwrapped, as tool calls already did. --- .../fastmcp/server/middleware/caching.py | 72 +++++++++--- tests/server/middleware/test_caching.py | 107 ++++++++++++++++++ 2 files changed, 164 insertions(+), 15 deletions(-) diff --git a/fastmcp_slim/fastmcp/server/middleware/caching.py b/fastmcp_slim/fastmcp/server/middleware/caching.py index cbf48b849..d470f1b90 100644 --- a/fastmcp_slim/fastmcp/server/middleware/caching.py +++ b/fastmcp_slim/fastmcp/server/middleware/caching.py @@ -18,8 +18,18 @@ from key_value.aio.wrappers.statistics.wrapper import ( from pydantic import Field from typing_extensions import NotRequired, Self, TypeVar, override -from fastmcp.prompts.base import Message, Prompt, PromptResult -from fastmcp.resources.base import Resource, ResourceContent, ResourceResult +from fastmcp.prompts.base import ( + InputRequiredPromptResult, + Message, + Prompt, + PromptResult, +) +from fastmcp.resources.base import ( + InputRequiredResourceResult, + Resource, + ResourceContent, + ResourceResult, +) from fastmcp.server.dependencies import get_access_token from fastmcp.server.middleware.middleware import CallNext, Middleware, MiddlewareContext from fastmcp.tools.base import InputRequiredToolResult, Tool, ToolResult @@ -28,6 +38,28 @@ from fastmcp.utilities.types import FastMCPBaseModel logger: Logger = get_logger(name=__name__) + +def _is_continuation_leg(context: MiddlewareContext[Any]) -> bool: + """Whether this request is answering a previous round's ask (SEP-2322). + + A continuation must bypass the cache entirely. Cache keys are built from the + component's identity and arguments alone, so a continuation shares its key + with a fresh call: reading could serve a prior flow's final answer to this + leg, and writing would serve THIS flow's final answer to a later fresh call, + which would then never be asked the questions at all. + + Either signal marks a continuation. A state-only round (one that carried + `request_state` without asking anything) retries with `input_responses` + still `None`. + """ + fastmcp_ctx = context.fastmcp_context + if fastmcp_ctx is None: + return False + return ( + fastmcp_ctx.input_responses is not None or fastmcp_ctx.request_state is not None + ) + + # Constants ONE_HOUR_IN_SECONDS = 3600 FIVE_MINUTES_IN_SECONDS = 300 @@ -413,19 +445,7 @@ class ResponseCachingMiddleware(Middleware): ) is False or not self._matches_tool_cache_settings(tool_name=tool_name): return await call_next(context) - # A multi-round continuation leg (SEP-2322) must bypass the cache - # entirely: the cache key is built from the tool name and arguments - # only, so a continuation shares its key with a fresh call. Reading - # could serve a prior flow's final answer to this leg; writing would - # serve THIS flow's final answer to a later fresh call — which would - # then never be asked the tool's questions. Either signal marks a - # continuation: a state-only round (request_state with no questions) - # retries with input_responses=None but still carries request_state. - fastmcp_ctx = context.fastmcp_context - if fastmcp_ctx is not None and ( - fastmcp_ctx.input_responses is not None - or fastmcp_ctx.request_state is not None - ): + if _is_continuation_leg(context): return await call_next(context) cache_key: str = _make_call_tool_cache_key( @@ -477,6 +497,9 @@ class ResponseCachingMiddleware(Middleware): if self._read_resource_settings.get("enabled") is False: return await call_next(context) + if _is_continuation_leg(context): + return await call_next(context) + cache_key: str = _make_read_resource_cache_key( msg=context.message, auth_key=_get_auth_partition_key() ) @@ -486,6 +509,14 @@ class ResponseCachingMiddleware(Middleware): return cached_value.unwrap() value: ResourceResult = await call_next(context) + + # Never cache a multi-round-trip ask (SEP-2322). An + # InputRequiredResourceResult is a request for client input on this leg, + # not a stable answer, and it carries no contents — wrapping it would + # cache an empty read and the client would never see the question. + if isinstance(value, InputRequiredResourceResult): + return value + cached_value = CacheableResourceResult.wrap(value) await self._read_resource_cache.put( @@ -507,6 +538,9 @@ class ResponseCachingMiddleware(Middleware): if self._get_prompt_settings.get("enabled") is False: return await call_next(context) + if _is_continuation_leg(context): + return await call_next(context) + cache_key: str = _make_get_prompt_cache_key( msg=context.message, auth_key=_get_auth_partition_key() ) @@ -515,6 +549,14 @@ class ResponseCachingMiddleware(Middleware): return cached_value.unwrap() value: PromptResult = await call_next(context) + + # Never cache a multi-round-trip ask (SEP-2322). An + # InputRequiredPromptResult is a request for client input on this leg, + # not a stable answer, and it carries no messages — wrapping it would + # cache an empty prompt and the client would never see the question. + if isinstance(value, InputRequiredPromptResult): + return value + cached_value = CacheablePromptResult.wrap(value) await self._get_prompt_cache.put( diff --git a/tests/server/middleware/test_caching.py b/tests/server/middleware/test_caching.py index b45961bcf..dd1782a7e 100644 --- a/tests/server/middleware/test_caching.py +++ b/tests/server/middleware/test_caching.py @@ -902,3 +902,110 @@ class TestAuthAwareCaching: assert {p.name for p in prompts} == {"public_prompt"} finally: auth_context_var.reset(tok) + + +class TestCachingWithInputRequiredResults: + """A multi-round-trip ask must survive `ResponseCachingMiddleware`. + + An ask carries no content of its own, so caching one would store an empty + result and the client would never see the question. Continuation legs must + also bypass the cache, since they share a cache key with a fresh call. + """ + + @staticmethod + def _answer(responses: mcp_types.InputResponses) -> str: + """The accepted value for the single question these guards ask.""" + result = responses["q"] + assert isinstance(result, mcp_types.ElicitResult) + assert result.content is not None + return str(result.content["q"]) + + @staticmethod + def _ask() -> mcp_types.InputRequiredResult: + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={ + "q": mcp_types.ElicitRequest( + method="elicitation/create", + params=mcp_types.ElicitRequestFormParams( + message="Which quarter?", + requested_schema={ + "type": "object", + "properties": {"q": {"type": "string"}}, + "required": ["q"], + }, + ), + ) + }, + ) + + @classmethod + def _server(cls) -> FastMCP: + mcp = FastMCP("cached-guards") + mcp.add_middleware(ResponseCachingMiddleware()) + + @mcp.tool + async def summarize_tool(ctx: Context) -> str | mcp_types.InputRequiredResult: + if ctx.input_responses is None: + return cls._ask() + return f"Summary for {cls._answer(ctx.input_responses)}" + + @mcp.prompt + async def summarize(ctx: Context) -> str | mcp_types.InputRequiredResult: + if ctx.input_responses is None: + return cls._ask() + return f"Summary for {cls._answer(ctx.input_responses)}" + + @mcp.resource("report://x") + async def report(ctx: Context) -> str | mcp_types.InputRequiredResult: + if ctx.input_responses is None: + return cls._ask() + return f"Report for {cls._answer(ctx.input_responses)}" + + return mcp + + @staticmethod + async def _handler(message, response_type, params, ctx): + from fastmcp.client.elicitation import ElicitResult + + return ElicitResult(action="accept", content=response_type(q="Q3")) + + async def test_tool_guard_completes_under_caching(self): + async with Client( + self._server(), mode="auto", elicitation_handler=self._handler + ) as client: + result = await client.call_tool("summarize_tool", {}) + + assert result.data == "Summary for Q3" + + async def test_prompt_guard_completes_under_caching(self): + async with Client( + self._server(), mode="auto", elicitation_handler=self._handler + ) as client: + result = await client.get_prompt("summarize") + + assert result.messages[0].content.text == "Summary for Q3" + + async def test_resource_guard_completes_under_caching(self): + async with Client( + self._server(), mode="auto", elicitation_handler=self._handler + ) as client: + result = await client.read_resource("report://x") + + assert result[0].text == "Report for Q3" + + async def test_guard_still_asks_on_a_second_fresh_call(self): + """The ask must not be cached away for the next caller. + + A second fresh flow has to be asked the same question; serving it a + cached final answer would skip the component's own per-round logic. + """ + mcp = self._server() + async with Client( + mcp, mode="auto", elicitation_handler=self._handler + ) as client: + first = await client.get_prompt("summarize") + second = await client.get_prompt("summarize") + + assert first.messages[0].content.text == "Summary for Q3" + assert second.messages[0].content.text == "Summary for Q3" From 2d3ad9ca0de9b26b6f55dff1c120c5b1aa845af3 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 16:40:33 -0400 Subject: [PATCH 5/9] Make a contended tasks/update wait instead of dropping its answer Partial fulfillment means two in-flight updates can carry different answers, so acknowledging the one that loses the update lock stranded the task on a key the client had already sent. --- fastmcp_tasks/fastmcp_tasks/handlers.py | 22 +++++-- tests/server/middleware/test_caching.py | 53 ++++++---------- tests/tasks/server/test_guard_reentrant.py | 74 ++++++++++++++++++++++ 3 files changed, 111 insertions(+), 38 deletions(-) diff --git a/fastmcp_tasks/fastmcp_tasks/handlers.py b/fastmcp_tasks/fastmcp_tasks/handlers.py index 836441c98..5193d5b84 100644 --- a/fastmcp_tasks/fastmcp_tasks/handlers.py +++ b/fastmcp_tasks/fastmcp_tasks/handlers.py @@ -39,7 +39,6 @@ from fastmcp_tasks.creation import ( registered_component_for_key, ) from fastmcp_tasks.input_store import ( - acquire_update_lock, acquire_update_lock_blocking, clear_outstanding, discard_outstanding, @@ -351,9 +350,24 @@ async def tasks_update( ) # Serialize concurrent updates for this task so two racing answers cannot - # each enqueue a next leg (double execution). A loser is an idempotent no-op. - if not await acquire_update_lock(docket, task_scope, task_id): - return UpdateTaskResult() + # each enqueue a next leg (double execution). Waiting rather than dropping + # the loser matters for partial fulfillment: SEP-2663 invites a client to + # answer a multi-request ask one key at a time, so two in-flight updates may + # carry *different* answers. Acknowledging the loser without storing its + # answer would strand the task waiting on a key the client believes it has + # already sent. Once the winner finishes, the loser re-reads the leg's + # outstanding state: a genuinely duplicate answer finds nothing left to + # match and is the idempotent no-op SEP-2663 asks for. + if not await acquire_update_lock_blocking(docket, task_scope, task_id): + # The holder is wedged. Report a retryable failure rather than a false + # acknowledgement, which would silently lose this answer. + raise MCPError( + code=mcp_types.INTERNAL_ERROR, + message=( + f"Task {task_id} has an update in progress that did not complete " + "in time; retry this update." + ), + ) try: # A cancelled task never re-enters: clearing outstanding on cancel makes # translate return None already, but check explicitly so a cancel that diff --git a/tests/server/middleware/test_caching.py b/tests/server/middleware/test_caching.py index dd1782a7e..a88efbd3f 100644 --- a/tests/server/middleware/test_caching.py +++ b/tests/server/middleware/test_caching.py @@ -26,6 +26,7 @@ from pydantic import AnyUrl, BaseModel from fastmcp import Context, FastMCP from fastmcp.client.client import CallToolResult, Client +from fastmcp.client.elicitation import ElicitResult from fastmcp.client.transports import FastMCPTransport from fastmcp.prompts.base import Message, Prompt from fastmcp.prompts.function_prompt import FunctionPrompt @@ -922,22 +923,18 @@ class TestCachingWithInputRequiredResults: @staticmethod def _ask() -> mcp_types.InputRequiredResult: - return mcp_types.InputRequiredResult( - result_type="input_required", - input_requests={ - "q": mcp_types.ElicitRequest( - method="elicitation/create", - params=mcp_types.ElicitRequestFormParams( - message="Which quarter?", - requested_schema={ - "type": "object", - "properties": {"q": {"type": "string"}}, - "required": ["q"], - }, - ), - ) + params = mcp_types.ElicitRequestFormParams( + message="Which quarter?", + requested_schema={ + "type": "object", + "properties": {"q": {"type": "string"}}, + "required": ["q"], }, ) + request = mcp_types.ElicitRequest(method="elicitation/create", params=params) + return mcp_types.InputRequiredResult( + result_type="input_required", input_requests={"q": request} + ) @classmethod def _server(cls) -> FastMCP: @@ -966,32 +963,24 @@ class TestCachingWithInputRequiredResults: @staticmethod async def _handler(message, response_type, params, ctx): - from fastmcp.client.elicitation import ElicitResult - return ElicitResult(action="accept", content=response_type(q="Q3")) - async def test_tool_guard_completes_under_caching(self): - async with Client( - self._server(), mode="auto", elicitation_handler=self._handler - ) as client: - result = await client.call_tool("summarize_tool", {}) + def _client(self) -> Client: + return Client(self._server(), mode="auto", elicitation_handler=self._handler) + async def test_tool_guard_completes_under_caching(self): + async with self._client() as client: + result = await client.call_tool("summarize_tool", {}) assert result.data == "Summary for Q3" async def test_prompt_guard_completes_under_caching(self): - async with Client( - self._server(), mode="auto", elicitation_handler=self._handler - ) as client: + async with self._client() as client: result = await client.get_prompt("summarize") - assert result.messages[0].content.text == "Summary for Q3" async def test_resource_guard_completes_under_caching(self): - async with Client( - self._server(), mode="auto", elicitation_handler=self._handler - ) as client: + async with self._client() as client: result = await client.read_resource("report://x") - assert result[0].text == "Report for Q3" async def test_guard_still_asks_on_a_second_fresh_call(self): @@ -1000,12 +989,8 @@ class TestCachingWithInputRequiredResults: A second fresh flow has to be asked the same question; serving it a cached final answer would skip the component's own per-round logic. """ - mcp = self._server() - async with Client( - mcp, mode="auto", elicitation_handler=self._handler - ) as client: + async with self._client() as client: first = await client.get_prompt("summarize") second = await client.get_prompt("summarize") - assert first.messages[0].content.text == "Summary for Q3" assert second.messages[0].content.text == "Summary for Q3" diff --git a/tests/tasks/server/test_guard_reentrant.py b/tests/tasks/server/test_guard_reentrant.py index 93123734e..fb4cc52a7 100644 --- a/tests/tasks/server/test_guard_reentrant.py +++ b/tests/tasks/server/test_guard_reentrant.py @@ -11,9 +11,12 @@ the real interceptor and handlers via `task_helpers`. from __future__ import annotations +import asyncio from typing import Any import mcp_types +from fastmcp_tasks.context import get_task_scope +from fastmcp_tasks.input_store import acquire_update_lock, release_update_lock from mcp.shared.exceptions import MCPError from mcp_types import INTERNAL_ERROR @@ -311,6 +314,77 @@ async def test_partial_update_keeps_task_parked_on_remaining_request(): assert final.result["content"][0]["text"] == "one+two" +async def test_partial_update_waits_for_a_held_update_lock(): + """An update that arrives while another holds the lock must still land. + + SEP-2663 invites a client to answer a multi-request ask one key at a time, + so two updates can be in flight carrying *different* answers. Acknowledging + the one that loses the lock without storing its answer would leave the task + waiting forever on a key the client believes it already sent. + + The lock is taken out of band here so the contention is deterministic rather + than dependent on scheduling. + """ + mcp = FastMCP("lock-contention") + mcp.add_extension(TasksExtension()) + + @mcp.tool(task=True) + async def two_questions(ctx: Context) -> str | mcp_types.InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _input_required( + { + "first": _elicit_request("First?"), + "second": _elicit_request("Second?"), + } + ) + return f"{_answer(responses, 'first')}+{_answer(responses, 'second')}" + + async with running_task_server(mcp): + created = await submit_task(mcp, "two_questions", {}) + parked = await wait_for_task( + mcp, created.task_id, target_states=frozenset({"input_required"}) + ) + assert parked.input_requests is not None + first = _key_asking(parked.input_requests, "First?") + second = _key_asking(parked.input_requests, "Second?") + + docket = mcp._docket + assert docket is not None + scope = get_task_scope() + + # Simulate a concurrent update in progress. + assert await acquire_update_lock(docket, scope, created.task_id) + pending = asyncio.create_task( + update_task( + mcp, + created.task_id, + {first: {"action": "accept", "content": {"value": "one"}}}, + ) + ) + await asyncio.sleep(0.05) + assert not pending.done(), "update returned while the lock was held" + await release_update_lock(docket, scope, created.task_id) + await pending + + # The blocked answer landed, so only the other key remains outstanding. + still_parked = await get_task(mcp, created.task_id) + assert still_parked.status == "input_required" + assert still_parked.input_requests is not None + assert list(still_parked.input_requests) == [second] + + await update_task( + mcp, + created.task_id, + {second: {"action": "accept", "content": {"value": "two"}}}, + ) + final = await wait_for_task(mcp, created.task_id) + + assert final.status == "completed" + assert final.result is not None + assert final.result["content"][0]["text"] == "one+two" + + async def test_final_answer_keeps_task_parked_until_next_leg_is_durable(): """The last answer must not retire its outstanding marker early. From 8768921fd9299891dfd6dcd2478c37b34fa8145a Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 16:46:21 -0400 Subject: [PATCH 6/9] Split guard caching tests into their own module Keeps test_caching.py under its size limit by topic rather than by compressing the new tests. --- tests/server/middleware/test_caching.py | 92 ------------- .../server/middleware/test_caching_guards.py | 123 ++++++++++++++++++ 2 files changed, 123 insertions(+), 92 deletions(-) create mode 100644 tests/server/middleware/test_caching_guards.py diff --git a/tests/server/middleware/test_caching.py b/tests/server/middleware/test_caching.py index a88efbd3f..b45961bcf 100644 --- a/tests/server/middleware/test_caching.py +++ b/tests/server/middleware/test_caching.py @@ -26,7 +26,6 @@ from pydantic import AnyUrl, BaseModel from fastmcp import Context, FastMCP from fastmcp.client.client import CallToolResult, Client -from fastmcp.client.elicitation import ElicitResult from fastmcp.client.transports import FastMCPTransport from fastmcp.prompts.base import Message, Prompt from fastmcp.prompts.function_prompt import FunctionPrompt @@ -903,94 +902,3 @@ class TestAuthAwareCaching: assert {p.name for p in prompts} == {"public_prompt"} finally: auth_context_var.reset(tok) - - -class TestCachingWithInputRequiredResults: - """A multi-round-trip ask must survive `ResponseCachingMiddleware`. - - An ask carries no content of its own, so caching one would store an empty - result and the client would never see the question. Continuation legs must - also bypass the cache, since they share a cache key with a fresh call. - """ - - @staticmethod - def _answer(responses: mcp_types.InputResponses) -> str: - """The accepted value for the single question these guards ask.""" - result = responses["q"] - assert isinstance(result, mcp_types.ElicitResult) - assert result.content is not None - return str(result.content["q"]) - - @staticmethod - def _ask() -> mcp_types.InputRequiredResult: - params = mcp_types.ElicitRequestFormParams( - message="Which quarter?", - requested_schema={ - "type": "object", - "properties": {"q": {"type": "string"}}, - "required": ["q"], - }, - ) - request = mcp_types.ElicitRequest(method="elicitation/create", params=params) - return mcp_types.InputRequiredResult( - result_type="input_required", input_requests={"q": request} - ) - - @classmethod - def _server(cls) -> FastMCP: - mcp = FastMCP("cached-guards") - mcp.add_middleware(ResponseCachingMiddleware()) - - @mcp.tool - async def summarize_tool(ctx: Context) -> str | mcp_types.InputRequiredResult: - if ctx.input_responses is None: - return cls._ask() - return f"Summary for {cls._answer(ctx.input_responses)}" - - @mcp.prompt - async def summarize(ctx: Context) -> str | mcp_types.InputRequiredResult: - if ctx.input_responses is None: - return cls._ask() - return f"Summary for {cls._answer(ctx.input_responses)}" - - @mcp.resource("report://x") - async def report(ctx: Context) -> str | mcp_types.InputRequiredResult: - if ctx.input_responses is None: - return cls._ask() - return f"Report for {cls._answer(ctx.input_responses)}" - - return mcp - - @staticmethod - async def _handler(message, response_type, params, ctx): - return ElicitResult(action="accept", content=response_type(q="Q3")) - - def _client(self) -> Client: - return Client(self._server(), mode="auto", elicitation_handler=self._handler) - - async def test_tool_guard_completes_under_caching(self): - async with self._client() as client: - result = await client.call_tool("summarize_tool", {}) - assert result.data == "Summary for Q3" - - async def test_prompt_guard_completes_under_caching(self): - async with self._client() as client: - result = await client.get_prompt("summarize") - assert result.messages[0].content.text == "Summary for Q3" - - async def test_resource_guard_completes_under_caching(self): - async with self._client() as client: - result = await client.read_resource("report://x") - assert result[0].text == "Report for Q3" - - async def test_guard_still_asks_on_a_second_fresh_call(self): - """The ask must not be cached away for the next caller. - - A second fresh flow has to be asked the same question; serving it a - cached final answer would skip the component's own per-round logic. - """ - async with self._client() as client: - first = await client.get_prompt("summarize") - second = await client.get_prompt("summarize") - assert first.messages[0].content.text == "Summary for Q3" - assert second.messages[0].content.text == "Summary for Q3" diff --git a/tests/server/middleware/test_caching_guards.py b/tests/server/middleware/test_caching_guards.py new file mode 100644 index 000000000..e552dc49e --- /dev/null +++ b/tests/server/middleware/test_caching_guards.py @@ -0,0 +1,123 @@ +"""Response caching around multi-round-trip asks (SEP-2322). + +A guard component answers a call by *returning* an `InputRequiredResult` — a +request for client input rather than a final answer. Two things follow for +`ResponseCachingMiddleware`, and they apply equally to tools, prompts, and +resources: + +- An ask must never be stored. It carries no content of its own, so caching one + writes an empty result, and every later caller is served that emptiness + instead of being asked the question. +- A continuation leg must bypass the cache entirely. Cache keys are built from + the component's identity and arguments alone, so a continuation shares its key + with a fresh call: reading could hand this leg a prior flow's final answer, and + writing would hand a later fresh caller *this* flow's answer, skipping the + questions altogether. +""" + +import mcp_types + +from fastmcp import Context, FastMCP +from fastmcp.client.client import Client +from fastmcp.client.elicitation import ElicitResult +from fastmcp.server.middleware.caching import ResponseCachingMiddleware + + +def _ask() -> mcp_types.InputRequiredResult: + """The single-question ask every guard in this module returns.""" + params = mcp_types.ElicitRequestFormParams( + message="Which quarter?", + requested_schema={ + "type": "object", + "properties": {"q": {"type": "string"}}, + "required": ["q"], + }, + ) + request = mcp_types.ElicitRequest(method="elicitation/create", params=params) + return mcp_types.InputRequiredResult( + result_type="input_required", + input_requests={"q": request}, + ) + + +def _answer(responses: mcp_types.InputResponses) -> str: + """The accepted value for the question `_ask` poses.""" + result = responses["q"] + assert isinstance(result, mcp_types.ElicitResult) + assert result.content is not None + return str(result.content["q"]) + + +async def _handler(message, response_type, params, ctx): + """An elicitation handler that always answers "Q3".""" + return ElicitResult(action="accept", content=response_type(q="Q3")) + + +def cached_guard_server() -> FastMCP: + """A caching server whose tool, prompt, and resource are all guards.""" + mcp = FastMCP("cached-guards") + mcp.add_middleware(ResponseCachingMiddleware()) + + @mcp.tool + async def summarize_tool(ctx: Context) -> str | mcp_types.InputRequiredResult: + if ctx.input_responses is None: + return _ask() + return f"Summary for {_answer(ctx.input_responses)}" + + @mcp.prompt + async def summarize(ctx: Context) -> str | mcp_types.InputRequiredResult: + if ctx.input_responses is None: + return _ask() + return f"Summary for {_answer(ctx.input_responses)}" + + @mcp.resource("report://x") + async def report(ctx: Context) -> str | mcp_types.InputRequiredResult: + if ctx.input_responses is None: + return _ask() + return f"Report for {_answer(ctx.input_responses)}" + + return mcp + + +def guard_client() -> Client: + """A client that answers each round automatically.""" + return Client(cached_guard_server(), mode="auto", elicitation_handler=_handler) + + +class TestGuardsCompleteUnderCaching: + """Each component type drives its loop to a real answer with caching on.""" + + async def test_tool(self): + async with guard_client() as client: + result = await client.call_tool("summarize_tool", {}) + + assert result.data == "Summary for Q3" + + async def test_prompt(self): + async with guard_client() as client: + result = await client.get_prompt("summarize") + + assert result.messages[0].content.text == "Summary for Q3" + + async def test_resource(self): + async with guard_client() as client: + result = await client.read_resource("report://x") + + assert result[0].text == "Report for Q3" + + +class TestAsksAreNotCached: + """A stored ask would poison every later caller.""" + + async def test_second_fresh_flow_is_asked_again(self): + """A second fresh flow must be asked the same question. + + Serving it a cached final answer would skip the component's own + per-round logic — it would receive an answer it never supplied input for. + """ + async with guard_client() as client: + first = await client.get_prompt("summarize") + second = await client.get_prompt("summarize") + + assert first.messages[0].content.text == "Summary for Q3" + assert second.messages[0].content.text == "Summary for Q3" From f7eed91aa8116f4a7c88e5524c037a5d73c34549 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 16:58:58 -0400 Subject: [PATCH 7/9] Forward guard asks through prompt, resource, and template proxies A proxy has no back-channel to the real user, so driving a backend ask inside it failed outright. Surface it as a result for the parent, as ProxyTool does. --- .../fastmcp/server/providers/proxy.py | 89 ++++++- tests/server/test_mrtr_guards.py | 130 ---------- tests/server/test_mrtr_guards_components.py | 232 ++++++++++++++++++ 3 files changed, 313 insertions(+), 138 deletions(-) create mode 100644 tests/server/test_mrtr_guards_components.py diff --git a/fastmcp_slim/fastmcp/server/providers/proxy.py b/fastmcp_slim/fastmcp/server/providers/proxy.py index 78f1b1f17..c368af077 100644 --- a/fastmcp_slim/fastmcp/server/providers/proxy.py +++ b/fastmcp_slim/fastmcp/server/providers/proxy.py @@ -42,9 +42,13 @@ from fastmcp.client.transports.base import TransportOptions from fastmcp.exceptions import ResourceError from fastmcp.mcp_config import MCPConfig from fastmcp.prompts import Message, Prompt, PromptResult -from fastmcp.prompts.base import PromptArgument +from fastmcp.prompts.base import InputRequiredPromptResult, PromptArgument from fastmcp.resources import Resource, ResourceTemplate -from fastmcp.resources.base import ResourceContent, ResourceResult +from fastmcp.resources.base import ( + InputRequiredResourceResult, + ResourceContent, + ResourceResult, +) from fastmcp.resources.template import expand_uri_template from fastmcp.server.context import Context from fastmcp.server.dependencies import fastmcp_request_ctx, get_context @@ -118,6 +122,36 @@ def _proxy_upstream_error(error: Exception) -> MCPError: ) +async def _relay_read_resource( + client: Client, uri: str, ctx: Context | None +) -> ( + list[mcp_types.TextResourceContents | mcp_types.BlobResourceContents] + | mcp_types.InputRequiredResult +): + """Read a backend resource, surfacing a guard ask rather than driving it. + + Mirrors `ProxyTool.run`: on a modern backend the low-level session is used + so an `InputRequiredResult` (SEP-2322) comes back as a result for the parent + to forward, instead of the high-level client trying to answer it here — the + proxy has no back-channel to the real user, so driving it fails outright. + The inbound request's continuation state travels down so the backend guard + sees the client's answers on its own `ctx.input_responses`. + """ + if client.protocol_version not in MODERN_PROTOCOL_VERSIONS: + return await client.read_resource(uri) + result = await client._await_with_session_monitoring( + client.session.read_resource( + uri, + input_responses=ctx.input_responses if ctx else None, + request_state=ctx.request_state if ctx else None, + allow_input_required=True, + ) + ) + if isinstance(result, mcp_types.InputRequiredResult): + return result + return list(result.contents) + + def _stash_proxy_request_context(client: Client, ctx: Context) -> None: """Stash the proxy's ``RequestContext`` on a ``ProxyClient`` before a backend call. @@ -418,9 +452,12 @@ class ProxyResource(Resource): ) as span: span.set_attribute("fastmcp.provider.type", "ProxyProvider") client = await self._get_client() + ctx = get_context() async with client: - _stash_proxy_request_context(client, get_context()) - result = await client.read_resource(backend_uri) + _stash_proxy_request_context(client, ctx) + result = await _relay_read_resource(client, backend_uri, ctx) + if isinstance(result, mcp_types.InputRequiredResult): + return InputRequiredResourceResult(result) if not result: raise ResourceError( f"Remote server returned empty content for {backend_uri}" @@ -516,9 +553,28 @@ class ProxyTemplate(ResourceTemplate): backend_template = self._backend_uri_template or self.uri_template parameterized_uri = expand_uri_template(backend_template, params) client = await self._get_client() + ctx = context or get_context() async with client: - _stash_proxy_request_context(client, context or get_context()) - result = await client.read_resource(parameterized_uri) + _stash_proxy_request_context(client, ctx) + result = await _relay_read_resource(client, parameterized_uri, ctx) + + if isinstance(result, mcp_types.InputRequiredResult): + # The backend template asked for input. `InputRequiredResourceResult` + # is a `ResourceResult`, so caching it on the returned resource lets + # the ask ride the ordinary read path out to the parent's wire + # handler, which unwraps it. + return ProxyResource( + client_factory=self._client_factory, + uri=parameterized_uri, + name=self.name, + title=self.title, + description=self.description, + mime_type=self.mime_type or "text/plain", + icons=self.icons, + meta=self.meta, + tags=get_fastmcp_metadata(self.meta).get("tags", []), + _cached_content=InputRequiredResourceResult(result), + ) if not result: raise ResourceError( @@ -635,9 +691,26 @@ class ProxyPrompt(Prompt): ) as span: span.set_attribute("fastmcp.provider.type", "ProxyProvider") client = await self._get_client() + ctx = get_context() async with client: - _stash_proxy_request_context(client, get_context()) - result = await client.get_prompt(backend_name, arguments) + _stash_proxy_request_context(client, ctx) + if client.protocol_version in MODERN_PROTOCOL_VERSIONS: + # See `_relay_read_resource`: surface a backend guard's ask + # instead of trying to answer it inside the proxy. + raw = await client._await_with_session_monitoring( + client.session.get_prompt( + backend_name, + arguments, + input_responses=ctx.input_responses if ctx else None, + request_state=ctx.request_state if ctx else None, + allow_input_required=True, + ) + ) + if isinstance(raw, mcp_types.InputRequiredResult): + return InputRequiredPromptResult(raw) + result = raw + else: + result = await client.get_prompt(backend_name, arguments) # Convert GetPromptResult to PromptResult, preserving meta from result # (not the static prompt meta which includes fastmcp tags) # Convert PromptMessages to Messages diff --git a/tests/server/test_mrtr_guards.py b/tests/server/test_mrtr_guards.py index 134eda8b4..8355bdebe 100644 --- a/tests/server/test_mrtr_guards.py +++ b/tests/server/test_mrtr_guards.py @@ -1215,133 +1215,3 @@ class TestHttpTransport: assert asked == ["Where would you like to fly?", "When to Paris?"] assert result.data == "Booked Paris on 2026-08-01" - - -class TestPromptGuard: - """`InputRequiredResult` is a result type, not a tools/call feature, so a - prompt can ask for input the same way a tool does (SEP-2322).""" - - @staticmethod - def _context_prompt_server() -> FastMCP: - mcp = FastMCP("prompt-guard") - - @mcp.prompt - async def summarize(ctx: Context) -> str | InputRequiredResult: - responses = ctx.input_responses - if responses is None: - return _ask( - _elicit("context", "What context?", "context"), - key="context", - request_state=None, - ) - return f"Summarizing with {_accepted(responses, 'context')['context']}" - - return mcp - - async def test_prompt_emits_input_required(self): - """The asking round reaches the wire as an InputRequiredResult.""" - async with Client(self._context_prompt_server(), mode="auto") as client: - result = await client.session.get_prompt( - "summarize", allow_input_required=True - ) - - assert isinstance(result, InputRequiredResult) - assert "context" in result.input_requests - - async def test_prompt_completes_with_responses(self): - """Answering the ask renders the prompt on the next round.""" - mcp = self._context_prompt_server() - async with Client(mcp, mode="auto") as client: - ask = await client.session.get_prompt( - "summarize", allow_input_required=True - ) - assert isinstance(ask, InputRequiredResult) - done = await client.session.get_prompt( - "summarize", - input_responses={ - "context": mcp_types.ElicitResult( - action="accept", content={"context": "quarterly report"} - ) - }, - ) - - assert done.messages[0].content.text == ("Summarizing with quarterly report") - - async def test_prompt_guard_rejected_on_handshake_era(self): - """The result type only exists at 2026-07-28, so an older connection - gets the era named rather than a generic invalid-result failure.""" - async with Client(self._context_prompt_server(), mode="legacy") as client: - with pytest.raises(MCPError, match="2026-07-28"): - await client.session.get_prompt("summarize") - - -class TestResourceGuard: - """Resources and templates ask for input the same way tools and prompts do.""" - - @staticmethod - def _resource_server() -> FastMCP: - mcp = FastMCP("resource-guard") - - @mcp.resource("data://report") - async def report(ctx: Context) -> str | InputRequiredResult: - responses = ctx.input_responses - if responses is None: - return _ask( - _elicit("context", "Which quarter?", "context"), - key="context", - request_state=None, - ) - return f"Report for {_accepted(responses, 'context')['context']}" - - @mcp.resource("data://report/{section}") - async def section_report( - section: str, ctx: Context - ) -> str | InputRequiredResult: - responses = ctx.input_responses - if responses is None: - return _ask( - _elicit("context", f"Which quarter for {section}?", "context"), - key="context", - request_state=None, - ) - quarter = _accepted(responses, "context")["context"] - return f"{section} for {quarter}" - - return mcp - - async def test_resource_emits_input_required(self): - async with Client(self._resource_server(), mode="auto") as client: - result = await client.session.read_resource( - "data://report", allow_input_required=True - ) - - assert isinstance(result, InputRequiredResult) - assert "context" in result.input_requests - - async def test_resource_completes_with_responses(self): - async with Client(self._resource_server(), mode="auto") as client: - done = await client.session.read_resource( - "data://report", - input_responses={ - "context": mcp_types.ElicitResult( - action="accept", content={"context": "Q3"} - ) - }, - ) - - assert done.contents[0].text == "Report for Q3" - - async def test_resource_template_emits_input_required(self): - """Templates share the converter, so the ask survives there too.""" - async with Client(self._resource_server(), mode="auto") as client: - result = await client.session.read_resource( - "data://report/revenue", allow_input_required=True - ) - - assert isinstance(result, InputRequiredResult) - assert "context" in result.input_requests - - async def test_resource_guard_rejected_on_handshake_era(self): - async with Client(self._resource_server(), mode="legacy") as client: - with pytest.raises(MCPError, match="2026-07-28"): - await client.session.read_resource("data://report") diff --git a/tests/server/test_mrtr_guards_components.py b/tests/server/test_mrtr_guards_components.py new file mode 100644 index 000000000..21a3d2a47 --- /dev/null +++ b/tests/server/test_mrtr_guards_components.py @@ -0,0 +1,232 @@ +"""Guard-mode multi-round-trip for prompts and resources (SEP-2322). + +`InputRequiredResult` is a *result type*, not a `tools/call` feature: any +request can resolve to one. A prompt or resource asks for client input exactly +the way a tool does — return the ask, read `ctx.input_responses` on the round +that follows. + +These tests cover the emission side for prompts, concrete resources, and +resource templates, the 2026-07-28 era gate, and the proxy path, where the ask +must be forwarded to the parent rather than answered inside the proxy (a proxy +has no back-channel to the real user). Tool guards live in +``tests/server/test_mrtr_guards.py``. +""" + +from __future__ import annotations + +import mcp_types +import pytest +from mcp.shared.exceptions import MCPError +from mcp_types import ElicitRequest, InputRequiredResult + +from fastmcp import Client, Context, FastMCP +from fastmcp.client.elicitation import ElicitResult + + +def _elicit(key: str, message: str, field: str) -> ElicitRequest: + """A single-field form elicitation request keyed by ``key``.""" + params = mcp_types.ElicitRequestFormParams( + message=message, + requested_schema={ + "type": "object", + "properties": {field: {"type": "string"}}, + "required": [field], + }, + ) + return ElicitRequest(method="elicitation/create", params=params) + + +def _ask( + request: ElicitRequest, key: str, request_state: str | None +) -> InputRequiredResult: + return InputRequiredResult( + result_type="input_required", + input_requests={key: request}, + request_state=request_state, + ) + + +def _accepted(responses: mcp_types.InputResponses, key: str) -> dict[str, object]: + """The accepted form content for one answered elicitation.""" + answer = responses[key] + assert isinstance(answer, mcp_types.ElicitResult) + assert answer.content is not None + return dict(answer.content) + + +def _modern_proxy(backend: FastMCP) -> FastMCP: + """A proxy whose backend client negotiates the modern era, so the backend + can emit an `InputRequiredResult` for the proxy to round-trip.""" + from fastmcp.server.providers.proxy import FastMCPProxy, ProxyClient + + return FastMCPProxy(client_factory=lambda: ProxyClient(backend, mode="auto")) + + + + +class TestPromptGuard: + """`InputRequiredResult` is a result type, not a tools/call feature, so a + prompt can ask for input the same way a tool does (SEP-2322).""" + + @staticmethod + def _context_prompt_server() -> FastMCP: + mcp = FastMCP("prompt-guard") + + @mcp.prompt + async def summarize(ctx: Context) -> str | InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _ask( + _elicit("context", "What context?", "context"), + key="context", + request_state=None, + ) + return f"Summarizing with {_accepted(responses, 'context')['context']}" + + return mcp + + async def test_prompt_emits_input_required(self): + """The asking round reaches the wire as an InputRequiredResult.""" + async with Client(self._context_prompt_server(), mode="auto") as client: + result = await client.session.get_prompt( + "summarize", allow_input_required=True + ) + + assert isinstance(result, InputRequiredResult) + assert "context" in result.input_requests + + async def test_prompt_completes_with_responses(self): + """Answering the ask renders the prompt on the next round.""" + mcp = self._context_prompt_server() + async with Client(mcp, mode="auto") as client: + ask = await client.session.get_prompt( + "summarize", allow_input_required=True + ) + assert isinstance(ask, InputRequiredResult) + done = await client.session.get_prompt( + "summarize", + input_responses={ + "context": mcp_types.ElicitResult( + action="accept", content={"context": "quarterly report"} + ) + }, + ) + + assert done.messages[0].content.text == ("Summarizing with quarterly report") + + async def test_prompt_guard_rejected_on_handshake_era(self): + """The result type only exists at 2026-07-28, so an older connection + gets the era named rather than a generic invalid-result failure.""" + async with Client(self._context_prompt_server(), mode="legacy") as client: + with pytest.raises(MCPError, match="2026-07-28"): + await client.session.get_prompt("summarize") + + +class TestResourceGuard: + """Resources and templates ask for input the same way tools and prompts do.""" + + @staticmethod + def _resource_server() -> FastMCP: + mcp = FastMCP("resource-guard") + + @mcp.resource("data://report") + async def report(ctx: Context) -> str | InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _ask( + _elicit("context", "Which quarter?", "context"), + key="context", + request_state=None, + ) + return f"Report for {_accepted(responses, 'context')['context']}" + + @mcp.resource("data://report/{section}") + async def section_report( + section: str, ctx: Context + ) -> str | InputRequiredResult: + responses = ctx.input_responses + if responses is None: + return _ask( + _elicit("context", f"Which quarter for {section}?", "context"), + key="context", + request_state=None, + ) + quarter = _accepted(responses, "context")["context"] + return f"{section} for {quarter}" + + return mcp + + async def test_resource_emits_input_required(self): + async with Client(self._resource_server(), mode="auto") as client: + result = await client.session.read_resource( + "data://report", allow_input_required=True + ) + + assert isinstance(result, InputRequiredResult) + assert "context" in result.input_requests + + async def test_resource_completes_with_responses(self): + async with Client(self._resource_server(), mode="auto") as client: + done = await client.session.read_resource( + "data://report", + input_responses={ + "context": mcp_types.ElicitResult( + action="accept", content={"context": "Q3"} + ) + }, + ) + + assert done.contents[0].text == "Report for Q3" + + async def test_resource_template_emits_input_required(self): + """Templates share the converter, so the ask survives there too.""" + async with Client(self._resource_server(), mode="auto") as client: + result = await client.session.read_resource( + "data://report/revenue", allow_input_required=True + ) + + assert isinstance(result, InputRequiredResult) + assert "context" in result.input_requests + + async def test_resource_guard_rejected_on_handshake_era(self): + async with Client(self._resource_server(), mode="legacy") as client: + with pytest.raises(MCPError, match="2026-07-28"): + await client.session.read_resource("data://report") + + +class TestProxyForwarding: + """A proxy forwards a backend guard's ask instead of answering it.""" + + async def test_guard_prompt_round_trips_through_proxy(self): + """A guard prompt behind a proxy surfaces its ask instead of the proxy + trying to answer it. The proxy has no back-channel to the real user, so + driving the ask internally fails with "Elicitation not supported".""" + backend = TestPromptGuard._context_prompt_server() + + async def answer(message, response_type, params, ctx): + return ElicitResult( + action="accept", content=response_type(context="quarterly report") + ) + + async with Client( + _modern_proxy(backend), mode="auto", elicitation_handler=answer + ) as client: + result = await client.get_prompt("summarize") + + assert result.messages[0].content.text == "Summarizing with quarterly report" + + async def test_guard_resource_round_trips_through_proxy(self): + """Concrete resources and templates forward the ask the same way.""" + backend = TestResourceGuard._resource_server() + + async def answer(message, response_type, params, ctx): + return ElicitResult(action="accept", content=response_type(context="Q3")) + + async with Client( + _modern_proxy(backend), mode="auto", elicitation_handler=answer + ) as client: + direct = await client.read_resource("data://report") + templated = await client.read_resource("data://report/revenue") + + assert direct[0].text == "Report for Q3" + assert templated[0].text == "revenue for Q3" From 46cd0c793357494545bd05873cc8293257a80bb9 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:03:04 -0400 Subject: [PATCH 8/9] Format the split guard-components test module --- tests/server/test_mrtr_guards_components.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/server/test_mrtr_guards_components.py b/tests/server/test_mrtr_guards_components.py index 21a3d2a47..ad896aa87 100644 --- a/tests/server/test_mrtr_guards_components.py +++ b/tests/server/test_mrtr_guards_components.py @@ -62,8 +62,6 @@ def _modern_proxy(backend: FastMCP) -> FastMCP: return FastMCPProxy(client_factory=lambda: ProxyClient(backend, mode="auto")) - - class TestPromptGuard: """`InputRequiredResult` is a result type, not a tools/call feature, so a prompt can ask for input the same way a tool does (SEP-2322).""" From 5745323ecd992c920096dcf7934c05c8830afba7 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 17:15:30 -0400 Subject: [PATCH 9/9] Cover trace propagation through the modern proxy relay --- .../fastmcp/server/providers/proxy.py | 4 +- .../server/telemetry/test_provider_tracing.py | 54 ++++++++++++++++++- 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/fastmcp_slim/fastmcp/server/providers/proxy.py b/fastmcp_slim/fastmcp/server/providers/proxy.py index c368af077..b3965446b 100644 --- a/fastmcp_slim/fastmcp/server/providers/proxy.py +++ b/fastmcp_slim/fastmcp/server/providers/proxy.py @@ -135,7 +135,9 @@ async def _relay_read_resource( to forward, instead of the high-level client trying to answer it here — the proxy has no back-channel to the real user, so driving it fails outright. The inbound request's continuation state travels down so the backend guard - sees the client's answers on its own `ctx.input_responses`. + sees the client's answers on its own `ctx.input_responses`. Trace context + still propagates: the SDK's JSON-RPC dispatcher injects it on every outgoing + request (SEP-414), below whichever client layer issued the call. """ if client.protocol_version not in MODERN_PROTOCOL_VERSIONS: return await client.read_resource(uri) diff --git a/tests/server/telemetry/test_provider_tracing.py b/tests/server/telemetry/test_provider_tracing.py index 7118044a2..50c33f8fe 100644 --- a/tests/server/telemetry/test_provider_tracing.py +++ b/tests/server/telemetry/test_provider_tracing.py @@ -4,7 +4,9 @@ from __future__ import annotations from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from fastmcp import FastMCP +from fastmcp import Client, Context, FastMCP +from fastmcp.server.providers.proxy import FastMCPProxy, ProxyClient +from fastmcp.telemetry import TRACE_PARENT_KEY class TestFastMCPProviderTracing: @@ -131,3 +133,53 @@ class TestProviderSpanHierarchy: assert child_span.parent is not None assert delegate_span.parent.span_id == parent_span.context.span_id assert child_span.parent.span_id == delegate_span.context.span_id + + +class TestModernProxyTracePropagation: + """A modern proxy relays resources and prompts through the low-level + session so a backend guard's ask can surface (SEP-2322). That path skips + the high-level client's trace injection, so the relay must stamp the + outgoing `_meta` itself — otherwise every modern proxy read breaks the + distributed trace, not only the guard rounds.""" + + @staticmethod + def _backend(seen: dict[str, dict]) -> FastMCP: + backend = FastMCP("trace-backend") + + def record(kind: str, ctx: Context) -> None: + rc = ctx.request_context + seen[kind] = dict(rc.meta) if rc is not None and rc.meta else {} + + @backend.resource("data://x") + async def concrete(ctx: Context) -> str: + record("resource", ctx) + return "ok" + + @backend.resource("data://{part}/y") + async def templated(part: str, ctx: Context) -> str: + record("template", ctx) + return "ok" + + @backend.prompt + async def greet(ctx: Context) -> str: + record("prompt", ctx) + return "ok" + + return backend + + async def test_traceparent_reaches_backend( + self, trace_exporter: InMemorySpanExporter + ): + seen: dict[str, dict] = {} + proxy = FastMCPProxy( + client_factory=lambda: ProxyClient(self._backend(seen), mode="auto") + ) + + async with Client(proxy, mode="auto") as client: + await client.read_resource("data://x") + await client.read_resource("data://p/y") + await client.get_prompt("greet") + + assert TRACE_PARENT_KEY in seen["resource"] + assert TRACE_PARENT_KEY in seen["template"] + assert TRACE_PARENT_KEY in seen["prompt"]