From 15b394c438a55f9ec6bdd8639b739555ed4bd8e5 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 05:27:13 +0000 Subject: [PATCH 1/8] fix(client): surface HTTP errors on resumption GET and SSE message POST Two client-side paths still swallowed non-2xx HTTP responses, leaving the caller hanging with no way to tell an auth failure from a slow server (#2110): - streamable HTTP: a non-2xx on the resumption GET (Last-Event-ID) hit a bare raise_for_status() inside the request's background task; the escaping HTTPStatusError tore down the transport's task group and every stream with it. - SSE transport: a non-2xx on the message POST raised into post_writer's catch-all, which logged and dropped it; the waiting caller hung forever and the write loop died. Both paths now resolve the waiting request with a JSON-RPC error correlated to its id, mirroring _handle_post_request's existing non-2xx handling: the caller gets a prompt INTERNAL_ERROR and the transport/session stays usable. A non-2xx on a notification POST has no waiter to resolve, so it is logged and contained. Regression tests drive both transports in-process (httpx MockTransport / ASGI) at 401/403/500 and pin that the error is correlated, prompt, and non-fatal to the session; all fail (hang into fail_after) without the fix. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL --- src/mcp/client/sse.py | 18 ++++++- src/mcp/client/streamable_http.py | 17 +++++-- tests/client/test_streamable_http.py | 42 +++++++++++++++ tests/shared/test_sse.py | 76 ++++++++++++++++++++++++++++ 4 files changed, 146 insertions(+), 7 deletions(-) diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py index 31d0f35391..65b4430a06 100644 --- a/src/mcp/client/sse.py +++ b/src/mcp/client/sse.py @@ -121,15 +121,29 @@ async def post_writer(endpoint_url: str): async def _send_message(session_message: SessionMessage) -> None: logger.debug(f"Sending client message: {session_message}") + message = session_message.message response = await client.post( endpoint_url, - json=session_message.message.model_dump( + json=message.model_dump( by_alias=True, mode="json", exclude_unset=True, ), ) - response.raise_for_status() + if response.status_code >= 400: + # Resolve the waiting caller with an error correlated to its + # request id, mirroring the streamable-HTTP transport: raising + # here would be swallowed by the post_writer handler below and + # the caller would hang forever (#2110). A notification has no + # waiter to resolve, so the failure is only logged. + logger.error(f"Message POST returned HTTP status {response.status_code}") + if isinstance(message, types.JSONRPCRequest): + error_data = types.ErrorData( + code=types.INTERNAL_ERROR, message="Server returned an error response" + ) + reply = types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data) + await read_stream_writer.send(SessionMessage(reply)) + return logger.debug(f"Client message sent successfully: {response.status_code}") async for session_message in write_stream_reader: diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..a8fab2bdd1 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -248,13 +248,20 @@ 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: a resumption token is only ever attached by a + # request's 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() + if event_source.response.status_code >= 400: + # 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 = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") + reply = JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data) + await ctx.read_stream_writer.send(SessionMessage(reply)) + return logger.debug("Resumption GET SSE connection established") async for sse in event_source: # pragma: no branch diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index d21f520daf..dd992b4c1e 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,6 +32,7 @@ from starlette.types import Receive, Scope, Send from mcp.client.streamable_http import ( + LAST_EVENT_ID, MAX_RECONNECTION_ATTEMPTS, RequestContext, StreamableHTTPTransport, @@ -132,6 +134,46 @@ def handler(request: httpx2.Request) -> httpx2.Response: assert reply.message.error.code == METHOD_NOT_FOUND +@pytest.mark.anyio +@pytest.mark.parametrize("status", [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). + """ + + 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_initialize_post_clears_cached_pv_header_and_unstamped_posts_read_it() -> None: """``initialize`` discards the cached protocol-version header; every other POST reads it. diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index c27dd69db3..d9d1b0ae9e 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -32,6 +32,7 @@ from starlette.requests import Request from starlette.responses import Response from starlette.routing import Mount, Route +from starlette.types import Receive, Scope, Send import mcp.client.sse from mcp.client.session import ClientSession @@ -108,6 +109,39 @@ 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).""" + server = Server(SERVER_NAME, on_read_resource=_handle_read_resource) + sse = SseServerTransport( + "/messages/", security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False) + ) + + async def handle_sse(request: Request) -> Response: + async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream): + await server.run(read_stream, write_stream, server.create_initialization_options()) + return Response() + + 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 sse.handle_post_message(scope, replay, send) + + return Starlette( + routes=[ + Route("/sse", endpoint=handle_sse), + Mount("/messages/", app=handle_post), + ] + ) + + @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,48 @@ async def test_sse_client_exception_handling( await session.read_resource(uri="xxx://will-not-work") +@pytest.mark.anyio +@pytest.mark.parametrize("status_code", [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). + """ + factory = in_process_client_factory(make_app_rejecting_posts({"resources/read": status_code})) + with anyio.fail_after(5): + async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams: + async with 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_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): + async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams: + async with 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.""" From e5fe739c59b932729954f726d344492c4fea94fc Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 05:35:35 +0000 Subject: [PATCH 2/8] test: use one parenthesized async-with to dodge py3.14 coverage phantom arc Separately nested async-with statements trip a phantom branch arc under coverage on Python 3.14 (the artifact already noted in mcp.client.sse), failing the 3.14 CI matrix legs at 99.99%. Collapse the two context managers into a single parenthesized async-with, the form the sibling streamable-http tests already use, instead of adding a pragma. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL --- tests/shared/test_sse.py | 46 +++++++++++++++++++++++----------------- 1 file changed, 27 insertions(+), 19 deletions(-) diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index d9d1b0ae9e..0beb171221 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -267,17 +267,21 @@ async def test_sse_client_request_post_http_error_reaches_caller_and_session_sur """ factory = in_process_client_factory(make_app_rejecting_posts({"resources/read": status_code})) with anyio.fail_after(5): - async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams: - async with ClientSession(*streams) as session: - await session.initialize() + # 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") + 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) + # The session survived the failed POST: the next request round-trips. + assert isinstance(await session.send_ping(), EmptyResult) @pytest.mark.anyio @@ -286,18 +290,22 @@ async def test_sse_client_notification_post_http_error_leaves_session_usable() - 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): - async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams: - async with ClientSession(*streams) as session: - await session.initialize() + # 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)) - ) + # 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) + # 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 From b4edbd4bb777e5029f844bebc7ec09bdf7b1293b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 05:56:37 +0000 Subject: [PATCH 3/8] fix(client): harden the resumption GET and SSE POST error paths per review Address the review findings on the previous revision: - Dispatch resumption on message type as well as metadata: a notification stamped with a resumption token is POSTed as usual instead of tripping the resumption path's request-only assertion and killing the write loop. - Treat any non-2xx as a failure (response.is_success), restoring the raise_for_status() semantics the checks replaced: an unfollowed redirect resolves the caller instead of being logged as success. - Map a 404 on the resumption GET while a session id is held to INVALID_REQUEST / "Session terminated", the POST path's session-expiry signal, so reconnect logic keyed on it works across both. - Contain the resumption read loop like _handle_sse_response: a stream dying mid-read or ending cleanly without a response resolves the waiter (CONNECTION_CLOSED) instead of tearing down the transport or hanging. - Resolve the resumption GET's status errors via _resolve_abandoned_request for its closed-stream containment instead of hand-building the error. - Surface network-level errors (httpx.HTTPError) on the SSE message POST through the same correlated path: on this transport nothing escapes loudly, so the caller previously hung forever. - Deduplicate the SSE test app wiring behind make_app(wrap_post=...). Seven new regression tests pin the above; each fails against the previous revision (hang into fail_after, transport teardown, or wrong error). Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL --- src/mcp/client/sse.py | 52 ++++++---- src/mcp/client/streamable_http.py | 69 ++++++++----- tests/client/test_streamable_http.py | 145 ++++++++++++++++++++++++++- tests/shared/test_sse.py | 114 +++++++++++++++------ 4 files changed, 304 insertions(+), 76 deletions(-) diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py index 65b4430a06..6606e1c02b 100644 --- a/src/mcp/client/sse.py +++ b/src/mcp/client/sse.py @@ -120,31 +120,39 @@ 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}") message = session_message.message - response = await client.post( - endpoint_url, - json=message.model_dump( - by_alias=True, - mode="json", - exclude_unset=True, - ), - ) - if response.status_code >= 400: - # Resolve the waiting caller with an error correlated to its - # request id, mirroring the streamable-HTTP transport: raising - # here would be swallowed by the post_writer handler below and - # the caller would hang forever (#2110). A notification has no - # waiter to resolve, so the failure is only logged. + try: + response = await client.post( + endpoint_url, + json=message.model_dump( + by_alias=True, + mode="json", + exclude_unset=True, + ), + ) + except httpx2.HTTPError as exc: + 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}") - if isinstance(message, types.JSONRPCRequest): - error_data = types.ErrorData( - code=types.INTERNAL_ERROR, message="Server returned an error response" - ) - reply = types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data) - await read_stream_writer.send(SessionMessage(reply)) - return - logger.debug(f"Client message sent successfully: {response.status_code}") + error = types.ErrorData( + code=types.INTERNAL_ERROR, message="Server returned an error response" + ) + # 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) + await read_stream_writer.send(SessionMessage(reply)) 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 a8fab2bdd1..3cfa9bfed0 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -248,32 +248,51 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None: else: raise ResumptionError("Resumption request requires a resumption token") # pragma: no cover - # Only requests resume: a resumption token is only ever attached by a - # request's metadata, so the original id is always available to map responses. + # 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: - if event_source.response.status_code >= 400: - # 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 = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") - reply = JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data) - await ctx.read_stream_writer.send(SessionMessage(reply)) - return - 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). + if event_source.response.status_code == 404 and self.session_id is not None: + # The GET carried our Mcp-Session-Id, so a 404 is the session-expiry + # signal reconnect logic keys on - same mapping as the POST path. + await self._resolve_abandoned_request( + ctx.read_stream_writer, original_request_id, "Session terminated", code=INVALID_REQUEST + ) + return + await self._resolve_abandoned_request( + ctx.read_stream_writer, + original_request_id, + "Server returned an error response", + code=INTERNAL_ERROR, + ) + 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". @@ -563,8 +582,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 dd992b4c1e..74c37299bd 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -34,6 +34,7 @@ from mcp.client.streamable_http import ( LAST_EVENT_ID, MAX_RECONNECTION_ATTEMPTS, + MCP_SESSION_ID, RequestContext, StreamableHTTPTransport, streamable_http_client, @@ -135,11 +136,12 @@ def handler(request: httpx2.Request) -> httpx2.Response: @pytest.mark.anyio -@pytest.mark.parametrize("status", [401, 403, 500]) +@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: @@ -174,6 +176,79 @@ def handler(request: httpx2.Request) -> httpx2.Response: 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. @@ -674,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 0beb171221..d3b69f2085 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 @@ -32,7 +33,7 @@ from starlette.requests import Request from starlette.responses import Response from starlette.routing import Mount, Route -from starlette.types import Receive, Scope, Send +from starlette.types import ASGIApp, Receive, Scope, Send import mcp.client.sse from mcp.client.session import ClientSession @@ -83,8 +84,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. @@ -97,10 +101,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), ] ) @@ -112,34 +120,23 @@ def make_server_app() -> Starlette: 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).""" - server = Server(SERVER_NAME, on_read_resource=_handle_read_resource) - sse = SseServerTransport( - "/messages/", security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False) - ) - async def handle_sse(request: Request) -> Response: - async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream): - await server.run(read_stream, write_stream, server.create_initialization_options()) - return Response() + 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 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} - async def replay() -> dict[str, Any]: - return {"type": "http.request", "body": body, "more_body": False} + await inner(scope, replay, send) - await sse.handle_post_message(scope, replay, send) + return handle_post - return Starlette( - routes=[ - Route("/sse", endpoint=handle_sse), - Mount("/messages/", app=handle_post), - ] - ) + return make_app(Server(SERVER_NAME, on_read_resource=_handle_read_resource), wrap_post=wrap) @pytest.mark.anyio @@ -259,11 +256,12 @@ async def test_sse_client_exception_handling( @pytest.mark.anyio -@pytest.mark.parametrize("status_code", [401, 403, 500]) +@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): @@ -284,6 +282,64 @@ async def test_sse_client_request_post_http_error_reaches_caller_and_session_sur 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_notification_post_http_error_leaves_session_usable() -> None: """A non-2xx on a notification's message POST resolves no caller (a notification has no From 51d99af6147ea0b2657fc44bf9f09a50e2e74c77 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 06:08:39 +0000 Subject: [PATCH 4/8] fix(client): cover OAuth failures and 404 session expiry on the SSE POST; document the new error contract Address the second review round: - Widen the SSE message POST's failure catch to (httpx.HTTPError, OAuthFlowError): an OAuthClientProvider re-auth failing inside client.post() previously took the same swallowed path and hung the waiting caller forever. - Map a 404 on the SSE message POST to INVALID_REQUEST / "Session terminated" when the endpoint URL carries a session id (the SSE analogue of the streamable transport's session check); keep the generic error when it does not. - Document the changed error behavior in docs/migration.md: resumption GET and SSE message POST outcome tables, and scope the "connect-level failures still escape" sentence to the streamable message POST, the one place it still holds. Three new regression tests; the OAuth and 404-session ones fail against the previous revision. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL --- docs/migration.md | 21 +++++++++- src/mcp/client/sse.py | 19 +++++++-- tests/shared/test_sse.py | 85 +++++++++++++++++++++++++++++++++++++++- 3 files changed, 119 insertions(+), 6 deletions(-) diff --git a/docs/migration.md b/docs/migration.md index b094d79f84..9eb80cebef 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* 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 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. ### `terminate_windows_process` removed diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py index 6606e1c02b..f7b8d637c1 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.auth.exceptions import OAuthFlowError 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 @@ -136,7 +137,9 @@ async def _send_message(session_message: SessionMessage) -> None: exclude_unset=True, ), ) - except httpx2.HTTPError as exc: + except (httpx2.HTTPError, OAuthFlowError) as exc: + # OAuthFlowError: OAuthClientProvider re-auth failing inside + # client.post() must resolve the waiter like any network error. logger.exception("Error POSTing message") error = types.ErrorData( code=types.CONNECTION_CLOSED, message=f"Failed to send message: {exc}" @@ -146,9 +149,17 @@ async def _send_message(session_message: SessionMessage) -> None: logger.debug(f"Client message sent successfully: {response.status_code}") return logger.error(f"Message POST returned HTTP status {response.status_code}") - error = types.ErrorData( - code=types.INTERNAL_ERROR, message="Server returned an error response" - ) + if ( + response.status_code == 404 + and _extract_session_id_from_endpoint(endpoint_url) is not None + ): + # The endpoint URL carries the session id, so a 404 is the + # session-expiry signal - same mapping as streamable HTTP. + error = types.ErrorData(code=types.INVALID_REQUEST, message="Session terminated") + else: + error = types.ErrorData( + code=types.INTERNAL_ERROR, message="Server returned an error response" + ) # 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) diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index d3b69f2085..134fd972bb 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -31,11 +31,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 @@ -340,6 +341,88 @@ def factory( 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 +async def test_sse_client_oauth_failure_on_post_reaches_caller_and_session_survives() -> None: + """An SDK OAuth flow failure raised from inside 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 — like any network error, it used to be swallowed inside post_writer).""" + + class _RefusingAuth(httpx2.Auth): + """Stands in for OAuthClientProvider whose re-auth fails 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 OAuthTokenError("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_notification_post_http_error_leaves_session_usable() -> None: """A non-2xx on a notification's message POST resolves no caller (a notification has no From 6472241329b623e050f4df2c01493f9233543004 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 06:26:42 +0000 Subject: [PATCH 5/8] fix(client): contain arbitrary POST failures and dedupe the status-error mapping Address the third review round: - Broaden the SSE message POST's failure catch to a terminal containment boundary (except Exception): user-supplied auth flows and hooks raise arbitrary types from inside client.post(), so an enumerated catch cannot keep the caller from hanging. - Extract the status -> JSON-RPC error mapping into mcp.client._transport.status_error_data and use it from the message POST handler, the resumption GET, and the SSE POST; the message POST keeps its pre-session 404 -> METHOD_NOT_FOUND case locally. Wire-identical. - Contain the SSE error-resolution send against a concurrently closed read stream (BrokenResourceError/ClosedResourceError -> debug log), mirroring _resolve_abandoned_request, so the teardown race cannot kill the write loop. Tests: the auth-failure test is parametrized over OAuthTokenError and RuntimeError, and a raw-stream teardown-race test pins that a failing POST whose error is undeliverable leaves the write loop serving later messages. Both fail against the previous revision. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL --- src/mcp/client/_transport.py | 14 ++++++++ src/mcp/client/sse.py | 35 +++++++++++--------- src/mcp/client/streamable_http.py | 33 +++++++------------ tests/shared/test_sse.py | 54 ++++++++++++++++++++++++++++--- 4 files changed, 93 insertions(+), 43 deletions(-) diff --git a/src/mcp/client/_transport.py b/src/mcp/client/_transport.py index 0163fef950..943f37bca8 100644 --- a/src/mcp/client/_transport.py +++ b/src/mcp/client/_transport.py @@ -5,6 +5,8 @@ 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 @@ -13,6 +15,18 @@ 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 f7b8d637c1..2d96d75a92 100644 --- a/src/mcp/client/sse.py +++ b/src/mcp/client/sse.py @@ -10,7 +10,7 @@ from anyio.abc import TaskStatus from httpx2 import SSEError -from mcp.client.auth.exceptions import OAuthFlowError +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 @@ -137,9 +137,11 @@ async def _send_message(session_message: SessionMessage) -> None: exclude_unset=True, ), ) - except (httpx2.HTTPError, OAuthFlowError) as exc: - # OAuthFlowError: OAuthClientProvider re-auth failing inside - # client.post() must resolve the waiter like any network error. + 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}" @@ -149,21 +151,22 @@ async def _send_message(session_message: SessionMessage) -> None: logger.debug(f"Client message sent successfully: {response.status_code}") return logger.error(f"Message POST returned HTTP status {response.status_code}") - if ( - response.status_code == 404 - and _extract_session_id_from_endpoint(endpoint_url) is not None - ): - # The endpoint URL carries the session id, so a 404 is the - # session-expiry signal - same mapping as streamable HTTP. - error = types.ErrorData(code=types.INVALID_REQUEST, message="Session terminated") - else: - error = types.ErrorData( - code=types.INTERNAL_ERROR, message="Server returned an error response" - ) + # 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) - await read_stream_writer.send(SessionMessage(reply)) + 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 3cfa9bfed0..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 @@ -259,18 +258,11 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None: # 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). - if event_source.response.status_code == 404 and self.session_id is not None: - # The GET carried our Mcp-Session-Id, so a 404 is the session-expiry - # signal reconnect logic keys on - same mapping as the POST path. - await self._resolve_abandoned_request( - ctx.read_stream_writer, original_request_id, "Session terminated", code=INVALID_REQUEST - ) - return + 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, - "Server returned an error response", - code=INTERNAL_ERROR, + ctx.read_stream_writer, original_request_id, error_data.message, code=error_data.code ) return logger.debug("Resumption GET SSE connection established") @@ -384,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 diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index 134fd972bb..f53a5fc91b 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -19,6 +19,7 @@ EmptyResult, Implementation, InitializeResult, + JSONRPCRequest, JSONRPCResponse, ListToolsResult, PaginatedRequestParams, @@ -44,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" @@ -392,17 +394,21 @@ async def handle_post(request: Request) -> Response: @pytest.mark.anyio -async def test_sse_client_oauth_failure_on_post_reaches_caller_and_session_survives() -> None: - """An SDK OAuth flow failure raised from inside a request's message POST reaches the waiting - caller promptly as a JSON-RPC error correlated to the request, and the session stays usable +@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 whose re-auth fails mid-session.""" + """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 OAuthTokenError("re-authentication failed") + raise exc_type("re-authentication failed") yield request factory = in_process_client_factory(make_server_app()) @@ -423,6 +429,44 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx 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 From af776219789ea2efe92a4947cebdf6bdd7b7f146 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 06:38:26 +0000 Subject: [PATCH 6/8] docs: scope the connect-failure escape note to request POSTs The sentence claimed connect-level failures on any streamable HTTP message POST still escape the transport context; that is only true for a request's POST (spawned on the transport task group). A notification or response POST runs inside post_writer's guarded loop: the failure is logged, does not escape, and kills the write loop - pre-existing behavior, now stated instead of implied away. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL --- docs/migration.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/migration.md b/docs/migration.md index 9eb80cebef..f54628ac87 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -2249,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)). On the streamable HTTP transport, connect-level failures such as `httpx2.ConnectError` on a 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. +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 From d25e6f8bcbd9bab117cbce9a2c99e90fa629f358 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 06:46:30 +0000 Subject: [PATCH 7/8] fix(client): export status_error_data in _transport's __all__ The module keeps an exhaustive export list; the helper is imported by sse.py and streamable_http.py. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL --- src/mcp/client/_transport.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mcp/client/_transport.py b/src/mcp/client/_transport.py index 943f37bca8..9b0c6451b3 100644 --- a/src/mcp/client/_transport.py +++ b/src/mcp/client/_transport.py @@ -10,7 +10,7 @@ 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]] From 66a928547c160e4aca20a558d28aa50c08151426 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 06:57:18 +0000 Subject: [PATCH 8/8] docs: align the SSE notification-POST sentence with its streamable sibling The SSE paragraph said "a notification"; the write loop also carries response POSTs, exactly as the streamable sentence already states. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL --- docs/migration.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/migration.md b/docs/migration.md index f54628ac87..78b713020a 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -2205,7 +2205,7 @@ The SSE transport (`sse_client`) applies the same rule to its message POST. In v | 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* has no caller to resolve; v2 logs and drops it, keeping the write loop (and every later send) alive. +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`.