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 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/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/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/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py b/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py index af85dd08c..5e96e37dc 100644 --- a/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py +++ b/fastmcp_slim/fastmcp/server/mixins/mcp_operations.py @@ -28,6 +28,8 @@ from fastmcp.exceptions import ( NotFoundError, 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 @@ -295,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 @@ -306,8 +308,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, ...) @@ -320,13 +326,30 @@ 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( 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 +374,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/providers/proxy.py b/fastmcp_slim/fastmcp/server/providers/proxy.py index cd6d888b2..42e89200e 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,38 @@ 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`. 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) + 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 +454,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 +555,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 +693,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/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..5193d5b84 100644 --- a/fastmcp_tasks/fastmcp_tasks/handlers.py +++ b/fastmcp_tasks/fastmcp_tasks/handlers.py @@ -39,9 +39,9 @@ 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, is_cancelled, load_current_leg, load_task_args, @@ -296,18 +296,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") @@ -342,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 @@ -352,21 +375,45 @@ 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) + # 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 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: + # 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/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" 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"] diff --git a/tests/server/test_mrtr_guards_components.py b/tests/server/test_mrtr_guards_components.py new file mode 100644 index 000000000..ad896aa87 --- /dev/null +++ b/tests/server/test_mrtr_guards_components.py @@ -0,0 +1,230 @@ +"""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" 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..fb4cc52a7 100644 --- a/tests/tasks/server/test_guard_reentrant.py +++ b/tests/tasks/server/test_guard_reentrant.py @@ -11,9 +11,14 @@ 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 from fastmcp import Context, FastMCP from fastmcp_tasks import TasksExtension @@ -58,6 +63,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"}) @@ -244,3 +257,187 @@ 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 + assert len(parked.input_requests) == 2 + + # 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, + {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_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. + + 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 + 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")