Share the tools-mode predicate across all sampling senders

The resolver path, the deprecated session.create_message, and the peer
sample API each hand-rolled the tools-mode test; the latter two still
keyed the answer model on tools alone, so a tool_choice-only request
was gated as tools-mode but its array-capable answer failed validation.
One wants_sampling_tools predicate in mcp.server.validation now drives
the capability check and the result model everywhere reachable, with
the overloads updated to match and tests on each surface.

Author: maxisbey

src/mcp/server/mcpserver/resolve.py

@@ -80,7 +80,7 @@
 from mcp.server.mcpserver.context import Context
 from mcp.server.mcpserver.exceptions import InvalidSignature, ToolError
 from mcp.server.request_state import compact_json
-from mcp.server.validation import validate_tool_use_result_messages
+from mcp.server.validation import validate_tool_use_result_messages, wants_sampling_tools
 from mcp.shared._callable_inspection import is_async_callable
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import ServerMessageMetadata
@@ -680,7 +680,7 @@ def _require_capability(context: Context[Any, Any], marker: _Marker, key: str) -
         name = "form elicitation"
     elif isinstance(marker, Sample):
         sampling = capabilities.sampling if capabilities is not None else None
-        wants_tools = _wants_tools(marker.params)
+        wants_tools = wants_sampling_tools(marker.params.tools, marker.params.tool_choice)
         if sampling is not None and (not wants_tools or sampling.tools is not None):
             return
         required = ClientCapabilities(
@@ -710,18 +710,17 @@ def _render_request(marker: _Marker) -> InputRequest:
     return ListRootsRequest()
 
 
-def _wants_tools(params: CreateMessageRequestParams) -> bool:
-    """Whether a sampling request is tools-mode: `sampling.tools` gated, array-capable answer."""
-    return params.tools is not None or params.tool_choice is not None
-
-
 def _result_type(
     marker: Sample | ListRoots,
 ) -> type[CreateMessageResult] | type[CreateMessageResultWithTools] | type[ListRootsResult]:
     """The result model a `Sample`/`ListRoots` response must validate against."""
     if isinstance(marker, ListRoots):
         return ListRootsResult
-    return CreateMessageResultWithTools if _wants_tools(marker.params) else CreateMessageResult
+    return (
+        CreateMessageResultWithTools
+        if wants_sampling_tools(marker.params.tools, marker.params.tool_choice)
+        else CreateMessageResult
+    )
 
 
 class _StateEntry(BaseModel):

src/mcp/server/session.py

@@ -14,7 +14,7 @@
 from typing_extensions import deprecated
 
 from mcp.server.connection import Connection
-from mcp.server.validation import validate_sampling_tools, validate_tool_use_result_messages
+from mcp.server.validation import validate_sampling_tools, validate_tool_use_result_messages, wants_sampling_tools
 from mcp.shared.dispatcher import CallOptions, DispatchContext, ProgressFnT
 from mcp.shared.exceptions import MCPDeprecationWarning
 from mcp.shared.message import ServerMessageMetadata
@@ -146,10 +146,10 @@ async def create_message(
         metadata: dict[str, Any] | None = None,
         model_preferences: types.ModelPreferences | None = None,
         tools: None = None,
-        tool_choice: types.ToolChoice | None = None,
+        tool_choice: None = None,
         related_request_id: types.RequestId | None = None,
     ) -> types.CreateMessageResult:
-        """Overload: Without tools, returns single content."""
+        """Overload: Without tools or tool_choice, returns single content."""
         ...
 
     @overload
@@ -165,11 +165,11 @@ async def create_message(
         stop_sequences: list[str] | None = None,
         metadata: dict[str, Any] | None = None,
         model_preferences: types.ModelPreferences | None = None,
-        tools: list[types.Tool],
+        tools: list[types.Tool] | None = None,
         tool_choice: types.ToolChoice | None = None,
         related_request_id: types.RequestId | None = None,
     ) -> types.CreateMessageResultWithTools:
-        """Overload: With tools, returns array-capable content."""
+        """Overload: With tools or tool_choice, returns array-capable content."""
         ...
 
     @deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
@@ -236,7 +236,7 @@ async def create_message(
         )
         metadata_obj = ServerMessageMetadata(related_request_id=related_request_id)
 
-        if tools is not None:
+        if wants_sampling_tools(tools, tool_choice):
             return await self.send_request(
                 request=request,
                 result_type=types.CreateMessageResultWithTools,

src/mcp/server/validation.py

@@ -26,6 +26,11 @@ def check_sampling_tools_capability(client_caps: ClientCapabilities | None) -> b
     return True
 
 
+def wants_sampling_tools(tools: list[Tool] | None, tool_choice: ToolChoice | None) -> bool:
+    """Whether a sampling request is tools-mode: `sampling.tools` gated, array-capable answer."""
+    return tools is not None or tool_choice is not None
+
+
 def validate_sampling_tools(
     client_caps: ClientCapabilities | None,
     tools: list[Tool] | None,
@@ -41,7 +46,7 @@ def validate_sampling_tools(
     Raises:
         MCPError: If tools/tool_choice are provided but client doesn't support them
     """
-    if tools is not None or tool_choice is not None:
+    if wants_sampling_tools(tools, tool_choice):
         if not check_sampling_tools_capability(client_caps):
             raise MCPError(code=INVALID_PARAMS, message="Client does not support sampling tools capability")
 

src/mcp/shared/peer.py

@@ -98,7 +98,7 @@ async def sample(
         metadata: dict[str, Any] | None = None,
         model_preferences: ModelPreferences | None = None,
         tools: None = None,
-        tool_choice: ToolChoice | None = None,
+        tool_choice: None = None,
         meta: Meta | None = None,
         opts: CallOptions | None = None,
     ) -> CreateMessageResult: ...
@@ -115,7 +115,7 @@ async def sample(
         stop_sequences: list[str] | None = None,
         metadata: dict[str, Any] | None = None,
         model_preferences: ModelPreferences | None = None,
-        tools: list[Tool],
+        tools: list[Tool] | None = None,
         tool_choice: ToolChoice | None = None,
         meta: Meta | None = None,
         opts: CallOptions | None = None,
@@ -157,7 +157,7 @@ async def sample(
             tool_choice=tool_choice,
         )
         result = await self.send_raw_request("sampling/createMessage", dump_params(params, meta), opts)
-        if tools is not None:
+        if tools is not None or tool_choice is not None:
             return CreateMessageResultWithTools.model_validate(result, by_name=False)
         return CreateMessageResult.model_validate(result, by_name=False)
 

tests/server/mcpserver/test_resolve.py

@@ -2757,7 +2757,8 @@ async def tool(login: Annotated[Login, Resolve(ask)]) -> str:
 
 
 @pytest.mark.anyio
-async def test_tool_choice_only_sample_validates_as_tools_mode():
+@pytest.mark.parametrize("mode", ["legacy", "auto"])
+async def test_tool_choice_only_sample_validates_as_tools_mode(mode: Literal["legacy", "auto"]):
     # Gate and answer model share one predicate: tool_choice alone is tools-mode,
     # so a single-content answer still validates (WithTools accepts both shapes).
     mcp = MCPServer(name="ToolChoiceOnly", request_state_security=RequestStateSecurity.ephemeral())
@@ -2769,12 +2770,12 @@ async def sampler(context: ClientRequestContext, params: CreateMessageRequestPar
     @mcp.tool()
     async def calc(answer: Annotated[CreateMessageResultWithTools, Resolve(_ask_with_tool_choice)]) -> str:
         assert isinstance(answer, CreateMessageResultWithTools)
-        content = answer.content[0] if isinstance(answer.content, list) else answer.content
-        assert isinstance(content, TextContent)
-        return content.text
+        assert isinstance(answer.content, TextContent)
+        return answer.content.text
 
     async with Client(
         mcp,
+        mode=mode,
         sampling_callback=sampler,
         sampling_capabilities=SamplingCapability(tools=SamplingToolsCapability()),
     ) as client:

tests/server/test_session.py

@@ -225,6 +225,21 @@ async def test_create_message_with_tools_returns_with_tools_result():
     assert params is not None and params["tools"][0]["name"] == "t"
 
 
+@pytest.mark.anyio
+async def test_create_message_with_tool_choice_only_returns_with_tools_result():
+    # tool_choice alone is tools-mode: the answer may carry array content.
+    outbound = StubOutbound(result={"role": "assistant", "content": [{"type": "text", "text": "ok"}], "model": "m"})
+    session = _make_session(
+        outbound, capabilities=ClientCapabilities(sampling=SamplingCapability(tools=SamplingToolsCapability()))
+    )
+    result = await session.create_message(  # pyright: ignore[reportDeprecated]
+        messages=[types.SamplingMessage(role="user", content=types.TextContent(type="text", text="hi"))],
+        max_tokens=10,
+        tool_choice=types.ToolChoice(mode="none"),
+    )
+    assert isinstance(result, types.CreateMessageResultWithTools)
+
+
 def test_check_client_capability_delegates_to_connection():
     outbound = StubOutbound()
     session = _make_session(outbound, capabilities=ClientCapabilities(sampling=SamplingCapability()))

tests/shared/test_peer.py

@@ -18,6 +18,7 @@
     SamplingMessage,
     TextContent,
     Tool,
+    ToolChoice,
 )
 
 from mcp.shared.dispatcher import DispatchContext
@@ -91,6 +92,21 @@ async def test_peer_sample_with_tools_returns_with_tools_result():
         assert isinstance(result, CreateMessageResultWithTools)
 
 
+@pytest.mark.anyio
+async def test_peer_sample_with_tool_choice_only_returns_with_tools_result():
+    # tool_choice alone is tools-mode: the answer may carry array content.
+    rec = _Recorder({"role": "assistant", "content": [{"type": "text", "text": "x"}], "model": "m"})
+    async with running_pair(direct_pair, server_on_request=rec.on_request) as (client, *_):
+        peer = ClientPeer(client)
+        with anyio.fail_after(5):
+            result = await peer.sample(  # pyright: ignore[reportDeprecated]
+                [SamplingMessage(role="user", content=TextContent(type="text", text="q"))],
+                max_tokens=5,
+                tool_choice=ToolChoice(mode="none"),
+            )
+        assert isinstance(result, CreateMessageResultWithTools)
+
+
 @pytest.mark.anyio
 async def test_peer_elicit_form_sends_elicitation_create_with_form_params():
     rec = _Recorder({"action": "accept", "content": {"name": "Max"}})