@@ -54,7 +54,7 @@
 from mcp.server.runner import modern_error_data, serve_one
 from mcp.server.streamable_http import check_accept_headers
 from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
-from mcp.shared.dispatcher import CallOptions
+from mcp.shared.dispatcher import CallOptions, request_id_in
 from mcp.shared.exceptions import NoBackChannelError
 from mcp.shared.inbound import (
     ERROR_CODE_HTTP_STATUS,
@@ -413,7 +413,10 @@ async def handle_modern_request(
     except ValidationError:
         # A batch, a posted response (clients MUST NOT send those: streamable-http
         # §Sending Messages item 4), or a request whose envelope is malformed.
-        await _write(_INVALID_BODY, scope, receive, send)
+        # Echo the original request id so envelope-invalid failures correlate.
+        request_id = request_id_in(decoded)
+        rej = JSONRPCError(jsonrpc="2.0", id=request_id, error=_INVALID_BODY.error)
+        await _write(rej, scope, receive, send)
         return
 
     if req.method == "subscriptions/listen" and not has_sse:
 
@@ -26,6 +26,7 @@ async def run_server():
 
 from mcp.os.win32.utilities import rebind_std_handle_to_fd
 from mcp.shared._context_streams import create_context_streams
+from mcp.shared.jsonrpc_dispatcher import UnparseableMessageError
 from mcp.shared.message import SessionMessage
 
 if sys.platform != "win32":  # pragma: no branch
@@ -188,7 +189,7 @@ async def stdin_reader():
                         try:
                             message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
                         except Exception as exc:
-                            await read_stream_writer.send(exc)
+                            await read_stream_writer.send(UnparseableMessageError(line, cause=exc))
                             continue
 
                         session_message = SessionMessage(message)
 
@@ -22,7 +22,6 @@
 from mcp_types import (
     DEFAULT_NEGOTIATED_VERSION,
     INTERNAL_ERROR,
-    INVALID_PARAMS,
     INVALID_REQUEST,
     PARSE_ERROR,
     ErrorData,
@@ -43,6 +42,7 @@
 from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
 from mcp.shared._stream_protocols import ReadStream, WriteStream
+from mcp.shared.dispatcher import request_id_in
 from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER
 from mcp.shared.message import CloseSSEStreamCallback, ServerMessageMetadata, SessionMessage
 
@@ -369,6 +369,7 @@ def _create_error_response(
         status_code: HTTPStatus,
         error_code: int = INVALID_REQUEST,
         headers: dict[str, str] | None = None,
+        request_id: RequestId | None = None,
     ) -> Response:
         """Create an error response with a simple string message."""
         response_headers = {"Content-Type": CONTENT_TYPE_JSON}
@@ -381,7 +382,7 @@ def _create_error_response(
         # Return a properly formatted JSON error response
         error_response = JSONRPCError(
             jsonrpc="2.0",
-            id=None,
+            id=request_id,
             error=ErrorData(code=error_code, message=error_message),
         )
 
@@ -546,10 +547,13 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
             try:
                 message = jsonrpc_message_adapter.validate_python(raw_message, by_name=False)
             except ValidationError as e:
+                # Echo the original request id so envelope-invalid failures correlate (#2848).
+                request_id = request_id_in(raw_message)
                 response = self._create_error_response(
                     f"Validation error: {str(e)}",
                     HTTPStatus.BAD_REQUEST,
-                    INVALID_PARAMS,
+                    INVALID_REQUEST,
+                    request_id=request_id,
                 )
                 await response(scope, receive, send)
                 return
 
@@ -40,6 +40,7 @@
     "ProgressFnT",
     "as_request_id",
     "coerce_request_id",
+    "request_id_in",
     "run_notify_intercept",
 ]
 
@@ -53,6 +54,16 @@ def as_request_id(value: object) -> RequestId | None:
     return None
 
 
+def request_id_in(message: Any) -> RequestId | None:
+    """A decoded JSON-RPC message's top-level request id, or None when the message is
+    not an object or carries no scalar string/int id."""
+    try:
+        rid: Any = message.get("id")
+    except AttributeError:
+        return None
+    return as_request_id(rid)
+
+
 def coerce_request_id(request_id: RequestId) -> RequestId:
     """Coerce a stringified int request id back to int so a peer-echoed id still correlates (matches the TS SDK).
 
 
@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import contextvars
+import json
 import logging
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass, field
@@ -22,6 +23,7 @@
     CONNECTION_CLOSED,
     INTERNAL_ERROR,
     INVALID_PARAMS,
+    INVALID_REQUEST,
     REQUEST_TIMEOUT,
     ErrorData,
     JSONRPCError,
@@ -49,6 +51,7 @@
     ProgressFnT,
     as_request_id,
     coerce_request_id,
+    request_id_in,
     run_notify_intercept,
 )
 from mcp.shared.exceptions import MCPError, NoBackChannelError
@@ -62,6 +65,7 @@
 
 __all__ = [
     "JSONRPCDispatcher",
+    "UnparseableMessageError",
     "cancelled_request_id_from_params",
     "handler_exception_to_error_data",
     "progress_token_from_params",
@@ -85,6 +89,33 @@
 answered - the handler's eventual result or error is dropped, not written."""
 
 
+class UnparseableMessageError(Exception):
+    """A transport failed to decode an inbound frame as a JSON-RPC message.
+
+    Transports send this instead of a bare parse exception so the receiving
+    side can still correlate an error response with the frame's JSON-RPC
+    request id, when one is recoverable from the raw payload.
+    """
+
+    def __init__(self, payload: str | bytes | None = None, *, cause: BaseException | None = None) -> None:
+        super().__init__(
+            f"failed to decode inbound frame: {cause!r}" if cause is not None else "failed to decode inbound frame"
+        )
+        self.payload = payload
+        self.__cause__ = cause
+
+    @property
+    def request_id(self) -> RequestId | None:
+        """Best-effort recovery of the frame's top-level request id, else None."""
+        if self.payload is None:
+            return None
+        try:
+            decoded = json.loads(self.payload)
+        except (ValueError, RecursionError):
+            return None
+        return request_id_in(decoded)
+
+
 def handler_exception_to_error_data(exc: BaseException) -> ErrorData | None:
     """Map a handler-raised exception to its wire `ErrorData`.
 
@@ -537,6 +568,17 @@ async def _dispatch(
         """
         if isinstance(item, Exception):
             if self.on_stream_exception is None:
+                # Answer an envelope-invalid frame whose request id is still
+                # recoverable; bare exceptions stay dropped as before.
+                request_id = item.request_id if isinstance(item, UnparseableMessageError) else None
+                if request_id is not None:
+                    self._spawn(
+                        self._write_error,
+                        request_id,
+                        ErrorData(code=INVALID_REQUEST, message="Invalid Request"),
+                        sender_ctx=sender_ctx,
+                    )
+                    return
                 logger.debug("transport yielded exception: %r", item)
                 return
             try:
 
@@ -15,6 +15,7 @@
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
     INVALID_PARAMS,
+    INVALID_REQUEST,
     PARSE_ERROR,
     PROTOCOL_VERSION_META_KEY,
     UNSUPPORTED_PROTOCOL_VERSION,
@@ -134,7 +135,7 @@ async def test_non_json_content_type_is_rejected() -> None:
 @requirement("hosting:http:parse-error-400")
 @requirement("hosting:http:batch")
 async def test_malformed_and_batched_bodies_return_400() -> None:
-    """A non-JSON body returns 400 Parse error; a JSON array of requests returns 400 Invalid params."""
+    """A non-JSON body returns 400 Parse error; a JSON array of requests returns 400 Invalid Request."""
     async with mounted_app(_server()) as (http, _):
         session_id = await initialize_via_http(http)
         not_json = await http.post(
@@ -154,7 +155,7 @@ async def test_malformed_and_batched_bodies_return_400() -> None:
     assert not_json.status_code == 400
     assert JSONRPCError.model_validate_json(not_json.text).error.code == PARSE_ERROR
     assert batched.status_code == 400
-    assert JSONRPCError.model_validate_json(batched.text).error.code == INVALID_PARAMS
+    assert JSONRPCError.model_validate_json(batched.text).error.code == INVALID_REQUEST
 
 
 @requirement("hosting:http:protocol-version-400")
 
@@ -181,9 +181,10 @@ async def test_modern_notification_post_is_acknowledged_202_and_a_posted_respons
     assert (acknowledged.status_code, acknowledged.content) == (202, b"")
     assert "mcp-session-id" not in acknowledged.headers
     assert refused.status_code == 400
+    # The posted response's own id is echoed so the client can correlate the refusal.
     assert JSONRPCError.model_validate(refused.json()) == JSONRPCError(
         jsonrpc="2.0",
-        id=None,
+        id=1,
         error=ErrorData(code=INVALID_REQUEST, message="Body must be a single JSON-RPC request or notification object"),
     )
 
 
@@ -24,6 +24,7 @@
 
 from mcp.server.mcpserver import MCPServer
 from mcp.server.stdio import stdio_server
+from mcp.shared.jsonrpc_dispatcher import UnparseableMessageError
 from mcp.shared.message import SessionMessage
 
 
@@ -75,6 +76,39 @@ async def test_stdio_server_round_trips_messages_over_injected_streams() -> None
     assert received_responses[1] == JSONRPCResponse(jsonrpc="2.0", id=4, result={})
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("line", "expected_id"),
+    [
+        pytest.param('{"jsonrpc": "1.0", "id": 3, "method": "ping", "params": {}}', 3, id="wrong-jsonrpc-version"),
+        pytest.param('{"id": 4, "method": "ping", "params": {}}', 4, id="missing-jsonrpc-field"),
+        pytest.param('{"jsonrpc": "2.0", "id": 8, "method": 12345, "params": {}}', 8, id="non-string-method"),
+    ],
+)
+async def test_stdio_server_envelope_invalid_line_surfaces_with_recoverable_request_id(
+    line: str, expected_id: int
+) -> None:
+    """A line that is valid JSON but not a valid JSON-RPC envelope surfaces as an
+    UnparseableMessageError carrying the raw payload, so the session can still
+    correlate an error response with the request's original id."""
+    stdin = io.StringIO(line + "\n")
+    stdout = io.StringIO()
+
+    with anyio.fail_after(5):
+        async with stdio_server(stdin=anyio.AsyncFile(stdin), stdout=anyio.AsyncFile(stdout)) as (
+            read_stream,
+            write_stream,
+        ):
+            async with read_stream:
+                received = await read_stream.receive()
+
+            assert isinstance(received, UnparseableMessageError)
+            assert received.request_id == expected_id
+
+            # Closing write_stream ends stdout_writer so the server context can join.
+            await write_stream.aclose()
+
+
 @pytest.mark.anyio
 async def test_stdio_server_invalid_utf8(monkeypatch: pytest.MonkeyPatch) -> None:
     """Non-UTF-8 stdin bytes surface as an in-stream exception without killing the stream."""
 
@@ -171,30 +171,35 @@ async def test_handle_modern_request_rejects_a_notification_post_at_an_unserved_
 
 
 @pytest.mark.parametrize(
-    "body",
+    ("body", "expected_id"),
     [
-        pytest.param({"jsonrpc": "2.0", "id": 1, "result": {}}, id="posted-response"),
-        pytest.param({"jsonrpc": "2.0", "id": 1, "error": {"code": -1, "message": "x"}}, id="posted-error"),
-        pytest.param([{"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 1}}], id="batch"),
-        pytest.param({"jsonrpc": "2.0", "id": None, "method": "tools/list"}, id="null-id-request"),
-        pytest.param({"jsonrpc": "2.0", "id": [1], "method": "tools/list"}, id="non-scalar-id-request"),
-        pytest.param({"jsonrpc": "2.0", "method": 7}, id="non-string-method-notification"),
-        pytest.param({"jsonrpc": "1.0", "method": "notifications/cancelled"}, id="wrong-jsonrpc-version"),
-        pytest.param("just a string", id="scalar"),
+        pytest.param({"jsonrpc": "2.0", "id": 1, "result": {}}, 1, id="posted-response"),
+        pytest.param({"jsonrpc": "2.0", "id": 1, "error": {"code": -1, "message": "x"}}, 1, id="posted-error"),
+        pytest.param(
+            [{"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 1}}], None, id="batch"
+        ),
+        pytest.param({"jsonrpc": "2.0", "id": None, "method": "tools/list"}, None, id="null-id-request"),
+        pytest.param({"jsonrpc": "2.0", "id": [1], "method": "tools/list"}, None, id="non-scalar-id-request"),
+        pytest.param({"jsonrpc": "2.0", "method": 7}, None, id="non-string-method-notification"),
+        pytest.param({"jsonrpc": "1.0", "method": "notifications/cancelled"}, None, id="wrong-jsonrpc-version"),
+        pytest.param("just a string", None, id="scalar"),
     ],
 )
-async def test_handle_modern_request_rejects_a_body_that_is_neither_request_nor_notification(body: Any) -> None:
+async def test_handle_modern_request_rejects_a_body_that_is_neither_request_nor_notification(
+    body: Any, expected_id: int | None
+) -> None:
     """Spec-mandated (streamable-http §Sending Messages item 4): the body MUST be a single request
     or notification and clients MUST NOT post responses. SDK-defined: anything else -- a posted
     response, a batch, a request whose `id` is malformed, a scalar -- is `INVALID_REQUEST` at
-    HTTP 400 with `id: null`, distinct from `PARSE_ERROR` (malformed JSON). A malformed-`id`
-    request in particular must not be mistaken for a notification and silently 202'd."""
+    HTTP 400 with the original id echoed when one is recoverable (else `id: null`), distinct
+    from `PARSE_ERROR` (malformed JSON). A malformed-`id` request in particular must not be
+    mistaken for a notification and silently 202'd."""
     async with _asgi_client(Server("test")) as http:
         response = await http.post("/mcp", json=body)
     assert response.status_code == 400
     assert response.json() == {
         "jsonrpc": "2.0",
-        "id": None,
+        "id": expected_id,
         "error": {"code": INVALID_REQUEST, "message": "Body must be a single JSON-RPC request or notification object"},
     }
 
@@ -216,6 +221,27 @@ async def test_handle_modern_request_rejects_malformed_body_with_parse_error() -
     }
 
 
+@pytest.mark.parametrize(
+    ("body", "expected_id"),
+    [
+        pytest.param({"jsonrpc": "1.0", "id": 3, "method": "ping", "params": {}}, 3, id="wrong-jsonrpc-version"),
+        pytest.param({"id": 4, "method": "ping", "params": {}}, 4, id="missing-jsonrpc-field"),
+        pytest.param({"jsonrpc": "2.0", "id": 8, "method": 12345, "params": {}}, 8, id="non-string-method"),
+    ],
+)
+async def test_handle_modern_request_envelope_invalid_request_echoes_the_original_id(
+    body: dict[str, Any], expected_id: int
+) -> None:
+    """An envelope-invalid but id-bearing request is answered -32600 with the
+    original request id, so the client can correlate the failure."""
+    async with _asgi_client(Server("test")) as http:
+        response = await http.post("/mcp", json=body)
+    assert response.status_code == 400
+    error = response.json()
+    assert error["id"] == expected_id
+    assert error["error"]["code"] == INVALID_REQUEST
+
+
 async def test_handle_modern_request_returns_transport_security_error_response() -> None:
     """The transport-security middleware's error response is sent verbatim and short-circuits."""
     settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["good.example"])