ClientSession receive loop swallows callback exceptions, replying "Invalid request parameters" to the server instead of propagating to the call_tool awaiter

Initial Checks

• I confirm that I'm using the latest version of MCP Python SDK
• I confirm that I searched for my issue in https://github.com/modelcontextprotocol/python-sdk/issues before opening this issue

Description

I hit this while wiring an MCP tool that uses ctx.elicit(...) into a
LangGraph agent — the natural pattern of calling interrupt() from the
client elicitation callback does not work because the callback's exception
is swallowed by _receive_loop and replied to the server as
Invalid request parameters. The same limitation blocks any framework that
relies on exception-based control flow on the awaiter's task (custom
async-cancellation strategies, Trio nursery patterns, retry/backoff
libraries, any human-in-the-loop integration), so a small SDK-level escape
hatch would unblock a broad class of MCP client developers, not just this
one use case.

Any exception raised inside a ClientSession callback
(elicitation_callback, sampling_callback, list_roots_callback) is
caught by the blanket except Exception in BaseSession._receive_loop
(src/mcp/shared/session.py) and turned into a JSON-RPC
Invalid request parameters reply to the server:

except Exception:
    logging.warning("Failed to validate request", exc_info=True)
    error_response = JSONRPCError(
        jsonrpc="2.0",
        id=message.message.id,
        error=ErrorData(code=INVALID_PARAMS, message="Invalid request parameters", data=""),
    )
    await self._write_stream.send(SessionMessage(message=error_response))

Consequences:

1. The server is blamed for "invalid params" when the failure was on the
   client. The original callback traceback is lost from the awaiter of
   session.call_tool(...).
2. Callbacks have no way to abort back to the awaiter. Frameworks that use
   exception-based flow control on the caller's task (LangGraph's
   interrupt() for human-in-the-loop pause/resume is the motivating
   case) cannot be wired through MCP without per-framework workarounds.

Expected: an opt-in escape hatch so a callback can raise an exception
that propagates out of the in-flight session.call_tool(...) instead of
becoming a JSON-RPC error to the server. Defaulting opt-in to off preserves
existing behaviour.

Proposed minimal fix (single file, ~30 lines, three coordinated edits):

# src/mcp/shared/session.py
#
# (1) BaseSession.__init__: add a per-session stash for marked exceptions.
self._propagate_errors: dict[RequestId, BaseException] = {}

# (2) Inside _receive_loop's request-handler `except Exception as e:` block,
#     branch on the marker before the existing INVALID_PARAMS path.
if getattr(e, "__mcp_propagate__", False):
    # Notify the peer so their request doesn't hang.
    error_response = JSONRPCError(
        jsonrpc="2.0",
        id=message.message.root.id,
        error=ErrorData(code=INTERNAL_ERROR, message="Handler raised", data=""),
    )
    await self._write_stream.send(SessionMessage(message=JSONRPCMessage(error_response)))
    # Surface to the awaiter of any in-flight outgoing request on this session.
    for in_flight_id, stream in list(self._response_streams.items()):
        self._propagate_errors[in_flight_id] = e
        await stream.aclose()
    continue
# ...existing INVALID_PARAMS path unchanged below this point.

# (3) Inside send_request, alongside the existing `except TimeoutError:`,
#     consume the stash on EndOfStream and re-raise the original exception.
except anyio.EndOfStream:
    propagate = self._propagate_errors.pop(request_id, None)
    if propagate is not None:
        raise propagate from None
    raise

I've already implemented and tested this locally against v1.x (and the
same shape on main for the V2 rework): ~33 lines added to
src/mcp/shared/session.py plus regression tests in
tests/shared/test_session.py. The patch is strictly additive — without the
__mcp_propagate__ marker, behaviour is byte-identical to today. End-to-end
verified by surfacing a LangGraph interrupt() from an elicitation callback
as a normal __interrupt__ on the agent's first invoke. ruff / pyright
clean, existing session tests still pass. Happy to open a PR once the issue
is accepted and you've confirmed the direction.

Example Code

# Self-contained reproduction. No third-party agent framework needed.
#
# Terminal A:  python repro_server.py
# Terminal B:  python repro_client.py
#
# Observed in A: WARNING:root:Failed to validate request: CallbackBoom(...)
# Observed in B: tool result.isError = True, CallbackBoom was NEVER raised
#                on the awaiter — silently converted to INVALID_PARAMS.

# --- repro_server.py ---
from mcp.server.fastmcp import Context, FastMCP
from pydantic import BaseModel

class Answer(BaseModel):
    value: str

server = FastMCP(name="repro", port=8765)

@server.tool()
async def ask(ctx: Context) -> str:
    result = await ctx.elicit(message="hello?", schema=Answer)
    return f"got: {result}"

if __name__ == "__main__":
    server.run(transport="streamable-http")


# --- repro_client.py ---
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

class CallbackBoom(Exception):
    """Stand-in for any framework's flow-control exception (e.g. GraphInterrupt)."""

async def on_elicit(ctx, params):
    raise CallbackBoom("callback wants to abort back to the awaiter")

async def main():
    async with streamablehttp_client("http://127.0.0.1:8765/mcp") as (r, w, _):
        async with ClientSession(r, w, elicitation_callback=on_elicit) as session:
            await session.initialize()
            try:
                result = await session.call_tool("ask", {})
            except CallbackBoom:
                print("OK: CallbackBoom propagated to the awaiter")
                return
            print("BUG: awaiter did not raise; result.isError =", result.isError)
            print("     content =", result.content)

asyncio.run(main())

Python & MCP Python SDK

Python: 3.13
mcp:    1.27.1

Start with src/mcp/shared/session.py, especially BaseSession._receive_loop and send_request, then review the regression tests in tests/shared/test_session.py. Reproduce the callback exception behavior and verify that marked callback exceptions reach the in-flight call_tool awaiter while unmarked exceptions retain the existing response behavior.