Address review feedback: ExceptionGroup bug, reverse dependency, type widening

- Replace task group + ctx.run(tg.start_soon, ...) with direct
  await sender_ctx.run(handler, msg) to avoid ExceptionGroup wrapping
  that would prevent ClosedResourceError from being caught
- Move ReadStream/WriteStream protocols to mcp.shared._stream_protocols
  so shared/server modules don't depend on client internals
- Restore write stream type narrowing in MessageStream (SessionMessage
  only, not SessionMessage | Exception)
- Remove unused T_Item TypeVar

Author: Kludex

src/mcp/client/_transport.py

@@ -3,57 +3,12 @@
 from __future__ import annotations
 
 from contextlib import AbstractAsyncContextManager
-from types import TracebackType
-from typing import Protocol, TypeVar, runtime_checkable
-
-from typing_extensions import Self
+from typing import Protocol
 
+from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.message import SessionMessage
 
-T_co = TypeVar("T_co", covariant=True)
-T_contra = TypeVar("T_contra", contravariant=True)
-
-
-@runtime_checkable
-class ReadStream(Protocol[T_co]):  # pragma: no branch
-    """Protocol for reading items from a stream.
-
-    Both ``MemoryObjectReceiveStream`` and ``ContextReceiveStream`` satisfy
-    this protocol.  Consumers that need the sender's context should use
-    ``getattr(stream, 'last_context', None)``.
-    """
-
-    async def receive(self) -> T_co: ...  # pragma: no branch
-    async def aclose(self) -> None: ...  # pragma: no branch
-    def __aiter__(self) -> ReadStream[T_co]: ...  # pragma: no branch
-    async def __anext__(self) -> T_co: ...  # pragma: no branch
-    async def __aenter__(self) -> Self: ...  # pragma: no branch
-    async def __aexit__(  # pragma: no branch
-        self,
-        exc_type: type[BaseException] | None,
-        exc_val: BaseException | None,
-        exc_tb: TracebackType | None,
-    ) -> bool | None: ...
-
-
-@runtime_checkable
-class WriteStream(Protocol[T_contra]):  # pragma: no branch
-    """Protocol for writing items to a stream.
-
-    Both ``MemoryObjectSendStream`` and ``ContextSendStream`` satisfy
-    this protocol.
-    """
-
-    async def send(self, item: T_contra, /) -> None: ...  # pragma: no branch
-    async def aclose(self) -> None: ...  # pragma: no branch
-    async def __aenter__(self) -> Self: ...  # pragma: no branch
-    async def __aexit__(  # pragma: no branch
-        self,
-        exc_type: type[BaseException] | None,
-        exc_val: BaseException | None,
-        exc_tb: TracebackType | None,
-    ) -> bool | None: ...
-
+__all__ = ["ReadStream", "WriteStream", "Transport", "TransportStreams"]
 
 TransportStreams = tuple[ReadStream[SessionMessage | Exception], WriteStream[SessionMessage]]
 

src/mcp/client/sse.py

@@ -143,8 +143,7 @@ async def _send_message(session_message: SessionMessage) -> None:
                                 async for session_message in write_stream_reader:
                                     sender_ctx = write_stream_reader.last_context
                                     if sender_ctx is not None:
-                                        async with anyio.create_task_group() as tg:
-                                            sender_ctx.run(tg.start_soon, _send_message, session_message)
+                                        await sender_ctx.run(_send_message, session_message)
                                     else:
                                         await _send_message(session_message)  # pragma: no cover
                         except Exception:  # pragma: lax no cover

src/mcp/client/streamable_http.py

@@ -482,8 +482,7 @@ async def handle_request_async():
                 async for session_message in write_stream_reader:
                     sender_ctx = write_stream_reader.last_context
                     if sender_ctx is not None:
-                        async with anyio.create_task_group() as tg_local:
-                            sender_ctx.run(tg_local.start_soon, _handle_message, session_message)
+                        await sender_ctx.run(_handle_message, session_message)
                     else:
                         await _handle_message(session_message)  # pragma: no cover
 

src/mcp/server/lowlevel/server.py

@@ -52,7 +52,6 @@ async def main():
 from typing_extensions import TypeVar
 
 from mcp import types
-from mcp.client._transport import ReadStream, WriteStream
 from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
@@ -66,6 +65,7 @@ async def main():
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
+from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.session import RequestResponder

src/mcp/server/session.py

@@ -37,10 +37,10 @@ async def handle_list_prompts(ctx: RequestContext, params) -> ListPromptsResult:
 from pydantic import AnyUrl, TypeAdapter
 
 from mcp import types
-from mcp.client._transport import ReadStream, WriteStream
 from mcp.server.experimental.session_features import ExperimentalServerSessionFeatures
 from mcp.server.models import InitializationOptions
 from mcp.server.validation import validate_sampling_tools, validate_tool_use_result_messages
+from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import StatelessModeNotSupported
 from mcp.shared.experimental.tasks.capabilities import check_tasks_capability
 from mcp.shared.experimental.tasks.helpers import RELATED_TASK_METADATA_KEY

src/mcp/server/streamable_http.py

@@ -24,9 +24,9 @@
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
-from mcp.client._transport import ReadStream, WriteStream
 from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
+from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
 from mcp.types import (

src/mcp/shared/_context_streams.py

@@ -20,7 +20,6 @@
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 
 T = TypeVar("T")
-T_Item = TypeVar("T_Item")
 
 # Internal payload carried through the underlying raw stream.
 _Envelope = tuple[contextvars.Context, T]

src/mcp/shared/_stream_protocols.py

@@ -0,0 +1,51 @@
+"""Stream protocols for MCP transports.
+
+These are general-purpose protocols satisfied by both ``MemoryObjectSendStream``/
+``MemoryObjectReceiveStream`` and the context-aware wrappers in ``_context_streams``.
+"""
+
+from __future__ import annotations
+
+from types import TracebackType
+from typing import Protocol, TypeVar, runtime_checkable
+
+from typing_extensions import Self
+
+T_co = TypeVar("T_co", covariant=True)
+T_contra = TypeVar("T_contra", contravariant=True)
+
+
+@runtime_checkable
+class ReadStream(Protocol[T_co]):  # pragma: no branch
+    """Protocol for reading items from a stream.
+
+    Consumers that need the sender's context should use
+    ``getattr(stream, 'last_context', None)``.
+    """
+
+    async def receive(self) -> T_co: ...  # pragma: no branch
+    async def aclose(self) -> None: ...  # pragma: no branch
+    def __aiter__(self) -> ReadStream[T_co]: ...  # pragma: no branch
+    async def __anext__(self) -> T_co: ...  # pragma: no branch
+    async def __aenter__(self) -> Self: ...  # pragma: no branch
+    async def __aexit__(  # pragma: no branch
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> bool | None: ...
+
+
+@runtime_checkable
+class WriteStream(Protocol[T_contra]):  # pragma: no branch
+    """Protocol for writing items to a stream."""
+
+    async def send(self, item: T_contra, /) -> None: ...  # pragma: no branch
+    async def aclose(self) -> None: ...  # pragma: no branch
+    async def __aenter__(self) -> Self: ...  # pragma: no branch
+    async def __aexit__(  # pragma: no branch
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> bool | None: ...

src/mcp/shared/memory.py

@@ -8,7 +8,7 @@
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
 from mcp.shared.message import SessionMessage
 
-MessageStream = tuple[ContextReceiveStream[SessionMessage | Exception], ContextSendStream[SessionMessage | Exception]]
+MessageStream = tuple[ContextReceiveStream[SessionMessage | Exception], ContextSendStream[SessionMessage]]
 
 
 @asynccontextmanager

src/mcp/shared/session.py

@@ -12,7 +12,7 @@
 from pydantic import BaseModel, TypeAdapter
 from typing_extensions import Self
 
-from mcp.client._transport import ReadStream, WriteStream
+from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage
 from mcp.shared.response_router import ResponseRouter
@@ -417,8 +417,8 @@ async def _handle_session_message(message: SessionMessage) -> None:
 
                     sender_ctx: contextvars.Context | None = getattr(self._read_stream, "last_context", None)
                     if sender_ctx is not None:
-                        async with anyio.create_task_group() as tg:
-                            sender_ctx.run(tg.start_soon, _handle_session_message, message)
+                        coro = sender_ctx.run(_handle_session_message, message)
+                        await coro
                     else:
                         await _handle_session_message(message)