diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 82de50fd05..6fd1ef8ab3 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -671,7 +671,9 @@ async def terminate_session(self, client: httpx2.AsyncClient) -> None: if response.status_code == 405: logger.debug("Server does not allow session termination") - elif response.status_code not in (200, 204): + elif not 200 <= response.status_code < 300: + # Any 2xx is a successful termination; servers may answer an + # asynchronous delete with 202 Accepted (spec only carves out 405). logger.warning(f"Session termination failed: {response.status_code}") # pragma: no cover except Exception as exc: # pragma: no cover logger.warning(f"Session termination failed: {exc}") diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py index 655d9941dc..a153d9d8d3 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -1207,6 +1207,56 @@ async def __aexit__(self, *args: Any) -> None: assert exc_info.value.error.message == "Session not found" +@pytest.mark.anyio +@pytest.mark.parametrize("delete_status,expect_warning", [(202, False), (500, True)]) +async def test_streamable_http_client_session_termination_status_handling( + basic_app: Starlette, caplog: pytest.LogCaptureFixture, delete_status: int, expect_warning: bool +) -> None: + """Any 2xx DELETE response is a successful termination; only other statuses warn. + + 202 Accepted is a valid answer for a server that processes termination asynchronously; + a warning there is a false positive (see #3546). 500 is a real failure and still warns. + """ + + class AnswerDeleteWithStatus(httpx2.AsyncBaseTransport): + def __init__(self, inner: StreamingASGITransport, status: int) -> None: + self.inner = inner + self.status = status + + async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response: + response = await self.inner.handle_async_request(request) + if request.method != "DELETE" or response.status_code != 200: + return response + await response.aread() + return httpx2.Response(self.status, headers=response.headers, request=request) + + async def __aenter__(self) -> AnswerDeleteWithStatus: + await self.inner.__aenter__() + return self + + async def __aexit__(self, *args: Any) -> None: + await self.inner.__aexit__(*args) + + httpx_client, _ = create_session_id_capturing_client( + basic_app, transport=AnswerDeleteWithStatus(StreamingASGITransport(basic_app), delete_status) + ) + + with caplog.at_level(logging.WARNING, logger="mcp.client.streamable_http"): + async with httpx_client: + async with streamable_http_client(f"{BASE_URL}/mcp", http_client=httpx_client) as ( + read_stream, + write_stream, + ): + async with ClientSession(read_stream, write_stream) as session: # pragma: no branch + await session.initialize() + await session.list_tools() + + warning_logged = "Session termination failed" in caplog.text + assert warning_logged is expect_warning, ( + f"delete_status={delete_status}: warning_logged={warning_logged}, expected={expect_warning}" + ) + + @pytest.mark.anyio async def test_streamable_http_client_resumption(event_app: tuple[SimpleEventStore, Starlette]) -> None: """A second client resumes an interrupted request with a resumption token and receives the rest."""