diff --git a/docs/migration.md b/docs/migration.md index b094d79f84..78b713020a 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -2188,6 +2188,25 @@ In v1, a non-2xx response to a message POST (other than 404) raised `httpx.HTTPS | 404, no session yet | `McpError` with positive code `32600` | `MCPError(-32601, 'Not Found')` | | Any other 4xx/5xx | `httpx.HTTPStatusError` escapes as `ExceptionGroup` | `MCPError(-32603, 'Server returned an error response')` | +The same contract covers the resumption GET — a request re-attached with a resumption token (`Last-Event-ID`). In v1 a failure there escaped the context as an `ExceptionGroup` that failed every pending request, or hung the resumed call forever: + +| Resumption GET outcome | v1 | v2 | +| --- | --- | --- | +| 404, session established | `httpx.HTTPStatusError` escapes as `ExceptionGroup` | `MCPError(-32600, 'Session terminated')` | +| Any other non-2xx | `httpx.HTTPStatusError` escapes as `ExceptionGroup` | `MCPError(-32603, 'Server returned an error response')` | +| Stream drops mid-read | error escapes as `ExceptionGroup` | `MCPError(-32000, 'resumption stream ended without a response')` | +| Stream ends cleanly with no response | resumed call hangs forever | `MCPError(-32000, 'resumption stream ended without a response')` | + +The SSE transport (`sse_client`) applies the same rule to its message POST. In v1 *any* POST failure — a non-2xx status, a network error, or an OAuth re-auth failure raised by the configured `auth` — was caught and logged inside the transport's writer task: the waiting caller hung forever and the write loop died, so every later send was silently dropped. In v2 the failing request resolves promptly and the session stays usable: + +| Message POST outcome | v1 | v2 | +| --- | --- | --- | +| 404, endpoint URL carries a session id | caller hangs forever; write loop dies | `MCPError(-32600, 'Session terminated')` | +| Any other non-2xx | caller hangs forever; write loop dies | `MCPError(-32603, 'Server returned an error response')` | +| Network-level failure (`httpx2.ConnectError`, timeouts) or OAuth flow failure | caller hangs forever; write loop dies | `MCPError(-32000, 'Failed to send message: ...')` | + +A failed POST of a *notification or response* has no caller to resolve; v2 logs and drops it, keeping the write loop (and every later send) alive. + Both common v1 patterns silently stop working: an `except* httpx.HTTPStatusError` around the transport context becomes dead code because status errors no longer escape the context, and a session-expiry check on `error.code == 32600` never matches again because the code is now the standard negative `-32600`. **Before (v1):** @@ -2230,7 +2249,7 @@ async with streamable_http_client(url) as (read, write): raise ``` -Move HTTP-status failure handling from around the transport context to around the individual calls, catching `MCPError` (see [`McpError` renamed to `MCPError`](#mcperror-renamed-to-mcperror)). Connect-level failures such as `httpx2.ConnectError` still escape the transport context as before; keep context-level handling for those only. +Move HTTP-status failure handling from around the transport context to around the individual calls, catching `MCPError` (see [`McpError` renamed to `MCPError`](#mcperror-renamed-to-mcperror)). On the streamable HTTP transport, connect-level failures such as `httpx2.ConnectError` on a *request's* message POST still escape the transport context as before — keep context-level handling for those; on the resumption GET and on the SSE transport's message POST they resolve the failing request instead, as above. A connect-level failure POSTing a *notification or response* on streamable HTTP is logged and does not escape the context, but it kills the transport's write loop — pre-existing behavior, unchanged from v1. ### `terminate_windows_process` removed diff --git a/src/mcp/client/_transport.py b/src/mcp/client/_transport.py index 0163fef950..9b0c6451b3 100644 --- a/src/mcp/client/_transport.py +++ b/src/mcp/client/_transport.py @@ -5,14 +5,28 @@ from contextlib import AbstractAsyncContextManager from typing import Protocol +from mcp_types import INTERNAL_ERROR, INVALID_REQUEST, ErrorData + from mcp.shared._stream_protocols import ReadStream, WriteStream from mcp.shared.message import SessionMessage -__all__ = ["ReadStream", "WriteStream", "Transport", "TransportStreams"] +__all__ = ["ReadStream", "WriteStream", "Transport", "TransportStreams", "status_error_data"] TransportStreams = tuple[ReadStream[SessionMessage | Exception], WriteStream[SessionMessage]] +def status_error_data(status_code: int, *, has_session: bool) -> ErrorData: + """Map a non-2xx HTTP status on a client transport request to the error its waiting caller receives. + + A 404 while a session is held is the session-expiry signal (`INVALID_REQUEST`, + "Session terminated"); anything else gets the generic stand-in. A call site with + an extra status mapping (e.g. the message POST's pre-session 404) branches first. + """ + if status_code == 404 and has_session: + return ErrorData(code=INVALID_REQUEST, message="Session terminated") + return ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") + + class Transport(AbstractAsyncContextManager[TransportStreams], Protocol): """Protocol for MCP transports. diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py index 31d0f35391..2d96d75a92 100644 --- a/src/mcp/client/sse.py +++ b/src/mcp/client/sse.py @@ -10,6 +10,7 @@ from anyio.abc import TaskStatus from httpx2 import SSEError +from mcp.client._transport import status_error_data from mcp.shared._compat import resync_tracer from mcp.shared._context_streams import create_context_streams from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client @@ -120,17 +121,52 @@ async def post_writer(endpoint_url: str): async with write_stream_reader, write_stream: async def _send_message(session_message: SessionMessage) -> None: + # A POST failure must not raise: the post_writer handler below + # would swallow it, hanging the waiting caller forever and killing + # the write loop (#2110). Mirror the streamable-HTTP transport + # instead: resolve the waiter with an error correlated to its + # request id, keeping the session usable. logger.debug(f"Sending client message: {session_message}") - response = await client.post( - endpoint_url, - json=session_message.message.model_dump( - by_alias=True, - mode="json", - exclude_unset=True, - ), - ) - response.raise_for_status() - logger.debug(f"Client message sent successfully: {response.status_code}") + message = session_message.message + try: + response = await client.post( + endpoint_url, + json=message.model_dump( + by_alias=True, + mode="json", + exclude_unset=True, + ), + ) + except Exception as exc: + # Terminal containment boundary: beyond httpx's own errors, + # user-supplied auth flows and hooks can raise arbitrary types + # from inside `client.post()`, so an enumerated catch cannot + # keep the caller from hanging. + logger.exception("Error POSTing message") + error = types.ErrorData( + code=types.CONNECTION_CLOSED, message=f"Failed to send message: {exc}" + ) + else: + if response.is_success: + logger.debug(f"Client message sent successfully: {response.status_code}") + return + logger.error(f"Message POST returned HTTP status {response.status_code}") + # The endpoint URL carrying a session id is this transport's + # "session established" signal, as `self.session_id` is for + # streamable HTTP. + error = status_error_data( + response.status_code, + has_session=_extract_session_id_from_endpoint(endpoint_url) is not None, + ) + # A notification has no waiter to resolve, so its failure is only logged. + if isinstance(message, types.JSONRPCRequest): + reply = types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error) + try: + await read_stream_writer.send(SessionMessage(reply)) + except (anyio.BrokenResourceError, anyio.ClosedResourceError): + # Teardown race: the reader is gone, so there is nobody + # left to resolve - contain it, keeping the write loop up. + logger.debug("read stream closed before request %r could be resolved", message.id) async for session_message in write_stream_reader: sender_ctx = write_stream_reader.last_context diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..40574c1f76 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -14,7 +14,6 @@ from httpx2 import EventSource, ServerSentEvent from mcp_types import ( CONNECTION_CLOSED, - INTERNAL_ERROR, INVALID_REQUEST, METHOD_NOT_FOUND, PARSE_ERROR, @@ -30,7 +29,7 @@ from mcp_types.version import MODERN_PROTOCOL_VERSIONS from pydantic import ValidationError -from mcp.client._transport import TransportStreams +from mcp.client._transport import TransportStreams, status_error_data from mcp.shared._compat import resync_tracer from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams from mcp.shared._httpx_utils import create_mcp_http_client @@ -248,25 +247,44 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None: else: raise ResumptionError("Resumption request requires a resumption token") # pragma: no cover - # Extract original request ID to map responses - original_request_id = None - if isinstance(ctx.session_message.message, JSONRPCRequest): # pragma: no branch - original_request_id = ctx.session_message.message.id + # Only requests resume: post_writer dispatches here on message type as well as + # metadata, so the original id is always available to map responses. + assert isinstance(ctx.session_message.message, JSONRPCRequest) + original_request_id = ctx.session_message.message.id - async with ctx.client.sse(self.url, headers=headers) as event_source: - event_source.response.raise_for_status() - logger.debug("Resumption GET SSE connection established") + try: + async with ctx.client.sse(self.url, headers=headers) as event_source: + if not event_source.response.is_success: + # Resolve the waiting caller with an error correlated to its request, + # mirroring `_handle_post_request`: an escaping `HTTPStatusError` would + # tear down the transport's task group and every stream with it (#2110). + error_data = status_error_data( + event_source.response.status_code, has_session=self.session_id is not None + ) + await self._resolve_abandoned_request( + ctx.read_stream_writer, original_request_id, error_data.message, code=error_data.code + ) + return + logger.debug("Resumption GET SSE connection established") - async for sse in event_source: # pragma: no branch - is_complete = await self._handle_sse_event( - sse, - ctx.read_stream_writer, - original_request_id, - ctx.metadata.on_resumption_token_update if ctx.metadata else None, - ) - if is_complete: - await event_source.response.aclose() - break + async for sse in event_source: + is_complete = await self._handle_sse_event( + sse, + ctx.read_stream_writer, + original_request_id, + ctx.metadata.on_resumption_token_update if ctx.metadata else None, + ) + if is_complete: + await event_source.response.aclose() + return + except Exception: + logger.debug("Resumption stream ended", exc_info=True) + + # Stream ended without a response, cleanly or mid-read: resolve the waiter, + # mirroring `_handle_sse_response`, else the caller would hang forever. + await self._resolve_abandoned_request( + ctx.read_stream_writer, original_request_id, "resumption stream ended without a response" + ) def _consume_modern_cancellation(self, session_message: SessionMessage) -> bool: """Translate an outbound `notifications/cancelled` at 2026; True means "do not POST". @@ -358,16 +376,13 @@ async def _handle_post_request(self, ctx: RequestContext) -> None: except (httpx2.StreamError, ValidationError): pass logger.debug("Non-2xx body was not a JSON-RPC error; using fallback") - if response.status_code == 404: - if self.session_id is None: - # No session yet → 404 is the HTTP-level spelling of - # METHOD_NOT_FOUND (gateway / legacy server doesn't know - # this method); "Session terminated" would be a lie here. - error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found") - else: - error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated") + if response.status_code == 404 and self.session_id is None: + # No session yet → 404 is the HTTP-level spelling of + # METHOD_NOT_FOUND (gateway / legacy server doesn't know + # this method); "Session terminated" would be a lie here. + error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found") else: - error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") + error_data = status_error_data(response.status_code, has_session=self.session_id is not None) session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) await ctx.read_stream_writer.send(session_message) return @@ -556,8 +571,10 @@ async def _handle_message(session_message: SessionMessage) -> None: else None ) - # Check if this is a resumption request - is_resumption = bool(metadata and metadata.resumption_token) + # Only a request resumes: the token names an interrupted request's + # stream, and `_handle_resumption_request` needs the id to correlate + # its outcome. A notification stamped with one is POSTed as usual. + is_resumption = bool(metadata and metadata.resumption_token) and isinstance(message, JSONRPCRequest) logger.debug(f"Sending client message: {message}") diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index d21f520daf..74c37299bd 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -19,6 +19,7 @@ CLIENT_CAPABILITIES_META_KEY, CLIENT_INFO_META_KEY, CONNECTION_CLOSED, + INTERNAL_ERROR, INVALID_REQUEST, METHOD_NOT_FOUND, PROTOCOL_VERSION_META_KEY, @@ -31,7 +32,9 @@ from starlette.types import Receive, Scope, Send from mcp.client.streamable_http import ( + LAST_EVENT_ID, MAX_RECONNECTION_ATTEMPTS, + MCP_SESSION_ID, RequestContext, StreamableHTTPTransport, streamable_http_client, @@ -132,6 +135,120 @@ def handler(request: httpx2.Request) -> httpx2.Response: assert reply.message.error.code == METHOD_NOT_FOUND +@pytest.mark.anyio +@pytest.mark.parametrize("status", [302, 401, 403, 500]) +async def test_resumption_get_http_error_resolves_caller_and_transport_survives(status: int) -> None: + """A non-2xx on the resumption GET resolves the waiting request with a JSON-RPC error + correlated to its id, and the transport stays usable for follow-up requests (SDK-defined; + #2110 — the status error used to escape into the task group and tear down every stream). + An unfollowed redirect counts: its body is no event stream, so no response can arrive. + """ + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.method == "GET" and LAST_EVENT_ID in request.headers: + return httpx2.Response(status) + body = json.loads(request.content) + return httpx2.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}}) + + with anyio.fail_after(5): + async with ( + httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http, + streamable_http_client("http://test/mcp", http_client=http) as (read, write), + ): + await write.send( + SessionMessage( + message=JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={}), + metadata=ClientMessageMetadata(resumption_token="token-1"), + ) + ) + reply = await read.receive() + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message, JSONRPCError) + assert reply.message.id == 1 + assert reply.message.error.code == INTERNAL_ERROR + assert reply.message.error.message == snapshot("Server returned an error response") + + # The transport survived: a plain follow-up request still round-trips. + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/list", params={}))) + follow_up = await read.receive() + assert isinstance(follow_up, SessionMessage) + assert isinstance(follow_up.message, JSONRPCResponse) + assert follow_up.message.id == 2 + + +@pytest.mark.anyio +async def test_resumption_get_404_with_session_reports_session_terminated() -> None: + """A 404 on the resumption GET while a session id is held reports "Session terminated" + (INVALID_REQUEST) to the waiter, the same session-expiry mapping as the POST path, so + reconnect logic keyed on that error works across both (SDK-defined).""" + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.method == "GET" and LAST_EVENT_ID in request.headers: + return httpx2.Response(404) + if request.method == "DELETE": # session termination on close + return httpx2.Response(200) + body = json.loads(request.content) + return httpx2.Response( + 200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}}, headers={MCP_SESSION_ID: "sess-1"} + ) + + with anyio.fail_after(5): + async with ( + httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http, + streamable_http_client("http://test/mcp", http_client=http) as (read, write), + ): + # An initialize round-trip stores the session id the server stamps on its response. + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="initialize", params={}))) + assert isinstance(await read.receive(), SessionMessage) + + await write.send( + SessionMessage( + message=JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/call", params={}), + metadata=ClientMessageMetadata(resumption_token="token-1"), + ) + ) + reply = await read.receive() + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message, JSONRPCError) + assert reply.message.id == 2 + assert reply.message.error.code == INVALID_REQUEST + assert reply.message.error.message == snapshot("Session terminated") + + +@pytest.mark.anyio +async def test_notification_with_resumption_token_is_posted_not_resumed() -> None: + """A notification stamped with a resumption token is POSTed like any notification, and the + write loop survives to serve the next request (SDK-defined: the token names an interrupted + request's stream, so resumption applies to requests only).""" + recorded: list[httpx2.Request] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + recorded.append(request) + body = json.loads(request.content) + if "id" not in body: + return httpx2.Response(202) + return httpx2.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}}) + + with anyio.fail_after(5): + async with ( + httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http, + streamable_http_client("http://test/mcp", http_client=http) as (read, write), + ): + await write.send( + SessionMessage( + message=JSONRPCNotification(jsonrpc="2.0", method="notifications/foo", params={}), + metadata=ClientMessageMetadata(resumption_token="token-1"), + ) + ) + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={}))) + reply = await read.receive() + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message, JSONRPCResponse) + assert reply.message.id == 1 + # The stamped notification went out as a plain POST, not a resumption GET. + assert [r.method for r in recorded] == ["POST", "POST"] + + @pytest.mark.anyio async def test_initialize_post_clears_cached_pv_header_and_unstamped_posts_read_it() -> None: """``initialize`` discards the cached protocol-version header; every other POST reads it. @@ -632,6 +749,74 @@ def handler(request: httpx2.Request) -> httpx2.Response: assert reply.message.error.code == CONNECTION_CLOSED +@pytest.mark.anyio +async def test_resumption_stream_dying_mid_read_resolves_caller_and_transport_survives() -> None: + """A resumption GET stream that dies mid-read resolves the waiter with CONNECTION_CLOSED + and the transport stays usable for follow-up requests (SDK-defined; #2110 — the read error + used to escape into the task group and tear down every stream).""" + dying = _DyingSSEStream() + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.method == "GET" and LAST_EVENT_ID in request.headers: + return httpx2.Response(200, headers={"content-type": "text/event-stream"}, stream=dying) + body = json.loads(request.content) + return httpx2.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}}) + + with anyio.fail_after(5): + async with ( + httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http, + streamable_http_client("http://test/mcp", http_client=http) as (read, write), + ): + await write.send( + SessionMessage( + message=JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={}), + metadata=ClientMessageMetadata(resumption_token="token-1"), + ) + ) + reply = await read.receive() + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message, JSONRPCError) + assert reply.message.id == 1 + assert reply.message.error.code == CONNECTION_CLOSED + assert reply.message.error.message == snapshot("resumption stream ended without a response") + + # The transport survived: a plain follow-up request still round-trips. + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/list", params={}))) + follow_up = await read.receive() + assert isinstance(follow_up, SessionMessage) + assert isinstance(follow_up.message, JSONRPCResponse) + assert follow_up.message.id == 2 + + +@pytest.mark.anyio +async def test_resumption_stream_clean_end_without_response_resolves_caller() -> None: + """A resumption GET stream that closes cleanly without delivering a response (e.g. the + server no longer holds the resumed request's events) resolves the waiter with an error + instead of hanging it forever (SDK-defined; #2110).""" + + def handler(request: httpx2.Request) -> httpx2.Response: + assert request.method == "GET" and LAST_EVENT_ID in request.headers + return httpx2.Response(200, headers={"content-type": "text/event-stream"}, content=b": nothing to replay\n\n") + + with anyio.fail_after(5): + async with ( + httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http, + streamable_http_client("http://test/mcp", http_client=http) as (read, write), + ): + await write.send( + SessionMessage( + message=JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={}), + metadata=ClientMessageMetadata(resumption_token="token-1"), + ) + ) + reply = await read.receive() + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message, JSONRPCError) + assert reply.message.id == 1 + assert reply.message.error.code == CONNECTION_CLOSED + assert reply.message.error.message == snapshot("resumption stream ended without a response") + + class _DeliverOnCommandSSEStream(httpx2.AsyncByteStream): """Parks after opening, then delivers one JSON-RPC response when told.""" diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index c27dd69db3..f53a5fc91b 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -1,7 +1,8 @@ """Tests for the SSE client and server transports, driven entirely in process.""" import json -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Callable +from types import TracebackType from typing import Any from unittest.mock import AsyncMock, MagicMock, Mock from urllib.parse import urlparse @@ -18,6 +19,7 @@ EmptyResult, Implementation, InitializeResult, + JSONRPCRequest, JSONRPCResponse, ListToolsResult, PaginatedRequestParams, @@ -30,10 +32,12 @@ ) from starlette.applications import Starlette from starlette.requests import Request -from starlette.responses import Response +from starlette.responses import Response, StreamingResponse from starlette.routing import Mount, Route +from starlette.types import ASGIApp, Receive, Scope, Send import mcp.client.sse +from mcp.client.auth.exceptions import OAuthTokenError from mcp.client.session import ClientSession from mcp.client.sse import _extract_session_id_from_endpoint, sse_client from mcp.server import Server, ServerRequestContext @@ -41,6 +45,7 @@ from mcp.server.transport_security import TransportSecuritySettings from mcp.shared._httpx_utils import McpHttpClientFactory from mcp.shared.exceptions import MCPError +from mcp.shared.message import SessionMessage from tests.interaction.transports import StreamingASGITransport SERVER_NAME = "test_server_for_SSE" @@ -82,8 +87,11 @@ async def _handle_read_resource(ctx: ServerRequestContext, params: ReadResourceR raise MCPError(code=404, message="OOPS! no resource with that URI was found") -def make_app(server: Server) -> Starlette: - """Mount `server` on a Starlette app exposing the SSE transport at /sse and /messages/.""" +def make_app(server: Server, wrap_post: Callable[[ASGIApp], ASGIApp] | None = None) -> Starlette: + """Mount `server` on a Starlette app exposing the SSE transport at /sse and /messages/. + + `wrap_post` optionally wraps the message-POST ASGI app (e.g. to inject HTTP failures). + """ # DNS-rebinding protection validates Host/Origin headers against a network attack that cannot # exist for an in-process app; the transport security behaviour itself is pinned by # tests/server/test_sse_security.py. @@ -96,10 +104,14 @@ async def handle_sse(request: Request) -> Response: await server.run(read_stream, write_stream, server.create_initialization_options()) return Response() + post_app: ASGIApp = sse.handle_post_message + if wrap_post is not None: + post_app = wrap_post(post_app) + return Starlette( routes=[ Route("/sse", endpoint=handle_sse), - Mount("/messages/", app=sse.handle_post_message), + Mount("/messages/", app=post_app), ] ) @@ -108,6 +120,28 @@ def make_server_app() -> Starlette: return make_app(Server(SERVER_NAME, on_read_resource=_handle_read_resource)) +def make_app_rejecting_posts(reject: dict[str, int]) -> Starlette: + """Like `make_server_app`, but the message POST is answered with a bare HTTP error + for JSON-RPC messages whose method appears in `reject` (they never reach the server).""" + + def wrap(inner: ASGIApp) -> ASGIApp: + async def handle_post(scope: Scope, receive: Receive, send: Send) -> None: + body = await Request(scope, receive).body() + status = reject.get(json.loads(body).get("method")) + if status is not None: + await Response(status_code=status)(scope, receive, send) + return + + async def replay() -> dict[str, Any]: + return {"type": "http.request", "body": body, "more_body": False} + + await inner(scope, replay, send) + + return handle_post + + return make_app(Server(SERVER_NAME, on_read_resource=_handle_read_resource), wrap_post=wrap) + + @pytest.mark.anyio async def test_raw_sse_connection() -> None: """The SSE GET responds 200 with an event-stream content type, announcing the session @@ -224,6 +258,239 @@ async def test_sse_client_exception_handling( await session.read_resource(uri="xxx://will-not-work") +@pytest.mark.anyio +@pytest.mark.parametrize("status_code", [302, 401, 403, 500]) +async def test_sse_client_request_post_http_error_reaches_caller_and_session_survives(status_code: int) -> None: + """A non-2xx on a request's message POST reaches the waiting caller promptly as a JSON-RPC + error correlated to the request, and the session stays usable (SDK-defined; #2110 — the + status error used to be swallowed inside post_writer, hanging the caller forever). + An unfollowed redirect counts: the message never reached the server, so no response can arrive. + """ + factory = in_process_client_factory(make_app_rejecting_posts({"resources/read": status_code})) + with anyio.fail_after(5): + # One parenthesized async-with: separately nested ones trip a phantom + # branch arc under coverage on Python 3.14 (see the note in mcp.client.sse). + async with ( + sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams, + ClientSession(*streams) as session, + ): + await session.initialize() + + with pytest.raises(MCPError) as exc_info: + await session.read_resource(uri="foobar://should-work") + assert exc_info.value.error.code == types.INTERNAL_ERROR + assert exc_info.value.error.message == snapshot("Server returned an error response") + + # The session survived the failed POST: the next request round-trips. + assert isinstance(await session.send_ping(), EmptyResult) + + +@pytest.mark.anyio +async def test_sse_client_request_post_network_error_reaches_caller_and_session_survives() -> None: + """A network-level failure on a request's message POST reaches the waiting caller promptly + as a JSON-RPC error correlated to the request, and the session stays usable (SDK-defined; + #2110 — the exception used to be swallowed inside post_writer, hanging the caller forever). + """ + + class _FlakyPostTransport(httpx2.AsyncBaseTransport): + """Serves the standard test app, but the POST of one JSON-RPC method never connects.""" + + def __init__(self) -> None: + self._inner = StreamingASGITransport(make_server_app(), cancel_on_close=False) + + async def __aenter__(self) -> "_FlakyPostTransport": + await self._inner.__aenter__() + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None = None, + exc_value: BaseException | None = None, + traceback: TracebackType | None = None, + ) -> None: + await self._inner.__aexit__(exc_type, exc_value, traceback) + + async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response: + if request.method == "POST" and json.loads(request.content).get("method") == "resources/read": + raise httpx2.ConnectError("connection refused", request=request) + return await self._inner.handle_async_request(request) + + def factory( + headers: dict[str, str] | None = None, + timeout: httpx2.Timeout | None = None, + auth: httpx2.Auth | None = None, + ) -> httpx2.AsyncClient: + return httpx2.AsyncClient( + transport=_FlakyPostTransport(), base_url=BASE_URL, headers=headers, timeout=timeout, auth=auth + ) + + with anyio.fail_after(5): + # One parenthesized async-with: separately nested ones trip a phantom + # branch arc under coverage on Python 3.14 (see the note in mcp.client.sse). + async with ( + sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams, + ClientSession(*streams) as session, + ): + await session.initialize() + + with pytest.raises(MCPError) as exc_info: + await session.read_resource(uri="foobar://should-work") + assert exc_info.value.error.code == types.CONNECTION_CLOSED + # The message embeds httpx's exception text; pin only the SDK-authored prefix. + assert exc_info.value.error.message.startswith("Failed to send message:") + + # The session survived the failed POST: the next request round-trips. + assert isinstance(await session.send_ping(), EmptyResult) + + +@pytest.mark.anyio +async def test_sse_client_post_404_with_session_endpoint_reports_session_terminated() -> None: + """A 404 on a request's message POST while the endpoint URL carries a session id reports + "Session terminated" (INVALID_REQUEST) to the caller, the same session-expiry mapping as + the streamable HTTP transport (SDK-defined).""" + factory = in_process_client_factory(make_app_rejecting_posts({"resources/read": 404})) + with anyio.fail_after(5): + # One parenthesized async-with: separately nested ones trip a phantom + # branch arc under coverage on Python 3.14 (see the note in mcp.client.sse). + async with ( + sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams, + ClientSession(*streams) as session, + ): + await session.initialize() + + with pytest.raises(MCPError) as exc_info: + await session.read_resource(uri="foobar://should-work") + assert exc_info.value.error.code == types.INVALID_REQUEST + assert exc_info.value.error.message == snapshot("Session terminated") + + +@pytest.mark.anyio +async def test_sse_client_post_404_without_session_endpoint_keeps_generic_error() -> None: + """A 404 on a request's message POST when the endpoint URL carries no session id keeps the + generic error: with no session to expire, "Session terminated" would be a lie (SDK-defined). + The raw endpoint is scripted because `SseServerTransport` always issues a session id.""" + + async def handle_sse(request: Request) -> StreamingResponse: + async def stream() -> AsyncGenerator[str, None]: + yield "event: endpoint\ndata: /messages/\n\n" + await anyio.Event().wait() # park until the client disconnects + + return StreamingResponse(stream(), media_type="text/event-stream") + + async def handle_post(request: Request) -> Response: + return Response(status_code=404) + + app = Starlette(routes=[Route("/sse", handle_sse), Route("/messages/", handle_post, methods=["POST"])]) + factory = in_process_client_factory(app) + with anyio.fail_after(5): + async with ( + sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams, + ClientSession(*streams) as session, + ): + with pytest.raises(MCPError) as exc_info: + await session.initialize() + assert exc_info.value.error.code == types.INTERNAL_ERROR + assert exc_info.value.error.message == snapshot("Server returned an error response") + + +@pytest.mark.anyio +@pytest.mark.parametrize("exc_type", [OAuthTokenError, RuntimeError]) +async def test_sse_client_auth_failure_on_post_reaches_caller_and_session_survives( + exc_type: type[Exception], +) -> None: + """A failure raised from inside a request's message POST by a user-supplied hook — an SDK + OAuth flow error, or any exception from a custom auth flow — reaches the waiting caller + promptly as a JSON-RPC error correlated to the request, and the session stays usable + (SDK-defined; #2110 — like any network error, it used to be swallowed inside post_writer).""" + + class _RefusingAuth(httpx2.Auth): + """Stands in for OAuthClientProvider (or any user auth hook) failing mid-session.""" + + async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Request, httpx2.Response]: + if request.method == "POST" and json.loads(request.content).get("method") == "resources/read": + raise exc_type("re-authentication failed") + yield request + + factory = in_process_client_factory(make_server_app()) + with anyio.fail_after(5): + async with ( + sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory, auth=_RefusingAuth()) as streams, + ClientSession(*streams) as session, + ): + await session.initialize() + + with pytest.raises(MCPError) as exc_info: + await session.read_resource(uri="foobar://should-work") + assert exc_info.value.error.code == types.CONNECTION_CLOSED + # The message embeds the auth exception's text; pin only the SDK-authored prefix. + assert exc_info.value.error.message.startswith("Failed to send message:") + + # The session survived the failed POST: the next request round-trips. + assert isinstance(await session.send_ping(), EmptyResult) + + +@pytest.mark.anyio +async def test_sse_client_post_error_after_reader_closed_is_contained() -> None: + """A failing POST whose error can no longer be delivered — the read stream already closed + with the server's SSE stream — is contained: the write loop survives and later messages + still reach the server (SDK-defined teardown-race guard). Raw streams, because the race + needs the read side closed while the write side keeps sending.""" + posted: list[str] = [] + second_post = anyio.Event() + + async def handle_sse(request: Request) -> StreamingResponse: + async def stream() -> AsyncGenerator[str, None]: + # The stream ends right after the endpoint event: the client's reader + # observes EOF and closes the read stream. + yield "event: endpoint\ndata: /messages/\n\n" + + return StreamingResponse(stream(), media_type="text/event-stream") + + async def handle_post(request: Request) -> Response: + posted.append(json.loads(await request.body())["method"]) + if len(posted) == 2: + second_post.set() + return Response(status_code=500) + + app = Starlette(routes=[Route("/sse", handle_sse), Route("/messages/", handle_post, methods=["POST"])]) + factory = in_process_client_factory(app) + with anyio.fail_after(5): + async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as (read, write): + # Wait for the reader to observe the server's EOF and close the read stream. + with pytest.raises(anyio.EndOfStream): + await read.receive() + + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="first/call", params={}))) + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=2, method="second/call", params={}))) + await second_post.wait() + # The first POST's undeliverable error was contained; the second still went out. + assert posted == ["first/call", "second/call"] + + +@pytest.mark.anyio +async def test_sse_client_notification_post_http_error_leaves_session_usable() -> None: + """A non-2xx on a notification's message POST resolves no caller (a notification has no + waiter) and leaves the session usable for subsequent requests (SDK-defined; #2110).""" + factory = in_process_client_factory(make_app_rejecting_posts({"notifications/cancelled": 500})) + with anyio.fail_after(5): + # One parenthesized async-with: separately nested ones trip a phantom + # branch arc under coverage on Python 3.14 (see the note in mcp.client.sse). + async with ( + sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams, + ClientSession(*streams) as session, + ): + await session.initialize() + + # Fire-and-forget: the rejected POST must neither raise nor stall the writer. + await session.send_notification( + types.CancelledNotification(params=types.CancelledNotificationParams(request_id=999)) + ) + + # The write loop is serialized, so this request's POST happens strictly after + # the rejected one; its success proves the failure was contained. + assert isinstance(await session.send_ping(), EmptyResult) + + @pytest.mark.anyio async def test_sse_client_basic_connection_mounted_app() -> None: """The SSE transport works unchanged when its app is mounted under a sub-path."""