examples(mrtr): handler-shape comparison deck (SEP-2322)

Python-SDK counterpart to typescript-sdk#1701. Seven ways to write the
same weather-lookup tool so the diff between files is the argument.

SDK primitives (src/mcp/server/experimental/mrtr.py):
- MrtrCtx.once() — idempotency guard tracked in request_state (Option F)
- ToolBuilder — structural step decomposition; end_step runs exactly once
  regardless of round count (Option G)
- input_response() — sugar for the guard-first pattern
- sse_retry_shim() — Option A comparison artifact (pragma no-cover until
  LATEST_PROTOCOL_VERSION bumps past the MRTR gate)
- dispatch_by_version() — Option D comparison artifact

Option examples (examples/servers/mrtr-options/):
- E (degrade-only): the SDK default. MRTR-native; pre-MRTR gets a default
  or error. Both quadrant rows collapse here.
- A (SSE shim): SDK emulates retry over SSE. Safe re-entry, hidden loop.
- B (await shim): exception-based. UNSAFE — hidden double-execution above
  await. Not a ship target; for contrast.
- C (version branch): explicit if/else in handler body.
- D (dual handler): two functions, SDK picks by version.
- F (ctx.once): idempotency guard, opt-in per side-effect.
- G (ToolBuilder): no above-the-guard zone; end_step structurally
  unreachable until all elicitations complete.

The invariant test (tests/experimental/test_mrtr.py) parametrises E/F/G
against the same Client + callback to prove identical wire behaviour —
the server's internal choice doesn't leak. The footgun test measures
audit_log count to prove F and G actually hold the guard (naive handler
fires twice; F and G fire once).

Both F and G depend on request_state integrity. The demos use plain
base64-JSON; a production SDK MUST HMAC-sign the blob.

Author: maxisbey

examples/servers/mrtr-options/README.md

@@ -0,0 +1,115 @@
+# MRTR handler-shape options (SEP-2322)
+
+Python-SDK counterpart to [typescript-sdk#1701]. Seven ways to write the same
+weather-lookup tool, so the diff between files is the argument.
+
+Unlike the TS demos, the lowlevel plumbing here is **real** — each option is
+an actual `mcp.server.Server` that round-trips `IncompleteResult` through the
+wire protocol. The invariant test at the bottom asserts they all produce
+identical client-observed behaviour.
+
+[typescript-sdk#1701]: https://github.com/modelcontextprotocol/typescript-sdk/pull/1701
+
+## The quadrant
+
+| Server infra                    | Pre-MRTR client                   | MRTR client |
+| ------------------------------- | --------------------------------- | ----------- |
+| Can hold SSE                    | E by default; A/C/D if you opt in | MRTR        |
+| MRTR-only (horizontally scaled) | E by necessity                    | MRTR        |
+
+Both rows *work* for old clients — version negotiation succeeds,
+`tools/list` is complete, tools that don't elicit are unaffected. Only
+elicitation inside a tool is unavailable. Bottom-left isn't "unresolvable";
+it's "E is the only option." Top-left is "E, unless you choose to carry SSE
+infra." The rows collapse for E, which is why it's the SDK default.
+
+## Options
+
+|                                | Author writes                   | SDK does                         | Hidden re-entry | Old client gets                   |
+| ------------------------------ | ------------------------------- | -------------------------------- | --------------- | --------------------------------- |
+| [E](mrtr_options/option_e_degrade.py)        | MRTR-native only                | Nothing                          | No              | Result w/ default, or error       |
+| [A](mrtr_options/option_a_sse_shim.py)       | MRTR-native only                | Retry-loop over SSE              | Yes, safe       | Full elicitation                  |
+| [B](mrtr_options/option_b_await_shim.py)     | `await elicit()`                | Exception → `IncompleteResult`   | **Yes, unsafe** | Full elicitation                  |
+| [C](mrtr_options/option_c_version_branch.py) | One handler, `if version` branch | Version accessor                | No              | Full elicitation                  |
+| [D](mrtr_options/option_d_dual_handler.py)   | Two handlers                    | Picks by version                 | No              | Full elicitation                  |
+| [F](mrtr_options/option_f_ctx_once.py)       | MRTR-native + `ctx.once` wraps  | `once()` guard in request_state  | No              | (same as E)                       |
+| [G](mrtr_options/option_g_tool_builder.py)   | Step functions + `.build()`     | Step-tracking in request_state   | No              | (same as E)                       |
+
+"Hidden re-entry" = the handler function is invoked more than once for a
+single logical tool call, and the author can't tell from the source text.
+
+**A is safe** because MRTR-native code has the re-entry guard (`if not
+prefs: return IncompleteResult(...)`) visible in source even though the
+*loop* is hidden.
+
+**B is unsafe** because `await elicit()` looks like a suspension point but
+is actually a re-entry point on MRTR sessions — see the `audit_log`
+landmine in that file.
+
+## Footgun prevention (F, G)
+
+A–E are about the dual-path axis (old client vs new). F and G address a
+different axis: even in a pure-MRTR world, the naive handler shape has a
+footgun. Code above the `if not prefs` guard runs on every retry. If that
+code is a DB write or HTTP POST, it executes N times for N-round
+elicitation. Nothing *enforces* putting side-effects below the guard —
+safety depends on the developer knowing the convention. The analogy from
+SDK-WG review: the naive MRTR handler is de-facto GOTO.
+
+**F (`MrtrCtx.once`)** keeps the monolithic handler but wraps side-effects
+in an idempotency guard. `ctx.once("audit", lambda: audit_log(...))` checks
+`request_state` — if the key is marked executed, skip. Opt-in: an unwrapped
+mutation still fires twice. The footgun is made *visually distinct*, which
+is reviewable.
+
+**G (`ToolBuilder`)** decomposes the handler into named step functions.
+`incomplete_step` may return `IncompleteResult` or data; `end_step` receives
+everything and runs exactly once. There is no "above the guard" zone because
+there is no guard — the SDK's step-tracking is the guard. Side-effects go in
+`end_step`, structurally unreachable until all elicitations complete.
+
+Both depend on `request_state` integrity. The demos use plain base64-JSON;
+a real SDK MUST HMAC-sign the blob, or the client can forge step-done
+markers and skip the guards. Per-session key derived from `initialize` keeps
+it stateless. Without signing, the safety story is advisory.
+
+## Trade-offs
+
+**E is the SDK default.** A horizontally-scaled server gets E for free —
+it's the only thing that works on that infra. A server that can hold SSE
+also gets E by default, and opts into A/C/D only if serving old-client
+elicitation is worth the extra infra dependency.
+
+**A vs E** is the core tension. Same author-facing code (MRTR-native), the
+only difference is whether old clients get elicitation. A requires shipping
+`sse_retry_shim`; E requires nothing. A also carries a deployment-time
+hazard E doesn't: the shim calls real SSE under the hood, so on MRTR-only
+infra it fails at runtime when an old client connects — a constraint that
+lives nowhere near the tool code.
+
+**B** is zero-migration but breaks silently for anything non-idempotent
+above the await. Not a ship target.
+
+**C vs D** is factoring: one function with a branch vs two functions with a
+dispatcher. Both put the dual-path burden on the tool author.
+
+**F vs G** is the footgun-prevention trade. F is minimal — one line per
+side-effect, composes with any handler shape. G is structural —
+double-execution impossible for `end_step`, but costs two function defs
+per tool. Likely SDK answer: ship F as a primitive on the context, ship G
+as an opt-in builder, recommend G for multi-round tools and F for
+single-question tools.
+
+## The invariant test
+
+`tests/server/experimental/test_mrtr_options.py` parametrises all seven
+servers against the same `Client` + `elicitation_callback`, asserting
+identical output. The footgun test measures `audit_count` to prove F and G
+hold the side-effect to one.
+
+## Not in scope
+
+- Persistent/Tasks workflow — `ServerTaskContext` already does
+  `input_required`; MRTR integration is a separate PR
+- `mrtrOnly` client flag — trivial to add, not demoed
+- requestState HMAC signing — called out in code comments

examples/servers/mrtr-options/mrtr_options/__init__.py

@@ -0,0 +1,7 @@
+"""MRTR handler-shape comparison — seven options on the same weather tool.
+
+See README.md for the trade-off matrix. Every option here is a real lowlevel
+``mcp.server.Server`` that produces identical wire behaviour to each client
+version — the server's internal choice doesn't leak. That's the argument
+against per-feature ``-mrtr`` capability flags.
+"""

examples/servers/mrtr-options/mrtr_options/_shared.py

@@ -0,0 +1,58 @@
+"""Domain logic shared across all options — *not* SDK machinery.
+
+The weather tool: given a location, asks which units, returns a temperature
+string. Same tool throughout so the diff between option files is the
+argument.
+
+``audit_log`` is the side-effect that makes the MRTR footgun concrete: under
+naive re-entry it fires once per round. Options F and G tame it.
+"""
+
+from __future__ import annotations
+
+from mcp import types
+from mcp.server import Server, ServerRequestContext
+
+UNITS_SCHEMA: types.ElicitRequestedSchema = {
+    "type": "object",
+    "properties": {"units": {"type": "string", "enum": ["metric", "imperial"], "title": "Units"}},
+    "required": ["units"],
+}
+
+UNITS_REQUEST = types.ElicitRequest(
+    params=types.ElicitRequestFormParams(message="Which units?", requested_schema=UNITS_SCHEMA)
+)
+
+
+def lookup_weather(location: str, units: str) -> str:
+    temp = "22°C" if units == "metric" else "72°F"
+    return f"Weather in {location}: {temp}, partly cloudy."
+
+
+_audit_count = 0
+
+
+def audit_log(location: str) -> None:
+    """The footgun. Under naive re-entry this fires N times for N-round MRTR."""
+    global _audit_count
+    _audit_count += 1
+    print(f"[audit] lookup requested for {location} (count={_audit_count})")
+
+
+def audit_count() -> int:
+    return _audit_count
+
+
+def reset_audit() -> None:
+    global _audit_count
+    _audit_count = 0
+
+
+async def no_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> types.ListToolsResult:
+    """Minimal tools/list handler so Client validation has something to call."""
+    return types.ListToolsResult(tools=[])
+
+
+def build_server(name: str, on_call_tool: object, **kwargs: object) -> Server:
+    """Consistent Server construction across option files."""
+    return Server(name, on_call_tool=on_call_tool, on_list_tools=no_tools, **kwargs)  # type: ignore[arg-type]

examples/servers/mrtr-options/mrtr_options/option_a_sse_shim.py

@@ -0,0 +1,60 @@
+"""Option A: SDK shim emulates the MRTR retry loop over SSE. Hidden loop.
+
+Tool author writes MRTR-native code only. The SDK wrapper detects the
+negotiated version:
+  - new client → pass ``IncompleteResult`` through, client drives retry
+  - old client → SDK runs the retry loop *locally*, fulfilling each
+    ``InputRequest`` via real SSE (``ctx.session.elicit_form()``),
+    re-invoking the handler until it returns a complete result
+
+Author experience: one code path. Re-entry is explicit in source (the
+``if not prefs`` guard), so the handler is safe to re-invoke by
+construction. But the *fact* that it's re-invoked for old clients is
+invisible — the shim is doing work the author can't see.
+
+What makes this "clunky but possible": the SDK runs a loop on the
+author's behalf. If the handler does something expensive before the
+guard, the author won't find out until an old client connects in prod.
+Works, but it's magic.
+
+Deployment hazard: ``sse_retry_shim`` calls real SSE under the hood.
+On MRTR-only infra it fails at runtime when an old client connects —
+a constraint that lives nowhere near the tool code. If that's the
+deployment, use Option E.
+"""
+
+from __future__ import annotations
+
+from mcp import types
+from mcp.server import ServerRequestContext
+from mcp.server.experimental.mrtr import input_response, sse_retry_shim
+
+from ._shared import UNITS_REQUEST, build_server, lookup_weather
+
+# ───────────────────────────────────────────────────────────────────────────
+# This is what the tool author writes. One function, MRTR-native. No
+# version check, no SSE awareness. The ``if not prefs`` guard IS the
+# re-entry contract; the author sees it, but doesn't see the shim
+# calling this in a loop for old-client sessions.
+# ───────────────────────────────────────────────────────────────────────────
+
+
+async def weather(
+    ctx: ServerRequestContext, params: types.CallToolRequestParams
+) -> types.CallToolResult | types.IncompleteResult:
+    location = (params.arguments or {}).get("location", "?")
+
+    prefs = input_response(params, "units")
+    if prefs is None:
+        return types.IncompleteResult(input_requests={"units": UNITS_REQUEST})
+
+    return types.CallToolResult(content=[types.TextContent(text=lookup_weather(location, prefs["units"]))])
+
+
+# ───────────────────────────────────────────────────────────────────────────
+# Registration applies the shim. In a real SDK this could be a flag on
+# ``add_tool`` or inferred from the handler signature — the author opts in
+# once at registration, not per-call.
+# ───────────────────────────────────────────────────────────────────────────
+
+server = build_server("mrtr-option-a", on_call_tool=sse_retry_shim(weather))

examples/servers/mrtr-options/mrtr_options/option_b_await_shim.py

@@ -0,0 +1,93 @@
+"""Option B: exception-based shim, ``await elicit()`` canonical. The footgun.
+
+Tool author writes today's ``await ctx.elicit(...)`` style. The shim routes:
+  - old client → native SSE, blocks inline (today's behaviour exactly)
+  - new client → ``elicit()`` raises ``NeedsInputSignal``, shim catches,
+    emits ``IncompleteResult``. On retry the handler runs *from the top*
+    and this time ``elicit()`` finds the answer in ``input_responses``.
+
+Author experience: zero migration. Handlers that work today keep working.
+The ``await`` reads linearly.
+
+The problem: the ``await`` is a lie on MRTR sessions. Everything above it
+re-executes on retry. Uncomment the ``audit_log()`` call below — an MRTR
+client triggers *two* audit entries for one tool call. A pre-MRTR client
+triggers one. Same source, different observable behaviour, nothing warns.
+
+Only safe if you can enforce "no side-effects before await" as a lint
+rule, which is hard in practice.
+
+**This is not a ship target — it's a cautionary comparison.**
+"""
+
+from __future__ import annotations
+
+from mcp import types
+from mcp.server import ServerRequestContext
+from mcp.server.experimental.mrtr import input_response
+
+from ._shared import UNITS_REQUEST, UNITS_SCHEMA, build_server, lookup_weather
+
+
+class NeedsInputSignal(Exception):
+    """Control-flow-by-exception. Unwound by the shim, packaged as IncompleteResult."""
+
+    def __init__(self, input_requests: types.InputRequests) -> None:
+        self.input_requests = input_requests
+        super().__init__("NeedsInputSignal (control flow, not an error)")
+
+
+async def elicit_or_signal(
+    ctx: ServerRequestContext, params: types.CallToolRequestParams, key: str
+) -> dict[str, str] | None:
+    """The ``await``-able elicit that looks linear but isn't on MRTR."""
+    version = ctx.session.client_params.protocol_version if ctx.session.client_params else None
+
+    # Old client: native SSE, no trickery.
+    if version is None or str(version) < "2026-06-01":
+        result = await ctx.session.elicit_form(message="Which units?", requested_schema=UNITS_SCHEMA)
+        if result.action != "accept" or not result.content:
+            return None
+        return {k: str(v) for k, v in result.content.items()}
+
+    # New client: check input_responses first.
+    prefs = input_response(params, key)
+    if prefs is not None:
+        return {k: str(v) for k, v in prefs.items()}
+
+    # Not pre-supplied → signal the shim. Everything on the stack unwinds.
+    # On retry the handler re-executes from line one.
+    raise NeedsInputSignal({key: UNITS_REQUEST})
+
+
+# ───────────────────────────────────────────────────────────────────────────
+# This is what the tool author writes. Looks linear. Isn't, on MRTR.
+# ───────────────────────────────────────────────────────────────────────────
+
+
+async def _weather_inner(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
+    location = (params.arguments or {}).get("location", "?")
+
+    # audit_log(location)
+    #   ^^^^^^^^^^^^^^^^^^
+    #   On pre-MRTR: runs once. On MRTR: runs once on the initial call,
+    #   once more on the retry. The await below isn't a suspension point
+    #   on MRTR — it's a re-entry point. Nothing in this syntax says so.
+
+    prefs = await elicit_or_signal(ctx, params, "units")
+    if not prefs:
+        return types.CallToolResult(content=[types.TextContent(text="Cancelled.")])
+
+    return types.CallToolResult(content=[types.TextContent(text=lookup_weather(location, prefs["units"]))])
+
+
+async def weather(
+    ctx: ServerRequestContext, params: types.CallToolRequestParams
+) -> types.CallToolResult | types.IncompleteResult:
+    try:
+        return await _weather_inner(ctx, params)
+    except NeedsInputSignal as signal:
+        return types.IncompleteResult(input_requests=signal.input_requests)
+
+
+server = build_server("mrtr-option-b", on_call_tool=weather)

examples/servers/mrtr-options/mrtr_options/option_c_version_branch.py

@@ -0,0 +1,52 @@
+"""Option C: explicit version branch in the handler body.
+
+No shim. Tool author checks the negotiated version themselves and writes
+both code paths inline. The SDK provides nothing except the version
+accessor and the raw primitives for each path.
+
+Author experience: everything is visible. Both protocol behaviours are
+right there in source, separated by an ``if``. No hidden re-entry, no
+magic wrappers. A reader traces exactly what happens for each client
+version.
+
+The cost is also visible: the elicitation schema is duplicated, the
+cancel-handling is duplicated, and there's a conditional at the top of
+every handler that uses elicitation. For one tool, fine. For twenty,
+it's twenty copies of the same branch.
+"""
+
+from __future__ import annotations
+
+from mcp import types
+from mcp.server import ServerRequestContext
+from mcp.server.experimental.mrtr import input_response
+
+from ._shared import UNITS_REQUEST, UNITS_SCHEMA, build_server, lookup_weather
+
+
+async def weather(
+    ctx: ServerRequestContext, params: types.CallToolRequestParams
+) -> types.CallToolResult | types.IncompleteResult:
+    location = (params.arguments or {}).get("location", "?")
+    version = ctx.session.client_params.protocol_version if ctx.session.client_params else None
+
+    # ───────────────────────────────────────────────────────────────────────
+    # The branch is the whole story.
+    # ───────────────────────────────────────────────────────────────────────
+
+    if version is not None and str(version) >= "2026-06-01":
+        # MRTR path: check input_responses, return IncompleteResult if missing.
+        prefs = input_response(params, "units")
+        if prefs is None:
+            return types.IncompleteResult(input_requests={"units": UNITS_REQUEST})
+        return types.CallToolResult(content=[types.TextContent(text=lookup_weather(location, prefs["units"]))])
+
+    # SSE path: inline await, blocks on the response stream.
+    result = await ctx.session.elicit_form(message="Which units?", requested_schema=UNITS_SCHEMA)
+    if result.action != "accept" or not result.content:
+        return types.CallToolResult(content=[types.TextContent(text="Cancelled.")])
+    units = str(result.content.get("units", "metric"))
+    return types.CallToolResult(content=[types.TextContent(text=lookup_weather(location, units))])
+
+
+server = build_server("mrtr-option-c", on_call_tool=weather)