From 704b74b3ab9a27b66ffaf52e01d1bb0f539f1bee Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 26 Jul 2026 14:38:26 -0400 Subject: [PATCH] Update tests for the removed sampling and roots server API --- tests/client/client/test_error_handling.py | 35 --- tests/client/test_roots.py | 30 +- tests/client/test_sampling.py | 227 ++++----------- tests/conformance/server.py | 10 - tests/server/middleware/test_middleware.py | 4 - .../middleware/test_middleware_nested.py | 10 +- .../providers/proxy/test_proxy_client.py | 48 +++- tests/server/test_context.py | 23 -- tests/server/test_protocol_eras.py | 263 ++---------------- 9 files changed, 142 insertions(+), 508 deletions(-) diff --git a/tests/client/client/test_error_handling.py b/tests/client/client/test_error_handling.py index 7bb500d8b..8f40e2ce6 100644 --- a/tests/client/client/test_error_handling.py +++ b/tests/client/client/test_error_handling.py @@ -21,7 +21,6 @@ from fastmcp.client import Client from fastmcp.client.mixins.tools import _parse_call_tool_result from fastmcp.client.transports import FastMCPTransport from fastmcp.exceptions import PromptError, ResourceError, ToolError -from fastmcp.server.sampling.run import SamplingTool, execute_tools from fastmcp.server.server import FastMCP @@ -416,37 +415,3 @@ class TestLogLevel: for record in caplog.records ) - async def test_sampling_tool_error_with_custom_log_level(self, caplog): - """ToolError with custom log_level in sampling should log at specified level.""" - - async def custom_level_sampling_tool(x: int) -> int: - raise ToolError("Expected sampling error", log_level=logging.WARNING) - - tool = SamplingTool.from_function(custom_level_sampling_tool) - tool_use = ToolUseContent( - type="tool_use", - id="test-id", - name="custom_level_sampling_tool", - input={"x": 42}, - ) - - with caplog.at_level(logging.WARNING): - results = await execute_tools( - tool_calls=[tool_use], - tool_map={"custom_level_sampling_tool": tool}, - mask_error_details=False, - ) - - assert len(results) == 1 - assert results[0].is_error - assert "Expected sampling error" in results[0].content[0].text # type: ignore - assert any( - "Error calling sampling tool" in record.message - and record.levelname == "WARNING" - for record in caplog.records - ) - assert not any( - "Error calling sampling tool" in record.message - and record.levelname == "ERROR" - for record in caplog.records - ) diff --git a/tests/client/test_roots.py b/tests/client/test_roots.py index e71f0c849..609bbdfd5 100644 --- a/tests/client/test_roots.py +++ b/tests/client/test_roots.py @@ -1,16 +1,23 @@ import pytest +from mcp_types import Root from fastmcp import Client, Context, FastMCP @pytest.fixture def fastmcp_server(): + """A server that issues a handshake-era `roots/list` request. + + `Context` has no `list_roots()` — server-initiated requests are not part of + FastMCP's server API. This server reaches the SDK session directly to stand + in for a legacy upstream, so the client's `roots=` handling stays covered. + """ mcp = FastMCP() @mcp.tool async def list_roots(context: Context) -> list[str]: - roots = await context.list_roots() - return [str(r.uri) for r in roots] + result = await context.session.list_roots() + return [str(r.uri) for r in result.roots] return mcp @@ -36,10 +43,27 @@ class TestClientRoots: @pytest.mark.parametrize("roots", [["file://x/y/z", "file://x/y/z"]]) async def test_valid_roots(self, fastmcp_server: FastMCP, roots: list[str]): - # ctx.list_roots is a legacy-era server-initiated feature. + # `roots/list` is a server-initiated request, so it only exists on the + # handshake era; SEP-2577 removed it from the modern protocol. async with Client(fastmcp_server, mode="legacy", roots=roots) as client: result = await client.call_tool("list_roots", {}) assert result.data == [ "file://x/y/z", "file://x/y/z", ] + + async def test_roots_handler_answers_a_legacy_server(self, fastmcp_server: FastMCP): + """A callable `roots=` handler still answers a legacy server's request.""" + calls: list[object] = [] + + async def roots_handler(ctx) -> list[Root]: + calls.append(ctx) + return [Root(uri="file://from/handler")] # ty: ignore[invalid-argument-type] + + async with Client( + fastmcp_server, mode="legacy", roots=roots_handler + ) as client: + result = await client.call_tool("list_roots", {}) + + assert len(calls) == 1 + assert result.data == ["file://from/handler"] diff --git a/tests/client/test_sampling.py b/tests/client/test_sampling.py index 98ec74494..e82e71093 100644 --- a/tests/client/test_sampling.py +++ b/tests/client/test_sampling.py @@ -1,71 +1,95 @@ import json -from typing import cast -from unittest.mock import AsyncMock +import mcp_types import pytest from mcp_types import TextContent from pydantic_core import to_json from fastmcp import Client, Context, FastMCP from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams -from fastmcp.server.sampling import SamplingResult, SamplingTool from fastmcp.utilities.types import Image +async def _sample( + context: Context, + messages: list[SamplingMessage], + *, + system_prompt: str | None = None, +) -> str: + """Issue a handshake-era `sampling/createMessage` request from a server. + + `Context` has no `sample()` — server-initiated sampling is not part of + FastMCP's server API. These tests cover the *client* side, which must keep + answering a legacy server, so the stand-in server reaches the SDK session + directly. + """ + result = await context.session.create_message( + messages=messages, + system_prompt=system_prompt, + max_tokens=512, + related_request_id=context.origin_request_id, + ) + assert isinstance(result.content, TextContent) + return result.content.text + + @pytest.fixture def fastmcp_server(): mcp = FastMCP() @mcp.tool async def simple_sample(message: str, context: Context) -> str: - result = await context.sample("Hello, world!") - assert isinstance(result, SamplingResult) - assert result.text is not None - return result.text + return await _sample( + context, + [ + SamplingMessage( + role="user", + content=TextContent(type="text", text="Hello, world!"), + ) + ], + ) @mcp.tool async def sample_with_system_prompt(message: str, context: Context) -> str: - result = await context.sample("Hello, world!", system_prompt="You love FastMCP") - assert isinstance(result, SamplingResult) - assert result.text is not None - return result.text + return await _sample( + context, + [ + SamplingMessage( + role="user", + content=TextContent(type="text", text="Hello, world!"), + ) + ], + system_prompt="You love FastMCP", + ) @mcp.tool async def sample_with_messages(message: str, context: Context) -> str: - result = await context.sample( + return await _sample( + context, [ - "Hello!", SamplingMessage( - content=TextContent( - type="text", text="How can I assist you today?" - ), - role="assistant", + role="user", content=TextContent(type="text", text="Hello!") ), - ] + SamplingMessage( + role="assistant", + content=TextContent(type="text", text="How can I assist you today?"), + ), + ], ) - assert isinstance(result, SamplingResult) - assert result.text is not None - return result.text @mcp.tool async def sample_with_image(image_bytes: bytes, context: Context) -> str: image = Image(data=image_bytes) - - result = await context.sample( + return await _sample( + context, [ SamplingMessage( content=TextContent(type="text", text="What's in this image?"), role="user", ), - SamplingMessage( - content=image.to_image_content(), - role="user", - ), - ] + SamplingMessage(content=image.to_image_content(), role="user"), + ], ) - assert isinstance(result, SamplingResult) - assert result.text is not None - return result.text return mcp @@ -123,26 +147,6 @@ async def test_sampling_with_messages(fastmcp_server: FastMCP): assert result.data == "I need to think." -async def test_sampling_with_fallback(fastmcp_server: FastMCP): - openai_sampling_handler = AsyncMock(return_value="But I need to think") - - fastmcp_server = FastMCP( - sampling_handler=openai_sampling_handler, - ) - - @fastmcp_server.tool - async def sample_with_fallback(context: Context) -> str: - sampling_result = await context.sample("Do not think.") - return cast(TextContent, sampling_result).text - - client = Client(fastmcp_server) - - async with client: - call_tool_result = await client.call_tool("sample_with_fallback") - - assert call_tool_result.data == "But I need to think" - - async def test_sampling_with_image(fastmcp_server: FastMCP): def sampling_handler( messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext @@ -193,8 +197,6 @@ class TestSamplingDefaultCapabilities: {"sampling": {"tools": {}}}, ensuring compatibility with servers that don't recognize the tools sub-field (e.g. older Java MCP SDK). """ - import mcp_types - server = FastMCP() def handler( @@ -209,8 +211,6 @@ class TestSamplingDefaultCapabilities: async def test_set_sampling_callback_default_capabilities_omit_tools(self): """set_sampling_callback should also default to no tools capability.""" - import mcp_types - server = FastMCP() client = Client(server) client.set_sampling_callback(lambda msgs, params, ctx: "ok") @@ -220,8 +220,6 @@ class TestSamplingDefaultCapabilities: async def test_explicit_tools_capability_is_preserved(self): """Explicitly passing tools capability should be respected.""" - import mcp_types - server = FastMCP() def handler( @@ -238,118 +236,3 @@ class TestSamplingDefaultCapabilities: caps = client._session_kwargs["sampling_capabilities"] assert isinstance(caps, mcp_types.SamplingCapability) assert caps.tools is not None - - -class TestSamplingWithTools: - """Tests for sampling with tools functionality.""" - - async def test_sampling_with_tools_requires_capability(self): - """Test that sampling with tools raises error when client lacks capability.""" - import mcp_types - - from fastmcp.exceptions import ToolError - - server = FastMCP() - - def search(query: str) -> str: - """Search the web.""" - return f"Results for: {query}" - - @server.tool - async def sample_with_tool(context: Context) -> str: - # This should fail because the client doesn't advertise tools capability - result = await context.sample( - messages="Search for Python tutorials", - tools=[search], - ) - return str(result) - - def sampling_handler( - messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext - ) -> str: - return "Response" - - # Explicitly disable tools capability by passing SamplingCapability without tools - async with Client( - server, - mode="legacy", - sampling_handler=sampling_handler, - sampling_capabilities=mcp_types.SamplingCapability(), # No tools - ) as client: - with pytest.raises(ToolError, match="sampling.tools capability"): - await client.call_tool("sample_with_tool", {}) - - async def test_sampling_with_tools_fallback_handler_can_return_string(self): - """Test that fallback handler can return a string even when tools are provided. - - The LLM might choose not to use any tools and just return a text response. - """ - # This handler returns a string - valid even when tools are provided - simple_handler = AsyncMock(return_value="Direct response without tools") - - mcp = FastMCP(sampling_handler=simple_handler) - - def search(query: str) -> str: - """Search the web.""" - return f"Results for: {query}" - - @mcp.tool - async def sample_with_tool(context: Context) -> str: - result = await context.sample( - messages="Search for Python tutorials", - tools=[search], - ) - return result.text or "no text" - - # Client without sampling handler - will use server's fallback - async with Client(mcp) as client: - result = await client.call_tool("sample_with_tool", {}) - - # Handler returned string directly, which is treated as final text response - assert result.data == "Direct response without tools" - - def test_sampling_tool_schema(self): - """Test that SamplingTool generates correct schema.""" - - def search(query: str, limit: int = 10) -> str: - """Search the web for results.""" - return f"Results for: {query}" - - tool = SamplingTool.from_function(search) - assert tool.name == "search" - assert tool.description == "Search the web for results." - assert "query" in tool.parameters.get("properties", {}) - assert "limit" in tool.parameters.get("properties", {}) - - async def test_sampling_tool_run(self): - """Test that SamplingTool.run() executes correctly.""" - - def add(a: int, b: int) -> int: - """Add two numbers.""" - return a + b - - tool = SamplingTool.from_function(add) - result = await tool.run({"a": 5, "b": 3}) - assert result == 8 - - async def test_sampling_tool_run_async(self): - """Test that SamplingTool.run() works with async functions.""" - - async def async_multiply(a: int, b: int) -> int: - """Multiply two numbers.""" - return a * b - - tool = SamplingTool.from_function(async_multiply) - result = await tool.run({"a": 4, "b": 7}) - assert result == 28 - - def test_tool_choice_parameter(self): - """Test that tool_choice parameter accepts string literals.""" - from fastmcp.server.context import ToolChoiceOption - - # Verify ToolChoiceOption type accepts the valid string values - choices: list[ToolChoiceOption] = ["auto", "required", "none"] - assert len(choices) == 3 - assert "auto" in choices - assert "required" in choices - assert "none" in choices diff --git a/tests/conformance/server.py b/tests/conformance/server.py index 9a6a97c28..ac6b20aef 100644 --- a/tests/conformance/server.py +++ b/tests/conformance/server.py @@ -134,16 +134,6 @@ async def test_tool_with_progress(ctx: Context) -> str: return "Progress test complete." -@server.tool(name="test_sampling") -async def test_sampling(prompt: str, ctx: Context) -> str: - """Requests LLM sampling via the client.""" - result = await ctx.sample( - messages=[prompt], - result_type=str, - ) - return f"Sampling result: {result}" - - class _UserInfo(BaseModel): username: str email: str diff --git a/tests/server/middleware/test_middleware.py b/tests/server/middleware/test_middleware.py index c625a7d55..715413017 100644 --- a/tests/server/middleware/test_middleware.py +++ b/tests/server/middleware/test_middleware.py @@ -147,10 +147,6 @@ def mcp_server(recording_middleware): async def log_tool(context: Context) -> None: await context.info(message="test log") - @mcp.tool - async def sample_tool(context: Context) -> None: - await context.sample("hello") - mcp.add_middleware(recording_middleware) # Register a progress notification handler (v2 API: (ctx, params)). diff --git a/tests/server/middleware/test_middleware_nested.py b/tests/server/middleware/test_middleware_nested.py index cc2eea6c0..374a07cc8 100644 --- a/tests/server/middleware/test_middleware_nested.py +++ b/tests/server/middleware/test_middleware_nested.py @@ -149,10 +149,6 @@ def mcp_server(recording_middleware): async def log_tool(context: Context) -> None: await context.info(message="test log") - @mcp.tool - async def sample_tool(context: Context) -> None: - await context.sample("hello") - mcp.add_middleware(recording_middleware) # Register a progress notification handler (v2 API: (ctx, params)). @@ -205,10 +201,6 @@ class TestNestedMiddlewareHooks: async def log_tool(context: Context) -> None: await context.info(message="test log") - @mcp.tool - async def sample_tool(context: Context) -> None: - await context.sample("hello") - mcp.add_middleware(nested_middleware) return mcp @@ -510,7 +502,7 @@ class TestProxyServer: async with Client(proxy_server) as client: await client.list_tools() - assert TAGS == [{"add-tool"}, set(), set(), set()] + assert TAGS == [{"add-tool"}, set(), set()] class TestToolCallDenial: diff --git a/tests/server/providers/proxy/test_proxy_client.py b/tests/server/providers/proxy/test_proxy_client.py index bc3b861fc..7638ef1b9 100644 --- a/tests/server/providers/proxy/test_proxy_client.py +++ b/tests/server/providers/proxy/test_proxy_client.py @@ -8,6 +8,7 @@ from mcp_types import ( LoggingLevel, ModelHint, ModelPreferences, + Root, TextContent, ) from pydantic import BaseModel, Field @@ -58,6 +59,34 @@ class TestProxyClientEraDefault: assert cast(Client, factory()).mode == "auto" +async def _backend_list_roots(context: Context) -> list[Root]: + """Issue a handshake-era `roots/list` from a backend server. + + `Context` has no `list_roots()`: server-initiated requests are not part of + FastMCP's server API. These helpers reach the SDK session directly to stand + in for a legacy upstream, which is the only thing the proxy relay forwards. + """ + result = await context.session.list_roots() + return result.roots + + +async def _backend_sample(context: Context) -> str: + """Issue a handshake-era `sampling/createMessage` from a backend server.""" + result = await context.session.create_message( + messages=[ + SamplingMessage( + role="user", content=TextContent(type="text", text="Hello, world!") + ) + ], + system_prompt="You love FastMCP", + temperature=0.5, + max_tokens=100, + model_preferences=ModelPreferences(hints=[ModelHint(name="gpt-4o")]), + related_request_id=context.origin_request_id, + ) + return result.content.text if isinstance(result.content, TextContent) else "" + + @pytest.fixture def fastmcp_server(): mcp = FastMCP("TestServer") @@ -68,21 +97,13 @@ def fastmcp_server(): @mcp.tool async def list_roots(context: Context) -> list[str]: - roots = await context.list_roots() - return [str(r.uri) for r in roots] + return [str(r.uri) for r in await _backend_list_roots(context)] @mcp.tool async def sampling( context: Context, ) -> str: - result = await context.sample( - "Hello, world!", - system_prompt="You love FastMCP", - temperature=0.5, - max_tokens=100, - model_preferences="gpt-4o", - ) - return result.text or "" + return await _backend_sample(context) @dataclass class Person: @@ -514,17 +535,16 @@ def roots_backend_server(): @mcp.resource("data://roots") async def roots_resource(context: Context) -> list[str]: - roots = await context.list_roots() - return [str(r.uri) for r in roots] + return [str(r.uri) for r in await _backend_list_roots(context)] @mcp.resource("data://roots/{key}") async def roots_template(key: str, context: Context) -> str: - roots = await context.list_roots() + roots = await _backend_list_roots(context) return ", ".join(f"{key}:{r.uri}" for r in roots) @mcp.prompt async def roots_prompt(context: Context) -> str: - roots = await context.list_roots() + roots = await _backend_list_roots(context) return ", ".join(str(r.uri) for r in roots) return mcp diff --git a/tests/server/test_context.py b/tests/server/test_context.py index 384a307ac..e37e5531b 100644 --- a/tests/server/test_context.py +++ b/tests/server/test_context.py @@ -8,7 +8,6 @@ from fastmcp.server.context import ( reset_transport, set_transport, ) -from fastmcp.server.sampling.run import _parse_model_preferences from fastmcp.server.server import FastMCP @@ -17,28 +16,6 @@ def context(): return Context(fastmcp=FastMCP()) -class TestParseModelPreferences: - def test_parse_model_preferences_string(self, context): - mp = _parse_model_preferences("claude-haiku-4-5") - assert isinstance(mp, ModelPreferences) - assert mp.hints is not None - assert mp.hints[0].name == "claude-haiku-4-5" - - def test_parse_model_preferences_list(self, context): - mp = _parse_model_preferences(["claude-haiku-4-5", "claude"]) - assert isinstance(mp, ModelPreferences) - assert mp.hints is not None - assert [h.name for h in mp.hints] == ["claude-haiku-4-5", "claude"] - - def test_parse_model_preferences_object(self, context): - obj = ModelPreferences(hints=[]) - assert _parse_model_preferences(obj) is obj - - def test_parse_model_preferences_invalid_type(self, context): - with pytest.raises(ValueError): - _parse_model_preferences(model_preferences=123) # pyright: ignore[reportArgumentType] # type: ignore[invalid-argument-type] # ty:ignore[invalid-argument-type] - - class TestSessionId: def test_session_id_with_http_headers(self, context): """Test that session_id returns the value from mcp-session-id header.""" diff --git a/tests/server/test_protocol_eras.py b/tests/server/test_protocol_eras.py index 972ba1733..47bb0ce75 100644 --- a/tests/server/test_protocol_eras.py +++ b/tests/server/test_protocol_eras.py @@ -30,7 +30,7 @@ from mcp.shared.exceptions import MCPError from pydantic import FileUrl from fastmcp import Client as FastMCPClient -from fastmcp import Context, FastMCP, settings +from fastmcp import Context, FastMCP from fastmcp.exceptions import PromptError, ResourceError from fastmcp.server.elicitation import AcceptedElicitation from fastmcp.server.middleware import Middleware @@ -215,16 +215,6 @@ def push_server() -> FastMCP: assert isinstance(result, AcceptedElicitation) return f"elicited {result.data}" - @mcp.tool - async def do_sample(ctx: Context) -> str: - result = await ctx.sample("hello") - return f"sampled {result.text}" - - @mcp.tool - async def do_list_roots(ctx: Context) -> str: - roots = await ctx.list_roots() - return f"roots {[str(r.uri) for r in roots]}" - @mcp.tool async def do_log(ctx: Context) -> str: await ctx.info("a log line") @@ -264,31 +254,12 @@ async def test_elicit_works_on_legacy(push_server): assert _texts(result.content) == ["elicited 7"] -async def test_sample_works_on_legacy(push_server): - async with SDKClient( - _server(push_server), mode="legacy", sampling_callback=_sampling_cb - ) as client: - result = await client.call_tool("do_sample", {}) - assert result.is_error is False - assert _texts(result.content) == ["sampled sampled-text"] - - -async def test_list_roots_works_on_legacy(push_server): - async with SDKClient( - _server(push_server), mode="legacy", list_roots_callback=_roots_cb - ) as client: - result = await client.call_tool("do_list_roots", {}) - assert result.is_error is False - assert _texts(result.content) == ["roots ['file:///tmp']"] - - @pytest.mark.parametrize("mode", MODERN_MODES) -@pytest.mark.parametrize("tool", ["do_elicit", "do_sample", "do_list_roots"]) -async def test_push_features_degrade_on_modern(push_server, mode, tool): - """Server-initiated requests (elicitation/sampling/roots) are removed at - 2026-07-28 (SEP-2577), so a tool that uses them must degrade to a surfaced - error rather than hang or crash the connection. This asserts the - degradation happens and reaches the caller as an isError result. +async def test_elicit_degrades_on_modern(push_server, mode): + """Elicitation is a server-initiated request, removed at 2026-07-28 + (SEP-2577), so a tool that uses it must degrade to a surfaced error rather + than hang or crash the connection. The connection survives: a subsequent + normal call still works. """ async with SDKClient( _server(push_server), @@ -297,230 +268,46 @@ async def test_push_features_degrade_on_modern(push_server, mode, tool): sampling_callback=_sampling_cb, list_roots_callback=_roots_cb, ) as client: - result = await client.call_tool(tool, {}) + result = await client.call_tool("do_elicit", {}) assert result.is_error is True - # A subsequent normal call still works: the connection survived the - # per-request failure rather than tearing down the whole session. log_result = await client.call_tool("do_log", {}) assert log_result.is_error is False -async def test_list_roots_degradation_message_is_clear_on_modern(push_server): - """`ctx.list_roots()` sends with no related_request_id, so the SDK selects - the connection's no-back-channel outbound and raises the self-explanatory - NoBackChannelError. This is the *good* degradation message and we assert it. - """ - async with SDKClient(_server(push_server), mode="2026-07-28") as client: - result = await client.call_tool("do_list_roots", {}) - assert result.is_error is True - message = " ".join(_texts(result.content)).lower() - assert "back-channel" in message and "server-initiated" in message - - -@pytest.mark.parametrize("tool", ["do_elicit", "do_sample"]) -async def test_elicit_sample_degradation_message_is_clear_on_modern(push_server, tool): - """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 def test_elicit_degradation_message_is_clear_on_modern(push_server): + """FastMCP era-gates elicit: on a 2026-07-28 connection it raises a clear, + era-aware error before hitting the wire, instead of the SDK's opaque + 'Method not found' (sdk-feedback.md #10). """ async with SDKClient( _server(push_server), mode="2026-07-28", elicitation_callback=_accept_elicit, - sampling_callback=_sampling_cb, ) as client: - result = await client.call_tool(tool, {}) - assert result.is_error is True - message = " ".join(_texts(result.content)).lower() - assert "server-initiated" in message - - -# --------------------------------------------------------------------------- -# 3a-bis. Server-configured sampling handler answers WITHOUT the client -# back-channel, so ctx.sample()/ctx.sample_step() must keep working on modern -# connections. The era-gate only fires when nothing can serve the request. -# --------------------------------------------------------------------------- - - -def _handler_server(behavior) -> FastMCP: - """A server whose sampling is answered by a server-side handler.""" - - def sampling_handler(messages, params, ctx) -> str: - return "handler-answer" - - mcp = FastMCP("handler", sampling_handler=sampling_handler) - if behavior is not None: - mcp.sampling_handler_behavior = behavior - - @mcp.tool - async def do_sample(ctx: Context) -> str: - result = await ctx.sample("hello") - return f"sampled {result.text}" - - @mcp.tool - async def do_sample_step(ctx: Context) -> str: - step = await ctx.sample_step("hello") - return f"stepped {step.text}" - - return mcp - - -@pytest.mark.parametrize("mode", MODERN_MODES) -@pytest.mark.parametrize("behavior", ["always", "fallback"]) -@pytest.mark.parametrize("method", ["do_sample", "do_sample_step"]) -async def test_server_sampling_handler_works_on_modern(mode, behavior, method): - """A server-side sampling handler answers entirely server-side, so it works - on modern (2026-07-28) connections regardless of behavior. The era-gate must - NOT block these — nothing touches the removed client back-channel. Crucially, - 'fallback' must go straight to the handler (no bare client-attempt failure).""" - server = _handler_server(behavior) - async with SDKClient(_server(server), mode=mode) as client: - result = await client.call_tool(method, {}) - assert result.is_error is False - assert "handler-answer" in " ".join(_texts(result.content)) - - -@pytest.mark.parametrize("behavior", ["always", "fallback"]) -@pytest.mark.parametrize("method", ["do_sample", "do_sample_step"]) -async def test_server_sampling_handler_works_on_legacy(behavior, method): - """Handshake-era behavior is unchanged: the server-side handler still answers - on legacy connections.""" - server = _handler_server(behavior) - async with SDKClient(_server(server), mode="legacy") as client: - result = await client.call_tool(method, {}) - assert result.is_error is False - assert "handler-answer" in " ".join(_texts(result.content)) - - -@pytest.mark.parametrize("mode", MODERN_MODES) -@pytest.mark.parametrize("method", ["do_sample", "do_sample_step"]) -async def test_sampling_without_handler_still_era_gated_on_modern( - push_server, mode, method -): - """With no server-side handler configured, the request would hit the removed - client back-channel, so the clear era error still fires on modern.""" - # push_server only defines do_sample; add a do_sample_step twin inline. - mcp = FastMCP("no-handler") - - @mcp.tool - async def do_sample(ctx: Context) -> str: - result = await ctx.sample("hello") - return f"sampled {result.text}" - - @mcp.tool - async def do_sample_step(ctx: Context) -> str: - step = await ctx.sample_step("hello") - return f"stepped {step.text}" - - async with SDKClient( - _server(mcp), mode=mode, sampling_callback=_sampling_cb - ) as client: - result = await client.call_tool(method, {}) + result = await client.call_tool("do_elicit", {}) assert result.is_error is True assert "server-initiated" in " ".join(_texts(result.content)).lower() # --------------------------------------------------------------------------- -# 3b. Sampling deprecation warning (SEP-2577): ctx.sample/ctx.sample_step warn +# 3a-bis. Sampling and roots are not in the server API at all # --------------------------------------------------------------------------- -@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 = set(context_module._sample_deprecation_warned) - context_module._sample_deprecation_warned.clear() - try: - yield - finally: - context_module._sample_deprecation_warned.clear() - context_module._sample_deprecation_warned.update(original) +@pytest.mark.parametrize("name", ["sample", "sample_step", "list_roots"]) +def test_removed_server_initiated_methods_are_absent(name): + """FastMCP 4 targets the modern protocol, so the capabilities SEP-2577 + removed are not in the server-authoring API — not deprecated, not era-gated, + absent. A server that calls them fails at attribute lookup, in every era. + """ + assert not hasattr(Context, name) -@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 - - from fastmcp.exceptions import FastMCPDeprecationWarning - - monkeypatch.setattr(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("kwarg", ["sampling_handler", "sampling_handler_behavior"]) +def test_server_sampling_handler_kwargs_are_rejected(kwarg): + """The server-side sampling handler existed only to answer `ctx.sample()`.""" + with pytest.raises(TypeError, match="SEP-2577"): + FastMCP("gone", **{kwarg: None}) @pytest.mark.parametrize("mode", MODERN_MODES)