Let call_tool drive an input-required exchange leg by leg

call_tool and call_tool_mcp take input_responses and request_state, and
allow_input_required hands back the ask instead of resolving it against the
client's own handlers. CallToolResult carries it on input_required, mirroring
the server's InputRequiredToolResult.

fastmcp_slim/fastmcp/client/client.py

@@ -234,13 +234,20 @@ class ClientSessionState:
 
 @dataclass
 class CallToolResult:
-    """Parsed result from a tool call."""
+    """Parsed result from a tool call.
+
+    A call that asked for input rather than completing carries the ask on
+    `input_required` and nothing else — `content` is empty and `data` is None.
+    That only happens when the caller passed `allow_input_required=True`; by
+    default the client resolves the exchange before returning.
+    """
 
     content: list[mcp_types.ContentBlock]
     structured_content: dict[str, Any] | None
     meta: dict[str, Any] | None
     data: Any = None
     is_error: bool = False
+    input_required: mcp_types.InputRequiredResult | None = None
 
 
 class Client(

fastmcp_slim/fastmcp/client/mixins/tools.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, Literal, cast, overload
 
 import mcp_types
 from mcp.client.caching import CacheMode
@@ -136,6 +136,7 @@ class ClientToolsMixin:
 
     # --- Call Tool ---
 
+    @overload
     async def call_tool_mcp(
         self: Client,
         name: str,
@@ -143,12 +144,67 @@ class ClientToolsMixin:
         progress_handler: ProgressHandler | None = None,
         timeout: datetime.timedelta | float | int | None = None,
         meta: dict[str, Any] | None = None,
-    ) -> mcp_types.CallToolResult:
+        *,
+        input_responses: mcp_types.InputResponses | None = None,
+        request_state: str | None = None,
+        allow_input_required: Literal[False] = False,
+    ) -> mcp_types.CallToolResult: ...
+
+    @overload
+    async def call_tool_mcp(
+        self: Client,
+        name: str,
+        arguments: dict[str, Any],
+        progress_handler: ProgressHandler | None = None,
+        timeout: datetime.timedelta | float | int | None = None,
+        meta: dict[str, Any] | None = None,
+        *,
+        input_responses: mcp_types.InputResponses | None = None,
+        request_state: str | None = None,
+        allow_input_required: Literal[True],
+    ) -> mcp_types.CallToolResult | mcp_types.InputRequiredResult: ...
+
+    async def call_tool_mcp(
+        self: Client,
+        name: str,
+        arguments: dict[str, Any],
+        progress_handler: ProgressHandler | None = None,
+        timeout: datetime.timedelta | float | int | None = None,
+        meta: dict[str, Any] | None = None,
+        *,
+        input_responses: mcp_types.InputResponses | None = None,
+        request_state: str | None = None,
+        allow_input_required: bool = False,
+    ) -> mcp_types.CallToolResult | mcp_types.InputRequiredResult:
         """Send a tools/call request and return the complete MCP protocol result.
 
         This method returns the raw CallToolResult object, which includes an isError flag
         and other metadata. It does not raise an exception if the tool call results in an error.
 
+        A tool that asks for client input answers with an `InputRequiredResult`
+        (SEP-2322) rather than a final result. By default that is resolved for
+        you, the same way `call_tool` does it — each embedded request is
+        dispatched to this client's handlers and the call is retried until it
+        completes. Pass `allow_input_required=True` to receive the ask instead
+        and drive the exchange one leg at a time, feeding the answers back
+        through `input_responses` and `request_state`:
+
+        ```python
+        async with Client(mcp) as client:
+            leg = await client.call_tool_mcp("book", {}, allow_input_required=True)
+            answers = {
+                key: mcp_types.ElicitResult(action="accept", content={"value": "Paris"})
+                for key in leg.input_requests
+            }
+            final = await client.call_tool_mcp(
+                "book",
+                {},
+                input_responses=answers,
+                request_state=leg.request_state,
+                allow_input_required=True,
+            )
+        ```
+
         Args:
             name (str): The name of the tool to call.
             arguments (dict[str, Any]): Arguments to pass to the tool.
@@ -160,8 +216,9 @@ class ClientToolsMixin:
                 can access this via `context.request_context.meta`. Defaults to None.
 
         Returns:
-            mcp_types.CallToolResult: The complete response object from the protocol,
-                containing the tool result and any additional metadata.
+            The complete response object from the protocol. An
+            `InputRequiredResult` when the tool asked for input and
+            `allow_input_required` is set; otherwise a `CallToolResult`.
 
         Raises:
             RuntimeError: If called while the client is not connected.
@@ -214,7 +271,15 @@ class ClientToolsMixin:
                     allow_claimed=has_claims,
                 )
 
-            first = await self._await_with_session_monitoring(_retry(None, None))
+            first = await self._await_with_session_monitoring(
+                _retry(input_responses, request_state)
+            )
+            if allow_input_required and isinstance(
+                first, mcp_types.InputRequiredResult
+            ):
+                # The caller is driving the exchange, so hand back the ask
+                # untouched rather than resolving it against our own handlers.
+                return first
             driven = await self._await_with_session_monitoring(
                 self._drive_input_required(first, _retry)
             )
@@ -281,6 +346,9 @@ class ClientToolsMixin:
         progress_handler: ProgressHandler | None = None,
         raise_on_error: bool = True,
         meta: dict[str, Any] | None = None,
+        input_responses: mcp_types.InputResponses | None = None,
+        request_state: str | None = None,
+        allow_input_required: bool = False,
     ) -> CallToolResult:
         """Call a tool on the server.
 
@@ -325,7 +393,21 @@ class ClientToolsMixin:
             timeout=timeout,
             progress_handler=progress_handler,
             meta=request_meta or None,
+            input_responses=input_responses,
+            request_state=request_state,
+            allow_input_required=cast("Literal[True]", allow_input_required),
         )
+        if isinstance(result, mcp_types.InputRequiredResult):
+            # The caller is driving; hand the ask back rather than parsing it as
+            # tool output, which it is not.
+            from fastmcp.client.client import CallToolResult
+
+            return CallToolResult(
+                content=[],
+                structured_content=None,
+                meta=None,
+                input_required=result,
+            )
         return await self._parse_call_tool_result(
             name, result, raise_on_error=raise_on_error
         )

tests/server/test_elicit_resolution.py

@@ -548,6 +548,127 @@ class TestConditionalResolvers:
                 return airport
 
 
+def accepted(**values) -> mcp_types.InputResponses:
+    """The `input_responses` map for one leg: an accepted answer per key."""
+    return {
+        key: mcp_types.ElicitResult(action="accept", content={"value": value})
+        for key, value in values.items()
+    }
+
+
+def questions(result) -> dict[str, str]:
+    """The message shown for each key on one leg of a call."""
+    leg = result.input_required
+    assert leg is not None, "expected an ask, got a terminal result"
+    assert leg.input_requests is not None
+    asked: dict[str, str] = {}
+    for key, request in leg.input_requests.items():
+        assert isinstance(request, mcp_types.ElicitRequest)
+        assert request.params is not None
+        asked[key] = request.params.message
+    return asked
+
+
+def carried(result) -> str | None:
+    """The opaque state to hand back on the next leg."""
+    assert result.input_required is not None
+    return result.input_required.request_state
+
+
+class TestDrivingLegsByHand:
+    """`allow_input_required=True` hands back each leg instead of resolving it,
+    so a test can assert on the wire shape a client would actually receive."""
+
+    async def test_each_leg_is_visible(self):
+        mcp = FastMCP("x")
+
+        def which_airport(destination: str) -> Elicit[str]:
+            return Elicit(f"Which airport in {destination}?", response_type=str)
+
+        @mcp.tool
+        async def book(
+            destination: Annotated[str, Elicit("Where would you like to fly?")],
+            airport: Annotated[str, Elicit(which_airport)],
+        ) -> str:
+            return f"Booked {destination}/{airport}"
+
+        # No elicitation_handler — nothing drives the exchange but this test.
+        async with Client(mcp) as client:
+            first = await client.call_tool("book", allow_input_required=True)
+            assert questions(first) == {"destination": "Where would you like to fly?"}
+
+            second = await client.call_tool(
+                "book",
+                input_responses=accepted(destination="Paris"),
+                request_state=carried(first),
+                allow_input_required=True,
+            )
+            # Only the airport — the destination is not asked again.
+            assert questions(second) == {"airport": "Which airport in Paris?"}
+
+            final = await client.call_tool(
+                "book",
+                input_responses=accepted(airport="CDG"),
+                request_state=carried(second),
+                allow_input_required=True,
+            )
+
+        assert final.input_required is None
+        assert final.data == "Booked Paris/CDG"
+
+    async def test_independent_questions_arrive_in_one_leg(self):
+        mcp = FastMCP("x")
+
+        @mcp.tool
+        async def book(
+            destination: Annotated[str, Elicit("Where to?")],
+            date: Annotated[str, Elicit("When?")],
+        ) -> str:
+            return f"{destination} on {date}"
+
+        async with Client(mcp) as client:
+            first = await client.call_tool("book", allow_input_required=True)
+            assert questions(first) == {
+                "destination": "Where to?",
+                "date": "When?",
+            }
+
+            final = await client.call_tool(
+                "book",
+                input_responses=accepted(destination="Paris", date="2026-08-01"),
+                request_state=carried(first),
+                allow_input_required=True,
+            )
+
+        assert final.data == "Paris on 2026-08-01"
+
+    async def test_a_resolver_that_knows_asks_nothing(self):
+        mcp = FastMCP("x")
+
+        def which_airport(destination: str) -> str | Elicit[str]:
+            return (
+                "LHR"
+                if destination == "London"
+                else Elicit("Which?", response_type=str)
+            )
+
+        @mcp.tool
+        async def book(
+            destination: str,
+            airport: Annotated[str, Elicit(which_airport)],
+        ) -> str:
+            return f"{destination}/{airport}"
+
+        async with Client(mcp) as client:
+            result = await client.call_tool(
+                "book", {"destination": "London"}, allow_input_required=True
+            )
+
+        # Terminal on the first leg — there was never anything to ask.
+        assert result.input_required is None
+        assert result.data == "London/LHR"
+
+
 class TestAskedOnce:
     """An answer already given satisfies its question on later rounds."""