@@ -11,18 +11,21 @@ async def run_server():
     ```
 """
 
+import json
 import os
+import re
 import sys
 import threading
 from collections.abc import Callable
 from contextlib import asynccontextmanager, suppress
 from dataclasses import dataclass
 from io import TextIOWrapper
-from typing import BinaryIO, Literal, TextIO
+from typing import BinaryIO, Literal, TextIO, cast
 
 import anyio
 import anyio.lowlevel
 import mcp_types as types
+from pydantic import ValidationError
 
 from mcp.os.win32.utilities import rebind_std_handle_to_fd
 from mcp.shared._context_streams import create_context_streams
@@ -158,6 +161,49 @@ def release() -> None:
     return os.fdopen(private_fd, mode, closefd=False), release
 
 
+_RAW_ID_PATTERN = re.compile(r'"id"\s*:\s*(-?\d+|"(?:[^"\\]|\\.)*")')
+
+
+def _request_id_from_raw_line(line: str) -> types.RequestId | None:
+    """Best-effort recovery of a request id from a line that failed validation.
+
+    A full parse is tried first; lines too malformed for it fall back to a
+    pattern scan, since the id sits ahead of the payload that usually breaks.
+    """
+    try:
+        parsed: object = json.loads(line)
+    except (json.JSONDecodeError, RecursionError):
+        match = _RAW_ID_PATTERN.search(line)
+        if match is None:
+            return None
+        try:
+            parsed = json.loads(match.group(1))
+        except json.JSONDecodeError:
+            return None
+    if isinstance(parsed, dict):
+        parsed = cast(dict[str, object], parsed).get("id")
+    if isinstance(parsed, str) or type(parsed) is int:
+        return parsed
+    return None
+
+
+def _validation_error_response(line: str, exc: Exception) -> SessionMessage:
+    """The JSON-RPC error frame for a line that failed message validation.
+
+    The message stays bare: the validation error text embeds the offending
+    input, which can be arbitrarily large.
+    """
+    if isinstance(exc, ValidationError) and any(error["type"] == "json_invalid" for error in exc.errors()):
+        code, message = types.PARSE_ERROR, "Parse error"
+    else:
+        code, message = types.INVALID_REQUEST, "Invalid request"
+    return SessionMessage(
+        types.JSONRPCError(
+            jsonrpc="2.0", id=_request_id_from_raw_line(line), error=types.ErrorData(code=code, message=message)
+        )
+    )
+
+
 @asynccontextmanager
 async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.AsyncFile[str] | None = None):
     """Serve MCP over the process's stdin and stdout.
@@ -188,7 +234,10 @@ 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)
+                            # A failed line must still complete its request: send the
+                            # error frame on the wire instead of surfacing a bare
+                            # exception the server dispatcher has no observer for.
+                            await write_stream.send(_validation_error_response(line, exc))
                             continue
 
                         session_message = SessionMessage(message)
 
@@ -1,5 +1,6 @@
 import gc
 import io
+import json
 import os
 import sys
 import threading
@@ -13,8 +14,12 @@
 from mcp_types import (
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
+    INVALID_REQUEST,
+    PARSE_ERROR,
     PROTOCOL_VERSION_META_KEY,
     SERVER_INFO_META_KEY,
+    ErrorData,
+    JSONRPCError,
     JSONRPCMessage,
     JSONRPCRequest,
     JSONRPCResponse,
@@ -75,26 +80,130 @@ async def test_stdio_server_round_trips_messages_over_injected_streams() -> None
     assert received_responses[1] == JSONRPCResponse(jsonrpc="2.0", id=4, result={})
 
 
+async def _serve_raw_lines(lines: list[str]) -> tuple[list[SessionMessage | Exception], list[JSONRPCMessage]]:
+    """Feed raw stdin lines through `stdio_server()`; return (read-stream items, parsed stdout frames).
+
+    Raw lines are the only way to exercise malformed input: the typed API cannot produce it.
+    """
+    stdin = io.StringIO("".join(line + "\n" for line in lines))
+    stdout = io.StringIO()
+
+    received: list[SessionMessage | Exception] = []
+    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:
+                try:
+                    while True:
+                        received.append(await read_stream.receive())
+                except anyio.EndOfStream:
+                    pass
+            await write_stream.aclose()
+
+    stdout.seek(0)
+    sent = [jsonrpc_message_adapter.validate_json(line.strip()) for line in stdout.readlines()]
+    return received, sent
+
+
 @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."""
+    """Non-UTF-8 stdin bytes are answered with a parse error frame without killing the stream.
+
+    Invalid bytes are replaced with U+FFFD, fail JSON parsing, and complete as a
+    JSON-RPC parse error with `id: null` (no id can be recovered); subsequent
+    valid messages are still processed.
+    """
     valid = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
     raw_stdin = io.BytesIO(b"\xff\xfe\n" + valid.model_dump_json(by_alias=True, exclude_none=True).encode() + b"\n")
+    stdout = io.StringIO()
 
     # stdio_server()'s default path wraps sys.stdin.buffer with errors='replace'.
     monkeypatch.setattr(sys, "stdin", TextIOWrapper(raw_stdin, encoding="utf-8"))
-    monkeypatch.setattr(sys, "stdout", TextIOWrapper(io.BytesIO(), encoding="utf-8"))
 
     with anyio.fail_after(5):
-        async with stdio_server() as (read_stream, write_stream):
-            await write_stream.aclose()
+        async with stdio_server(stdout=anyio.AsyncFile(stdout)) as (read_stream, write_stream):
             async with read_stream:  # pragma: no branch
-                first = await read_stream.receive()
-                assert isinstance(first, Exception)
+                received = await read_stream.receive()
+                assert isinstance(received, SessionMessage)
+                assert received.message == valid
+            await write_stream.aclose()
 
-                second = await read_stream.receive()
-                assert isinstance(second, SessionMessage)
-                assert second.message == valid
+    stdout.seek(0)
+    sent = [jsonrpc_message_adapter.validate_json(line.strip()) for line in stdout.readlines()]
+    assert sent == [JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error"))]
+
+
+@pytest.mark.anyio
+async def test_stdio_server_deeply_nested_request_completes_with_a_parse_error() -> None:
+    """A request too deeply nested to validate completes with a parse error carrying its id.
+
+    Regression for #2751: the failure used to surface only as a bare exception on
+    the read stream, which the server dispatcher drops, leaving the request id
+    pending until the client timed out.
+    """
+    params: object = {"leaf": True}
+    for index in reversed(range(256)):
+        params = {f"p{index}": params}
+    deep_line = json.dumps({"jsonrpc": "2.0", "id": 900256, "method": "ping", "params": params})
+    shallow = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+
+    received, sent = await _serve_raw_lines([deep_line, shallow.model_dump_json(by_alias=True, exclude_none=True)])
+
+    # The shallow request after the failed one still arrives; the failed one does not.
+    assert len(received) == 1
+    assert isinstance(received[0], SessionMessage)
+    assert received[0].message == shallow
+    assert sent == [JSONRPCError(jsonrpc="2.0", id=900256, error=ErrorData(code=PARSE_ERROR, message="Parse error"))]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("line", "expected"),
+    [
+        # Truncated JSON: the id is recovered from the raw line, ahead of the break.
+        (
+            '{"jsonrpc":"2.0","id":"bad-1","method":"ping","params":[',
+            JSONRPCError(jsonrpc="2.0", id="bad-1", error=ErrorData(code=PARSE_ERROR, message="Parse error")),
+        ),
+        (
+            '{"jsonrpc":"2.0","id":42,"method":"ping","params":[',
+            JSONRPCError(jsonrpc="2.0", id=42, error=ErrorData(code=PARSE_ERROR, message="Parse error")),
+        ),
+        # No id anywhere in the line: the parse error falls back to `id: null`.
+        (
+            "not json",
+            JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error")),
+        ),
+        # An id whose string token is itself invalid JSON is not recoverable either.
+        (
+            '{"jsonrpc":"2.0","id":"a\\q","method":"ping"}',
+            JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error")),
+        ),
+        # Valid JSON that is not a valid message: invalid request, id preserved where present.
+        (
+            '{"jsonrpc":"2.0","id":"bad-method","method":1}',
+            JSONRPCError(
+                jsonrpc="2.0", id="bad-method", error=ErrorData(code=INVALID_REQUEST, message="Invalid request")
+            ),
+        ),
+        (
+            '{"jsonrpc":"2.0","id":null,"method":1}',
+            JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=INVALID_REQUEST, message="Invalid request")),
+        ),
+        (
+            "[]",
+            JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=INVALID_REQUEST, message="Invalid request")),
+        ),
+    ],
+)
+async def test_stdio_server_unparseable_line_completes_with_an_error_frame(line: str, expected: JSONRPCError) -> None:
+    """A line that fails validation is answered on stdout with a single JSON-RPC error frame."""
+    received, sent = await _serve_raw_lines([line])
+
+    assert received == []
+    assert sent == [expected]
 
 
 @contextmanager
@@ -608,14 +717,19 @@ def close(self) -> None:
 
 
 def _serve_stdio_and_collect(
-    monkeypatch: pytest.MonkeyPatch, server: MCPServer, frames: list[JSONRPCRequest], responses: int
+    monkeypatch: pytest.MonkeyPatch,
+    server: MCPServer,
+    frames: list[JSONRPCRequest],
+    responses: int,
+    raw_prefix: bytes = b"",
 ) -> list[JSONRPCMessage]:
     """Serve `frames` over process stdio and return the parsed response lines.
 
     Runs the blocking server.run("stdio") in a daemon thread and signals stdin EOF only after
     `responses` lines arrive - as a real client would - so handlers never race EOF cancellation.
+    `raw_prefix` prepends untyped wire bytes (e.g. a malformed line) the typed frames cannot express.
     """
-    payload = "".join(f.model_dump_json(by_alias=True, exclude_none=True) + "\n" for f in frames).encode()
+    payload = raw_prefix + "".join(f.model_dump_json(by_alias=True, exclude_none=True) + "\n" for f in frames).encode()
     stdin = _GatedStdin(payload)
     stdout = _NotifyingStdout()
     monkeypatch.setattr(sys, "stdin", TextIOWrapper(stdin, encoding="utf-8"))
@@ -643,6 +757,31 @@ def test_mcpserver_run_stdio_serves_until_stdin_closes(monkeypatch: pytest.Monke
     assert responses == [JSONRPCResponse(jsonrpc="2.0", id=1, result={})]
 
 
+def test_mcpserver_run_stdio_answers_a_deeply_nested_request_with_a_parse_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A deeply nested request over real stdio gets a parse error response instead of hanging.
+
+    End-to-end regression for #2751 through `MCPServer.run("stdio")`: the client
+    waiting on the nested request's id receives a response, and a later shallow
+    request on the same connection is still served.
+    """
+    params: object = {"leaf": True}
+    for index in reversed(range(256)):
+        params = {f"p{index}": params}
+    deep_line = (json.dumps({"jsonrpc": "2.0", "id": 900256, "method": "ping", "params": params}) + "\n").encode()
+    ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+
+    responses = _serve_stdio_and_collect(
+        monkeypatch, MCPServer(name="DeepRequestStdioServer"), [ping], 2, raw_prefix=deep_line
+    )
+
+    assert responses == [
+        JSONRPCError(jsonrpc="2.0", id=900256, error=ErrorData(code=PARSE_ERROR, message="Parse error")),
+        JSONRPCResponse(jsonrpc="2.0", id=1, result={}),
+    ]
+
+
 def test_mcpserver_run_stdio_runs_lifespan_cleanup_after_stdin_closes(monkeypatch: pytest.MonkeyPatch) -> None:
     """Code after `yield` in a lifespan runs when stdin EOF ends `run("stdio")`.