fix: exit stdio server cleanly on interrupt

src/mcp/server/mcpserver/server.py

@@ -291,13 +291,16 @@ def run(
         if transport not in TRANSPORTS.__args__:  # type: ignore  # pragma: no cover
             raise ValueError(f"Unknown transport: {transport}")
 
-        match transport:
-            case "stdio":
-                anyio.run(self.run_stdio_async)
-            case "sse":  # pragma: no cover
-                anyio.run(lambda: self.run_sse_async(**kwargs))
-            case "streamable-http":  # pragma: no cover
-                anyio.run(lambda: self.run_streamable_http_async(**kwargs))
+        try:
+            match transport:
+                case "stdio":
+                    anyio.run(self.run_stdio_async)
+                case "sse":
+                    anyio.run(lambda: self.run_sse_async(**kwargs))
+                case "streamable-http":  # pragma: no branch
+                    anyio.run(lambda: self.run_streamable_http_async(**kwargs))
+        except KeyboardInterrupt:
+            return
 
     async def _handle_list_tools(
         self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None

tests/interaction/transports/test_stdio.py

@@ -21,6 +21,7 @@
 import sys
 import tempfile
 from pathlib import Path
+from typing import TextIO, cast
 
 import anyio
 import pytest
@@ -71,7 +72,7 @@ async def collect(params: LoggingMessageNotificationParams) -> None:
                 # so the server module is measured. Empty when not running under coverage.
                 env={key: value for key, value in os.environ.items() if key.startswith("COVERAGE_")},
             ),
-            errlog=errlog,
+            errlog=cast(TextIO, errlog),
         )
 
         with anyio.fail_after(10):
@@ -92,7 +93,8 @@ async def collect(params: LoggingMessageNotificationParams) -> None:
     # seeing it proves the process exited on its own rather than via the transport's terminate
     # escalation, without a timing-based assertion. The capture itself proves stderr passthrough:
     # the transport routes the child's stderr to the caller's `errlog` without consuming it.
-    assert captured_stderr == snapshot("stdio-echo: clean exit\n")
+    # Prerelease Python/lowest-direct dependency runs may print warnings before the server marker.
+    assert captured_stderr.endswith("stdio-echo: clean exit\n")
 
 
 @requirement("transport:stdio:stream-purity")

tests/server/mcpserver/test_server.py

@@ -1,6 +1,6 @@
 import base64
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -74,6 +74,28 @@ def test_dependencies(self):
         mcp_no_deps = MCPServer("test")
         assert mcp_no_deps.dependencies == []
 
+    @pytest.mark.parametrize("transport", ["stdio", "sse", "streamable-http"])
+    def test_keyboard_interrupt_exits_cleanly(self, transport: Literal["stdio", "sse", "streamable-http"]):
+        mcp = MCPServer("test")
+
+        with patch("mcp.server.mcpserver.server.anyio.run", side_effect=KeyboardInterrupt) as run:
+            mcp.run(transport)
+
+        assert run.call_count == 1
+        if transport == "stdio":
+            run.assert_called_once_with(mcp.run_stdio_async)
+
+    def test_run_propagates_non_interrupt_errors(self):
+        mcp = MCPServer("test")
+
+        with (
+            patch("mcp.server.mcpserver.server.anyio.run", side_effect=RuntimeError("boom")) as run,
+            pytest.raises(RuntimeError, match="boom"),
+        ):
+            mcp.run("stdio")
+
+        run.assert_called_once_with(mcp.run_stdio_async)
+
     async def test_sse_app_returns_starlette_app(self):
         """Test that sse_app returns a Starlette application with correct routes."""
         mcp = MCPServer("test")