Move allow_input_required overload+gate down to ClientSession.call_tool

The gate now lives once on ClientSession (mechanics layer); Client and
ClientSessionGroup are pure passthroughs that forward allow_input_required.
Third 'bool' overload on ClientSession.call_tool lets the passthrough impls
type-check. Reverts the isinstance narrowing in examples/ and tests/shared/;
default return is CallToolResult everywhere, so the change is additive.

Author: maxisbey

README.v2.md

@@ -2166,7 +2166,6 @@ async def run():
 
             # Call a tool (add tool from mcpserver_quickstart)
             result = await session.call_tool("add", arguments={"a": 5, "b": 3})
-            assert isinstance(result, types.CallToolResult)
             result_unstructured = result.content[0]
             if isinstance(result_unstructured, types.TextContent):
                 print(f"Tool result: {result_unstructured.text}")
@@ -2432,22 +2431,19 @@ async def parse_tool_results():
 
             # Example 1: Parsing text content
             result = await session.call_tool("get_data", {"format": "text"})
-            assert isinstance(result, types.CallToolResult)
             for content in result.content:
                 if isinstance(content, types.TextContent):
                     print(f"Text: {content.text}")
 
             # Example 2: Parsing structured content from JSON tools
             result = await session.call_tool("get_user", {"id": "123"})
-            assert isinstance(result, types.CallToolResult)
-            if result.structured_content:
+            if hasattr(result, "structured_content") and result.structured_content:
                 # Access structured data directly
                 user_data = result.structured_content
                 print(f"User: {user_data.get('name')}, Age: {user_data.get('age')}")
 
             # Example 3: Parsing embedded resources
             result = await session.call_tool("read_config", {})
-            assert isinstance(result, types.CallToolResult)
             for content in result.content:
                 if isinstance(content, types.EmbeddedResource):
                     resource = content.resource
@@ -2458,14 +2454,12 @@ async def parse_tool_results():
 
             # Example 4: Parsing image content
             result = await session.call_tool("generate_chart", {"data": [1, 2, 3]})
-            assert isinstance(result, types.CallToolResult)
             for content in result.content:
                 if isinstance(content, types.ImageContent):
                     print(f"Image ({content.mime_type}): {len(content.data)} bytes")
 
             # Example 5: Handling errors
             result = await session.call_tool("failing_tool", {})
-            assert isinstance(result, types.CallToolResult)
             if result.is_error:
                 print("Tool execution failed!")
                 for content in result.content:

docs/migration.md

@@ -364,11 +364,9 @@ For an in-process `Client(server)` (where `server` is a `Server` or `MCPServer`
 
 `Client.send_ping()` is deprecated (ping is removed in 2026-07-28); pin `mode='legacy'` if you need it.
 
-### `ClientSession.call_tool` returns `CallToolResult | InputRequiredResult`
+### `call_tool` can return `InputRequiredResult` (opt-in)
 
-For protocol 2026-07-28, a `tools/call` request may return an `InputRequiredResult` asking the client to supply additional input and retry. `ClientSession.call_tool` now returns `CallToolResult | InputRequiredResult` to reflect this; narrow with `isinstance(result, CallToolResult)` before reading `.content` / `.is_error` / `.structured_content`.
-
-The high-level `Client.call_tool` still returns `CallToolResult` by default (and raises `RuntimeError` if the server requests input). Pass `allow_input_required=True` to receive the `InputRequiredResult` and retry with `input_responses=` / `request_state=`.
+For protocol 2026-07-28, a `tools/call` request may return an `InputRequiredResult` asking the client to supply additional input and retry. By default `call_tool` (on `ClientSession`, `Client`, and `ClientSessionGroup`) still returns `CallToolResult` and raises `RuntimeError` if the server requests input. Pass `allow_input_required=True` to receive the `InputRequiredResult` instead, then retry with `input_responses=` / `request_state=`.
 
 ### `McpError` renamed to `MCPError`
 

examples/clients/simple-auth-client/mcp_simple_auth_client/main.py

@@ -25,7 +25,6 @@
 from mcp.client.streamable_http import streamable_http_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 from mcp.shared.message import SessionMessage
-from mcp.types import CallToolResult
 
 
 class InMemoryTokenStorage(TokenStorage):
@@ -294,7 +293,7 @@ async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None = Non
         try:
             result = await self.session.call_tool(tool_name, arguments or {})
             print(f"\n🔧 Tool '{tool_name}' result:")
-            if isinstance(result, CallToolResult):
+            if hasattr(result, "content"):
                 for content in result.content:
                     if content.type == "text":
                         print(content.text)

examples/clients/sse-polling-client/mcp_sse_polling_client/main.py

@@ -20,7 +20,6 @@
 import click
 from mcp import ClientSession
 from mcp.client.streamable_http import streamable_http_client
-from mcp.types import CallToolResult
 
 
 async def run_demo(url: str, items: int, checkpoint_every: int) -> None:
@@ -56,7 +55,6 @@ async def run_demo(url: str, items: int, checkpoint_every: int) -> None:
             )
 
             print("-" * 40)
-            assert isinstance(result, CallToolResult)
             if result.content:
                 content = result.content[0]
                 text = getattr(content, "text", str(content))

examples/snippets/clients/parsing_tool_results.py

@@ -16,22 +16,19 @@ async def parse_tool_results():
 
             # Example 1: Parsing text content
             result = await session.call_tool("get_data", {"format": "text"})
-            assert isinstance(result, types.CallToolResult)
             for content in result.content:
                 if isinstance(content, types.TextContent):
                     print(f"Text: {content.text}")
 
             # Example 2: Parsing structured content from JSON tools
             result = await session.call_tool("get_user", {"id": "123"})
-            assert isinstance(result, types.CallToolResult)
-            if result.structured_content:
+            if hasattr(result, "structured_content") and result.structured_content:
                 # Access structured data directly
                 user_data = result.structured_content
                 print(f"User: {user_data.get('name')}, Age: {user_data.get('age')}")
 
             # Example 3: Parsing embedded resources
             result = await session.call_tool("read_config", {})
-            assert isinstance(result, types.CallToolResult)
             for content in result.content:
                 if isinstance(content, types.EmbeddedResource):
                     resource = content.resource
@@ -42,14 +39,12 @@ async def parse_tool_results():
 
             # Example 4: Parsing image content
             result = await session.call_tool("generate_chart", {"data": [1, 2, 3]})
-            assert isinstance(result, types.CallToolResult)
             for content in result.content:
                 if isinstance(content, types.ImageContent):
                     print(f"Image ({content.mime_type}): {len(content.data)} bytes")
 
             # Example 5: Handling errors
             result = await session.call_tool("failing_tool", {})
-            assert isinstance(result, types.CallToolResult)
             if result.is_error:
                 print("Tool execution failed!")
                 for content in result.content:

examples/snippets/clients/stdio_client.py

@@ -64,7 +64,6 @@ async def run():
 
             # Call a tool (add tool from mcpserver_quickstart)
             result = await session.call_tool("add", arguments={"a": 5, "b": 3})
-            assert isinstance(result, types.CallToolResult)
             result_unstructured = result.content[0]
             if isinstance(result_unstructured, types.TextContent):
                 print(f"Tool result: {result_unstructured.text}")

examples/snippets/clients/url_elicitation_client.py

@@ -150,7 +150,6 @@ async def call_tool_with_error_handling(
     """
     try:
         result = await session.call_tool(tool_name, arguments)
-        assert isinstance(result, types.CallToolResult)
 
         # Check if the tool returned an error in the result
         if result.is_error:

src/mcp/client/client.py

@@ -439,22 +439,17 @@ async def call_tool(
             RuntimeError: If the server returns an `InputRequiredResult` and
                 ``allow_input_required`` is ``False``.
         """
-        result = await self.session.call_tool(
+        # TODO(L84): stop forwarding allow_input_required; run the MRTR auto-loop driver here (S6).
+        return await self.session.call_tool(
             name=name,
             arguments=arguments,
             read_timeout_seconds=read_timeout_seconds,
             progress_callback=progress_callback,
             input_responses=input_responses,
             request_state=request_state,
             meta=meta,
+            allow_input_required=allow_input_required,
         )
-        if isinstance(result, InputRequiredResult) and not allow_input_required:
-            # TODO(L84): replace this raise with the MRTR auto-loop driver (S6).
-            raise RuntimeError(
-                "Server returned InputRequiredResult; pass allow_input_required=True to receive it "
-                "and retry call_tool(..., input_responses=..., request_state=result.request_state)."
-            )
-        return result
 
     async def list_prompts(
         self,

src/mcp/client/session.py

@@ -4,7 +4,7 @@
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from types import TracebackType
-from typing import Any, Protocol, cast
+from typing import Any, Literal, Protocol, cast, overload
 
 import anyio
 import anyio.abc
@@ -602,6 +602,7 @@ async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None
             types.EmptyResult,
         )
 
+    @overload
     async def call_tool(
         self,
         name: str,
@@ -612,8 +613,62 @@ async def call_tool(
         input_responses: types.InputResponses | None = None,
         request_state: str | None = None,
         meta: RequestParamsMeta | None = None,
+        allow_input_required: Literal[False] = False,
+    ) -> types.CallToolResult: ...
+
+    @overload
+    async def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        read_timeout_seconds: float | None = None,
+        progress_callback: ProgressFnT | None = None,
+        *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: Literal[True],
+    ) -> types.CallToolResult | types.InputRequiredResult: ...
+
+    @overload
+    async def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        read_timeout_seconds: float | None = None,
+        progress_callback: ProgressFnT | None = None,
+        *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: bool,
+    ) -> types.CallToolResult | types.InputRequiredResult: ...
+
+    async def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        read_timeout_seconds: float | None = None,
+        progress_callback: ProgressFnT | None = None,
+        *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: bool = False,
     ) -> types.CallToolResult | types.InputRequiredResult:
-        """Send a tools/call request with optional progress callback support."""
+        """Send a tools/call request with optional progress callback support.
+
+        Args:
+            input_responses: Responses to a prior `InputRequiredResult.input_requests`.
+            request_state: Opaque state echoed from a prior `InputRequiredResult`.
+            allow_input_required: When ``False`` (default), an `InputRequiredResult`
+                from the server raises `RuntimeError`; when ``True``, it is returned
+                so the caller can resolve the requests and retry.
+
+        Raises:
+            RuntimeError: If the server returns an `InputRequiredResult` and
+                ``allow_input_required`` is ``False``.
+        """
 
         result = await self.send_request(
             types.CallToolRequest(
@@ -633,6 +688,11 @@ async def call_tool(
         if isinstance(result, types.CallToolResult) and not result.is_error:
             await self._validate_tool_result(name, result)
 
+        if isinstance(result, types.InputRequiredResult) and not allow_input_required:
+            raise RuntimeError(
+                "Server returned InputRequiredResult; pass allow_input_required=True to receive it "
+                "and retry call_tool(..., input_responses=..., request_state=result.request_state)."
+            )
         return result
 
     async def _validate_tool_result(self, name: str, result: types.CallToolResult) -> None:

src/mcp/client/session_group.py

@@ -238,22 +238,16 @@ async def call_tool(
         """
         session = self._tool_to_session[name]
         session_tool_name = self.tools[name].name
-        result = await session.call_tool(
+        return await session.call_tool(
             session_tool_name,
             arguments=arguments,
             read_timeout_seconds=read_timeout_seconds,
             progress_callback=progress_callback,
             input_responses=input_responses,
             request_state=request_state,
             meta=meta,
+            allow_input_required=allow_input_required,
         )
-        if isinstance(result, types.InputRequiredResult) and not allow_input_required:
-            # TODO(L84): replace this raise with the MRTR auto-loop driver (S6).
-            raise RuntimeError(
-                "Server returned InputRequiredResult; pass allow_input_required=True to receive it "
-                "and retry call_tool(..., input_responses=..., request_state=result.request_state)."
-            )
-        return result
 
     async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
         """Disconnects from a single MCP server."""