Client auto-resolves InputRequiredResult via existing callbacks (SEP-2322) (#2998)

.github/actions/conformance/client.py

@@ -22,7 +22,7 @@ Scenarios:
     http-standard-headers                   - Connect, call a tool (Mcp-* headers checked)
     http-invalid-tool-headers               - List tools, call every surfaced tool (x-mcp-header filter)
     elicitation-sep1034-client-defaults     - Elicitation with default accept callback
-    sep-2322-client-request-state           - Drive the manual MRTR retry surface
+    sep-2322-client-request-state           - Drive the MRTR auto-loop (SEP-2322)
     auth/client-credentials-jwt             - Client credentials with private_key_jwt
     auth/client-credentials-basic           - Client credentials with client_secret_basic
     auth/*                                  - Authorization code flow (default for auth scenarios)
@@ -374,46 +374,28 @@ async def run_elicitation_defaults(server_url: str) -> None:
 
 @register("sep-2322-client-request-state")
 async def run_mrtr_client(server_url: str) -> None:
-    """Drive the manual MRTR retry surface against the SEP-2322 client mock.
+    """Drive the SEP-2322 client mock through `Client.call_tool`'s auto-loop.
 
-    The mock speaks the modern lifecycle (server/discover, no initialize) and
+    The mock inspects raw `tools/call` params, so registering an
-    inspects the wire params of each tools/call round, so this exercises the
+    `elicitation_callback` and letting the driver run is enough to satisfy
-    explicit allow_input_required=True path rather than an auto-loop: round 1
+    all five wire-shape checks: the driver echoes `request_state` byte-exact
-    receives an InputRequiredResult, the fixture fulfils the elicitation
+    and omits it when the server sent none, every retry mints a fresh
-    locally, then round 2 retries with input_responses + the echoed
+    JSON-RPC id, the unrelated call between auto-loops carries no MRTR
-    request_state. Passing request_state straight off the typed result -- a
+    params, and the no-`resultType` response parses as a terminal
-    str when the server sent one, None when it didn't -- lets the
+    `CallToolResult` so the driver never retries it.
-    serializer's exclude_none drop the key in the no-state case without a
-    branch here. The unrelated call between rounds proves MRTR params don't
-    leak across tools, and the no-result-type call must parse as a complete
-    CallToolResult with no retry.
     """
-    async with Client(server_url, mode=client_mode()) as client:
+
+    async def confirm(
+        context: ClientRequestContext, params: types.ElicitRequestParams
+    ) -> types.ElicitResult | types.ErrorData:
+        return types.ElicitResult(action="accept", content={"confirmed": True})
+
+    async with Client(server_url, mode=client_mode(), elicitation_callback=confirm) as client:
         await client.list_tools()
-        confirm = {"confirm": types.ElicitResult(action="accept", content={"confirmed": True})}
-
-        r1 = await client.call_tool("test_mrtr_echo_state", {}, allow_input_required=True)
-        assert isinstance(r1, types.InputRequiredResult)
 
+        await client.call_tool("test_mrtr_echo_state", {})
         await client.call_tool("test_mrtr_unrelated", {})
-
+        await client.call_tool("test_mrtr_no_state", {})
-        await client.call_tool(
-            "test_mrtr_echo_state",
-            {},
-            input_responses=confirm,
-            request_state=r1.request_state,
-            allow_input_required=True,
-        )
-
-        r2 = await client.call_tool("test_mrtr_no_state", {}, allow_input_required=True)
-        assert isinstance(r2, types.InputRequiredResult)
-        await client.call_tool(
-            "test_mrtr_no_state",
-            {},
-            input_responses=confirm,
-            request_state=r2.request_state,
-            allow_input_required=True,
-        )
 
         result = await client.call_tool("test_mrtr_no_result_type", {})
         assert isinstance(result, types.CallToolResult)

docs/advanced/deprecated.md

@@ -8,7 +8,7 @@ The table below names each deprecated feature, why it is going away, and the rep
 
 | Deprecated | Why | What you do instead |
 |---|---|---|
-| **Roots**: `ctx.session.list_roots()`, `client.send_roots_list_changed()`, the `list_roots_callback=` you pass to `Client(...)` | [SEP-2577](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577) retires the capability. | Take the paths that matter as ordinary tool arguments or resource URIs. |
+| **Roots**: `ctx.session.list_roots()`, `client.send_roots_list_changed()`, the `list_roots_callback=` you pass to `Client(...)` | [SEP-2577](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577) retires the capability. | Take the paths as ordinary tool arguments or resource URIs, or embed a `ListRootsRequest` in an `InputRequiredResult` (see **Multi-round-trip requests**). |
 | **Server-initiated sampling**: `ctx.session.create_message()`, the `sampling_callback=` you pass to `Client(...)` | SEP-2577 retires the capability. | Return `InputRequiredResult` and let the client retry the call (see **Multi-round-trip requests**). |
 | **Protocol logging**: `ctx.log()`, `ctx.debug()`, `ctx.info()`, `ctx.warning()`, `ctx.error()`, `ctx.session.send_log_message()`, `client.set_logging_level()` | SEP-2577 retires the capability. Nothing in-protocol replaces it. | Ordinary `import logging` to stderr (see **Logging**). |
 | **`ping`**: `client.send_ping()` | **Removed** from the protocol, not merely deprecated. There is no `ping` method in 2026-07-28. | Nothing. It only works against a `mode="legacy"` connection. |
@@ -17,7 +17,7 @@ The table below names each deprecated feature, why it is going away, and the rep
 Three things fall out of that table:
 
 * Roots, sampling, and logging go together. One proposal, **SEP-2577**, deprecates all three capabilities at once.
-* Sampling and roots share a deeper problem: they are the two places a **server** sends a **request** to the **client**. That whole direction is what 2026-07-28 replaces with **Multi-round-trip requests**.
+* Sampling and roots share a deeper problem: they are places a **server** sends a **request** to the **client**. That whole direction is what 2026-07-28 replaces with **Multi-round-trip requests**. It is the standalone RPC methods (`sampling/createMessage`, `roots/list`, and push-style `elicitation/create`) that are gone; the `CreateMessageRequest` / `ListRootsRequest` / `ElicitRequest` payload types survive, embedded in `InputRequiredResult.input_requests`, and on the client they hit the same callbacks.
 * `ping` is the odd one out. The protocol does not deprecate it, it removes it. The SDK method still warns (its message says *removed*, not *deprecated*) and calling it on a modern connection answers with *"Method not found"*.
 
 ## Deprecated is advisory
@@ -82,7 +82,7 @@ That is the whole API. There is no per-method switch, and you don't want one: th
 ## Recap
 
 * The 2026-07-28 spec deprecates **roots**, server-initiated **sampling**, and protocol **logging** (all SEP-2577), restricts **progress** to server-to-client, and removes **`ping`**.
-* The replacement column points you onward: **Multi-round-trip requests** for sampling, **Logging** for logging, **Progress** for progress. Roots needs no chapter (pass the paths as arguments) and `ping` needs nothing at all.
+* The replacement column points you onward: **Multi-round-trip requests** for sampling and roots, **Logging** for logging, **Progress** for progress. `ping` needs nothing at all.
 * Deprecated is advisory: no wire changes, everything keeps working against pre-2026 sessions, and you get a visible `MCPDeprecationWarning` (a `UserWarning`, so it is on by default).
 * Sampling and roots additionally need a back-channel that a 2026-07-28 session does not have. On a modern connection they warn and then they raise.
 * `warnings.filterwarnings("ignore", category=MCPDeprecationWarning)` silences the whole category; `"error::mcp.MCPDeprecationWarning"` in pytest turns it into a test failure.

docs/advanced/multi-round-trip.md

@@ -33,40 +33,43 @@ Everything else in that file (the explicit `input_schema`, the hand-built `CallT
 
 ## The client side
 
-`call_tool` will not hand you an `InputRequiredResult` unless you opt in.
+`Client` runs the loop for you.
 
-!!! check
+Register the callbacks the server might ask for (`elicitation_callback`, `sampling_callback`, `list_roots_callback`) and call the tool. When an `InputRequiredResult` arrives, `Client` dispatches each entry in `input_requests` to the matching callback, retries with the answers and the echoed `request_state`, and keeps going until a `CallToolResult` comes back:
-    Call a tool that needs input without opting in and `call_tool` raises:
 
-    ```text
+```python title="client.py" hl_lines="12 13"
-    Server returned InputRequiredResult; pass allow_input_required=True to receive it and retry call_tool(..., input_responses=..., request_state=result.request_state).
+--8<-- "docs_src/mrtr/tutorial003.py"
-    ```
-
-    That is deliberate. Most call sites expect a result or an exception, not a third thing in the
-    middle of the happy path, and pyright agrees: without the flag, `call_tool` is typed to return
-    a plain `CallToolResult`.
-
-Pass `allow_input_required=True` and the result reaches you intact:
-
-```python
-result.result_type     # 'input_required'
-result.request_state   # 'provision-v1'
-result.input_requests  # {'region': ElicitRequest(method='elicitation/create', params=ElicitRequestFormParams(...))}
 ```
 
-### The retry loop
+* That `elicitation_callback` is the same one a pre-2026 server's back-channel `elicitation/create` would have hit. The same is true of `sampling_callback` for `sampling/createMessage` and `list_roots_callback` for `roots/list`: at 2026-07-28 the standalone server->client RPCs are gone, but the identical `ElicitRequest` / `CreateMessageRequest` / `ListRootsRequest` payloads ride inside `input_requests` and dispatch to the same three callbacks. One set of callbacks serves both eras.
+* `call_tool` returns a plain `CallToolResult`. The intermediate rounds are invisible to the caller.
+* `get_prompt` and `read_resource` drive the same loop.
 
-Now you own the loop. There is no automatic driver yet; `while isinstance(result, InputRequiredResult)` **is** the API:
+!!! check
+    Leave the callback off and the loop fails on the first round: the SDK's stand-in callback
+    answers every elicitation with an error, and `call_tool` raises `MCPError` with the message
+    *"Elicitation not supported"*.
 
-```python title="client.py" hl_lines="13-15 17-20"
+The loop is bounded. `Client(..., input_required_max_rounds=10)` is the default cap; a server that keeps returning `InputRequiredResult` past it makes `call_tool` raise. If a round carries only `request_state` and no `input_requests`, `Client` sleeps briefly (50ms doubling to a 250ms ceiling) before retrying, so a server that is just saying *"not done yet"* isn't busy-polled.
+
+### Driving the loop yourself
+
+The auto-loop is enough for a single-process client. Own the loop instead when:
+
+* Your client is **distributed**: the process that renders the question to the user is not the process that called `call_tool`, so a different worker issues the retry. `request_state` is the persistable token you carry across that boundary, through your own storage, and `input_responses` is what the other side sends back with it.
+* You want to **inspect** each round: log or audit every `input_requests` entry, refuse certain request kinds, or apply your own backoff between legs.
+* You want a **wall-clock** bound rather than a round-count bound: wrap your own loop in `anyio.fail_after(...)` instead of relying on `input_required_max_rounds`.
+
+Drop to the underlying session, where `allow_input_required=True` hands you the union directly:
+
+```python title="client.py" hl_lines="13 14 20"
 --8<-- "docs_src/mrtr/tutorial002.py"
 ```
 
-* `allow_input_required=True` widens the return type to `CallToolResult | InputRequiredResult`. That union is exactly what the `isinstance` is narrowing.
+* `client.session.call_tool(..., allow_input_required=True)` widens the return type to `CallToolResult | InputRequiredResult`. The `isinstance` is what narrows it back.
+* `request_state` is now in your hands. Write it down between legs and the conversation can resume from a fresh process.
 * For every entry in `input_requests` you put an `InputResponse` under the **same key** in `input_responses`. `fulfil` is where your UI goes; this one hard-codes the answer.
 * Same tool name, same `arguments`, every leg. The retry is the original call carried out again, not a new method.
-* `request_state=result.request_state`: copy it across. Never inspect it, never invent it.
-* When the server has everything it needs it returns a `CallToolResult` and the loop exits.
 
 ## A 2026-07-28 result
 
@@ -88,9 +91,8 @@ Now you own the loop. There is no automatic driver yet; `while isinstance(result
 
 * At 2026-07-28 a server that needs input mid-call **returns** an `InputRequiredResult`. It never opens a request to the client.
 * `input_requests` is what it needs. `request_state` is an opaque resume token only the server reads.
-* The client answers by calling the **same tool again** with `input_responses=` and `request_state=`.
+* `Client` runs the retry loop for you: register `elicitation_callback` / `sampling_callback` / `list_roots_callback` and `call_tool` returns a plain `CallToolResult`. `input_required_max_rounds` (default 10) bounds it.
-* By default `call_tool` raises on an `InputRequiredResult`; `allow_input_required=True` opts in and widens the return type.
+* To inspect or persist rounds, use `client.session.call_tool(..., allow_input_required=True)` and own the `while isinstance(result, InputRequiredResult)` loop yourself.
-* The manual `while isinstance(result, InputRequiredResult)` loop is the whole client API; there is no auto-retry driver yet.
 * The server side is the **low-level** `Server` only; `@mcp.tool()` has no sugar for this yet.
 
 This is the mechanism that replaces server-initiated sampling and the rest of the push-style back-channel; see **Deprecated features**.

docs/client/callbacks.md

@@ -61,6 +61,10 @@ One `tools/call` from you, one `elicitation/create` back from the server, answer
     protocol does, in-memory and over a URL alike. Pin `mode="legacy"` whenever your client has
     to answer one; every test behind this page does. **Protocol versions** has the whole story.
 
+    On a 2026-07-28 session the callback isn't dead, it's fed differently: when a tool returns an
+    `InputRequiredResult` carrying an `ElicitRequest`, `Client` dispatches that entry to the same
+    `elicitation_callback` and retries the call for you. That flow is **Multi-round-trip requests**.
+
 ## A callback is a capability
 
 You never told the server that your client can answer elicitation requests. The SDK did.
@@ -109,7 +113,7 @@ Pass all three callbacks and you get `['elicitation', 'sampling', 'roots']`. Pas
 
 `sampling_callback` answers `sampling/createMessage`: the server asking *your* model to complete something. `list_roots_callback` answers `roots/list`: the server asking which directories it may work in.
 
-Both work. Both follow the rule above. And both serve features the **2026-07-28 spec deprecates**: a modern server doesn't call back into your model mid-request, it hands the request back to you as part of the tool result (**Multi-round-trip requests**), and roots give way to plain tool arguments and resource URIs. The whole list is in **Deprecated features**.
+Both work. Both follow the rule above. And both serve RPCs the **2026-07-28 spec removes**: a modern server doesn't call back into your client mid-request, it hands the request back to you as part of the tool result (**Multi-round-trip requests**). The callbacks themselves are not dead. When an `InputRequiredResult` carries a `CreateMessageRequest` or a `ListRootsRequest`, `Client`'s auto-loop dispatches it to the same `sampling_callback` or `list_roots_callback` you registered here. The whole list is in **Deprecated features**.
 
 You still need the callbacks to talk to servers that haven't moved. The signatures:
 

docs/migration.md

@@ -395,9 +395,13 @@ 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.
 
-### `call_tool` can return `InputRequiredResult` (opt-in)
+### `InputRequiredResult` handling differs between `Client` and `ClientSession`
 
-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=`.
+For protocol 2026-07-28, `tools/call`, `prompts/get`, and `resources/read` may return an `InputRequiredResult` asking the client to supply additional input (sampling, elicitation, roots) and retry.
+
+On the high-level `Client`, `call_tool`, `get_prompt`, and `read_resource` resolve this automatically: they dispatch each requested input to the matching callback (`sampling_callback`, `elicitation_callback`, `list_roots_callback`) and retry until a final result is returned, so the call still returns the bare `CallToolResult` / `GetPromptResult` / `ReadResourceResult`. The round limit is `Client(input_required_max_rounds=...)` (default 10). Earlier v2 prereleases exposed an `allow_input_required` parameter on these `Client` methods; that parameter has been removed. For manual control use `client.session.call_tool(..., allow_input_required=True)`. Note that `read_timeout_seconds` now bounds each underlying round, not the whole loop; wrap the call in `anyio.fail_after(...)` for a whole-loop bound.
+
+On `ClientSession`, `call_tool` / `get_prompt` / `read_resource` still return the bare result and raise `RuntimeError` if the server requests input. Pass `allow_input_required=True` to receive the `InputRequiredResult` instead, then drive the loop yourself with `input_responses=` / `request_state=`. `ClientSessionGroup.call_tool` accepts the same flag.
 
 ### `call_tool` mirrors `x-mcp-header` arguments into `Mcp-Param-*` headers (SEP-2243)
 

docs/tutorial/elicitation.md

@@ -149,5 +149,6 @@ Now swap in the URL-mode `server.py` and point the same `main()` at `pay_deposit
 * `result.action` is `"accept"`, `"decline"` or `"cancel"`; `result.data` exists only on accept.
 * `await ctx.elicit_url(message, url, elicitation_id)` is for everything that must not pass through the model; `ctx.session.send_elicit_complete(elicitation_id)` says the out-of-band part is done.
 * The client answers with one `elicitation_callback`, branching on the params type; registering it is what declares the capability.
+* On a 2026-07-28 connection the server returns the question instead of pushing it; the same callback is fed by **Multi-round-trip requests**.
 
 A tool that can ask is good. A tool that says how far along it is (**Progress**) is next.

docs_src/mrtr/tutorial002.py

@@ -10,10 +10,10 @@ def fulfil(request: InputRequest) -> InputResponse:
 
 
 async def provision(client: Client, name: str) -> CallToolResult:
-    result = await client.call_tool("provision", {"name": name}, allow_input_required=True)
+    result = await client.session.call_tool("provision", {"name": name}, allow_input_required=True)
     while isinstance(result, InputRequiredResult):
         responses = {key: fulfil(request) for key, request in (result.input_requests or {}).items()}
-        result = await client.call_tool(
+        result = await client.session.call_tool(
             "provision",
             {"name": name},
             input_responses=responses,

docs_src/mrtr/tutorial003.py

@@ -0,0 +1,14 @@
+from mcp_types import ElicitRequestParams, ElicitResult
+
+from mcp import Client
+from mcp.client import ClientRequestContext
+
+
+async def handle_elicitation(context: ClientRequestContext, params: ElicitRequestParams) -> ElicitResult:
+    return ElicitResult(action="accept", content={"region": "eu-west-1"})
+
+
+async def main() -> None:
+    async with Client("http://127.0.0.1:8000/mcp", elicitation_callback=handle_elicitation) as client:
+        result = await client.call_tool("provision", {"name": "orders"})
+        print(result.content)

examples/stories/manifest.toml

@@ -34,6 +34,9 @@ multi_connection = true
 # progress + log notifications dropped on the modern streamable-HTTP path pending SSE wiring
 xfail = ["http-asgi:modern"]
 
+[story.mrtr]
+era = "modern"
+
 [story.legacy_elicitation]
 era    = "legacy"
 status = "legacy"
@@ -140,7 +143,6 @@ fixed_port    = 8000                            # issuer/PRM metadata bake in :8
 
 [deferred]
 caching       = "client honouring + per-result override unlanded"
-mrtr          = "#2898 — InputRequiredResult runtime"
 subscriptions = "#2901 — Client.listen / ServerEventBus"
 tasks         = "extensions capability map + tasks runtime"
 apps          = "#2896 — extensions capability map"

examples/stories/mrtr/README.md

@@ -1,30 +1,54 @@
 # mrtr
 
-Multi-round tool results: a 2026-era tool call returns
+Multi-round tool result: on the 2026-07-28 protocol a tool that needs user
-`resultType: "input_required"` with a `requestState` HMAC instead of pushing an
+input mid-call **returns** `resultType: "input_required"` with embedded
-`elicitation/create` request. The client fulfils the input and resubmits, and
+`inputRequests` and an opaque `requestState`, instead of pushing a
-the server resumes from the carried state. The story will show both the
+server→client request. The client fulfils the embedded requests and retries the
-auto-fulfil helper and a manual resubmit loop.
+original `tools/call` carrying `inputResponses` and the echoed `requestState`.
+The story shows both the `Client` auto-loop (one `await call_tool`, callbacks
+fired transparently) and a manual `client.session` loop (the persistable form).
 
-**Status: not yet implemented** ([#2898](https://github.com/modelcontextprotocol/python-sdk/issues/2898)).
+## Run it
-The lowlevel registration surface is in this base —
+
-[#2967](https://github.com/modelcontextprotocol/python-sdk/pull/2967)
+```bash
-(`ae13ede`) widened the tool/prompt/resource handler return types to include
+# HTTP — the client self-hosts the server on a free port, runs, then tears it
-`InputRequiredResult`. The runnable story is deliberately a follow-up PR to
+# down (the InputRequiredResult round-trip is 2026-era only)
-keep this one reviewable.
+uv run python -m stories.mrtr.client --http
+# same, against the lowlevel-API server variant
+uv run python -m stories.mrtr.client --http --server server_lowlevel
+```
+
+## What to look at
+
+- `client.py` `main` — the auto-loop is invisible at the call site:
+  `Client(target, mode=mode, elicitation_callback=on_elicit)` then
+  `await client.call_tool("deploy", ...)`. The same `on_elicit` callback the
+  legacy push path uses is dispatched for each embedded `inputRequests` entry.
+- `client.py` manual block — `client.session.call_tool(...,
+  allow_input_required=True)` returns the raw `InputRequiredResult` so
+  `request_state` can be persisted between rounds; the retry is just another
+  `tools/call` with `input_responses=` / `request_state=`.
+- `server.py` `deploy` — `ctx.input_responses` / `ctx.request_state` read the
+  retry payload; the first round returns
+  `InputRequiredResult(input_requests={...}, request_state=...)`, the second
+  returns the final string.
+- `server_lowlevel.py` — same wire contract via `params.input_responses` /
+  `params.request_state` and a hand-built `InputRequiredResult`.
+
+## Caveats
+
+- **Loop bound.** The auto-loop gives up after `input_required_max_rounds`
+  (default 10) with `InputRequiredRoundsExceededError`; raise it on the
+  `Client` ctor or drop to the manual loop.
+- **`requestState` integrity is the server's job.** The client echoes it
+  byte-exact and never inspects it; the server MUST treat it as
+  attacker-controlled. The SDK ships no signing helper yet.
 
 ## Spec
 
-[Multi-round tool results — server features](https://modelcontextprotocol.io/specification/draft/server/tools#multi-round-results)
+[Multi-round results — server features](https://modelcontextprotocol.io/specification/draft/server/tools#multi-round-results)
-
-## Working example elsewhere
-
-The TypeScript SDK ships a runnable `mrtr` story:
-[typescript-sdk/examples/mrtr](https://github.com/modelcontextprotocol/typescript-sdk/tree/main/examples/mrtr).
 
 ## See also
 
-`legacy_elicitation/` and `sampling/` — the handshake-era push equivalents that
+`legacy_elicitation/` and `sampling/` — the handshake-era push equivalents this
-this mechanism replaces on the 2026 protocol. The TypeScript SDK ships a single
+mechanism replaces on the 2026 protocol.
-dual-era `elicitation/` story covering both eras in one place; we re-merge
-`legacy_elicitation/` back into `elicitation/` once MRTR lands.

examples/stories/mrtr/client.py

@@ -0,0 +1,46 @@
+"""Drive the deploy tool both ways: the Client auto-loop, and a manual session-level loop."""
+
+import mcp_types as types
+
+from mcp.client import Client, ClientRequestContext
+from stories._harness import Target, run_client
+
+
+async def on_elicit(context: ClientRequestContext, params: types.ElicitRequestParams) -> types.ElicitResult:
+    # The same callback serves legacy push-style elicitation/create requests AND embedded
+    # InputRequiredResult.input_requests entries — the driver dispatches both here.
+    assert isinstance(params, types.ElicitRequestFormParams)
+    assert "confirm" in params.requested_schema["properties"]
+    return types.ElicitResult(action="accept", content={"confirm": True})
+
+
+async def main(target: Target, *, mode: str = "auto") -> None:
+    async with Client(target, mode=mode, elicitation_callback=on_elicit) as client:
+        # ── auto-loop: Client.call_tool dispatches input_requests to on_elicit and retries
+        # internally; the caller just sees the final CallToolResult.
+        deployed = await client.call_tool("deploy", {"env": "production"})
+        assert isinstance(deployed.content[0], types.TextContent)
+        assert deployed.content[0].text == "deployed to production", deployed
+
+        # ── manual loop: drop to client.session for the raw InputRequiredResult so the
+        # request_state can be persisted between rounds (e.g. across a process restart).
+        first = await client.session.call_tool("deploy", {"env": "staging"}, allow_input_required=True)
+        assert isinstance(first, types.InputRequiredResult)
+        assert first.input_requests is not None and "confirm" in first.input_requests
+        assert first.request_state == "awaiting-confirm"
+        # Decline this time so the path diverges from the auto-loop run above.
+        responses: types.InputResponses = {"confirm": types.ElicitResult(action="decline")}
+        second = await client.session.call_tool(
+            "deploy",
+            {"env": "staging"},
+            input_responses=responses,
+            request_state=first.request_state,
+            allow_input_required=True,
+        )
+        assert isinstance(second, types.CallToolResult)
+        assert isinstance(second.content[0], types.TextContent)
+        assert second.content[0].text == "deployment to staging cancelled", second
+
+
+if __name__ == "__main__":
+    run_client(main)

examples/stories/mrtr/server.py

@@ -0,0 +1,39 @@
+"""Multi-round tool result (2026 era): a tool returns input_required and resumes from echoed state."""
+
+from mcp_types import ElicitRequest, ElicitRequestedSchema, ElicitRequestFormParams, ElicitResult, InputRequiredResult
+
+from mcp.server.mcpserver import Context, MCPServer
+from stories._hosting import run_server_from_args
+
+CONFIRM_SCHEMA: ElicitRequestedSchema = {
+    "type": "object",
+    "properties": {"confirm": {"type": "boolean", "description": "Proceed with the deployment?"}},
+    "required": ["confirm"],
+}
+
+
+def build_server() -> MCPServer:
+    mcp = MCPServer("mrtr-example")
+
+    @mcp.tool(description="Deploy to an environment, asking the user to confirm first.")
+    async def deploy(env: str, ctx: Context) -> str | InputRequiredResult:
+        responses = ctx.input_responses
+        if responses is None or "confirm" not in responses:
+            # First round: ask the client to elicit confirmation. request_state is opaque
+            # to the client; here it carries the step name so the retry can verify the echo.
+            ask = ElicitRequest(
+                params=ElicitRequestFormParams(message=f"Deploy to {env}?", requested_schema=CONFIRM_SCHEMA)
+            )
+            return InputRequiredResult(input_requests={"confirm": ask}, request_state="awaiting-confirm")
+        # Retry round: the client echoed request_state byte-exact and supplied the answer.
+        assert ctx.request_state == "awaiting-confirm", ctx.request_state
+        answer = responses["confirm"]
+        if isinstance(answer, ElicitResult) and answer.action == "accept" and (answer.content or {}).get("confirm"):
+            return f"deployed to {env}"
+        return f"deployment to {env} cancelled"
+
+    return mcp
+
+
+if __name__ == "__main__":
+    run_server_from_args(build_server)

examples/stories/mrtr/server_lowlevel.py

@@ -0,0 +1,62 @@
+"""Multi-round tool result (2026 era) against the low-level Server."""
+
+from typing import Any
+
+import mcp_types as types
+
+from mcp.server.context import ServerRequestContext
+from mcp.server.lowlevel import Server
+from stories._hosting import run_server_from_args
+
+CONFIRM_SCHEMA: types.ElicitRequestedSchema = {
+    "type": "object",
+    "properties": {"confirm": {"type": "boolean", "description": "Proceed with the deployment?"}},
+    "required": ["confirm"],
+}
+DEPLOY_INPUT_SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "properties": {"env": {"type": "string"}},
+    "required": ["env"],
+}
+
+
+def build_server() -> Server[Any]:
+    async def list_tools(
+        ctx: ServerRequestContext[Any], params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(
+            tools=[
+                types.Tool(
+                    name="deploy",
+                    description="Deploy to an environment, asking the user to confirm first.",
+                    input_schema=DEPLOY_INPUT_SCHEMA,
+                )
+            ]
+        )
+
+    async def call_tool(
+        ctx: ServerRequestContext[Any], params: types.CallToolRequestParams
+    ) -> types.CallToolResult | types.InputRequiredResult:
+        assert params.name == "deploy" and params.arguments is not None
+        env = params.arguments["env"]
+        responses = params.input_responses
+        if responses is None or "confirm" not in responses:
+            ask = types.ElicitRequest(
+                params=types.ElicitRequestFormParams(message=f"Deploy to {env}?", requested_schema=CONFIRM_SCHEMA)
+            )
+            return types.InputRequiredResult(input_requests={"confirm": ask}, request_state="awaiting-confirm")
+        assert params.request_state == "awaiting-confirm", params.request_state
+        answer = responses["confirm"]
+        if (
+            isinstance(answer, types.ElicitResult)
+            and answer.action == "accept"
+            and (answer.content or {}).get("confirm")
+        ):
+            return types.CallToolResult(content=[types.TextContent(text=f"deployed to {env}")])
+        return types.CallToolResult(content=[types.TextContent(text=f"deployment to {env} cancelled")])
+
+    return Server("mrtr-example", on_list_tools=list_tools, on_call_tool=call_tool)
+
+
+if __name__ == "__main__":
+    run_server_from_args(build_server)

src/mcp-types/mcp_types/_types.py

@@ -2061,7 +2061,7 @@ class InputRequiredResult(Result):
 
     @model_validator(mode="after")
     def _require_one_field(self) -> Self:
-        if self.input_requests is None and self.request_state is None:
+        if not self.input_requests and self.request_state is None:
             raise ValueError("InputRequiredResult requires at least one of input_requests or request_state")
         return self
 

src/mcp/__init__.py

@@ -58,6 +58,7 @@ from mcp_types import (
 )
 from mcp_types import Role as SamplingRole
 
+from .client._input_required import InputRequiredRoundsExceededError
 from .client.client import Client
 from .client.session import ClientSession
 from .client.session_group import ClientSessionGroup
@@ -87,6 +88,7 @@ __all__ = [
     "InitializeRequest",
     "InitializeResult",
     "InitializedNotification",
+    "InputRequiredRoundsExceededError",
     "JSONRPCError",
     "JSONRPCRequest",
     "JSONRPCResponse",

src/mcp/client/__init__.py

@@ -1,8 +1,9 @@
 """MCP Client module."""
 
+from mcp.client._input_required import InputRequiredRoundsExceededError
 from mcp.client._transport import Transport
 from mcp.client.client import Client
 from mcp.client.context import ClientRequestContext
 from mcp.client.session import ClientSession
 
-__all__ = ["Client", "ClientRequestContext", "ClientSession", "Transport"]
+__all__ = ["Client", "ClientRequestContext", "ClientSession", "InputRequiredRoundsExceededError", "Transport"]

src/mcp/client/_input_required.py

@@ -0,0 +1,127 @@
+"""SEP-2322 client-side multi-round-trip driver.
+
+When a server returns `InputRequiredResult` instead of the normal result of a
+`tools/call` / `prompts/get` / `resources/read`, the client fulfils the
+embedded `input_requests` (sampling, elicitation, roots) and retries the
+original request carrying the responses and the echoed opaque `request_state`.
+This module implements that retry loop as a pure function so it can drive any
+of the three methods identically; `Client` builds the `dispatch` and `retry`
+closures, `ClientSession` stays mechanics-only.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import TypeVar
+
+import anyio
+import anyio.abc
+from mcp_types import ErrorData, InputRequest, InputRequiredResult, InputResponse, InputResponses
+
+from mcp.shared.exceptions import MCPError
+
+DEFAULT_INPUT_REQUIRED_MAX_ROUNDS = 10
+"""Default cap on `InputRequiredResult` retry rounds before the driver gives up.
+
+Matches the typescript-sdk default; csharp-sdk and go-sdk use the same value
+as a hard constant.
+"""
+
+_STATE_ONLY_BACKOFF_INITIAL_SECONDS = 0.05
+"""First sleep when an `InputRequiredResult` carries only `request_state` (no input requests)."""
+
+_STATE_ONLY_BACKOFF_CAP_SECONDS = 0.25
+"""Upper bound on the state-only backoff sleep; reached after three consecutive state-only legs."""
+
+
+ResultT = TypeVar("ResultT")
+
+
+class InputRequiredRoundsExceededError(RuntimeError):
+    """The server kept returning `InputRequiredResult` past the configured `max_rounds`."""
+
+    def __init__(self, max_rounds: int) -> None:
+        super().__init__(
+            f"Server returned InputRequiredResult for more than {max_rounds} rounds; "
+            "raise input_required_max_rounds on the Client, or use "
+            "client.session.<method>(..., allow_input_required=True) to drive the loop manually."
+        )
+        self.max_rounds = max_rounds
+
+
+async def run_input_required_driver(
+    first: InputRequiredResult,
+    *,
+    dispatch: Callable[[str, InputRequest], Awaitable[InputResponse | ErrorData]],
+    retry: Callable[[InputResponses | None, str | None], Awaitable[ResultT | InputRequiredResult]],
+    max_rounds: int = DEFAULT_INPUT_REQUIRED_MAX_ROUNDS,
+) -> ResultT:
+    """Resolve an `InputRequiredResult` to its terminal result.
+
+    Loops until `retry` returns a non-`InputRequiredResult`, or `max_rounds` is
+    exhausted. Each round either dispatches all `input_requests` concurrently
+    and retries with the collected responses, or — when the server sent only
+    `request_state` — sleeps with exponential backoff (50ms doubling to a 250ms
+    cap, reset by any leg that carries input requests) and retries empty.
+    `request_state` is passed through byte-exact and never inspected.
+
+    Args:
+        first: The `InputRequiredResult` the original call returned.
+        dispatch: Runs one embedded `InputRequest` through the client's
+            sampling / elicitation / roots callbacks. Called concurrently per
+            request key. An `ErrorData` return aborts the loop as an `MCPError`.
+        retry: Re-issues the original request with the collected responses and
+            the latest `request_state`. Each call mints a fresh JSON-RPC id.
+        max_rounds: Cap on retry rounds.
+
+    Raises:
+        InputRequiredRoundsExceededError: `max_rounds` exhausted.
+        MCPError: A `dispatch` call returned `ErrorData`.
+    """
+    rounds = 0
+    state_only_delay = _STATE_ONLY_BACKOFF_INITIAL_SECONDS
+    current: ResultT | InputRequiredResult = first
+    while isinstance(current, InputRequiredResult):
+        rounds += 1
+        if rounds > max_rounds:
+            raise InputRequiredRoundsExceededError(max_rounds)
+        if current.input_requests:
+            state_only_delay = _STATE_ONLY_BACKOFF_INITIAL_SECONDS
+            responses: InputResponses | None = await _dispatch_all(current.input_requests, dispatch)
+        else:
+            await anyio.sleep(state_only_delay)
+            state_only_delay = min(state_only_delay * 2, _STATE_ONLY_BACKOFF_CAP_SECONDS)
+            responses = None
+        current = await retry(responses, current.request_state)
+    return current
+
+
+async def _dispatch_all(
+    requests: dict[str, InputRequest],
+    dispatch: Callable[[str, InputRequest], Awaitable[InputResponse | ErrorData]],
+) -> InputResponses:
+    """Run `dispatch` concurrently for every key, raising `MCPError` on the first `ErrorData`.
+
+    The first task to return `ErrorData` cancels its siblings via the task
+    group's cancel scope, so a refused input does not wait on a slow peer.
+    A callback that *raises* propagates as an `ExceptionGroup` like any other
+    task-group failure.
+    """
+    responses: InputResponses = {}
+    refused: ErrorData | None = None
+
+    async def run_one(tg: anyio.abc.TaskGroup, key: str, req: InputRequest) -> None:
+        nonlocal refused
+        result = await dispatch(key, req)
+        if isinstance(result, ErrorData):
+            refused = result
+            tg.cancel_scope.cancel()
+        else:
+            responses[key] = result
+
+    async with anyio.create_task_group() as tg:
+        for key, req in requests.items():
+            tg.start_soon(run_one, tg, key, req)
+    if refused is not None:
+        raise MCPError.from_error_data(refused)
+    return responses

src/mcp/client/client.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 from collections.abc import Awaitable, Callable, Mapping
 from contextlib import AsyncExitStack
 from dataclasses import KW_ONLY, dataclass, field
-from typing import Any, Literal, TypeVar, overload
+from typing import Any, Literal, TypeVar
 
 import anyio
 import mcp_types as types
@@ -13,9 +13,12 @@ from mcp_types import (
     CallToolResult,
     CompleteResult,
     EmptyResult,
+    ErrorData,
     GetPromptResult,
     Implementation,
+    InputRequest,
     InputRequiredResult,
+    InputResponse,
     InputResponses,
     ListPromptsResult,
     ListResourcesResult,
@@ -32,10 +35,19 @@ from mcp_types import (
 from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
 from typing_extensions import deprecated
 
+from mcp.client._input_required import DEFAULT_INPUT_REQUIRED_MAX_ROUNDS, run_input_required_driver
 from mcp.client._memory import InMemoryTransport
 from mcp.client._probe import negotiate_auto
 from mcp.client._transport import Transport
-from mcp.client.session import ClientSession, ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
+from mcp.client.session import (
+    ClientRequestContext,
+    ClientSession,
+    ElicitationFnT,
+    ListRootsFnT,
+    LoggingFnT,
+    MessageHandlerFnT,
+    SamplingFnT,
+)
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
@@ -51,6 +63,7 @@ initialize), or a modern protocol-version string (adopt directly). The ``str`` a
 forward-compat; ``Client.__post_init__`` rejects anything outside that set at construction."""
 
 _T = TypeVar("_T")
+_ResultT = TypeVar("_ResultT")
 
 _Connector = Callable[[AsyncExitStack, ConnectMode, bool], Awaitable["Dispatcher[Any]"]]
 """Resolved at ``__post_init__`` from the shape of ``server`` alone: enter whatever resources
@@ -199,6 +212,11 @@ class Client:
     elicitation_callback: ElicitationFnT | None = None
     """Callback for handling elicitation requests."""
 
+    input_required_max_rounds: int = DEFAULT_INPUT_REQUIRED_MAX_ROUNDS
+    """Cap on `InputRequiredResult` retry rounds before `call_tool` / `get_prompt` /
+    `read_resource` give up. Use `client.session.<method>(..., allow_input_required=True)`
+    to drive the loop manually instead."""
+
     _entered: bool = field(init=False, default=False)
     _session: ClientSession | None = field(init=False, default=None)
     _exit_stack: AsyncExitStack | None = field(init=False, default=None)
@@ -356,17 +374,42 @@ class Client:
         """List available resource templates from the server."""
         return await self.session.list_resource_templates(params=PaginatedRequestParams(cursor=cursor, _meta=meta))
 
-    async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> ReadResourceResult:
+    async def read_resource(
+        self,
+        uri: str,
+        *,
+        input_responses: InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+    ) -> ReadResourceResult:
         """Read a resource from the server.
 
+        If the server returns an `InputRequiredResult`, the embedded input
+        requests are dispatched to this client's sampling / elicitation / roots
+        callbacks and the read is retried automatically (up to
+        `input_required_max_rounds`).
+
         Args:
             uri: The URI of the resource to read.
+            input_responses: Responses to seed the first call with (e.g. when
+                resuming from a persisted `InputRequiredResult`).
+            request_state: Opaque state to seed the first call with.
             meta: Additional metadata for the request.
 
         Returns:
             The resource content.
+
+        Raises:
+            InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
+            MCPError: A callback returned `ErrorData` for an embedded input request.
         """
-        return await self.session.read_resource(uri, meta=meta)
+
+        async def retry(r: InputResponses | None, s: str | None) -> ReadResourceResult | InputRequiredResult:
+            return await self.session.read_resource(
+                uri, input_responses=r, request_state=s, meta=meta, allow_input_required=True
+            )
+
+        return await self._drive_input_required(await retry(input_responses, request_state), retry)
 
     async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
         """Subscribe to resource updates."""
@@ -376,7 +419,6 @@ class Client:
         """Unsubscribe from resource updates."""
         return await self.session.unsubscribe_resource(uri, meta=meta)
 
-    @overload
     async def call_tool(
         self,
         name: str,
@@ -387,69 +429,47 @@ class Client:
         input_responses: InputResponses | None = None,
         request_state: str | None = None,
         meta: RequestParamsMeta | None = None,
-        allow_input_required: Literal[False] = False,
+    ) -> CallToolResult:
-    ) -> 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: bool,
-    ) -> 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.
 
+        If the server returns an `InputRequiredResult`, the embedded input
+        requests are dispatched to this client's sampling / elicitation / roots
+        callbacks and the call is retried automatically (up to
+        `input_required_max_rounds`). To drive the loop yourself — e.g. to
+        persist `request_state` across process restarts — use
+        `client.session.call_tool(..., allow_input_required=True)`.
+
         Args:
-            name: The name of the tool to call
+            name: The name of the tool to call.
-            arguments: Arguments to pass to the tool
+            arguments: Arguments to pass to the tool.
-            read_timeout_seconds: Timeout for the tool call
+            read_timeout_seconds: Timeout for each underlying `tools/call` round.
-            progress_callback: Callback for progress updates
+            progress_callback: Callback for progress updates.
-            input_responses: Responses to a prior `InputRequiredResult.input_requests`
+            input_responses: Responses to seed the first call with (e.g. when
-            request_state: Opaque state echoed from a prior `InputRequiredResult`
+                resuming from a persisted `InputRequiredResult`).
-            meta: Additional metadata for the request
+            request_state: Opaque state to seed the first call with.
-            allow_input_required: When ``False`` (default), an `InputRequiredResult`
+            meta: Additional metadata for the request.
-                from the server raises `RuntimeError`; when ``True``, it is returned
-                so the caller can resolve the requests and retry.
 
         Returns:
-            The tool result. When ``allow_input_required=True``, may instead be an
+            The tool result.
-            `InputRequiredResult` carrying the server's input requests and opaque
-            ``request_state`` for the retry.
 
         Raises:
-            RuntimeError: If the server returns an `InputRequiredResult` and
+            InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
-                ``allow_input_required`` is ``False``.
+            MCPError: A callback returned `ErrorData` for an embedded input request.
         """
-        # TODO(L84): stop forwarding allow_input_required; run the MRTR auto-loop driver here (S6).
+
-        return await self.session.call_tool(
+        async def retry(r: InputResponses | None, s: str | None) -> CallToolResult | InputRequiredResult:
-            name=name,
+            return await self.session.call_tool(
-            arguments=arguments,
+                name,
-            read_timeout_seconds=read_timeout_seconds,
+                arguments,
-            progress_callback=progress_callback,
+                read_timeout_seconds=read_timeout_seconds,
-            input_responses=input_responses,
+                progress_callback=progress_callback,
-            request_state=request_state,
+                input_responses=r,
-            meta=meta,
+                request_state=s,
-            allow_input_required=allow_input_required,
+                meta=meta,
-        )
+                allow_input_required=True,
+            )
+
+        return await self._drive_input_required(await retry(input_responses, request_state), retry)
 
     async def list_prompts(
         self,
@@ -461,19 +481,66 @@ class Client:
         return await self.session.list_prompts(params=PaginatedRequestParams(cursor=cursor, _meta=meta))
 
     async def get_prompt(
-        self, name: str, arguments: dict[str, str] | None = None, *, meta: RequestParamsMeta | None = None
+        self,
+        name: str,
+        arguments: dict[str, str] | None = None,
+        *,
+        input_responses: InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
     ) -> GetPromptResult:
         """Get a prompt from the server.
 
+        If the server returns an `InputRequiredResult`, the embedded input
+        requests are dispatched to this client's sampling / elicitation / roots
+        callbacks and the get is retried automatically (up to
+        `input_required_max_rounds`).
+
         Args:
-            name: The name of the prompt
+            name: The name of the prompt.
-            arguments: Arguments to pass to the prompt
+            arguments: Arguments to pass to the prompt.
-            meta: Additional metadata for the request
+            input_responses: Responses to seed the first call with (e.g. when
+                resuming from a persisted `InputRequiredResult`).
+            request_state: Opaque state to seed the first call with.
+            meta: Additional metadata for the request.
 
         Returns:
             The prompt content.
+
+        Raises:
+            InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
+            MCPError: A callback returned `ErrorData` for an embedded input request.
         """
-        return await self.session.get_prompt(name=name, arguments=arguments, meta=meta)
+
+        async def retry(r: InputResponses | None, s: str | None) -> GetPromptResult | InputRequiredResult:
+            return await self.session.get_prompt(
+                name, arguments, input_responses=r, request_state=s, meta=meta, allow_input_required=True
+            )
+
+        return await self._drive_input_required(await retry(input_responses, request_state), retry)
+
+    async def _drive_input_required(
+        self,
+        first: _ResultT | InputRequiredResult,
+        retry: Callable[[InputResponses | None, str | None], Awaitable[_ResultT | InputRequiredResult]],
+    ) -> _ResultT:
+        """Hand an `InputRequiredResult` to the SEP-2322 driver, or pass a terminal result through.
+
+        `dispatch` routes each embedded request through the same callback table
+        that serves legacy server→client RPCs, so the two paths stay
+        behaviourally identical by construction.
+        """
+        if not isinstance(first, InputRequiredResult):
+            return first
+        session = self.session
+
+        async def dispatch(key: str, req: InputRequest) -> InputResponse | ErrorData:
+            ctx = ClientRequestContext(session=session, request_id=key, meta=req.params.meta if req.params else None)
+            return await session._dispatch_input_request(ctx, req)  # pyright: ignore[reportPrivateUsage]
+
+        return await run_input_required_driver(
+            first, dispatch=dispatch, retry=retry, max_rounds=self.input_required_max_rounds
+        )
 
     async def complete(
         self,

src/mcp/client/session.py

@@ -185,6 +185,19 @@ ClientResponse: TypeAdapter[types.ClientResult | types.ErrorData] = TypeAdapter(
 _CallToolResultAdapter: TypeAdapter[types.CallToolResult | types.InputRequiredResult] = TypeAdapter(
     types.CallToolResult | types.InputRequiredResult
 )
+_GetPromptResultAdapter: TypeAdapter[types.GetPromptResult | types.InputRequiredResult] = TypeAdapter(
+    types.GetPromptResult | types.InputRequiredResult
+)
+_ReadResourceResultAdapter: TypeAdapter[types.ReadResourceResult | types.InputRequiredResult] = TypeAdapter(
+    types.ReadResourceResult | types.InputRequiredResult
+)
+
+
+def _input_required_unexpected(method: str) -> RuntimeError:
+    return RuntimeError(
+        "Server returned InputRequiredResult; pass allow_input_required=True to receive it "
+        f"and retry {method}(..., input_responses=..., request_state=result.request_state)."
+    )
 
 
 class ClientSession:
@@ -591,12 +604,64 @@ class ClientSession:
             types.ListResourceTemplatesResult,
         )
 
-    async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.ReadResourceResult:
+    @overload
-        """Send a resources/read request."""
+    async def read_resource(
-        return await self.send_request(
+        self,
-            types.ReadResourceRequest(params=types.ReadResourceRequestParams(uri=uri, _meta=meta)),
+        uri: str,
-            types.ReadResourceResult,
+        *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: Literal[False] = False,
+    ) -> types.ReadResourceResult: ...
+
+    @overload
+    async def read_resource(
+        self,
+        uri: str,
+        *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: bool,
+    ) -> types.ReadResourceResult | types.InputRequiredResult: ...
+
+    async def read_resource(
+        self,
+        uri: str,
+        *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: bool = False,
+    ) -> types.ReadResourceResult | types.InputRequiredResult:
+        """Send a resources/read request.
+
+        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.ReadResourceRequest(
+                params=types.ReadResourceRequestParams(
+                    uri=uri,
+                    input_responses=input_responses,
+                    request_state=request_state,
+                    _meta=meta,
+                ),
+            ),
+            _ReadResourceResultAdapter,
         )
+        if isinstance(result, types.InputRequiredResult) and not allow_input_required:
+            raise _input_required_unexpected("read_resource")
+        return result
 
     async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
         """Send a resources/subscribe request."""
@@ -689,10 +754,7 @@ class ClientSession:
             await self._validate_tool_result(name, result)
 
         if isinstance(result, types.InputRequiredResult) and not allow_input_required:
-            raise RuntimeError(
+            raise _input_required_unexpected("call_tool")
-                "Server returned InputRequiredResult; pass allow_input_required=True to receive it "
-                "and retry call_tool(..., input_responses=..., request_state=result.request_state)."
-            )
         return result
 
     def _resolve_param_headers(self, name: str, arguments: Mapping[str, Any]) -> dict[str, str]:
@@ -734,18 +796,68 @@ class ClientSession:
         """
         return await self.send_request(types.ListPromptsRequest(params=params), types.ListPromptsResult)
 
+    @overload
     async def get_prompt(
         self,
         name: str,
         arguments: dict[str, str] | None = None,
         *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
         meta: RequestParamsMeta | None = None,
-    ) -> types.GetPromptResult:
+        allow_input_required: Literal[False] = False,
-        """Send a prompts/get request."""
+    ) -> types.GetPromptResult: ...
-        return await self.send_request(
+
-            types.GetPromptRequest(params=types.GetPromptRequestParams(name=name, arguments=arguments, _meta=meta)),
+    @overload
-            types.GetPromptResult,
+    async def get_prompt(
+        self,
+        name: str,
+        arguments: dict[str, str] | None = None,
+        *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: bool,
+    ) -> types.GetPromptResult | types.InputRequiredResult: ...
+
+    async def get_prompt(
+        self,
+        name: str,
+        arguments: dict[str, str] | None = None,
+        *,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+        meta: RequestParamsMeta | None = None,
+        allow_input_required: bool = False,
+    ) -> types.GetPromptResult | types.InputRequiredResult:
+        """Send a prompts/get request.
+
+        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.GetPromptRequest(
+                params=types.GetPromptRequestParams(
+                    name=name,
+                    arguments=arguments,
+                    input_responses=input_responses,
+                    request_state=request_state,
+                    _meta=meta,
+                ),
+            ),
+            _GetPromptResultAdapter,
         )
+        if isinstance(result, types.InputRequiredResult) and not allow_input_required:
+            raise _input_required_unexpected("get_prompt")
+        return result
 
     async def complete(
         self,
@@ -829,13 +941,7 @@ class ClientSession:
             ctx = ClientRequestContext(
                 session=self, request_id=dctx.request_id, meta=request.params.meta if request.params else None
             )
-            match request:
+            response = await self._dispatch_input_request(ctx, request)
-                case types.CreateMessageRequest(params=sampling_params):
-                    response = await self._sampling_callback(ctx, sampling_params)
-                case types.ElicitRequest(params=elicit_params):
-                    response = await self._elicitation_callback(ctx, elicit_params)
-                case types.ListRootsRequest():  # pragma: no branch
-                    response = await self._list_roots_callback(ctx)
         client_response = ClientResponse.validate_python(response)
         if isinstance(client_response, types.ErrorData):
             raise MCPError.from_error_data(client_response)
@@ -847,6 +953,23 @@ class ClientSession:
             raise MCPError(code=INTERNAL_ERROR, message="Client callback returned an invalid result") from None
         return dumped
 
+    async def _dispatch_input_request(
+        self, ctx: ClientRequestContext, req: types.InputRequest
+    ) -> types.InputResponse | types.ErrorData:
+        """Route a server-initiated input request to the matching constructor callback.
+
+        Shared by the legacy server→client RPC path (`_on_request`) and the
+        2026-07-28 multi-round-trip driver, which dispatches the embedded
+        `InputRequiredResult.input_requests` through the same callbacks.
+        """
+        match req:
+            case types.CreateMessageRequest(params=p):
+                return await self._sampling_callback(ctx, p)
+            case types.ElicitRequest(params=p):
+                return await self._elicitation_callback(ctx, p)
+            case types.ListRootsRequest():  # pragma: no branch
+                return await self._list_roots_callback(ctx)
+
     async def _on_notify(
         self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
     ) -> None: