Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 10 additions & 7 deletions src/mcp/server/mcpserver/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
10 changes: 6 additions & 4 deletions tests/interaction/transports/test_stdio.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,10 +89,12 @@ async def collect(params: LoggingMessageNotificationParams) -> None:
[LoggingMessageNotificationParams(level="info", logger="echo", data="echoing across\nprocesses")]
)
# The server writes this line only after its run loop returns, which happens when stdin closes:
# 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")
# seeing it as the final stderr line 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. Some lowest-direct dependency combinations can emit import-time warnings before
# the server starts, so do not require this to be the only stderr line.
assert captured_stderr.splitlines(keepends=True)[-1:] == snapshot(["stdio-echo: clean exit\n"])


@requirement("transport:stdio:stream-purity")
Expand Down
21 changes: 21 additions & 0 deletions tests/server/mcpserver/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Loading