Max Isbey
c54fe3b36a
The Option B footgun was: await elicit() looks like a suspension point but
is actually a re-entry point, so everything above it runs twice. Option H
fixes that by making it a REAL suspension point — the coroutine frame is
held in a ContinuationStore across MRTR rounds, keyed by request_state.
Handler code stays exactly as it was in the SSE era:
async def my_tool(ctx: LinearCtx, location: str) -> str:
audit_log(location) # runs exactly once
units = await ctx.elicit("Which units?", UnitsSchema)
return f"{location}: 22°{units.u}"
The wrapper linear_mrtr(my_tool, store=...) translates this into a standard
MRTR on_call_tool handler. Round 1 starts the coroutine; elicit() sends
IncompleteResult back through the wrapper and parks on a stream. Round 2's
retry wakes it with the answer. The coroutine continues from where it
stopped — no re-entry, no double-execution.
Trade-off: server holds the frame in memory between rounds. Client sees
pure MRTR (no SSE, independent requests), but server is stateful within
a single tool call. Horizontally-scaled deployments need sticky routing on
the request_state token. Same operational shape as Option A's SSE hold,
without the long-lived connection.
SDK pieces (src/mcp/server/experimental/mrtr/linear.py):
- LinearCtx with async elicit(message, PydanticSchema) -> instance
- ContinuationStore — owns the task group, TTL-based frame expiry
- linear_mrtr(handler, store=...) — the wrapper
- ElicitDeclined raised when user declines/cancels
7 E2E tests including the key assertion: side-effects above await fire
exactly once (the test measures audit_log count).
Python SDK Examples
This folders aims to provide simple examples of using the Python SDK. Please refer to the servers repository for real-world servers.