feat(session): propagate callback exceptions to the awaiter

[Ar-maan05]

Allow exceptions raised inside user-defined client callbacks (elicitation, sampling, list roots) to propagate back to the awaiter of the outgoing request (e.g., session.call_tool), instead of being silently swallowed by the receive loop and converted into a generic INVALID_PARAMS JSON-RPC error.

Usage Example

Users opt-in to propagation by adding a __mcp_propagate__ marker attribute to their custom exceptions:

class MyControlFlowException(Exception):
__mcp_propagate__ = True

async def on_elicit(ctx, params):
# This exception will bubble up to the caller of send_request / call_tool
raise MyControlFlowException("abort")

Solution

• Added self._propagate_errors: dict[RequestId, BaseException] = {} to BaseSession to stash exceptions marked with mcp_propagate = True .
• In _receive_loop , if a callback exception has mcp_propagate = True :
• Send an INTERNAL_ERROR response back to the server so it doesn't hang.
• Populate _propagate_errors for active request IDs.
• Close active outgoing request streams and exit the receive loop.
• In send_request , catch anyio.EndOfStream and raise the stashed exception on the caller's task.
• Added test_callback_exception_propagation in tests/shared/test_session.py to prevent regressions.

Closes #2673