diff --git a/docs/get-started/real-host.md b/docs/get-started/real-host.md
index 159c1f56f..3a1b67bbe 100644
--- a/docs/get-started/real-host.md
+++ b/docs/get-started/real-host.md
@@ -165,7 +165,7 @@ Once that command sits and waits, what's left is almost always one of three thin
 
 * **A relative path.** The host launches your server from *its* working directory, not the one you registered from. `server.py` where `/absolute/path/to/server.py` is needed is the single most common failure. If the host can't find `uv` either, that path has to be absolute too.
 * **The host is still running its old config.** Hosts read their config at launch. Claude Desktop in particular has to be *fully quit* (not just its window closed) and reopened before an edit to `claude_desktop_config.json` takes effect.
-* **Something reached stdout.** On stdio, stdout *is* the protocol. One stray `print()` and the host reads a corrupt message and drops the connection. Log with the `logging` module, which writes to stderr. **[Logging](../handlers/logging.md)** has the whole story.
+* **Something reached stdout before serving began.** On stdio, stdout *is* the protocol. The SDK diverts stray output to stderr while serving, but anything that reaches stdout before then (a wrapper script echoing, an import-time `print()` in an unbuffered process) hands the host a corrupt message and it drops the connection. Log with the `logging` module, which writes to stderr. **[Logging](../handlers/logging.md)** has the whole story.
 
 Claude Desktop keeps a log per server: `mcp-server-<NAME>.log` is your server's stderr, next to `mcp.log` for connections, under `~/Library/Logs/Claude` on macOS and `%APPDATA%\Claude\logs` on Windows.
 
diff --git a/docs/handlers/logging.md b/docs/handlers/logging.md
index 945aa60d5..fa8a305f4 100644
--- a/docs/handlers/logging.md
+++ b/docs/handlers/logging.md
@@ -33,8 +33,9 @@ For a **stdio** server this question matters more than usual. The host launched
 The standard library already does the right thing: log output goes to `sys.stderr` by default. Your `logger.info(...)` lines land in the terminal (or wherever the host collects the subprocess's stderr), and the protocol stream stays clean.
 
 !!! tip
-    Never `print()` in a stdio server. `print` writes to **stdout**, and stdout *is* the wire: one stray
-    line and the client is trying to parse it as JSON-RPC.
+    Don't `print()` in a stdio server. `print` writes to **stdout**, and stdout belongs to the protocol:
+    while serving, the SDK diverts stray stdout to stderr so it can't corrupt the wire, but that leaves
+    your line interleaved raw among the log output, with no level, no logger name, and no way to filter it.
 
     `logger.debug("got here")` is the same one line of effort and goes to the right place.
 
@@ -72,7 +73,7 @@ went to standard error: the terminal, not the wire.
 * The MCP protocol's logging capability is deprecated by the 2026-07-28 spec and not replaced. Don't build on it.
 * `logger = logging.getLogger(__name__)` at module level, `logger.info(...)` in the tool. That's the whole pattern.
 * Log output never reaches the model. Only the value you `return` does.
-* Standard error is yours; stdout belongs to the protocol. Never `print()` in a stdio server.
+* Standard error is yours; stdout belongs to the protocol. The SDK diverts a stray `print()` to stderr while serving, but it arrives unlabeled; use `logging`.
 * `MCPServer(..., log_level="DEBUG")` sets the level, and a logging configuration you made first is left alone.
 
 Telling connected clients that something on your server changed (the tool list, a resource) is **[Subscriptions](subscriptions.md)**.
diff --git a/docs/migration.md b/docs/migration.md
index 876608db7..5697d1460 100644
--- a/docs/migration.md
+++ b/docs/migration.md
@@ -1855,6 +1855,22 @@ group (spawned with `start_new_session=True`); the `getpgid()` lookup and the
 per-process terminate/kill fallback are gone. The win32 utilities logger is now
 named `mcp.os.win32.utilities` (was `client.stdio.win32`).
 
+### `stdio_server` keeps the protocol streams on private descriptors
+
+While serving, the stdio transport moves the wire to private descriptors and points
+fd 0 at the null device and fd 1 at stderr, restoring both on exit. Subprocesses and
+handler code can no longer read protocol bytes or write into the stream (the
+[#671](https://github.com/modelcontextprotocol/python-sdk/issues/671) fix). Ordinary
+servers have nothing to do, and code that inspects or manipulates fd 0/1 directly
+during a session now sees the diversions, not the wire.
+
+One pattern needs migrating: a watchdog thread that `poll()`s fd 0 for `POLLHUP` to
+detect a vanished client will no longer fire, because the null device never reports
+it (a POSIX-specific pattern; `select.poll` does not exist on Windows). Watch the
+parent process instead: on POSIX, exit when `os.getppid()` changes, which happens
+when the client dies because orphaned processes are reparented. That works on both
+v1 and v2 and does not depend on descriptor layout.
+
 ### WebSocket transport removed
 
 The WebSocket transport has been removed: `mcp.client.websocket.websocket_client`, `mcp.server.websocket.websocket_server`, and the `ws` optional dependency extra (`mcp[ws]`) no longer exist. WebSocket was never part of the MCP specification. Use the streamable HTTP transport instead (`mcp.client.streamable_http.streamable_http_client` on the client, `streamable_http_app()` on the server), which supports bidirectional communication with server-to-client streaming over standard HTTP.
diff --git a/docs/run/index.md b/docs/run/index.md
index f92090fee..8e88b2d9e 100644
--- a/docs/run/index.md
+++ b/docs/run/index.md
@@ -39,31 +39,7 @@ python server.py
 
 Nothing prints, and it doesn't return. It is waiting on stdin for a host to speak first.
 
-That also means stdout **is the wire**. A stray `print()` corrupts the stream; the `logging` module writes to stderr and is the right tool. That story is in **[Logging](../handlers/logging.md)**.
-
-On Windows, the same rule applies to child processes your tools start. A child
-that inherits the stdio server's stdin can block behind the server's protocol
-reader. If your tool starts a subprocess and you do not intend to feed it input,
-redirect the child's stdin:
-
-```python
-import asyncio
-import subprocess
-import sys
-
-
-async def run_script() -> tuple[bytes, bytes]:
-    process = await asyncio.create_subprocess_exec(
-        sys.executable,
-        "script.py",
-        stdin=subprocess.DEVNULL,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-    )
-    return await process.communicate()
-```
-
-The matching troubleshooting entry is **[My stdio tool hangs when it starts a subprocess on Windows](../troubleshooting.md#my-stdio-tool-hangs-when-it-starts-a-subprocess-on-windows)**.
+That also means stdout **is the wire**. While serving, the SDK moves the wire to a private descriptor and diverts stray output (a `print()`, or a subprocess writing to its inherited stdout) to stderr, where it can't corrupt the stream. Output that reaches stdout *before* serving begins (a wrapper script echoing, an unbuffered import-time print) still lands on the wire. For output you actually want, the `logging` module is the right tool. That story is in **[Logging](../handlers/logging.md)**.
 
 ### Try it
 
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
index 127944835..9fea1e7dc 100644
--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -137,45 +137,10 @@ There is no error string for this, which is exactly why it is hard to search. Th
 * **Is the tool on the `mcp` the host is running?** A second `MCPServer(...)` in another module is a different, empty server. Check which object the host's command actually imports.
 * **Did two tools share a name?** Then one of them is gone. Look for `Tool already exists:` in the server log.
 * **Is the host's list stale?** Adding a tool after startup only reaches clients that handle `notifications/tools/list_changed`. Restarting the host is the blunt fix.
-* **Did something write to `stdout`?** On a stdio transport, stdout *is* the protocol: one stray `print()` and the host drops the connection, which some hosts render as a server with nothing in it. Log with the `logging` module instead. The rest of the host-side checklist is on **[Connect to a real host](get-started/real-host.md)**.
+* **Did something write to `stdout` before the server started serving?** While serving, the SDK diverts stray stdout to stderr (best-effort: an environment that replaces the standard streams is served as-is), but output flushed to stdout earlier (a wrapper script echoing, an import-time `print()` in an unbuffered process) lands on the protocol stream, and one junk line can make the host drop the connection, which some hosts render as a server with nothing in it. Log with the `logging` module instead. The rest of the host-side checklist is on **[Connect to a real host](get-started/real-host.md)**.
 
 An "invalid" tool name is *not* on that list: a non-conforming name logs a warning but the tool is registered and listed anyway.
 
-## My stdio tool hangs when it starts a subprocess on Windows
-
-Your server is running over `stdio`, and a tool starts another process with
-`asyncio.create_subprocess_exec`, `asyncio.create_subprocess_shell`, or
-`subprocess.Popen`. The tool call never returns on Windows, while the same code
-works over an HTTP transport.
-
-The child inherited the server's stdin. In a stdio server, stdin is the protocol
-pipe and the server is already waiting on it for the next JSON-RPC message. A
-Python child process on Windows can block during startup when it inherits that
-same pipe.
-
-If you do not intend to send input to the child, redirect its stdin:
-
-```python
-import asyncio
-import subprocess
-import sys
-
-
-async def run_script() -> tuple[bytes, bytes]:
-    process = await asyncio.create_subprocess_exec(
-        sys.executable,
-        "script.py",
-        stdin=subprocess.DEVNULL,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-    )
-    return await process.communicate()
-```
-
-Use the same idea with `subprocess.Popen(..., stdin=subprocess.DEVNULL)`. Also
-capture or redirect the child's stdout. The stdio server's stdout is the MCP
-wire, so a child that writes there can corrupt the connection.
-
 ## `MCPError: Server returned an error response`
 
 The server refused the HTTP request outright, with a body that is not JSON-RPC, so the python `Client` has nothing better to show you than this stand-in.
diff --git a/src/mcp/os/win32/utilities.py b/src/mcp/os/win32/utilities.py
index 1cc867d4f..d5a642971 100644
--- a/src/mcp/os/win32/utilities.py
+++ b/src/mcp/os/win32/utilities.py
@@ -1,4 +1,4 @@
-"""Windows-specific functionality for stdio client operations."""
+"""Windows-specific functionality for stdio transport operations."""
 
 import logging
 import shutil
@@ -20,14 +20,36 @@
     import pywintypes
     import win32api
     import win32con
+    import win32file
     import win32job
 else:
     # Type stubs for non-Windows platforms
     win32api = None
     win32con = None
+    win32file = None
     win32job = None
     pywintypes = None
 
+
+def rebind_std_handle_to_fd(fd: int) -> None:
+    """Points the Win32 standard-handle slot for fd 0, 1, or 2 at fd's current OS handle.
+
+    os.dup2 updates only the CRT descriptor table; subprocess handle inheritance
+    reads the Win32 slot, so it must be repointed too.
+
+    Raises:
+        OSError: The slot could not be set.
+    """
+    if sys.platform != "win32" or not win32api or not win32file or not pywintypes:
+        return
+    std_ids = {0: win32api.STD_INPUT_HANDLE, 1: win32api.STD_OUTPUT_HANDLE, 2: win32api.STD_ERROR_HANDLE}
+    try:
+        win32api.SetStdHandle(std_ids[fd], win32file._get_osfhandle(fd))
+    except pywintypes.error as exc:
+        # Normalized so callers' OSError-based best-effort handling covers it.
+        raise OSError(f"SetStdHandle failed for fd {fd}") from exc
+
+
 # How often FallbackProcess polls the underlying Popen for exit.
 _EXIT_POLL_INTERVAL = 0.01
 
diff --git a/src/mcp/server/stdio.py b/src/mcp/server/stdio.py
index 876d256dd..d6916fa53 100644
--- a/src/mcp/server/stdio.py
+++ b/src/mcp/server/stdio.py
@@ -1,15 +1,9 @@
-"""Stdio Server Transport Module
-
-This module provides functionality for creating an stdio-based transport layer
-that can be used to communicate with an MCP client through standard input/output
-streams.
+"""Stdio server transport for MCP.
 
 Example:
     ```python
     async def run_server():
         async with stdio_server() as (read_stream, write_stream):
-            # read_stream contains incoming JSONRPCMessages from stdin
-            # write_stream allows sending JSONRPCMessages to stdout
             server = await create_my_server()
             await server.run(read_stream, write_stream, init_options)
 
@@ -17,61 +11,169 @@ async def run_server():
     ```
 """
 
+import os
 import sys
-from contextlib import asynccontextmanager
+import threading
+from collections.abc import Callable
+from contextlib import asynccontextmanager, suppress
 from io import TextIOWrapper
+from typing import BinaryIO, Literal, TextIO
 
 import anyio
 import anyio.lowlevel
 import mcp_types as types
 
+from mcp.os.win32.utilities import rebind_std_handle_to_fd
 from mcp.shared._context_streams import create_context_streams
 from mcp.shared.message import SessionMessage
 
+# fds whose real stream a serving transport owns, whether diverted or served in place.
+_claimed: set[int] = set()
+_claimed_lock = threading.Lock()
+
+
+def _is_backed_by_fd(stream: TextIO, fd: int) -> bool:
+    try:
+        return stream.buffer.fileno() == fd
+    except (AttributeError, OSError, ValueError):
+        return False
+
+
+def _std_descriptors_open() -> bool:
+    """Whether fds 0-2 are all open, so a dup cannot land in the standard range."""
+    try:
+        for fd in (0, 1, 2):
+            os.fstat(fd)
+    except OSError:
+        return False
+    return True
+
+
+def _open_stdin_diversion() -> int:
+    return os.open(os.devnull, os.O_RDONLY)
+
+
+def _open_stdout_diversion() -> int:
+    try:
+        return os.dup(2)
+    except OSError:
+        return os.open(os.devnull, os.O_WRONLY)
+
+
+def _claim_fd(
+    fd: int, stream: TextIO, mode: Literal["rb", "wb"], open_diversion: Callable[[], int]
+) -> tuple[BinaryIO, Callable[[], None] | None]:
+    """Move the protocol pipe to a private descriptor and divert fd while serving.
+
+    Raises:
+        RuntimeError: fd is already claimed by another transport in this process.
+    """
+    if not _is_backed_by_fd(stream, fd):
+        return stream.buffer, None
+    with _claimed_lock:
+        if fd in _claimed:
+            raise RuntimeError(f"another stdio_server() in this process has already claimed fd {fd}")
+        _claimed.add(fd)
+
+    def unclaim() -> None:
+        with _claimed_lock:
+            _claimed.discard(fd)
+
+    if not _std_descriptors_open():
+        return stream.buffer, unclaim
+    private_fd = None
+    try:
+        private_fd = os.dup(fd)
+        diversion_fd = open_diversion()
+        try:
+            os.dup2(diversion_fd, fd)
+        finally:
+            os.close(diversion_fd)
+        if sys.platform == "win32":  # pragma: no cover
+            rebind_std_handle_to_fd(fd)
+    except OSError:
+        if private_fd is not None:
+            _restore_fd(fd, private_fd)
+            os.close(private_fd)
+        return stream.buffer, unclaim
+
+    def restore() -> None:
+        # Drain text buffered during the claim (a stray print) to the diversion.
+        with suppress(OSError, ValueError):
+            stream.flush()
+        # A failed restore leaves fd diverted; keep it claimed so later transports are refused.
+        if _restore_fd(fd, private_fd):
+            unclaim()
+
+    # closefd=False: a blocked worker thread may still read this descriptor after exit.
+    return os.fdopen(private_fd, mode, closefd=False), restore
+
+
+def _restore_fd(fd: int, private_fd: int) -> bool:
+    """Point fd back at the protocol pipe; a failure never masks the transport's exit."""
+    try:
+        os.dup2(private_fd, fd)
+        if sys.platform == "win32":  # pragma: no cover
+            rebind_std_handle_to_fd(fd)
+    except OSError:
+        return False
+    return True
+
 
 @asynccontextmanager
 async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.AsyncFile[str] | None = None):
-    """Server transport for stdio: this communicates with an MCP client by reading
-    from the current process' stdin and writing to stdout.
+    """Serve MCP over the process's stdin and stdout.
+
+    While serving, fd 0 points at the null device and fd 1 at stderr, so handlers
+    and children read EOF and their stray output misses the wire; both descriptors
+    are restored on exit. Explicit streams skip the claim, and a second concurrent
+    stdio_server() raises RuntimeError.
     """
-    # Purposely not using context managers for these, as we don't want to close
-    # standard process handles. Encoding of stdin/stdout as text streams on
-    # python is platform-dependent (Windows is particularly problematic), so we
-    # re-wrap the underlying binary stream to ensure UTF-8.
-    if not stdin:
-        stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace"))
-    if not stdout:
-        stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))
-
-    read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
-    write_stream, write_stream_reader = create_context_streams[SessionMessage](0)
-
-    async def stdin_reader():
-        try:
-            async with read_stream_writer:
-                async for line in stdin:
-                    try:
-                        message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
-                    except Exception as exc:
-                        await read_stream_writer.send(exc)
-                        continue
-
-                    session_message = SessionMessage(message)
-                    await read_stream_writer.send(session_message)
-        except anyio.ClosedResourceError:  # pragma: no cover
-            await anyio.lowlevel.checkpoint()
-
-    async def stdout_writer():
-        try:
-            async with write_stream_reader:
-                async for session_message in write_stream_reader:
-                    json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True)
-                    await stdout.write(json + "\n")
-                    await stdout.flush()
-        except anyio.ClosedResourceError:  # pragma: no cover
-            await anyio.lowlevel.checkpoint()
-
-    async with anyio.create_task_group() as tg:
-        tg.start_soon(stdin_reader)
-        tg.start_soon(stdout_writer)
-        yield read_stream, write_stream
+    # Re-wrap the binary buffers as UTF-8 text; the std handles' platform encodings are unreliable.
+    restore_stdin: Callable[[], None] | None = None
+    restore_stdout: Callable[[], None] | None = None
+    try:
+        if not stdin:
+            stdin_buffer, restore_stdin = _claim_fd(0, sys.stdin, "rb", _open_stdin_diversion)
+            stdin = anyio.wrap_file(TextIOWrapper(stdin_buffer, encoding="utf-8", errors="replace"))
+        if not stdout:
+            stdout_buffer, restore_stdout = _claim_fd(1, sys.stdout, "wb", _open_stdout_diversion)
+            stdout = anyio.wrap_file(TextIOWrapper(stdout_buffer, encoding="utf-8"))
+
+        read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
+        write_stream, write_stream_reader = create_context_streams[SessionMessage](0)
+
+        async def stdin_reader():
+            try:
+                async with read_stream_writer:
+                    async for line in stdin:
+                        try:
+                            message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
+                        except Exception as exc:
+                            await read_stream_writer.send(exc)
+                            continue
+
+                        session_message = SessionMessage(message)
+                        await read_stream_writer.send(session_message)
+            except anyio.ClosedResourceError:  # pragma: no cover
+                await anyio.lowlevel.checkpoint()
+
+        async def stdout_writer():
+            try:
+                async with write_stream_reader:
+                    async for session_message in write_stream_reader:
+                        json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True)
+                        await stdout.write(json + "\n")
+                        await stdout.flush()
+            except anyio.ClosedResourceError:  # pragma: no cover
+                await anyio.lowlevel.checkpoint()
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(stdin_reader)
+            tg.start_soon(stdout_writer)
+            yield read_stream, write_stream
+    finally:
+        if restore_stdout is not None:
+            restore_stdout()
+        if restore_stdin is not None:
+            restore_stdin()
diff --git a/tests/interaction/_requirements.py b/tests/interaction/_requirements.py
index 752c17aa4..f2a7a9ba7 100644
--- a/tests/interaction/_requirements.py
+++ b/tests/interaction/_requirements.py
@@ -3916,9 +3916,12 @@ def __post_init__(self) -> None:
         note="Only observable over stdio: stdin/stdout purity is stdio-specific.",
         divergence=Divergence(
             note=(
-                "stdio_server's own writes satisfy this, but it does not redirect or guard sys.stdout: "
-                "handler code that calls print() writes directly to the protocol stream and corrupts the "
-                "framing. The spec MUST is satisfied only as long as application code behaves."
+                "While serving, stdio_server moves the wire to private descriptors and diverts fd 0/1, so "
+                "handler code and its child processes can neither read protocol bytes nor write into the "
+                "stream (pinned by tests/server/test_stdio.py). Remaining gaps: output flushed to stdout "
+                "before the transport enters can still precede the first frame, and the claim is "
+                "best-effort - skipped for explicitly injected streams and for processes without "
+                "normal standard descriptors."
             ),
         ),
     ),
diff --git a/tests/interaction/transports/test_stdio.py b/tests/interaction/transports/test_stdio.py
index dbb7de345..2c34f5a0b 100644
--- a/tests/interaction/transports/test_stdio.py
+++ b/tests/interaction/transports/test_stdio.py
@@ -113,7 +113,8 @@ async def test_stdio_server_writes_one_jsonrpc_message_per_line() -> None:
     """Every `stdio_server` write is one valid JSON-RPC message on its own line.
 
     Each line is newline-terminated with payload newlines JSON-escaped. This proves the
-    transport's own framing; it does not guard `sys.stdout` against handler code (see the
+    transport's own framing over injected streams; the descriptor-level guard that keeps
+    handler code off the wire is pinned by tests/server/test_stdio.py (see the narrowed
     divergence on `transport:stdio:stream-purity`).
     """
     captured = io.StringIO()
diff --git a/tests/server/test_stdio.py b/tests/server/test_stdio.py
index 218e34d5a..ee6ecbfbb 100644
--- a/tests/server/test_stdio.py
+++ b/tests/server/test_stdio.py
@@ -1,11 +1,13 @@
 import io
+import os
 import sys
 import threading
-from collections.abc import AsyncIterator
-from contextlib import asynccontextmanager
+from collections.abc import AsyncIterator, Iterator
+from contextlib import asynccontextmanager, contextmanager
 from io import TextIOWrapper
 
 import anyio
+import anyio.to_thread
 import pytest
 from mcp_types import (
     CLIENT_CAPABILITIES_META_KEY,
@@ -25,11 +27,7 @@
 
 @pytest.mark.anyio
 async def test_stdio_server_round_trips_messages_over_injected_streams() -> None:
-    """stdio_server frames JSON-RPC messages as one line each in both directions.
-
-    Parses one message per stdin line and writes each outgoing message as exactly one
-    line, driven over injected in-process streams.
-    """
+    """stdio_server frames JSON-RPC messages as one line each in both directions."""
     stdin = io.StringIO()
     stdout = io.StringIO()
 
@@ -77,17 +75,11 @@ async def test_stdio_server_round_trips_messages_over_injected_streams() -> None
 
 @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.
-
-    Invalid bytes are replaced with U+FFFD, fail JSON parsing, and arrive as an in-stream
-    exception; subsequent valid messages are still processed.
-    """
-    # \xff\xfe are invalid UTF-8 start bytes.
+    """Non-UTF-8 stdin bytes surface as an in-stream exception without killing the stream."""
     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")
 
-    # Replace sys.stdin with a wrapper whose .buffer is our raw bytes, so that
-    # stdio_server()'s default path wraps it with errors='replace'.
+    # 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"))
 
@@ -95,24 +87,400 @@ async def test_stdio_server_invalid_utf8(monkeypatch: pytest.MonkeyPatch) -> Non
         async with stdio_server() as (read_stream, write_stream):
             await write_stream.aclose()
             async with read_stream:  # pragma: no branch
-                # First line: \xff\xfe -> U+FFFD U+FFFD -> JSON parse fails -> exception in stream
                 first = await read_stream.receive()
                 assert isinstance(first, Exception)
 
-                # Second line: valid message still comes through
                 second = await read_stream.receive()
                 assert isinstance(second, SessionMessage)
                 assert second.message == valid
 
 
+@contextmanager
+def _pipe_planted_on_fd0(monkeypatch: pytest.MonkeyPatch) -> Iterator[tuple[int, int]]:
+    """Plants a fresh pipe on fd 0 and rebinds sys.stdin over it; yields (read_fd, write_fd).
+
+    Close ownership: the caller closes the write end exactly once; the helper closes only what
+    it created - a double close lands on a recycled fd and destroys whatever unrelated file now
+    lives there (pytest's capture files). os.dup2 is captured up front to survive monkeypatching.
+    """
+    real_dup2 = os.dup2
+    in_r, in_w = os.pipe()
+    saved0 = os.dup(0)
+    os.dup2(in_r, 0)
+    # Created after the plant so its cached stream state describes the pipe.
+    stdin_double = TextIOWrapper(open(0, "rb", closefd=False), encoding="utf-8")
+    try:
+        monkeypatch.setattr(sys, "stdin", stdin_double)
+        yield in_r, in_w
+    finally:
+        stdin_double.close()
+        real_dup2(saved0, 0)
+        os.close(saved0)
+        os.close(in_r)
+
+
+@contextmanager
+def _pipe_planted_on_fd1(monkeypatch: pytest.MonkeyPatch) -> Iterator[tuple[int, int]]:
+    """Like _pipe_planted_on_fd0 but plants fd 1 and rebinds sys.stdout; yields (read_fd, write_fd)."""
+    real_dup2 = os.dup2
+    out_r, out_w = os.pipe()
+    saved1 = os.dup(1)
+    os.dup2(out_w, 1)
+    stdout_double = TextIOWrapper(open(1, "wb", closefd=False), encoding="utf-8")
+    try:
+        monkeypatch.setattr(sys, "stdout", stdout_double)
+        yield out_r, out_w
+    finally:
+        stdout_double.close()
+        real_dup2(saved1, 1)
+        os.close(saved1)
+        os.close(out_r)
+        os.close(out_w)
+
+
+@contextmanager
+def _pipe_planted_on_fd2() -> Iterator[int]:
+    """Like _pipe_planted_on_fd0 but plants fd 2 to observe the stdout diversion; yields the read end."""
+    real_dup2 = os.dup2
+    err_r, err_w = os.pipe()
+    saved2 = os.dup(2)
+    try:
+        os.dup2(err_w, 2)
+        yield err_r
+    finally:
+        real_dup2(saved2, 2)
+        os.close(saved2)
+        os.close(err_r)
+        os.close(err_w)
+
+
+def _frame(message: JSONRPCRequest | JSONRPCResponse) -> bytes:
+    """One JSON-RPC message as the newline-terminated wire line the transport reads."""
+    return (message.model_dump_json(by_alias=True, exclude_none=True) + "\n").encode()
+
+
+async def _read_from(fd: int) -> bytes:
+    """One os.read from fd, in a worker thread.
+
+    A regression can leave the pipe empty; abandoning turns a read that would outlive
+    fail_after on the loop thread into a red TimeoutError instead.
+    """
+    return await anyio.to_thread.run_sync(os.read, fd, 65536, abandon_on_cancel=True)
+
+
+@pytest.mark.anyio
+async def test_stdio_server_takes_stdin_off_the_descriptor_table_while_serving(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """On the real process stdin, the transport claims the protocol pipe and releases it on exit.
+
+    SDK-defined behavior: while serving, fd 0 is the null device, so an inheriting child
+    cannot consume protocol bytes or, on Windows, hang at startup (CPython gh-78961).
+    """
+    with _pipe_planted_on_fd0(monkeypatch) as (in_r, in_w):
+        out_r, out_w = os.pipe()
+        stdout_double = TextIOWrapper(open(out_w, "wb", closefd=False), encoding="utf-8")
+        try:
+            monkeypatch.setattr(sys, "stdout", stdout_double)
+
+            request = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+            response = JSONRPCResponse(jsonrpc="2.0", id=1, result={})
+            with anyio.fail_after(5):
+                async with stdio_server() as (read_stream, write_stream):
+                    async with read_stream:
+                        # fd 0 is the null device: instant EOF instead of protocol bytes.
+                        assert await anyio.to_thread.run_sync(os.read, 0, 1, abandon_on_cancel=True) == b""
+
+                        os.write(in_w, _frame(request))
+                        received = await read_stream.receive()
+                        assert isinstance(received, SessionMessage)
+                        assert received.message == request
+
+                        await write_stream.send(SessionMessage(response))
+                        line = await _read_from(out_r)
+                        assert jsonrpc_message_adapter.validate_json(line.decode().strip()) == response
+
+                        os.close(in_w)  # EOF lets the reader finish so the context can exit
+                        await write_stream.aclose()
+
+                # samestat is trivially-true for pipes on Windows; POSIX legs carry these assertions.
+                assert os.path.sameopenfile(0, in_r)
+        finally:
+            stdout_double.close()
+            os.close(out_r)
+            os.close(out_w)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("failing_call", ["dup", "dup2"])
+async def test_stdio_server_reads_stdin_in_place_when_descriptor_isolation_fails(
+    failing_call: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A descriptor-table failure while claiming stdin degrades to reading sys.stdin in place.
+
+    SDK-defined behavior: isolation is best-effort; the dup2 variant fails after the private duplicate exists.
+    """
+    request = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+    with _pipe_planted_on_fd0(monkeypatch) as (in_r, in_w):
+        os.write(in_w, _frame(request))
+        os.close(in_w)
+        monkeypatch.setattr(sys, "stdout", TextIOWrapper(io.BytesIO(), encoding="utf-8"))
+
+        # Injectors fire once, then pass through: pytest capture also calls os.dup/os.dup2,
+        # and a still-armed injector detonating there corrupts every later test's capture.
+        if failing_call == "dup":
+            real_dup = os.dup
+            armed = [True]
+
+            def failing_dup(fd: int) -> int:
+                if fd == 0 and armed[0]:
+                    armed[0] = False
+                    raise OSError("injected descriptor failure")
+                return real_dup(fd)
+
+            monkeypatch.setattr(os, "dup", failing_dup)
+        else:
+            real_dup2 = os.dup2
+            armed = [True]
+
+            def failing_dup2(fd: int, fd2: int, inheritable: bool = True) -> int:
+                if armed[0]:
+                    armed[0] = False
+                    raise OSError("injected descriptor failure")
+                return real_dup2(fd, fd2, inheritable)
+
+            monkeypatch.setattr(os, "dup2", failing_dup2)
+
+        with anyio.fail_after(5):
+            async with stdio_server() as (read_stream, write_stream):  # pragma: no branch
+                async with read_stream:  # pragma: no branch
+                    # Isolation was skipped: fd 0 is still the protocol pipe.
+                    assert os.path.sameopenfile(0, in_r)
+                    # In-place transports still own the stream: a second one is refused.
+                    with pytest.raises(RuntimeError, match="already claimed fd 0"):
+                        async with stdio_server():
+                            pytest.fail("unreachable")  # pragma: no cover
+                    # The spent injector passes calls through untouched.
+                    os.close(os.dup(0))
+                    received = await read_stream.receive()
+                    assert isinstance(received, SessionMessage)
+                    assert received.message == request
+                    await write_stream.aclose()
+
+
+@pytest.mark.anyio
+async def test_stdio_server_exits_cleanly_when_the_stdin_restore_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A failed fd 0 restore on exit is swallowed, not raised, and the fd stays claimed.
+
+    SDK-defined behavior: the restore must never mask what ended the transport, and a
+    still-diverted fd must refuse later transports rather than serve them the diversion.
+    """
+    request = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+    fresh_claims: set[int] = set()
+    monkeypatch.setattr("mcp.server.stdio._claimed", fresh_claims)  # this test leaves fd 0 claimed
+    with _pipe_planted_on_fd0(monkeypatch) as (_, in_w):
+        os.write(in_w, _frame(request))
+        os.close(in_w)
+        monkeypatch.setattr(sys, "stdout", TextIOWrapper(io.BytesIO(), encoding="utf-8"))
+
+        # The claim's dup2 (first call) succeeds; only the restore's (second) fails.
+        real_dup2 = os.dup2
+        dup2_calls: list[tuple[int, int]] = []
+
+        def flaky_dup2(fd: int, fd2: int, inheritable: bool = True) -> int:
+            dup2_calls.append((fd, fd2))
+            if len(dup2_calls) == 2:
+                raise OSError("injected restore failure")
+            return real_dup2(fd, fd2, inheritable)
+
+        monkeypatch.setattr(os, "dup2", flaky_dup2)
+
+        with anyio.fail_after(5):
+            async with stdio_server() as (read_stream, write_stream):  # pragma: no branch
+                async with read_stream:  # pragma: no branch
+                    received = await read_stream.receive()
+                    assert isinstance(received, SessionMessage)
+                    assert received.message == request
+                    await write_stream.aclose()
+
+        # Restore attempted (second dup2), failure swallowed, fd 0 left on the null device.
+        assert dup2_calls[1] == (dup2_calls[1][0], 0)
+        devnull_probe = os.open(os.devnull, os.O_RDONLY)
+        try:
+            assert os.path.sameopenfile(0, devnull_probe)
+        finally:
+            os.close(devnull_probe)
+
+        # The still-diverted fd stays claimed: a later transport is refused,
+        # not handed the null device as its wire.
+        with pytest.raises(RuntimeError, match="already claimed fd 0"):
+            async with stdio_server():
+                pytest.fail("unreachable")  # pragma: no cover
+
+
+@pytest.mark.anyio
+async def test_stdio_server_takes_stdout_off_the_descriptor_table_while_serving(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """On the real process stdout, the transport claims the wire and diverts fd 1 to stderr.
+
+    SDK-defined behavior: stray writes to fd 1 land in the client's log, not the JSON-RPC stream.
+    """
+    response = JSONRPCResponse(jsonrpc="2.0", id=1, result={})
+    with _pipe_planted_on_fd1(monkeypatch) as (out_r, out_w), _pipe_planted_on_fd2() as err_r:
+        with anyio.fail_after(5):
+            async with stdio_server(stdin=anyio.AsyncFile(io.StringIO())) as (read_stream, write_stream):
+                read_stream.close()
+                os.write(1, b"stray child output\n")
+                assert await _read_from(err_r) == b"stray child output\n"
+
+                # The text layer writes os.linesep, hence CRLF on Windows.
+                print("stray print", flush=True)
+                assert await _read_from(err_r) == b"stray print" + os.linesep.encode()
+
+                await write_stream.send(SessionMessage(response))
+                line = await _read_from(out_r)
+                assert jsonrpc_message_adapter.validate_json(line.decode().strip()) == response
+
+                await write_stream.aclose()
+
+        assert os.path.sameopenfile(1, out_w)
+
+
+@pytest.mark.anyio
+async def test_stdio_server_diverts_stdout_to_the_null_device_when_stderr_is_unusable(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """When stderr cannot be duplicated, fd 1 is diverted to the null device instead.
+
+    SDK-defined behavior: a process with unusable fd 2 still gets stdout claimed; the wire stays pure.
+    """
+    response = JSONRPCResponse(jsonrpc="2.0", id=1, result={})
+    with _pipe_planted_on_fd1(monkeypatch) as (out_r, out_w):
+        # One-shot injector, as in the isolation-failure test.
+        real_dup = os.dup
+        armed = [True]
+
+        def failing_dup(fd: int) -> int:
+            if fd == 2 and armed[0]:
+                armed[0] = False
+                raise OSError("injected stderr failure")
+            return real_dup(fd)
+
+        monkeypatch.setattr(os, "dup", failing_dup)
+
+        with anyio.fail_after(5):
+            async with stdio_server(stdin=anyio.AsyncFile(io.StringIO())) as (read_stream, write_stream):
+                read_stream.close()
+                devnull_probe = os.open(os.devnull, os.O_WRONLY)
+                try:
+                    assert os.path.sameopenfile(1, devnull_probe)
+                finally:
+                    os.close(devnull_probe)
+
+                os.write(1, b"discarded\n")
+                await write_stream.send(SessionMessage(response))
+                line = await _read_from(out_r)
+                assert jsonrpc_message_adapter.validate_json(line.decode().strip()) == response
+                await write_stream.aclose()
+
+        assert os.path.sameopenfile(1, out_w)
+
+
+@pytest.mark.anyio
+async def test_a_second_stdio_server_on_the_same_process_streams_is_refused(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A concurrent stdio_server() on already-claimed streams raises instead of contending."""
+    request = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+    response = JSONRPCResponse(jsonrpc="2.0", id=1, result={})
+    with _pipe_planted_on_fd0(monkeypatch) as (in_r, in_w), _pipe_planted_on_fd1(monkeypatch) as (out_r, out_w):
+        with anyio.fail_after(5):
+            async with stdio_server() as (read_stream, write_stream):
+                async with read_stream:  # pragma: no branch
+                    with pytest.raises(RuntimeError, match="already claimed fd 0"):
+                        async with stdio_server():
+                            pytest.fail("unreachable")  # pragma: no cover
+
+                    os.write(in_w, _frame(request))
+                    received = await read_stream.receive()
+                    assert isinstance(received, SessionMessage)
+                    assert received.message == request
+                    await write_stream.send(SessionMessage(response))
+                    line = await _read_from(out_r)
+                    assert jsonrpc_message_adapter.validate_json(line.decode().strip()) == response
+                    os.close(in_w)
+                    await write_stream.aclose()
+
+        assert os.path.sameopenfile(0, in_r)
+        assert os.path.sameopenfile(1, out_w)
+
+
+@pytest.mark.anyio
+async def test_a_refused_claim_releases_the_stream_it_already_took(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A transport refused halfway through claiming restores what it claimed first.
+
+    The first transport claims only stdout; the second claims stdin, is refused on stdout, and must release stdin.
+    """
+    with _pipe_planted_on_fd0(monkeypatch) as (in_r, in_w), _pipe_planted_on_fd1(monkeypatch) as (_, out_w):
+        with anyio.fail_after(5):
+            async with stdio_server(stdin=anyio.AsyncFile(io.StringIO())) as (read_stream, write_stream):
+                read_stream.close()
+                with pytest.raises(RuntimeError, match="already claimed fd 1"):
+                    async with stdio_server():
+                        pytest.fail("unreachable")  # pragma: no cover
+
+                # fd 0 is back on the protocol pipe, not the null device.
+                assert os.path.sameopenfile(0, in_r)
+                await write_stream.aclose()
+
+        assert os.path.sameopenfile(1, out_w)
+        os.close(in_w)
+
+
+@pytest.mark.anyio
+async def test_stdio_server_serves_in_place_when_the_standard_descriptors_are_incomplete(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A process missing a standard descriptor is served in place, without surgery.
+
+    With fd 2 closed, a duplicate could land in the standard range: the transport must not touch the table.
+    """
+    request = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+    response = JSONRPCResponse(jsonrpc="2.0", id=1, result={})
+    with _pipe_planted_on_fd0(monkeypatch) as (in_r, in_w), _pipe_planted_on_fd1(monkeypatch) as (out_r, out_w):
+        saved2 = os.dup(2)
+        os.close(2)
+        try:
+            with anyio.fail_after(5):
+                async with stdio_server() as (read_stream, write_stream):
+                    async with read_stream:  # pragma: no branch
+                        assert os.path.sameopenfile(0, in_r)
+                        assert os.path.sameopenfile(1, out_w)
+
+                        os.write(in_w, _frame(request))
+                        received = await read_stream.receive()
+                        assert isinstance(received, SessionMessage)
+                        assert received.message == request
+                        await write_stream.send(SessionMessage(response))
+                        line = await _read_from(out_r)
+                        assert jsonrpc_message_adapter.validate_json(line.decode().strip()) == response
+                        os.close(in_w)
+                        await write_stream.aclose()
+        finally:
+            os.dup2(saved2, 2)
+            os.close(saved2)
+
+
 class _GatedStdin(io.RawIOBase):
     """Raw stdin double: serves its frames, then blocks until released before EOF.
 
-    A real stdio client keeps stdin open until it has read the responses it is
-    awaiting; an immediate EOF after the last frame races the dispatcher's
-    EOF-time cancellation of in-flight handlers (only inline-handled methods
-    would deterministically answer first). The blocked read sits in
-    `stdio_server`'s reader worker thread and unblocks on `release()`.
+    A real client holds stdin open until it reads its responses; instant EOF races the
+    dispatcher's EOF-time cancellation of in-flight handlers.
     """
 
     name = "<gated-stdin>"
@@ -131,8 +499,7 @@ def readinto(self, b: Buffer) -> int:
             view[:n] = self._pending[:n]
             self._pending = self._pending[n:]
             return n
-        # A missed release falls through to EOF after the bound; the caller's
-        # own response assertions then report what actually arrived.
+        # A missed release falls through to EOF after the bound.
         self._released.wait(5)
         return 0
 
@@ -143,8 +510,7 @@ def release(self) -> None:
 class _NotifyingStdout(io.RawIOBase):
     """Raw stdout double that counts newline-terminated lines and can be awaited on.
 
-    Survives wrapper close (`close()` is a no-op) so the test can read what was
-    written after `run()` has torn its TextIOWrapper down.
+    close() is a no-op so the test can read what was written after run() tears down.
     """
 
     name = "<notifying-stdout>"
@@ -182,14 +548,8 @@ def _serve_stdio_and_collect(
 ) -> list[JSONRPCMessage]:
     """Serve `frames` over process stdio and return the parsed response lines.
 
-    Runs the blocking `server.run("stdio")` in a daemon thread (it creates its
-    own event loop, so a sync test cannot arm `anyio.fail_after`) and signals
-    stdin EOF only after `responses` lines arrive on stdout - the way a real
-    client closes the pipe - so spawned in-flight handlers never race the
-    dispatcher's EOF cancellation. The join bound turns a run loop that never
-    returns on stdin EOF into a red test instead of a silent CI hang; an
-    exception escaping `run()` still fails the test via pytest's
-    unhandled-thread warning, escalated by `filterwarnings = ["error"]`.
+    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.
     """
     payload = "".join(f.model_dump_json(by_alias=True, exclude_none=True) + "\n" for f in frames).encode()
     stdin = _GatedStdin(payload)
@@ -211,11 +571,7 @@ def target() -> None:
 
 
 def test_mcpserver_run_stdio_serves_until_stdin_closes(monkeypatch: pytest.MonkeyPatch) -> None:
-    """`MCPServer.run("stdio")` serves over process stdio and returns at stdin EOF.
-
-    Answers a request over the process's stdio and returns when stdin reaches EOF,
-    rather than serving forever.
-    """
+    """`MCPServer.run("stdio")` serves over process stdio and returns at stdin EOF."""
     ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
 
     responses = _serve_stdio_and_collect(monkeypatch, MCPServer(name="RunStdioServer"), [ping], 1)
@@ -226,8 +582,7 @@ def test_mcpserver_run_stdio_serves_until_stdin_closes(monkeypatch: pytest.Monke
 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")`.
 
-    Regression lock for the issue #1027 shutdown chain: the run loop must end on
-    stdin EOF and unwind the lifespan rather than be killed before returning.
+    Regression lock for the issue #1027 shutdown chain.
     """
     events: list[str] = []
 
@@ -251,10 +606,7 @@ async def lifespan(server: MCPServer) -> AsyncIterator[None]:
 def test_mcpserver_run_stdio_serves_a_modern_connection(monkeypatch: pytest.MonkeyPatch) -> None:
     """`MCPServer.run("stdio")` serves the modern era over process stdio.
 
-    A `server/discover` probe gets a DiscoverResult (no initialize handshake)
-    and a subsequent envelope-bearing request is served at the discovered
-    version - the wire exchange `Client(mode='auto')` drives against a stdio
-    server.
+    A `server/discover` probe (no initialize handshake), then a request served at the discovered version.
     """
     envelope = {
         PROTOCOL_VERSION_META_KEY: "2026-07-28",
@@ -270,7 +622,6 @@ def test_mcpserver_run_stdio_serves_a_modern_connection(monkeypatch: pytest.Monk
     assert "2026-07-28" in responses[0].result["supportedVersions"]
     assert responses[0].result["serverInfo"]["name"] == "ModernStdioServer"
     assert isinstance(responses[1], JSONRPCResponse) and responses[1].id == 2
-    # `resultType` is the modern-only wire field: its presence proves the
-    # request was served at the discovered version, not the handshake era.
+    # resultType is modern-only: proves the request was served at the discovered version.
     assert responses[1].result["tools"] == []
     assert responses[1].result["resultType"] == "complete"
diff --git a/tests/transports/stdio/test_lifecycle.py b/tests/transports/stdio/test_lifecycle.py
index 8a370c10f..1d5ec3b47 100644
--- a/tests/transports/stdio/test_lifecycle.py
+++ b/tests/transports/stdio/test_lifecycle.py
@@ -15,12 +15,15 @@
 import threading
 from contextlib import AsyncExitStack
 from pathlib import Path
+from textwrap import dedent
 
 import anyio
 import anyio.abc
 import pytest
+from mcp_types import TextContent
 
 from mcp.client import stdio
+from mcp.client.client import Client
 from mcp.client.stdio import StdioServerParameters, stdio_client
 from mcp.os.win32.utilities import FallbackProcess
 from tests.transports.stdio._liveness import (
@@ -274,3 +277,42 @@ async def test_fallback_process_wait_is_cancellable_while_the_child_lives() -> N
         popen.wait()
         popen.stdin.close()
         popen.stdout.close()
+
+
+@pytest.mark.anyio
+async def test_a_tool_spawned_childs_stdout_writes_never_reach_the_wire(tmp_path: Path) -> None:
+    """A child writing to its inherited stdout pollutes the server's stderr, never the protocol.
+
+    Pre-isolation the junk line landed in the JSON-RPC stream (fails on base);
+    fd 1 has exactly one target, so stderr delivery proves the wire never saw it.
+    """
+    server = dedent(
+        """
+        import subprocess, sys
+        from mcp.server import MCPServer
+
+        mcp = MCPServer("noisy-spawner")
+
+        @mcp.tool()
+        def run_noisy_child() -> str:
+            # No redirection: the child inherits the server's stdout.
+            proc = subprocess.run([sys.executable, "-c", "print('this is not json')"], timeout=20)
+            return str(proc.returncode)
+
+        mcp.run()
+        """
+    )
+
+    with (tmp_path / "server-stderr.txt").open("w+") as errlog:
+        transport = stdio_client(StdioServerParameters(command=sys.executable, args=["-c", server]), errlog=errlog)
+        # Bound covers three interpreter cold starts; a regressed Windows leg hangs rather than corrupts.
+        with anyio.fail_after(40):
+            async with Client(transport) as client:
+                result = await client.call_tool("run_noisy_child")
+        errlog.seek(0)
+        server_stderr = errlog.read()
+
+    content = result.content[0]
+    assert isinstance(content, TextContent)
+    assert content.text == "0"
+    assert "this is not json" in server_stderr
diff --git a/tests/transports/stdio/test_windows.py b/tests/transports/stdio/test_windows.py
index 2d4eeac82..a5c9b4e7b 100644
--- a/tests/transports/stdio/test_windows.py
+++ b/tests/transports/stdio/test_windows.py
@@ -15,12 +15,14 @@
 import sys
 from contextlib import AsyncExitStack
 from pathlib import Path
+from textwrap import dedent
 
 import anyio
 import anyio.abc
 import pytest
-from mcp_types import JSONRPCRequest, JSONRPCResponse
+from mcp_types import JSONRPCRequest, JSONRPCResponse, TextContent
 
+from mcp.client.client import Client
 from mcp.client.stdio import StdioServerParameters, stdio_client
 from mcp.os.win32.utilities import FallbackProcess
 from mcp.shared.message import SessionMessage
@@ -238,3 +240,46 @@ async def test_a_native_server_emitting_crlf_line_endings_round_trips_messages()
             # here instead of a parsed message.
             assert isinstance(received, SessionMessage)
             assert received.message == JSONRPCResponse(jsonrpc="2.0", id=1, result={})
+
+
+async def test_a_tool_spawned_python_child_with_default_stdin_completes_promptly() -> None:  # pragma: no cover
+    """A tool that runs a Python subprocess without redirecting stdin returns promptly.
+
+    Regression for #671: pre-isolation the child inherited the protocol stdin pipe
+    and hung in interpreter startup (CPython gh-78961) until the next inbound message.
+    """
+    server = dedent(
+        """
+        import subprocess, sys
+        from mcp.server import MCPServer
+
+        mcp = MCPServer("spawner")
+
+        @mcp.tool()
+        def run_child() -> str:
+            proc = subprocess.run([sys.executable, "-c", "print('ok')"], capture_output=True, timeout=20)
+            return proc.stdout.decode().strip()
+
+        @mcp.tool()
+        def run_child_bare() -> str:
+            # Even without redirection the console subsystem hands the child the standard handles.
+            proc = subprocess.run([sys.executable, "-c", "pass"], timeout=20)
+            return str(proc.returncode)
+
+        mcp.run()
+        """
+    )
+    transport = stdio_client(StdioServerParameters(command=sys.executable, args=["-c", server]))
+
+    # A regression hangs forever, so the bound only has to beat "never".
+    with anyio.fail_after(40.0):
+        async with Client(transport) as client:
+            result = await client.call_tool("run_child")
+            bare = await client.call_tool("run_child_bare")
+
+    content = result.content[0]
+    assert isinstance(content, TextContent)
+    assert content.text == "ok"
+    bare_content = bare.content[0]
+    assert isinstance(bare_content, TextContent)
+    assert bare_content.text == "0"