Client call_tool: accept input_responses/request_state; return InputRequiredResult via allow_input_required opt-in

maxisbey · maxisbey · commit d9a6317306c7 · 2026-06-25T14:22:55.000Z
- ClientSession.send_request: accept TypeAdapter[T] alongside type[T] for result_type so callers can parse union results. - ClientSession.call_tool (mechanics): add input_responses= / request_state= retry kwargs; return CallToolResult | InputRequiredResult; gate output-schema validation on isinstance(result, CallToolResult). - Client.call_tool / ClientSessionGroup.call_tool (policy): @overload on allow_input_required — Literal[False] (default) returns CallToolResult; Literal[True] returns the union. Default raises RuntimeError on InputRequiredResult with a retry steer (TODO(L80) marks where the auto-loop driver replaces this). - Examples and tests that call ClientSession.call_tool directly narrow with isinstance(result, CallToolResult); README.v2.md regenerated from snippets.
diff --git a/README.v2.md b/README.v2.md
@@ -2166,6 +2166,7 @@ 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}")
@@ -2431,19 +2432,22 @@ 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"})
-            if hasattr(result, "structured_content") and result.structured_content:
+            assert isinstance(result, types.CallToolResult)
+            if 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
@@ -2454,12 +2458,14 @@ 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:
diff --git a/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py b/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py
@@ -25,6 +25,7 @@
 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):
@@ -293,7 +294,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 hasattr(result, "content"):
+            if isinstance(result, CallToolResult):
                 for content in result.content:
                     if content.type == "text":
                         print(content.text)
diff --git a/examples/clients/sse-polling-client/mcp_sse_polling_client/main.py b/examples/clients/sse-polling-client/mcp_sse_polling_client/main.py
@@ -20,6 +20,7 @@
 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:
@@ -55,6 +56,7 @@ 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))
diff --git a/examples/snippets/clients/parsing_tool_results.py b/examples/snippets/clients/parsing_tool_results.py
@@ -16,19 +16,22 @@ 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"})
-            if hasattr(result, "structured_content") and result.structured_content:
+            assert isinstance(result, types.CallToolResult)
+            if 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
@@ -39,12 +42,14 @@ 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:
diff --git a/examples/snippets/clients/stdio_client.py b/examples/snippets/clients/stdio_client.py
@@ -64,6 +64,7 @@ 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}")
diff --git a/examples/snippets/clients/url_elicitation_client.py b/examples/snippets/clients/url_elicitation_client.py
@@ -150,6 +150,7 @@ 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:
diff --git a/src/mcp/client/client.py b/src/mcp/client/client.py
@@ -5,7 +5,7 @@
 from collections.abc import Awaitable, Callable, Mapping
 from contextlib import AsyncExitStack
 from dataclasses import KW_ONLY, dataclass, field
-from typing import Any, Literal, TypeVar
+from typing import Any, Literal, TypeVar, overload
 
 import anyio
 from typing_extensions import deprecated
@@ -30,6 +30,8 @@
     EmptyResult,
     GetPromptResult,
     Implementation,
+    InputRequiredResult,
+    InputResponses,
     ListPromptsResult,
     ListResourcesResult,
     ListResourceTemplatesResult,
@@ -374,34 +376,85 @@ async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None
         """Unsubscribe from resource updates."""
         return await self.session.unsubscribe_resource(uri, meta=meta)
 
+    @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: InputResponses | None = None,
+        request_state: str | None = None,
         meta: RequestParamsMeta | None = None,
-    ) -> CallToolResult:
+        allow_input_required: Literal[False] = False,
+    ) -> 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: InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: Literal[True],
+    ) -> CallToolResult | 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: InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: bool = False,
+    ) -> CallToolResult | InputRequiredResult:
         """Call a tool on the server.
 
         Args:
             name: The name of the tool to call
             arguments: Arguments to pass to the tool
             read_timeout_seconds: Timeout for the tool call
             progress_callback: Callback for progress updates
+            input_responses: Responses to a prior `InputRequiredResult.input_requests`
+            request_state: Opaque state echoed from a prior `InputRequiredResult`
             meta: Additional metadata for the request
+            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.
 
         Returns:
-            The tool result.
+            The tool result. When ``allow_input_required=True``, may instead be an
+            `InputRequiredResult` carrying the server's input requests and opaque
+            ``request_state`` for the retry.
+
+        Raises:
+            RuntimeError: If the server returns an `InputRequiredResult` and
+                ``allow_input_required`` is ``False``.
         """
-        return await self.session.call_tool(
+        result = 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,
         )
+        if isinstance(result, InputRequiredResult) and not allow_input_required:
+            # TODO(L80): 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,
diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py
@@ -173,6 +173,10 @@ async def _default_logging_callback(
 
 ClientResponse: TypeAdapter[types.ClientResult | types.ErrorData] = TypeAdapter(types.ClientResult | types.ErrorData)
 
+_CallToolResultAdapter: TypeAdapter[types.CallToolResult | types.InputRequiredResult] = TypeAdapter(
+    types.CallToolResult | types.InputRequiredResult
+)
+
 
 class ClientSession:
     """Client half of an MCP connection, running on a `Dispatcher`.
@@ -269,7 +273,7 @@ async def __aexit__(
     async def send_request(
         self,
         request: types.ClientRequest,
-        result_type: type[ReceiveResultT],
+        result_type: type[ReceiveResultT] | TypeAdapter[ReceiveResultT],
         request_read_timeout_seconds: float | None = None,
         metadata: ClientMessageMetadata | None = None,
         progress_callback: ProgressFnT | None = None,
@@ -308,6 +312,8 @@ async def send_request(
             _methods.validate_server_result(method, version, raw)
         except KeyError:
             pass
+        if isinstance(result_type, TypeAdapter):
+            return result_type.validate_python(raw)
         return result_type.model_validate(raw, by_name=False)
 
     async def send_notification(self, notification: types.ClientNotification) -> None:
@@ -603,20 +609,28 @@ async def call_tool(
         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,
-    ) -> types.CallToolResult:
+    ) -> types.CallToolResult | types.InputRequiredResult:
         """Send a tools/call request with optional progress callback support."""
 
         result = await self.send_request(
             types.CallToolRequest(
-                params=types.CallToolRequestParams(name=name, arguments=arguments, _meta=meta),
+                params=types.CallToolRequestParams(
+                    name=name,
+                    arguments=arguments,
+                    input_responses=input_responses,
+                    request_state=request_state,
+                    _meta=meta,
+                ),
             ),
-            types.CallToolResult,
+            _CallToolResultAdapter,
             request_read_timeout_seconds=read_timeout_seconds,
             progress_callback=progress_callback,
         )
 
-        if not result.is_error:
+        if isinstance(result, types.CallToolResult) and not result.is_error:
             await self._validate_tool_result(name, result)
 
         return result
diff --git a/src/mcp/client/session_group.py b/src/mcp/client/session_group.py
@@ -11,7 +11,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 from types import TracebackType
-from typing import Any, TypeAlias
+from typing import Any, Literal, TypeAlias, overload
 
 import anyio
 import httpx
@@ -190,25 +190,70 @@ def tools(self) -> dict[str, types.Tool]:
         """Returns the tools as a dictionary of names to tools."""
         return self._tools
 
+    @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: types.RequestParamsMeta | None = None,
-    ) -> types.CallToolResult:
-        """Executes a tool given its name and arguments."""
+        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: types.RequestParamsMeta | None = None,
+        allow_input_required: Literal[True],
+    ) -> 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: types.RequestParamsMeta | None = None,
+        allow_input_required: bool = False,
+    ) -> types.CallToolResult | types.InputRequiredResult:
+        """Executes a tool given its name and arguments.
+
+        Raises:
+            RuntimeError: If the server returns an `InputRequiredResult` and
+                ``allow_input_required`` is ``False``.
+        """
         session = self._tool_to_session[name]
         session_tool_name = self.tools[name].name
-        return await session.call_tool(
+        result = 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,
         )
+        if isinstance(result, types.InputRequiredResult) and not allow_input_required:
+            # TODO(L80): 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."""
diff --git a/tests/client/test_http_unicode.py b/tests/client/test_http_unicode.py
@@ -142,6 +142,7 @@ async def test_streamable_http_client_unicode_tool_call() -> None:
         # Test 2: Send Unicode text in tool call (client→server→client)
         for test_name, test_string in UNICODE_TEST_STRINGS.items():
             result = await session.call_tool("echo_unicode", arguments={"text": test_string})
+            assert isinstance(result, types.CallToolResult)
 
             # Verify server correctly received and echoed back Unicode
             assert len(result.content) == 1
diff --git a/tests/client/test_session.py b/tests/client/test_session.py
diff --git a/tests/client/test_session_group.py b/tests/client/test_session_group.py
diff --git a/tests/issues/test_88_random_error.py b/tests/issues/test_88_random_error.py
diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py
diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py
