diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 82de50fd05..fa82566115 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -352,83 +352,100 @@ async def _handle_post_request(self, ctx: RequestContext) -> None: if ctx.metadata is not None and ctx.metadata.headers is not None: headers.update(ctx.metadata.headers) - async with stream_within_origin( - ctx.client, - "POST", - self.url, - json=message.model_dump(by_alias=True, mode="json", exclude_unset=True), - headers=headers, - ) as response: - if response.status_code == 202: - logger.debug("Received 202 Accepted") - if isinstance(message, JSONRPCRequest): - # A request's response arrives on this POST's body; 202 says - # none will follow. Resolve rather than park the caller forever. - await self._resolve_abandoned_request( - ctx.read_stream_writer, - message.id, - "server answered a request with 202 Accepted", - code=INVALID_REQUEST, - ) - return + try: + async with stream_within_origin( + ctx.client, + "POST", + self.url, + json=message.model_dump(by_alias=True, mode="json", exclude_unset=True), + headers=headers, + ) as response: + if response.status_code == 202: + logger.debug("Received 202 Accepted") + if isinstance(message, JSONRPCRequest): + # A request's response arrives on this POST's body; 202 says + # none will follow. Resolve rather than park the caller forever. + await self._resolve_abandoned_request( + ctx.read_stream_writer, + message.id, + "server answered a request with 202 Accepted", + code=INVALID_REQUEST, + ) + return - if (redirect := _unfollowed_redirect(response)) is not None: - logger.warning(redirect) - if isinstance(message, JSONRPCRequest): - await self._resolve_abandoned_request( - ctx.read_stream_writer, message.id, redirect, code=INVALID_REQUEST - ) - return + if (redirect := _unfollowed_redirect(response)) is not None: + logger.warning(redirect) + if isinstance(message, JSONRPCRequest): + await self._resolve_abandoned_request( + ctx.read_stream_writer, message.id, redirect, code=INVALID_REQUEST + ) + return - if response.status_code >= 400: - if isinstance(message, JSONRPCRequest): - # A spec-correct server may return the JSON-RPC error in the - # body at a non-2xx status (e.g. 400 for INVALID_PARAMS, 404 - # for METHOD_NOT_FOUND). Surface that error rather than the - # status-derived stand-in below. - if response.headers.get("content-type", "").lower().startswith("application/json"): - try: - body = await response.aread() - parsed = jsonrpc_message_adapter.validate_json(body, by_name=False) - if isinstance(parsed, JSONRPCError): - # The server may have set `id: null` (request rejected before its - # id was parsed); use this request's id so correlation works. - reply = JSONRPCError(jsonrpc="2.0", id=message.id, error=parsed.error) - await ctx.read_stream_writer.send(SessionMessage(reply)) - return - 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") + if response.status_code >= 400: + if isinstance(message, JSONRPCRequest): + # A spec-correct server may return the JSON-RPC error in the + # body at a non-2xx status (e.g. 400 for INVALID_PARAMS, 404 + # for METHOD_NOT_FOUND). Surface that error rather than the + # status-derived stand-in below. + if response.headers.get("content-type", "").lower().startswith("application/json"): + try: + body = await response.aread() + parsed = jsonrpc_message_adapter.validate_json(body, by_name=False) + if isinstance(parsed, JSONRPCError): + # The server may have set `id: null` (request rejected before its + # id was parsed); use this request's id so correlation works. + reply = JSONRPCError(jsonrpc="2.0", id=message.id, error=parsed.error) + await ctx.read_stream_writer.send(SessionMessage(reply)) + return + 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") else: - error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated") - 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)) - await ctx.read_stream_writer.send(session_message) - return + 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)) + await ctx.read_stream_writer.send(session_message) + return - if self._is_initialization_request(message): - self._maybe_extract_session_id_from_response(response) + if self._is_initialization_request(message): + self._maybe_extract_session_id_from_response(response) - # Per https://modelcontextprotocol.io/specification/2025-06-18/basic#notifications: - # The server MUST NOT send a response to notifications. + # Per https://modelcontextprotocol.io/specification/2025-06-18/basic#notifications: + # The server MUST NOT send a response to notifications. + if isinstance(message, JSONRPCRequest): + content_type = response.headers.get("content-type", "").lower() + if content_type.startswith("application/json"): + await self._handle_json_response(response, ctx.read_stream_writer, request_id=message.id) + elif content_type.startswith("text/event-stream"): + await self._handle_sse_response(response, ctx) + else: + logger.error(f"Unexpected content type: {content_type}") + error_data = ErrorData(code=INVALID_REQUEST, message=f"Unexpected content type: {content_type}") + error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) + await ctx.read_stream_writer.send(error_msg) + + except httpx2.TransportError as exc: + # The HTTP exchange itself failed (connection dropped before a + # response, connect/read errors, timeouts). This must fail only + # the request that was in flight: resolve its waiter with a + # synthesized error so the session and its transport survive, + # mirroring how non-resumable SSE drops are handled above. if isinstance(message, JSONRPCRequest): - content_type = response.headers.get("content-type", "").lower() - if content_type.startswith("application/json"): - await self._handle_json_response(response, ctx.read_stream_writer, request_id=message.id) - elif content_type.startswith("text/event-stream"): - await self._handle_sse_response(response, ctx) - else: - logger.error(f"Unexpected content type: {content_type}") - error_data = ErrorData(code=INVALID_REQUEST, message=f"Unexpected content type: {content_type}") - error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) - await ctx.read_stream_writer.send(error_msg) + logger.warning(f"HTTP transport error on POST for request {message.id}: {exc!r}") + await self._resolve_abandoned_request( + ctx.read_stream_writer, + message.id, + f"HTTP transport error: {exc}", + ) + elif isinstance(message, JSONRPCNotification): + logger.warning(f"HTTP transport error on POST for {message.method}: {exc!r}") async def _handle_json_response( self, diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index c6e62ad94a..ee971ddc18 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -968,3 +968,39 @@ async def test_https_endpoint_redirected_to_plain_http_elsewhere_never_suggests_ The server is likely behind a TLS-terminating proxy whose forwarded headers it does not trust, often combined with a trailing-slash difference. Try https://backend.lan:8000/mcp/ instead, or fix the proxy settings.\ """) + + +@pytest.mark.anyio +async def test_a_post_transport_error_fails_only_that_request_and_keeps_the_session() -> None: + """A POST whose HTTP exchange itself fails (server drops the connection before + responding) must resolve only that request with an error; the session and its + transport stay usable, matching the TypeScript client's recovery behavior.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + body = json.loads(request.content) + if body.get("id") == "drop-1": + raise httpx2.RemoteProtocolError("Server disconnected without sending a response.") + return httpx2.Response( + 200, + headers={"content-type": "application/json"}, + 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(JSONRPCRequest(jsonrpc="2.0", id="drop-1", method="tools/call", params={}))) + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="after-1", method="ping"))) + first = await read.receive() + second = await read.receive() + + assert isinstance(first, SessionMessage) + assert isinstance(first.message, JSONRPCError) + assert first.message.id == "drop-1" + assert first.message.error.code == CONNECTION_CLOSED + + assert isinstance(second, SessionMessage) + assert isinstance(second.message, JSONRPCResponse) + assert second.message.id == "after-1"