feat(mrtr): add IncompleteResult types and client retry loop (SEP-2322)

Lowlevel plumbing for Multi Round-Trip Requests:

Types:
- IncompleteResult with result_type discriminator, input_requests, request_state
- InputRequest/InputResponse unions (elicitation, sampling, roots)
- input_responses + request_state fields on RequestParams

Server (lowlevel):
- on_call_tool return widened to include IncompleteResult

Session:
- send_request accepts TypeAdapter (overload) for union result parsing
- call_tool_mrtr() returns CallToolResult | IncompleteResult
- call_tool() stays narrow, raises on IncompleteResult with migration hint

Client:
- call_tool() drives MRTR retry loop internally — dispatches embedded
  input requests to elicitation/sampling/list_roots callbacks, retries
  with collected responses + echoed request_state
- max_mrtr_rounds bound (default 8)

The client-side delta from today's code is zero: elicitation_callback is the
same function whether it fires from SSE push or MRTR retry.

Author: maxisbey

src/mcp/client/client.py

@@ -12,17 +12,24 @@
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
+from mcp.shared._context import RequestContext
 from mcp.shared.session import ProgressFnT
 from mcp.types import (
     CallToolResult,
     CompleteResult,
+    CreateMessageRequest,
+    ElicitRequest,
     EmptyResult,
+    ErrorData,
     GetPromptResult,
     Implementation,
+    IncompleteResult,
     InitializeResult,
+    InputResponses,
     ListPromptsResult,
     ListResourcesResult,
     ListResourceTemplatesResult,
+    ListRootsRequest,
     ListToolsResult,
     LoggingLevel,
     PaginatedRequestParams,
@@ -32,6 +39,9 @@
     ResourceTemplateReference,
 )
 
+MRTR_MAX_ROUNDS = 8
+"""Bound on MRTR retry rounds. A well-formed handler converges; an unbounded loop is a bug."""
+
 
 @dataclass
 class Client:
@@ -95,6 +105,9 @@ async def main():
     elicitation_callback: ElicitationFnT | None = None
     """Callback for handling elicitation requests."""
 
+    max_mrtr_rounds: int = MRTR_MAX_ROUNDS
+    """Maximum MRTR retry rounds before raising (SEP-2322). A server that never converges is a bug."""
+
     _session: ClientSession | None = field(init=False, default=None)
     _exit_stack: AsyncExitStack | None = field(init=False, default=None)
     _transport: Transport = field(init=False)
@@ -238,6 +251,12 @@ async def call_tool(
     ) -> CallToolResult:
         """Call a tool on the server.
 
+        If the server returns an ``IncompleteResult`` (SEP-2322 MRTR), this
+        method drives the retry loop internally: each embedded input request
+        is dispatched to the matching callback (``elicitation_callback``,
+        ``sampling_callback``, or ``list_roots_callback``) and the tool is
+        re-called with the collected responses plus echoed ``request_state``.
+
         Args:
             name: The name of the tool to call
             arguments: Arguments to pass to the tool
@@ -248,14 +267,65 @@ async def call_tool(
         Returns:
             The tool result.
         """
-        return await self.session.call_tool(
-            name=name,
-            arguments=arguments,
-            read_timeout_seconds=read_timeout_seconds,
-            progress_callback=progress_callback,
-            meta=meta,
+        input_responses: InputResponses | None = None
+        request_state: str | None = None
+
+        for _round in range(self.max_mrtr_rounds):
+            result = await self.session.call_tool_mrtr(
+                name=name,
+                arguments=arguments,
+                read_timeout_seconds=read_timeout_seconds,
+                progress_callback=progress_callback,
+                meta=meta,
+                input_responses=input_responses,
+                request_state=request_state,
+            )
+
+            if isinstance(result, CallToolResult):
+                return result
+
+            input_responses = await self._fulfil_input_requests(result)
+            request_state = result.request_state
+
+        raise RuntimeError(
+            f"MRTR retry loop for tool {name!r} exceeded {self.max_mrtr_rounds} rounds without converging"
         )
 
+    async def _fulfil_input_requests(self, incomplete: IncompleteResult) -> InputResponses | None:
+        """Dispatch each embedded input request to the matching callback."""
+        if not incomplete.input_requests:
+            return None
+
+        ctx = RequestContext[ClientSession](session=self.session)
+        responses: InputResponses = {}
+
+        for key, req in incomplete.input_requests.items():
+            match req:
+                case ElicitRequest(params=params):
+                    if self.elicitation_callback is None:
+                        raise RuntimeError(
+                            f"Server sent elicitation input request {key!r} but no elicitation_callback is configured"
+                        )
+                    result = await self.elicitation_callback(ctx, params)
+                case CreateMessageRequest(params=params):
+                    if self.sampling_callback is None:  # pragma: no cover
+                        raise RuntimeError(
+                            f"Server sent sampling input request {key!r} but no sampling_callback is configured"
+                        )
+                    result = await self.sampling_callback(ctx, params)
+                case ListRootsRequest():
+                    if self.list_roots_callback is None:  # pragma: no cover
+                        raise RuntimeError(
+                            f"Server sent roots input request {key!r} but no list_roots_callback is configured"
+                        )
+                    result = await self.list_roots_callback(ctx)
+
+            if isinstance(result, ErrorData):
+                raise RuntimeError(f"Input request {key!r} failed: {result.message}")
+            responses[key] = result
+
+        return responses
+
     async def list_prompts(
         self,
         *,

src/mcp/client/session.py

@@ -97,6 +97,10 @@ async def _default_logging_callback(
 
 ClientResponse: TypeAdapter[types.ClientResult | types.ErrorData] = TypeAdapter(types.ClientResult | types.ErrorData)
 
+_call_tool_result_adapter: TypeAdapter[types.IncompleteResult | types.CallToolResult] = TypeAdapter(
+    types.IncompleteResult | types.CallToolResult
+)
+
 
 class ClientSession(
     BaseSession[
@@ -305,18 +309,60 @@ async def call_tool(
         *,
         meta: RequestParamsMeta | None = None,
     ) -> types.CallToolResult:
-        """Send a tools/call request with optional progress callback support."""
+        """Send a tools/call request with optional progress callback support.
 
-        result = await self.send_request(
+        Raises:
+            RuntimeError: If the server returns an IncompleteResult. Use
+                ``Client.call_tool`` or ``call_tool_mrtr`` to handle MRTR flows.
+        """
+        result = await self.call_tool_mrtr(
+            name,
+            arguments,
+            read_timeout_seconds,
+            progress_callback,
+            meta=meta,
+        )
+        if isinstance(result, types.IncompleteResult):
+            raise RuntimeError(
+                f"Server returned IncompleteResult for tool {name!r}. "
+                "Use Client.call_tool or ClientSession.call_tool_mrtr to handle MRTR flows."
+            )
+        return result
+
+    async def call_tool_mrtr(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        read_timeout_seconds: float | None = None,
+        progress_callback: ProgressFnT | None = None,
+        *,
+        meta: RequestParamsMeta | None = None,
+        input_responses: types.InputResponses | None = None,
+        request_state: str | None = None,
+    ) -> types.CallToolResult | types.IncompleteResult:
+        """Send a single tools/call request; returns IncompleteResult if server needs input.
+
+        This is the MRTR-aware variant (SEP-2322). One request → one response.
+        Higher-level ``mcp.client.Client.call_tool`` drives the retry loop; this
+        method just surfaces whatever the server sent.
+        """
+
+        result: types.CallToolResult | types.IncompleteResult = 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,
+            _call_tool_result_adapter,
             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

src/mcp/server/lowlevel/server.py

@@ -118,7 +118,7 @@ def __init__(
         | None = None,
         on_call_tool: Callable[
             [ServerRequestContext[LifespanResultT], types.CallToolRequestParams],
-            Awaitable[types.CallToolResult | types.CreateTaskResult],
+            Awaitable[types.CallToolResult | types.IncompleteResult | types.CreateTaskResult],
         ]
         | None = None,
         on_list_resources: Callable[

src/mcp/shared/session.py

@@ -4,7 +4,7 @@
 from collections.abc import Callable
 from contextlib import AsyncExitStack
 from types import TracebackType
-from typing import Any, Generic, Protocol, TypeVar
+from typing import Any, Generic, Protocol, TypeVar, overload
 
 import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
@@ -230,14 +230,34 @@ async def __aexit__(
         self._task_group.cancel_scope.cancel()
         return await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
 
+    @overload
     async def send_request(
         self,
         request: SendRequestT,
         result_type: type[ReceiveResultT],
         request_read_timeout_seconds: float | None = None,
         metadata: MessageMetadata = None,
         progress_callback: ProgressFnT | None = None,
-    ) -> ReceiveResultT:
+    ) -> ReceiveResultT: ...
+
+    @overload
+    async def send_request(
+        self,
+        request: SendRequestT,
+        result_type: TypeAdapter[Any],
+        request_read_timeout_seconds: float | None = None,
+        metadata: MessageMetadata = None,
+        progress_callback: ProgressFnT | None = None,
+    ) -> Any: ...
+
+    async def send_request(
+        self,
+        request: SendRequestT,
+        result_type: type[ReceiveResultT] | TypeAdapter[Any],
+        request_read_timeout_seconds: float | None = None,
+        metadata: MessageMetadata = None,
+        progress_callback: ProgressFnT | None = None,
+    ) -> ReceiveResultT | Any:
         """Sends a request and waits for a response.
 
         Raises an MCPError if the response contains an error. If a request read timeout is provided, it will take
@@ -280,6 +300,8 @@ async def send_request(
 
             if isinstance(response_or_error, JSONRPCError):
                 raise MCPError.from_jsonrpc_error(response_or_error)
+            elif isinstance(result_type, TypeAdapter):
+                return result_type.validate_python(response_or_error.result)
             else:
                 return result_type.model_validate(response_or_error.result, by_name=False)
 

src/mcp/types/__init__.py

@@ -74,10 +74,15 @@
     ImageContent,
     Implementation,
     IncludeContext,
+    IncompleteResult,
     InitializedNotification,
     InitializeRequest,
     InitializeRequestParams,
     InitializeResult,
+    InputRequest,
+    InputRequests,
+    InputResponse,
+    InputResponses,
     ListPromptsRequest,
     ListPromptsResult,
     ListResourcesRequest,
@@ -179,6 +184,7 @@
     client_notification_adapter,
     client_request_adapter,
     client_result_adapter,
+    input_request_adapter,
     server_notification_adapter,
     server_request_adapter,
     server_result_adapter,
@@ -342,6 +348,13 @@
     "SubscribeRequestParams",
     "UnsubscribeRequest",
     "UnsubscribeRequestParams",
+    # MRTR (SEP-2322)
+    "IncompleteResult",
+    "InputRequest",
+    "InputRequests",
+    "InputResponse",
+    "InputResponses",
+    "input_request_adapter",
     # Results
     "CallToolResult",
     "CancelTaskResult",

src/mcp/types/_types.py

@@ -76,6 +76,19 @@ class RequestParams(MCPModel):
     for task augmentation of specific request types in their capabilities.
     """
 
+    input_responses: dict[str, Any] | None = None
+    """Responses to input requests from a prior IncompleteResult (SEP-2322 MRTR).
+
+    Keys mirror the server's inputRequests keys; values are the corresponding
+    ElicitResult, CreateMessageResult, or ListRootsResult payloads.
+    """
+
+    request_state: str | None = None
+    """Opaque state echoed from a prior IncompleteResult (SEP-2322 MRTR).
+
+    Clients MUST NOT inspect or modify this value.
+    """
+
     meta: RequestParamsMeta | None = Field(alias="_meta", default=None)
 
 
@@ -1716,6 +1729,44 @@ class ElicitationRequiredErrorData(MCPModel):
     """List of URL mode elicitations that must be completed."""
 
 
+# ─── Multi Round-Trip Requests (SEP-2322) ───────────────────────────────────
+
+InputRequest: TypeAlias = CreateMessageRequest | ElicitRequest | ListRootsRequest
+"""A server-initiated request embedded in an IncompleteResult."""
+
+InputResponse: TypeAlias = CreateMessageResult | CreateMessageResultWithTools | ElicitResult | ListRootsResult
+"""A client's response to an InputRequest, sent on the retry."""
+
+InputRequests: TypeAlias = dict[str, InputRequest]
+"""Keyed map of input requests. Keys are server-assigned and opaque to the client."""
+
+InputResponses: TypeAlias = dict[str, InputResponse]
+"""Keyed map of input responses. Keys mirror the server's InputRequests keys."""
+
+
+class IncompleteResult(Result):
+    """A result indicating the server needs more input before completing (SEP-2322).
+
+    The client MUST retry the original request with ``input_responses`` populated
+    for each key in ``input_requests``, and ``request_state`` echoed verbatim.
+
+    At least one of ``input_requests`` or ``request_state`` must be present.
+    """
+
+    result_type: Literal["incomplete"] = "incomplete"
+    """Discriminator marking this as an incomplete result."""
+
+    input_requests: InputRequests | None = None
+    """Server-initiated requests the client must fulfil before retrying."""
+
+    request_state: str | None = None
+    """Opaque state the client must echo back on retry. Not inspected by the client."""
+
+
+input_request_adapter: TypeAdapter[InputRequest] = TypeAdapter(InputRequest)
+"""Type adapter for validating embedded InputRequest payloads."""
+
+
 ClientResult = (
     EmptyResult
     | CreateMessageResult
@@ -1774,5 +1825,6 @@ class ElicitationRequiredErrorData(MCPModel):
     | ListTasksResult
     | CancelTaskResult
     | CreateTaskResult
+    | IncompleteResult
 )
 server_result_adapter = TypeAdapter[ServerResult](ServerResult)