From 17954f569dd2881ecf2013dd6f940132c0aedbaf Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Mon, 6 Jul 2026 21:30:03 -0400 Subject: [PATCH] Deprecate ctx.sample/sample_step and era-gate sampling+elicit on modern connections --- fastmcp_slim/fastmcp/server/context.py | 80 ++++++++++++++-- tests/server/test_protocol_eras.py | 121 ++++++++++++++++++++++--- 2 files changed, 181 insertions(+), 20 deletions(-) diff --git a/fastmcp_slim/fastmcp/server/context.py b/fastmcp_slim/fastmcp/server/context.py index 3d0709736..e76c6ab6b 100644 --- a/fastmcp_slim/fastmcp/server/context.py +++ b/fastmcp_slim/fastmcp/server/context.py @@ -21,12 +21,13 @@ from mcp_types import ( ) from mcp_types import Prompt as SDKPrompt from mcp_types import Resource as SDKResource +from mcp_types.version import MODERN_PROTOCOL_VERSIONS from pydantic.networks import AnyUrl from typing_extensions import TypeVar from uncalled_for import SharedContext import fastmcp -from fastmcp.exceptions import FastMCPDeprecationWarning +from fastmcp.exceptions import FastMCPDeprecationWarning, ToolError from fastmcp.resources.base import ResourceResult from fastmcp.server.dependencies import FastMCPRequestContext, fastmcp_request_ctx from fastmcp.server.elicitation import ( @@ -79,6 +80,47 @@ ResultT = TypeVar("ResultT", default=str) # Import ToolChoiceOption from sampling module (after other imports) from fastmcp.server.sampling.run import ToolChoiceOption # noqa: E402 +# Warn-once guards for the sampling deprecation. Server-initiated createMessage +# was removed from MCP as of 2026-07-28 (SEP-2577); the warning fires a single +# time per process to flag that ctx.sample/ctx.sample_step are on their way out. +_sample_deprecation_warned = False + +_SAMPLING_DEPRECATION_MESSAGE = ( + "ctx.sample() and ctx.sample_step() are deprecated and will be removed in a " + "future FastMCP release. They rely on server-initiated createMessage " + "requests, which were removed from MCP as of 2026-07-28 (SEP-2577), so they " + "work only on session-based (handshake-era) connections. Call an LLM " + "directly from your server instead." +) + +_SAMPLING_MODERN_ERROR = ( + "server-initiated sampling is not available on MCP 2026-07-28 connections; " + "SEP-2577 removed it — call an LLM from your server instead." +) + +_ELICIT_MODERN_ERROR = ( + "elicitation via server-initiated requests is unavailable on 2026-07-28 " + "connections." +) + + +def _warn_sampling_deprecated() -> None: + """Emit the sampling deprecation warning once per process. + + Gated on ``settings.deprecation_warnings`` like every other FastMCP + deprecation; fires a single time (module-level flag) rather than per call. + """ + global _sample_deprecation_warned + if _sample_deprecation_warned or not fastmcp.settings.deprecation_warnings: + return + _sample_deprecation_warned = True + warnings.warn( + _SAMPLING_DEPRECATION_MESSAGE, + FastMCPDeprecationWarning, + stacklevel=3, + ) + + _current_context: ContextVar[Context | None] = ContextVar("context", default=None) TransportType = Literal["stdio", "sse", "streamable-http"] @@ -862,6 +904,20 @@ class Context: return await self.request_context.close_sse_stream() + def _is_modern_protocol(self) -> bool: + """True when the negotiated MCP protocol era removed the back-channel. + + Reads the negotiated protocol version from the active request context. + The 2026-07-28 era (SEP-2577) has no back-channel for server-initiated + requests such as sampling/elicitation. Returns False when no request + context is available (e.g. background task or pre-session), leaving the + existing wire path to surface its own error. + """ + rc = self.request_context + if rc is None: + return False + return rc.protocol_version in MODERN_PROTOCOL_VERSIONS + async def sample_step( self, messages: str | Sequence[str | SamplingMessage], @@ -926,6 +982,9 @@ class Context: # Continue with tool results messages = step.history """ + _warn_sampling_deprecated() + if self._is_modern_protocol(): + raise ToolError(_SAMPLING_MODERN_ERROR) return await sample_step_impl( self, messages=messages, @@ -1027,12 +1086,15 @@ class Context: - .result: The typed result (str for text, parsed object for structured) - .history: All messages exchanged during sampling - Note: - Background task support for sampling is planned for a future release. - Currently, sampling in background tasks requires using the low-level - session.create_message() API directly. + Deprecated: + Server-initiated sampling relies on the createMessage back-channel, + which MCP removed as of 2026-07-28 (SEP-2577). This method works only + on session-based (handshake-era) connections and will be removed in a + future FastMCP release. Call an LLM directly from your server instead. """ - # TODO: Add background task support similar to elicit() when is_background_task + _warn_sampling_deprecated() + if self._is_modern_protocol(): + raise ToolError(_SAMPLING_MODERN_ERROR) return await sample_impl( # ty: ignore[invalid-return-type] self, messages=messages, @@ -1213,6 +1275,12 @@ class Context: schema=config.schema, ) else: + # Foreground push path: server-initiated elicitation needs a + # back-channel, which the 2026-07-28 era removed (SEP-2577). Raise a + # clear era-aware error before hitting the wire instead of the SDK's + # opaque "Method not found". Handshake-era behavior is unchanged. + if self._is_modern_protocol(): + raise ToolError(_ELICIT_MODERN_ERROR) # Standard request mode: use session.elicit directly result = await self.session.elicit( message=message, diff --git a/tests/server/test_protocol_eras.py b/tests/server/test_protocol_eras.py index ac56cc960..e0df7013c 100644 --- a/tests/server/test_protocol_eras.py +++ b/tests/server/test_protocol_eras.py @@ -321,21 +321,12 @@ async def test_list_roots_degradation_message_is_clear_on_modern(push_server): assert "back-channel" in message and "server-initiated" in message -@pytest.mark.xfail( - strict=True, - reason=( - "sdk-feedback.md #10: elicitation/sampling attach a related_request_id, " - "so at 2026 the request reaches the client _on_request and fails with a " - "bare 'Method not found' KeyError instead of an era-aware " - "NoBackChannelError. list_roots (no related id) already degrades " - "clearly; elicit/sample do not." - ), -) @pytest.mark.parametrize("tool", ["do_elicit", "do_sample"]) async def test_elicit_sample_degradation_message_is_clear_on_modern(push_server, tool): - """Characterizes the inconsistency in #10: we WANT elicit/sample to surface - an era-aware message (like list_roots does). Currently they surface a bare - 'Method not found', so this xfails strict until the SDK unifies the path. + """FastMCP era-gates elicit/sample: on a 2026-07-28 connection they raise a + clear, era-aware error before hitting the wire, instead of the SDK's opaque + 'Method not found' (sdk-feedback.md #10). Both messages name the removed + server-initiated capability so the caller knows why the request degraded. """ async with SDKClient( _server(push_server), @@ -346,7 +337,109 @@ async def test_elicit_sample_degradation_message_is_clear_on_modern(push_server, result = await client.call_tool(tool, {}) assert result.is_error is True message = " ".join(_texts(result.content)).lower() - assert "back-channel" in message or "server-initiated" in message + assert "server-initiated" in message + + +# --------------------------------------------------------------------------- +# 3b. Sampling deprecation warning (SEP-2577): ctx.sample/ctx.sample_step warn +# --------------------------------------------------------------------------- + + +@pytest.fixture +def reset_sample_warn_flag(): + """Reset the process-wide warn-once flag so a warning can be observed.""" + import fastmcp.server.context as context_module + + original = context_module._sample_deprecation_warned + context_module._sample_deprecation_warned = False + try: + yield + finally: + context_module._sample_deprecation_warned = original + + +@pytest.mark.parametrize("method", ["do_sample", "do_sample_step"]) +async def test_sampling_emits_deprecation_warning(reset_sample_warn_flag, method): + """`ctx.sample()` and `ctx.sample_step()` emit a FastMCPDeprecationWarning + naming SEP-2577 and the server-side-LLM migration path.""" + from fastmcp.exceptions import FastMCPDeprecationWarning + + mcp = FastMCP("warn") + + @mcp.tool + async def do_sample(ctx: Context) -> str: + await ctx.sample("hello") + return "ok" + + @mcp.tool + async def do_sample_step(ctx: Context) -> str: + await ctx.sample_step("hello") + return "ok" + + with pytest.warns(FastMCPDeprecationWarning, match="SEP-2577"): + async with SDKClient( + _server(mcp), mode="legacy", sampling_callback=_sampling_cb + ) as client: + await client.call_tool(method, {}) + + +async def test_sampling_deprecation_warning_fires_once_per_process( + reset_sample_warn_flag, +): + """The deprecation warning is warn-once: a second sample call in the same + process does not re-warn.""" + from fastmcp.exceptions import FastMCPDeprecationWarning + + mcp = FastMCP("warn-once") + + @mcp.tool + async def do_sample(ctx: Context) -> str: + await ctx.sample("hello") + return "ok" + + with pytest.warns(FastMCPDeprecationWarning): + async with SDKClient( + _server(mcp), mode="legacy", sampling_callback=_sampling_cb + ) as client: + await client.call_tool("do_sample", {}) + + import warnings as _warnings + + with _warnings.catch_warnings(): + _warnings.simplefilter("error", FastMCPDeprecationWarning) + async with SDKClient( + _server(mcp), mode="legacy", sampling_callback=_sampling_cb + ) as client: + result = await client.call_tool("do_sample", {}) + assert result.is_error is False + + +async def test_sampling_deprecation_warning_suppressible_via_settings( + reset_sample_warn_flag, monkeypatch +): + """Setting `deprecation_warnings=False` suppresses the sampling warning, + matching the house pattern for every other FastMCP deprecation.""" + import warnings as _warnings + + import fastmcp + from fastmcp.exceptions import FastMCPDeprecationWarning + + monkeypatch.setattr(fastmcp.settings, "deprecation_warnings", False) + + mcp = FastMCP("no-warn") + + @mcp.tool + async def do_sample(ctx: Context) -> str: + await ctx.sample("hello") + return "ok" + + with _warnings.catch_warnings(): + _warnings.simplefilter("error", FastMCPDeprecationWarning) + async with SDKClient( + _server(mcp), mode="legacy", sampling_callback=_sampling_cb + ) as client: + result = await client.call_tool("do_sample", {}) + assert result.is_error is False @pytest.mark.parametrize("mode", MODERN_MODES)