Deprecate ctx.sample/sample_step and era-gate sampling+elicit on modern connections

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,

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)