From ce03a03fef1f2d734e0831ae6a0a4dff6533c9f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:35:56 +0800 Subject: [PATCH 1/5] fix(client): reinitialize expired streamable HTTP sessions --- src/mcp/client/_transport.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/mcp/client/_transport.py b/src/mcp/client/_transport.py index 0163fef950..868151abe1 100644 --- a/src/mcp/client/_transport.py +++ b/src/mcp/client/_transport.py @@ -3,7 +3,7 @@ from __future__ import annotations from contextlib import AbstractAsyncContextManager -from typing import Protocol +from typing import Final, Protocol from mcp.shared._stream_protocols import ReadStream, WriteStream from mcp.shared.message import SessionMessage @@ -12,6 +12,11 @@ TransportStreams = tuple[ReadStream[SessionMessage | Exception], WriteStream[SessionMessage]] +# SDK-private signal emitted by StreamableHTTPTransport after a request with an +# established MCP session receives HTTP 404. It never crosses the wire. +SESSION_EXPIRED: Final = -32003 +SESSION_EXPIRED_MARKER: Final = "mcp.client.session_expired" + class Transport(AbstractAsyncContextManager[TransportStreams], Protocol): """Protocol for MCP transports. From 053e798f1d58a76c52be6d00fa5edbb2c2f9e7c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:36:32 +0800 Subject: [PATCH 2/5] fix(client): reinitialize expired streamable HTTP sessions --- src/mcp/client/session.py | 95 ++++++++++++++++++++++++++++++++------- 1 file changed, 78 insertions(+), 17 deletions(-) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index 895339ca18..48335de4ca 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -37,7 +37,7 @@ from pydantic import BaseModel, Discriminator, Tag, TypeAdapter, ValidationError from typing_extensions import Self, TypeVar, deprecated -from mcp.client._transport import ReadStream, WriteStream +from mcp.client._transport import SESSION_EXPIRED, SESSION_EXPIRED_MARKER, ReadStream, WriteStream from mcp.client.extension import NotificationBinding, ResultClaim, UnexpectedClaimedResult from mcp.client.subscriptions import ListenRoute from mcp.shared._compat import resync_tracer @@ -415,6 +415,8 @@ def __init__( self._negotiated_version: str | None = None self._stamp: Callable[[dict[str, Any], CallOptions], None] = _preconnect_stamp self._task_group: anyio.abc.TaskGroup | None = None + self._session_recovery_lock = anyio.Lock() + self._session_generation = 0 # subscriptions/listen demux routes; membership decides ack consumption (raw listens are never registered) self._listen_routes: dict[RequestId, ListenRoute] = {} if dispatcher is not None: @@ -504,26 +506,15 @@ async def _deliver_bound_notifications( # A raising handler costs only that delivery, as in _on_notify. logger.exception("notification binding handler for %r raised", binding.method) - async def send_request( + async def _send_request_once( self, request: types.ClientRequest | types.Request[Any, Any], result_type: type[ReceiveResultT] | TypeAdapter[ReceiveResultT], - request_read_timeout_seconds: float | None = None, - metadata: ClientMessageMetadata | None = None, - progress_callback: ProgressFnT | None = None, + request_read_timeout_seconds: float | None, + metadata: ClientMessageMetadata | None, + progress_callback: ProgressFnT | None, ) -> ReceiveResultT: - """Send a request and wait for its typed result. - - Args: - metadata: Streamable HTTP resumption hints. - - Raises: - MCPError: Error response, read timeout, or connection closed. - RuntimeError: Called before entering the context manager. - ValueError: The request declares `name_param` but its params carry no string name. - pydantic.ValidationError: The server returned a result that does not - conform to the negotiated protocol version. - """ + """Send one typed request without session-expiry recovery.""" data = request.model_dump(by_alias=True, mode="json", exclude_none=True) method: str = data["method"] opts: CallOptions = {} @@ -562,6 +553,76 @@ async def send_request( return result_type.validate_python(raw, by_name=False) return result_type.model_validate(raw, by_name=False) + async def _recover_expired_session(self, generation: int) -> None: + """Reinitialize once for all requests that observed one expired legacy session.""" + async with self._session_recovery_lock: + if generation != self._session_generation: + return + self._initialize_result = None + self._discover_result = None + self._discover_server_info = None + self._negotiated_version = None + self._stamp = _preconnect_stamp + self._active_claims = {} + self._call_tool_adapter = _CallToolResultAdapter + self._x_mcp_header_maps.clear() + self._tool_output_schemas.clear() + self._tool_output_validators.clear() + await self.initialize() + self._session_generation += 1 + + async def send_request( + self, + request: types.ClientRequest | types.Request[Any, Any], + result_type: type[ReceiveResultT] | TypeAdapter[ReceiveResultT], + request_read_timeout_seconds: float | None = None, + metadata: ClientMessageMetadata | None = None, + progress_callback: ProgressFnT | None = None, + ) -> ReceiveResultT: + """Send a request and wait for its typed result. + + An established legacy Streamable HTTP session that receives a 404 is + reinitialized once and the original request is retried once, as required + by the transport specification. + + Args: + metadata: Streamable HTTP resumption hints. + + Raises: + MCPError: Error response, read timeout, connection closed, or a + repeated session-expiry response. + RuntimeError: Called before entering the context manager. + ValueError: The request declares `name_param` but its params carry no string name. + pydantic.ValidationError: The server returned a result that does not + conform to the negotiated protocol version. + """ + generation = self._session_generation + sent_on_legacy_session = self._initialize_result is not None and self._discover_result is None + try: + return await self._send_request_once( + request, + result_type, + request_read_timeout_seconds, + metadata, + progress_callback, + ) + except MCPError as exc: + error_data = exc.data + is_transport_expiry = ( + isinstance(error_data, Mapping) + and cast(Mapping[str, object], error_data).get(SESSION_EXPIRED_MARKER) is True + ) + if exc.code != SESSION_EXPIRED or not is_transport_expiry or not sent_on_legacy_session: + raise + await self._recover_expired_session(generation) + return await self._send_request_once( + request, + result_type, + request_read_timeout_seconds, + metadata, + progress_callback, + ) + async def send_notification(self, notification: types.ClientNotification) -> None: """Send a one-way notification. Usable before entering the context manager. From 9436aceb0cc4826a06fbe15014d6dc78de13523c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:36:36 +0800 Subject: [PATCH 3/5] fix(client): reinitialize expired streamable HTTP sessions --- src/mcp/client/streamable_http.py | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..db19b986c5 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -30,7 +30,7 @@ from mcp_types.version import MODERN_PROTOCOL_VERSIONS from pydantic import ValidationError -from mcp.client._transport import TransportStreams +from mcp.client._transport import SESSION_EXPIRED, SESSION_EXPIRED_MARKER, TransportStreams 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 @@ -359,13 +359,24 @@ async def _handle_post_request(self, ctx: RequestContext) -> None: 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: + request_session_id = headers.get(MCP_SESSION_ID) + if request_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. + # this method); session recovery 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") + # A post-session 404 means the server discarded this + # request's session. Clear only if this request still + # owns the current generation: a delayed old 404 must + # not erase a session another request just recovered. + if self.session_id == request_session_id: + self.session_id = None + error_data = ErrorData( + code=SESSION_EXPIRED, + message="Session expired", + data={SESSION_EXPIRED_MARKER: True}, + ) else: error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) From 0dbac726a36b2d9f5ba0567822134aeef268ae56 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:36:39 +0800 Subject: [PATCH 4/5] fix(client): reinitialize expired streamable HTTP sessions --- tests/client/test_notification_response.py | 180 ++++++++++++++++++++- 1 file changed, 175 insertions(+), 5 deletions(-) diff --git a/tests/client/test_notification_response.py b/tests/client/test_notification_response.py index b21e734fa3..1f9168a5e9 100644 --- a/tests/client/test_notification_response.py +++ b/tests/client/test_notification_response.py @@ -6,6 +6,7 @@ import json +import anyio import httpx2 import mcp_types as types import pytest @@ -17,6 +18,7 @@ from mcp import ClientSession, MCPError from mcp.client import IncomingMessage +from mcp.client._transport import SESSION_EXPIRED, SESSION_EXPIRED_MARKER from mcp.client.streamable_http import streamable_http_client pytestmark = pytest.mark.anyio @@ -254,10 +256,12 @@ async def test_client_falls_back_to_generic_error_when_non_2xx_body_is_a_jsonrpc assert exc.value.error.code == types.INTERNAL_ERROR -async def test_client_falls_back_to_session_terminated_when_404_body_is_malformed_json() -> None: - """SDK-defined: an unparseable ``application/json`` body on a 404 response is swallowed - and the status-derived ``INVALID_REQUEST`` (session-terminated) fallback resolves the - pending request — the parse failure never propagates.""" +async def test_client_reports_session_expiry_after_a_404_recovery_retry_has_malformed_json() -> None: + """SDK-defined: a malformed 404 body still triggers one session recovery attempt before failing. + + The parse failure is not surfaced because HTTP 404 is the transport's session-expiry signal. A second + 404 after recovery is bounded and returns the private session-expired error rather than looping. + """ app = _create_non_2xx_json_body_app(404, b"not valid json{{{") async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client: async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream): @@ -265,4 +269,170 @@ async def test_client_falls_back_to_session_terminated_when_404_body_is_malforme await session.initialize() with pytest.raises(MCPError) as exc: await session.list_tools() - assert exc.value.error.code == types.INVALID_REQUEST + assert exc.value.error.code == SESSION_EXPIRED + + +def _create_expired_session_recovery_app(requests: list[tuple[str, str | None]]) -> Starlette: + """Return a fresh session after rejecting one established-session request.""" + initialize_count = 0 + expire_once = True + + async def handle_mcp_request(request: Request) -> Response: + nonlocal expire_once, initialize_count + data = json.loads(await request.body()) + method = data.get("method") + session_id = request.headers.get("mcp-session-id") + requests.append((method, session_id)) + + if method == "initialize": + initialize_count += 1 + return JSONResponse( + {"jsonrpc": "2.0", "id": data["id"], "result": INIT_RESPONSE}, + headers={"mcp-session-id": f"session-{initialize_count}"}, + ) + if method == "notifications/initialized": + return Response(status_code=202) + if method == "tools/list" and expire_once: + expire_once = False + assert session_id == "session-1" + return Response(status_code=404) + if method == "tools/list": + assert session_id == "session-2" + return JSONResponse({"jsonrpc": "2.0", "id": data["id"], "result": {"tools": []}}) + return Response(status_code=500) + + return Starlette(debug=True, routes=[Route("/mcp", handle_mcp_request, methods=["POST"])]) + + +async def test_client_reinitializes_once_after_an_established_session_returns_404() -> None: + """Spec-mandated: a 404 for an established legacy session creates a fresh session and retries once. + + The recovery initialize must omit the expired session id; its initialized notification and the retried + request must carry the new id. This drives the public ``ClientSession`` API through an in-process ASGI app. + """ + requests: list[tuple[str, str | None]] = [] + app = _create_expired_session_recovery_app(requests) + + async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client: + async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream): + async with ClientSession(read_stream, write_stream) as session: + result = await session.initialize() + assert result.server_info.name == "test-non-sdk-server" + + tools = await session.list_tools() + + assert tools.tools == [] + assert requests == [ + ("initialize", None), + ("notifications/initialized", "session-1"), + ("tools/list", "session-1"), + ("initialize", None), + ("notifications/initialized", "session-2"), + ("tools/list", "session-2"), + ] + + +def _create_repeated_expired_session_app(requests: list[tuple[str, str | None]]) -> Starlette: + """Always expire requests from an established session.""" + initialize_count = 0 + + async def handle_mcp_request(request: Request) -> Response: + nonlocal initialize_count + data = json.loads(await request.body()) + method = data.get("method") + session_id = request.headers.get("mcp-session-id") + requests.append((method, session_id)) + + if method == "initialize": + initialize_count += 1 + return JSONResponse( + {"jsonrpc": "2.0", "id": data["id"], "result": INIT_RESPONSE}, + headers={"mcp-session-id": f"session-{initialize_count}"}, + ) + if method == "notifications/initialized": + return Response(status_code=202) + if method == "tools/list": + return Response(status_code=404) + return Response(status_code=500) + + return Starlette(debug=True, routes=[Route("/mcp", handle_mcp_request, methods=["POST"])]) + + +async def test_client_retries_an_expired_session_request_only_once() -> None: + """SDK-defined: a retry that also receives 404 surfaces an error instead of opening a recovery loop.""" + requests: list[tuple[str, str | None]] = [] + app = _create_repeated_expired_session_app(requests) + + async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client: + async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream): + async with ClientSession(read_stream, write_stream) as session: + await session.initialize() + with pytest.raises(MCPError) as exc_info: + await session.list_tools() + + assert exc_info.value.code == SESSION_EXPIRED + assert exc_info.value.data == {SESSION_EXPIRED_MARKER: True} + assert requests == [ + ("initialize", None), + ("notifications/initialized", "session-1"), + ("tools/list", "session-1"), + ("initialize", None), + ("notifications/initialized", "session-2"), + ("tools/list", "session-2"), + ] + + +async def test_concurrent_expired_session_requests_share_one_reinitialization() -> None: + """SDK-defined: concurrent 404 responses recover one session generation, not one per caller.""" + requests: list[tuple[str, str | None]] = [] + old_requests_started = 0 + old_requests_ready = anyio.Event() + release_old_requests = anyio.Event() + initialize_count = 0 + + async def handle_mcp_request(request: Request) -> Response: + nonlocal initialize_count, old_requests_started + data = json.loads(await request.body()) + method = data.get("method") + session_id = request.headers.get("mcp-session-id") + requests.append((method, session_id)) + + if method == "initialize": + initialize_count += 1 + return JSONResponse( + {"jsonrpc": "2.0", "id": data["id"], "result": INIT_RESPONSE}, + headers={"mcp-session-id": f"session-{initialize_count}"}, + ) + if method == "notifications/initialized": + return Response(status_code=202) + if method == "tools/list" and session_id == "session-1": + old_requests_started += 1 + if old_requests_started == 2: + old_requests_ready.set() + await release_old_requests.wait() + return Response(status_code=404) + if method == "tools/list" and session_id == "session-2": + return JSONResponse({"jsonrpc": "2.0", "id": data["id"], "result": {"tools": []}}) + return Response(status_code=500) + + app = Starlette(debug=True, routes=[Route("/mcp", handle_mcp_request, methods=["POST"])]) + results: list[types.ListToolsResult] = [] + + async def append_list_tools_result(session: ClientSession) -> None: + results.append(await session.list_tools()) + + async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client: + async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream): + async with ClientSession(read_stream, write_stream) as session: + await session.initialize() + async with anyio.create_task_group() as task_group: + task_group.start_soon(append_list_tools_result, session) + task_group.start_soon(append_list_tools_result, session) + with anyio.fail_after(5): + await old_requests_ready.wait() + release_old_requests.set() + + assert [result.tools for result in results] == [[], []] + assert requests.count(("initialize", None)) == 2 + assert requests.count(("notifications/initialized", "session-2")) == 1 + assert requests.count(("tools/list", "session-2")) == 2 From 12323d557fde69a43a46cde49b008bff59febf10 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:36:42 +0800 Subject: [PATCH 5/5] fix(client): reinitialize expired streamable HTTP sessions --- .../transports/test_client_transport_http.py | 41 ++++++++++--------- 1 file changed, 22 insertions(+), 19 deletions(-) diff --git a/tests/interaction/transports/test_client_transport_http.py b/tests/interaction/transports/test_client_transport_http.py index 625fcaad8d..caa6f66525 100644 --- a/tests/interaction/transports/test_client_transport_http.py +++ b/tests/interaction/transports/test_client_transport_http.py @@ -14,10 +14,9 @@ import mcp_types as types import pytest from inline_snapshot import snapshot -from mcp_types import INVALID_REQUEST, CallToolResult, ErrorData, ListToolsResult, TextContent, Tool +from mcp_types import CallToolResult, ListToolsResult, TextContent, Tool from starlette.types import Receive, Scope, Send -from mcp import MCPError from mcp.client.client import Client from mcp.client.streamable_http import streamable_http_client from mcp.server import Server, ServerRequestContext @@ -215,38 +214,42 @@ async def record(request: httpx2.Request) -> None: assert resumption_gets == [] -@requirement("client-transport:http:404-surfaces") -async def test_a_404_mid_session_surfaces_as_a_session_terminated_error() -> None: - """A 404 in response to a request after initialization is reported to the caller as an MCP error. +@requirement("client-transport:http:session-404-reinitialize") +async def test_a_404_mid_session_reinitializes_before_retrying_the_request() -> None: + """Spec-mandated: a request carrying an expired session id initializes a fresh session and retries once. - The spec says the client MUST start a new session in this situation; the SDK instead surfaces a - `Session terminated` error to the caller. The spec's MUST is tracked at - client-transport:http:session-404-reinitialize; this test pins the SDK's current behaviour. + The injected 404 applies only to the first established ``tools/list`` request. Recovery initialization + reaches the real server without an MCP session header, then the retry succeeds through the new session. """ server = _tooled_server() real_app = server.streamable_http_app(transport_security=NO_DNS_REBINDING_PROTECTION) - initialize_seen = anyio.Event() + expired_once = False - async def first_post_then_404(scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] == "http" and scope["method"] == "POST" and initialize_seen.is_set(): - await send({"type": "http.response.start", "status": 404, "headers": []}) - await send({"type": "http.response.body", "body": b""}) - return + async def expire_first_established_tools_list(scope: Scope, receive: Receive, send: Send) -> None: + nonlocal expired_once if scope["type"] == "http" and scope["method"] == "POST": - initialize_seen.set() + headers = dict(scope["headers"]) + if headers.get(b"mcp-session-id") is not None and not expired_once: + expired_once = True + await send({"type": "http.response.start", "status": 404, "headers": []}) + await send({"type": "http.response.body", "body": b""}) + return await real_app(scope, receive, send) async with ( server.session_manager.run(), - httpx2.AsyncClient(transport=StreamingASGITransport(first_post_then_404), base_url=BASE_URL) as http_client, + httpx2.AsyncClient( + transport=StreamingASGITransport(expire_first_established_tools_list), + base_url=BASE_URL, + ) as http_client, ): transport = streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client) with anyio.fail_after(5): # pragma: no branch async with Client(transport, mode="legacy") as client: # pragma: no branch - with pytest.raises(MCPError) as exc_info: # pragma: no branch - await client.list_tools() + result = await client.list_tools() - assert exc_info.value.error == snapshot(ErrorData(code=INVALID_REQUEST, message="Session terminated")) + assert expired_once is True + assert [tool.name for tool in result.tools] == ["echo"] def _blocking_server(started: anyio.Event, cancelled: anyio.Event) -> Server: