Address review: durable re-entry, resource guards, deterministic test

Keep the final outstanding input marker until the next task leg is durable,
so a racing tasks/get cannot read a parked leg as complete. Let resources and
resource templates return InputRequiredResult like tools and prompts. Identify
parked requests by their question rather than sort order.

fastmcp_slim/fastmcp/resources/base.py

@@ -210,6 +210,35 @@ class ResourceResult(pydantic.BaseModel):
         )
 
 
+class InputRequiredResourceResult(ResourceResult):
+    """The full result of a single multi-round-trip resource read (SEP-2322).
+
+    `InputRequiredResult` is a result type, not a `tools/call` feature: any
+    request may resolve to one. When a resource or resource template returns an
+    `InputRequiredResult` from its body to ask the client for input, that ask is
+    the legitimate result of this `resources/read` — so FastMCP wraps it in this
+    `ResourceResult` subclass, mirroring `InputRequiredToolResult` and
+    `InputRequiredPromptResult`, and it flows through the middleware chain as an
+    ordinary return value.
+
+    Invariant: the wrapped `InputRequiredResult` is never serialized as resource
+    contents. `contents` is always empty; the wire handler (`_on_read_resource`)
+    reads `.input_required` and returns it to the runner.
+    """
+
+    input_required: mcp_types.InputRequiredResult = pydantic.Field(
+        description="The client-input request this read resolved to (SEP-2322)"
+    )
+
+    def __init__(self, input_required: mcp_types.InputRequiredResult) -> None:
+        # Bypass ResourceResult's content-normalizing __init__: an
+        # input-required read carries no contents (see the invariant above), and
+        # `input_required` is a required field ResourceResult.__init__ can't set.
+        pydantic.BaseModel.__init__(
+            self, contents=[], meta=None, input_required=input_required
+        )
+
+
 def _public_content_meta(meta: dict[str, Any] | None) -> dict[str, Any] | None:
     """Strip FastMCP's internal bookkeeping out of component meta.
 
@@ -259,6 +288,12 @@ def convert_raw_to_resource_result(
     if isinstance(raw_value, ResourceResult):
         return raw_value
 
+    if isinstance(raw_value, mcp_types.InputRequiredResult):
+        # The resource asked the client for input (SEP-2322). Wrap it so the
+        # ask travels the middleware chain as an ordinary result; the wire
+        # handler unwraps it.
+        return InputRequiredResourceResult(raw_value)
+
     meta = _public_content_meta(meta)
 
     # For plain str/bytes returns, wrap in ResourceContent with the

fastmcp_slim/fastmcp/server/mixins/mcp_operations.py

@@ -29,6 +29,7 @@ from fastmcp.exceptions import (
     to_mcp_error,
 )
 from fastmcp.prompts.base import InputRequiredPromptResult
+from fastmcp.resources.base import InputRequiredResourceResult
 from fastmcp.server.completions import CompletionValues, normalize_completion
 from fastmcp.server.dependencies import bind_request_context, extract_version_spec
 from fastmcp.tools.base import InputRequiredToolResult, ToolResult
@@ -296,7 +297,7 @@ class MCPOperationsMixin:
         self: FastMCP,
         ctx: ServerRequestContext,
         params: ReadResourceRequestParams,
-    ) -> mcp_types.ReadResourceResult:
+    ) -> mcp_types.ReadResourceResult | mcp_types.InputRequiredResult:
         """Handle MCP 'resources/read' requests."""
         with bind_request_context(ctx):
             uri = params.uri
@@ -325,6 +326,23 @@ class MCPOperationsMixin:
                 # already happened inside read_resource.
                 raise to_mcp_error(e) from e
 
+            if isinstance(result, InputRequiredResourceResult):
+                # The resource requested client input (SEP-2322). As with tools
+                # and prompts, the multi-round-trip result type only exists at
+                # 2026-07-28, so name the era problem on an older connection
+                # rather than failing as a generic "invalid result".
+                if ctx.protocol_version not in MODERN_PROTOCOL_VERSIONS:
+                    raise MCPError(
+                        code=INVALID_PARAMS,
+                        message=(
+                            f"Resource {str(uri)!r} returned an InputRequiredResult "
+                            "to request client input, but the multi-round-trip "
+                            "result type (SEP-2322) only exists at MCP 2026-07-28; "
+                            f"this connection negotiated {ctx.protocol_version!r}."
+                        ),
+                    )
+                return result.input_required
+
             return result.to_mcp_result(uri)
 
     async def _on_get_prompt(

fastmcp_tasks/fastmcp_tasks/handlers.py

@@ -374,17 +374,25 @@ async def tasks_update(
         # may answer a multi-request ask one update at a time.
         await store_input_responses(docket, task_scope, task_id, translated)
 
-        # Retire only what this update answered. While anything is still
-        # outstanding the task stays `input_required` and `tasks/get` surfaces
-        # the remaining keys — the leg re-enters only once every request has an
-        # answer (SEP-2663 partial fulfillment).
-        await discard_outstanding(
-            docket, task_scope, task_id, leg_number, answered_keys
-        )
-        still_pending = await read_outstanding_inputs(
+        # A partial update retires only the keys it answered, leaving the task
+        # `input_required` with the rest surfaced; the leg re-enters only once
+        # every request has an answer (SEP-2663 partial fulfillment).
+        #
+        # The *last* answer deliberately leaves its marker in place. Outstanding
+        # requests are what make a completed-but-parked leg read as
+        # `input_required` rather than as a finished task, so retiring the final
+        # one before the next leg is durable would let a racing `tasks/get`
+        # report the task complete with a `None` result — and would strand it
+        # there for good if the enqueue below failed, since a retried update
+        # would no longer match any key. `clear_outstanding` runs after the
+        # pointer swap instead.
+        outstanding = await read_outstanding_inputs(
             docket, task_scope, task_id, leg_number
         )
-        if still_pending:
+        if set(outstanding) - set(answered_keys):
+            await discard_outstanding(
+                docket, task_scope, task_id, leg_number, answered_keys
+            )
             return UpdateTaskResult()
 
         # Every request is answered, so enqueue the next leg. Ordering matters:

tests/server/test_mrtr_guards.py

@@ -1273,3 +1273,75 @@ class TestPromptGuard:
         async with Client(self._context_prompt_server(), mode="legacy") as client:
             with pytest.raises(MCPError, match="2026-07-28"):
                 await client.session.get_prompt("summarize")
+
+
+class TestResourceGuard:
+    """Resources and templates ask for input the same way tools and prompts do."""
+
+    @staticmethod
+    def _resource_server() -> FastMCP:
+        mcp = FastMCP("resource-guard")
+
+        @mcp.resource("data://report")
+        async def report(ctx: Context) -> str | InputRequiredResult:
+            responses = ctx.input_responses
+            if responses is None:
+                return _ask(
+                    _elicit("context", "Which quarter?", "context"),
+                    key="context",
+                    request_state=None,
+                )
+            return f"Report for {_accepted(responses, 'context')['context']}"
+
+        @mcp.resource("data://report/{section}")
+        async def section_report(
+            section: str, ctx: Context
+        ) -> str | InputRequiredResult:
+            responses = ctx.input_responses
+            if responses is None:
+                return _ask(
+                    _elicit("context", f"Which quarter for {section}?", "context"),
+                    key="context",
+                    request_state=None,
+                )
+            quarter = _accepted(responses, "context")["context"]
+            return f"{section} for {quarter}"
+
+        return mcp
+
+    async def test_resource_emits_input_required(self):
+        async with Client(self._resource_server(), mode="auto") as client:
+            result = await client.session.read_resource(
+                "data://report", allow_input_required=True
+            )
+
+        assert isinstance(result, InputRequiredResult)
+        assert "context" in result.input_requests
+
+    async def test_resource_completes_with_responses(self):
+        async with Client(self._resource_server(), mode="auto") as client:
+            done = await client.session.read_resource(
+                "data://report",
+                input_responses={
+                    "context": mcp_types.ElicitResult(
+                        action="accept", content={"context": "Q3"}
+                    )
+                },
+            )
+
+        assert done.contents[0].text == "Report for Q3"
+
+    async def test_resource_template_emits_input_required(self):
+        """Templates share the converter, so the ask survives there too."""
+        async with Client(self._resource_server(), mode="auto") as client:
+            result = await client.session.read_resource(
+                "data://report/revenue", allow_input_required=True
+            )
+
+        assert isinstance(result, InputRequiredResult)
+        assert "context" in result.input_requests
+
+    async def test_resource_guard_rejected_on_handshake_era(self):
+        async with Client(self._resource_server(), mode="legacy") as client:
+            with pytest.raises(MCPError, match="2026-07-28"):
+                await client.session.read_resource("data://report")

tests/tasks/server/test_guard_reentrant.py

@@ -60,6 +60,14 @@ def _input_required(
     )
 
 
+def _key_asking(input_requests: dict[str, Any], message: str) -> str:
+    """The surfaced key whose parked request asks *message*."""
+    for key, payload in input_requests.items():
+        if payload["params"]["message"] == message:
+            return key
+    raise AssertionError(f"no parked request asks {message!r}")
+
+
 async def _park_key(mcp: FastMCP, task_id: str) -> str:
     parked = await wait_for_task(
         mcp, task_id, target_states=frozenset({"input_required"})
@@ -273,10 +281,12 @@ async def test_partial_update_keeps_task_parked_on_remaining_request():
             mcp, created.task_id, target_states=frozenset({"input_required"})
         )
         assert parked.input_requests is not None
-        keys = sorted(parked.input_requests)
-        assert len(keys) == 2
+        assert len(parked.input_requests) == 2
 
-        answered, pending = keys[0], keys[1]
+        # Surfaced keys are freshly minted per request, so they carry no order
+        # a test can rely on. Identify each by the question it asks.
+        answered = _key_asking(parked.input_requests, "First?")
+        pending = _key_asking(parked.input_requests, "Second?")
         await update_task(
             mcp,
             created.task_id,
@@ -301,6 +311,41 @@ async def test_partial_update_keeps_task_parked_on_remaining_request():
     assert final.result["content"][0]["text"] == "one+two"
 
 
+async def test_final_answer_keeps_task_parked_until_next_leg_is_durable():
+    """The last answer must not retire its outstanding marker early.
+
+    Outstanding requests are what make a completed-but-parked leg read as
+    `input_required`. Discarding the final one before the next leg is enqueued
+    would let a `tasks/get` landing in that window see a finished execution with
+    no result and report the task complete.
+    """
+    mcp = FastMCP("durable-reentry")
+    mcp.add_extension(TasksExtension())
+
+    @mcp.tool(task=True)
+    async def one_question(ctx: Context) -> str | mcp_types.InputRequiredResult:
+        responses = ctx.input_responses
+        if responses is None:
+            return _input_required({"only": _elicit_request("Only?")})
+        return f"got {_answer(responses, 'only')}"
+
+    async with running_task_server(mcp):
+        created = await submit_task(mcp, "one_question", {})
+        key = await _park_key(mcp, created.task_id)
+
+        await update_task(
+            mcp,
+            created.task_id,
+            {key: {"action": "accept", "content": {"value": "answer"}}},
+        )
+        final = await wait_for_task(mcp, created.task_id)
+
+    # The task must land on the real result, never on a phantom completion.
+    assert final.status == "completed"
+    assert final.result is not None
+    assert final.result["content"][0]["text"] == "got answer"
+
+
 async def test_protocol_error_fails_the_task_with_inlined_error():
     """SEP-2663 reserves `failed` for protocol faults: an `MCPError` raised by
     the body is inlined as a JSON-RPC error rather than reported as a completed