Tidy naming and doc nits from review

- Rename _SHIELDED_WRITE_TIMEOUT to _ABANDON_WRITE_TIMEOUT; the
  timed-out arm passes it unshielded, so the old name lied there
- Document that on_stream_exception is awaited inline in the read loop
- Note in run()'s docstring that the dispatcher is single-shot
- Import ProgressFnT from mcp.shared.dispatcher (its home) instead of
  the mcp.shared.session compat shim
- Migration guide: scope the optional-fields section to
  ServerRequestContext, and correct the null-id bullet (v1 surfaced
  the transport's ValidationError, not an MCPError)

Author: maxisbey

docs/migration.md

@@ -1120,9 +1120,9 @@ async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestPar
     )
 ```
 
-### `RequestContext`: request-specific fields are now optional
+### `ServerRequestContext`: request-specific fields are now optional
 
-The `RequestContext` class now uses optional fields for request-specific data (`request_id`, `meta`, etc.) so it can be used for both request and notification handlers. In notification handlers, these fields are `None`.
+`ServerRequestContext` now uses optional fields for request-specific data (`request_id`, `meta`, etc.) so it can be used for both request and notification handlers. In notification handlers, these fields are `None`.
 
 ```python
 from mcp.server import ServerRequestContext
@@ -1177,7 +1177,7 @@ Behavior changes:
 - **Session shutdown now answers in-flight server-initiated requests with `CONNECTION_CLOSED` (-32000)**; v1 left them unanswered. The write is bounded (~1s) so closing stays fast.
 - **Notification callbacks are concurrent.** `logging_callback`, `progress_callback`, and `message_handler` deliveries start in arrival order but each runs as its own task: they may interleave, and a `progress_callback` delivery may finish after the request it reports on has returned. Callbacks that need strict sequencing must coordinate themselves, and there is no built-in bound on concurrent deliveries (v1's inline loop processed one message at a time).
 - **Transport-level `Exception` items are delivered to `message_handler` the same way** — as their own task, without blocking the receive loop — and a `message_handler` that raises on one is logged, not fatal to the session.
-- **Stray responses are no longer surfaced to `message_handler`.** Responses with an unknown id are ignored (as the spec asks; v1 surfaced a `RuntimeError`), and error responses with a null `id` — a peer reporting a parse error — are dropped with a debug log (v1 surfaced an `MCPError`).
+- **Stray responses are no longer surfaced to `message_handler`.** Responses with an unknown id are ignored (as the spec asks; v1 surfaced a `RuntimeError`), and error responses with a null `id` — a peer reporting a parse error — are dropped with a debug log (v1 surfaced the transport's `ValidationError`).
 - **A raising request callback** is answered with `code=0` and the exception text; v1 flattened every callback exception to `INVALID_PARAMS`. For a specific error response, return `ErrorData` (unchanged) or raise `MCPError`. One carve-out: pydantic's `ValidationError` is still answered with `INVALID_PARAMS`, as in v1.
 - **`send_request` before entering the context manager** raises `RuntimeError` immediately; v1 wrote to the transport and hung until the timeout. After the connection has closed it raises `MCPError` (`CONNECTION_CLOSED`) instead. `send_notification` before entry still works.
 - **`send_notification` no longer takes `related_request_id`, and `send_request` no longer accepts `ServerMessageMetadata`.** No client transport ever serialized these hints; progress and response correlation via `progressToken` and the request id is unaffected.

src/mcp/client/client.py

@@ -12,7 +12,7 @@
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
-from mcp.shared.session import ProgressFnT
+from mcp.shared.dispatcher import ProgressFnT
 from mcp.types import (
     CallToolResult,
     CompleteResult,

src/mcp/client/session.py

@@ -15,11 +15,11 @@
 from mcp import types
 from mcp.client._transport import ReadStream, WriteStream
 from mcp.shared._compat import resync_tracer
-from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher
+from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher, ProgressFnT
 from mcp.shared.exceptions import MCPError
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
-from mcp.shared.session import ProgressFnT, RequestResponder
+from mcp.shared.session import RequestResponder
 from mcp.shared.transport_context import TransportContext
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
 from mcp.types import RequestId, RequestParamsMeta

src/mcp/shared/jsonrpc_dispatcher.py

@@ -51,9 +51,9 @@
 
 logger = logging.getLogger(__name__)
 
-_SHIELDED_WRITE_TIMEOUT: float = 5
-"""Bound for courtesy abandon-path writes; without it a wedged transport
-would turn the shielded write into an uncancellable hang."""
+_ABANDON_WRITE_TIMEOUT: float = 5
+"""Bound for courtesy-cancel writes on the abandon paths; the caller-cancel
+arm shields its write, so a wedged transport would otherwise hang it uncancellably."""
 
 _SHUTDOWN_WRITE_TIMEOUT: float = 1
 """Tighter bound for the shutdown-arm error write so a wedged transport can't hold session close."""
@@ -232,7 +232,8 @@ def __init__(
                 message is dequeued (e.g. `initialize`); an inline handler
                 that awaits the peer deadlocks the parked loop.
             on_stream_exception: Observer for `Exception` items on the read
-                stream; without it they are debug-logged and dropped.
+                stream; without it they are debug-logged and dropped. Awaited
+                inline in the read loop, so a slow observer stalls dispatch.
         """
         self._read_stream = read_stream
         self._write_stream = write_stream
@@ -332,7 +333,7 @@ async def send_raw_request(
                         _related_request_id,
                     ),
                     shield=False,
-                    timeout=_SHIELDED_WRITE_TIMEOUT,
+                    timeout=_ABANDON_WRITE_TIMEOUT,
                     describe=f"courtesy cancel for timed-out request {request_id!r}",
                 )
             raise MCPError(code=REQUEST_TIMEOUT, message=f"Request {method!r} timed out") from None
@@ -343,7 +344,7 @@ async def send_raw_request(
                 await self._final_write(
                     partial(self._cancel_outbound, request_id, "caller cancelled", _related_request_id),
                     shield=True,
-                    timeout=_SHIELDED_WRITE_TIMEOUT,
+                    timeout=_ABANDON_WRITE_TIMEOUT,
                     describe=f"courtesy cancel for caller-cancelled request {request_id!r}",
                 )
             raise
@@ -382,6 +383,7 @@ async def run(
         """Drive the receive loop until the read stream closes.
 
         `task_status.started()` fires once `send_raw_request` is usable.
+        Single-shot: once the loop ends the dispatcher stays closed and cannot be restarted.
         """
         try:
             # LIFO exits: the write stream closes only after the task-group join, so teardown writes still land.

tests/shared/test_jsonrpc_dispatcher.py

@@ -727,7 +727,7 @@ async def test_caller_cancel_courtesy_write_is_bounded_when_the_transport_is_wed
     caplog: pytest.LogCaptureFixture,
 ):
     """A wedged transport write cannot turn caller cancellation into an unbounded shielded hang:
-    `_SHIELDED_WRITE_TIMEOUT` abandons the courtesy-cancel write (SDK-defined bound). On regression
+    `_ABANDON_WRITE_TIMEOUT` abandons the courtesy-cancel write (SDK-defined bound). On regression
     the test hangs rather than failing fast - fail_after cannot cancel through the shield."""
     c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](0)
     s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](0)
@@ -745,7 +745,7 @@ async def caller() -> None:
         gave_up.set()
 
     try:
-        # Both bounds exceed the in-loop _SHIELDED_WRITE_TIMEOUT (5s); the virtual clock makes them instant.
+        # Both bounds exceed the in-loop _ABANDON_WRITE_TIMEOUT (5s); the virtual clock makes them instant.
         with anyio.fail_after(30):
             async with anyio.create_task_group() as tg:  # pragma: no branch
                 await tg.start(client.run, on_request, on_notify)
@@ -775,7 +775,7 @@ async def test_timeout_courtesy_cancel_write_is_bounded_when_the_transport_is_we
     caplog: pytest.LogCaptureFixture,
 ):
     """A wedged transport write cannot delay the REQUEST_TIMEOUT error indefinitely (SDK-defined
-    bound): `_SHIELDED_WRITE_TIMEOUT` abandons the courtesy cancel so the error still surfaces."""
+    bound): `_ABANDON_WRITE_TIMEOUT` abandons the courtesy cancel so the error still surfaces."""
     c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](0)
     s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](0)
     client: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(s2c_recv, c2s_send)
@@ -799,7 +799,7 @@ async def caller() -> None:
                 request = await c2s_recv.receive()
             assert isinstance(request, SessionMessage)
             assert isinstance(request.message, JSONRPCRequest)
-            # Exceeds the request timeout (1s) plus _SHIELDED_WRITE_TIMEOUT (5s); virtual clock, no wall time.
+            # Exceeds the request timeout (1s) plus _ABANDON_WRITE_TIMEOUT (5s); virtual clock, no wall time.
             with anyio.fail_after(10):
                 await gave_up.wait()
             tg.cancel_scope.cancel()
@@ -835,7 +835,7 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) ->
         raise NotImplementedError
 
     try:
-        # 3s sits between _SHUTDOWN_WRITE_TIMEOUT (1s) and _SHIELDED_WRITE_TIMEOUT (5s): pins the tighter bound.
+        # 3s sits between _SHUTDOWN_WRITE_TIMEOUT (1s) and _ABANDON_WRITE_TIMEOUT (5s): pins the tighter bound.
         with anyio.fail_after(3):
             async with anyio.create_task_group() as tg:  # pragma: no branch
                 await tg.start(server.run, park, on_notify)