Handle KeyboardInterrupt in server run

Author: koriyoshi2041

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":  # 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))
+        except KeyboardInterrupt:
+            return
 
     async def _handle_list_tools(
         self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None

tests/server/mcpserver/test_server.py

@@ -73,6 +73,27 @@ def test_dependencies(self):
         mcp_no_deps = MCPServer("test")
         assert mcp_no_deps.dependencies == []
 
+    def test_run_suppresses_keyboard_interrupt(self, monkeypatch: pytest.MonkeyPatch):
+        mcp = MCPServer("test")
+
+        def raise_keyboard_interrupt(*args: Any, **kwargs: Any) -> None:
+            raise KeyboardInterrupt
+
+        monkeypatch.setattr("mcp.server.mcpserver.server.anyio.run", raise_keyboard_interrupt)
+
+        assert mcp.run(transport="stdio") is None
+
+    def test_run_reraises_other_exceptions(self, monkeypatch: pytest.MonkeyPatch):
+        mcp = MCPServer("test")
+
+        def raise_runtime_error(*args: Any, **kwargs: Any) -> None:
+            raise RuntimeError("startup failed")
+
+        monkeypatch.setattr("mcp.server.mcpserver.server.anyio.run", raise_runtime_error)
+
+        with pytest.raises(RuntimeError, match="startup failed"):
+            mcp.run(transport="stdio")
+
     async def test_sse_app_returns_starlette_app(self):
         """Test that sse_app returns a Starlette application with correct routes."""
         mcp = MCPServer("test")