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

[Ar-maan05]

Merged the latest main to resolve the conflict.

The only collision was a one-line coverage pragma in src/mcp/shared/session.py: #2691 added # pragma: lax no cover to the except Exception: in the receive loop's connection-closed cleanup. Kept that pragma (took main's side), this PR has no reason to drop it. The exception-propagation logic and tests/shared/test_session.py merged cleanly with no other changes.

Verified locally: uv run --frozen pytest tests/shared/test_session.py -> 11 passed.

[Ar-maan05]

Friendly ping on this one; CI is green (27/27) and it's rebased on latest main. Quick recap of the approach: callback exceptions are only propagated when the user opts in via a __mcp_propagate__ = True marker, so default behavior is unchanged. On a marked exception the receive loop still sends an INTERNAL_ERROR back to the server (so it doesn't hang), stashes the exception per active request id in _propagate_errors, closes the outgoing request streams, and exits; send_request then catches anyio.EndOfStream and re-raises the stashed exception to the awaiter. Covered by test_callback_exception_propagation. Closes #2673 , happy to adjust the marker approach if maintainers prefer a different opt-in mechanism. Would love a review when someone has a moment 🙏

[Ar-maan05]

Re-merged latest main to clear a fresh conflict. Main's "Remove Tasks (SEP-1686)" (#2714) took the ResponseRouter machinery out of session.py, which this branch still carried from an earlier base. The merge dropped all of that automatically except one spot: the class-attribute block where this PR adds _propagate_errors sat right next to the old _response_routers declaration. I kept _propagate_errors and dropped the now-orphaned _response_routers line; nothing else references ResponseRouter anymore.

Re-verified on top of current main: tests/shared/test_session.py passes (11, including test_callback_exception_propagation and test_send_request_end_of_stream_without_propagated_error), and ./scripts/test reports 100% coverage with strict-no-cover clean. Back to mergeable. Happy to walk through the approach if a review slot opens up.