fix: don't send log notification on transport error (#2257)
This commit is contained in:
parent
31a38b5078
commit
62eb08e5b2
3 changed files with 58 additions and 31 deletions
|
|
@ -414,11 +414,6 @@ class Server(Generic[LifespanResultT]):
|
||||||
)
|
)
|
||||||
case Exception():
|
case Exception():
|
||||||
logger.error(f"Received exception from stream: {message}")
|
logger.error(f"Received exception from stream: {message}")
|
||||||
await session.send_log_message(
|
|
||||||
level="error",
|
|
||||||
data="Internal Server Error",
|
|
||||||
logger="mcp.server.exception_handler",
|
|
||||||
)
|
|
||||||
if raise_exceptions:
|
if raise_exceptions:
|
||||||
raise message
|
raise message
|
||||||
case _:
|
case _:
|
||||||
|
|
|
||||||
|
|
@ -334,7 +334,7 @@ class BaseSession(
|
||||||
async with self._read_stream, self._write_stream:
|
async with self._read_stream, self._write_stream:
|
||||||
try:
|
try:
|
||||||
async for message in self._read_stream:
|
async for message in self._read_stream:
|
||||||
if isinstance(message, Exception): # pragma: no cover
|
if isinstance(message, Exception):
|
||||||
await self._handle_incoming(message)
|
await self._handle_incoming(message)
|
||||||
elif isinstance(message.message, JSONRPCRequest):
|
elif isinstance(message.message, JSONRPCRequest):
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
|
|
@ -1,55 +1,42 @@
|
||||||
from unittest.mock import AsyncMock, Mock
|
from unittest.mock import AsyncMock, Mock
|
||||||
|
|
||||||
|
import anyio
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from mcp import types
|
from mcp import types
|
||||||
from mcp.server.lowlevel.server import Server
|
from mcp.server.lowlevel.server import Server
|
||||||
from mcp.server.session import ServerSession
|
from mcp.server.session import ServerSession
|
||||||
|
from mcp.shared.message import SessionMessage
|
||||||
from mcp.shared.session import RequestResponder
|
from mcp.shared.session import RequestResponder
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_exception_handling_with_raise_exceptions_true():
|
async def test_exception_handling_with_raise_exceptions_true():
|
||||||
"""Test that exceptions are re-raised when raise_exceptions=True"""
|
"""Transport exceptions are re-raised when raise_exceptions=True."""
|
||||||
server = Server("test-server")
|
server = Server("test-server")
|
||||||
session = Mock(spec=ServerSession)
|
session = Mock(spec=ServerSession)
|
||||||
session.send_log_message = AsyncMock()
|
|
||||||
|
|
||||||
test_exception = RuntimeError("Test error")
|
test_exception = RuntimeError("Test error")
|
||||||
|
|
||||||
with pytest.raises(RuntimeError, match="Test error"):
|
with pytest.raises(RuntimeError, match="Test error"):
|
||||||
await server._handle_message(test_exception, session, {}, raise_exceptions=True)
|
await server._handle_message(test_exception, session, {}, raise_exceptions=True)
|
||||||
|
|
||||||
session.send_log_message.assert_called_once()
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
@pytest.mark.parametrize(
|
async def test_exception_handling_with_raise_exceptions_false():
|
||||||
"exception_class,message",
|
"""Transport exceptions are logged locally but not sent to the client.
|
||||||
[
|
|
||||||
(ValueError, "Test validation error"),
|
The transport that reported the error is likely broken; writing back
|
||||||
(RuntimeError, "Test runtime error"),
|
through it races with stream closure (#1967, #2064). The TypeScript,
|
||||||
(KeyError, "Test key error"),
|
Go, and C# SDKs all log locally only.
|
||||||
(Exception, "Basic error"),
|
"""
|
||||||
],
|
|
||||||
)
|
|
||||||
async def test_exception_handling_with_raise_exceptions_false(exception_class: type[Exception], message: str):
|
|
||||||
"""Test that exceptions are logged when raise_exceptions=False"""
|
|
||||||
server = Server("test-server")
|
server = Server("test-server")
|
||||||
session = Mock(spec=ServerSession)
|
session = Mock(spec=ServerSession)
|
||||||
session.send_log_message = AsyncMock()
|
session.send_log_message = AsyncMock()
|
||||||
|
|
||||||
test_exception = exception_class(message)
|
await server._handle_message(RuntimeError("Test error"), session, {}, raise_exceptions=False)
|
||||||
|
|
||||||
await server._handle_message(test_exception, session, {}, raise_exceptions=False)
|
session.send_log_message.assert_not_called()
|
||||||
|
|
||||||
# Should send log message
|
|
||||||
session.send_log_message.assert_called_once()
|
|
||||||
call_args = session.send_log_message.call_args
|
|
||||||
|
|
||||||
assert call_args.kwargs["level"] == "error"
|
|
||||||
assert call_args.kwargs["data"] == "Internal Server Error"
|
|
||||||
assert call_args.kwargs["logger"] == "mcp.server.exception_handler"
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
|
|
@ -72,3 +59,48 @@ async def test_normal_message_handling_not_affected():
|
||||||
|
|
||||||
# Verify _handle_request was called
|
# Verify _handle_request was called
|
||||||
server._handle_request.assert_called_once()
|
server._handle_request.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_server_run_exits_cleanly_when_transport_yields_exception_then_closes():
|
||||||
|
"""Regression test for #1967 / #2064.
|
||||||
|
|
||||||
|
Exercises the real Server.run() path with real memory streams, reproducing
|
||||||
|
what happens in stateless streamable HTTP when a POST handler throws:
|
||||||
|
|
||||||
|
1. Transport yields an Exception into the read stream
|
||||||
|
(streamable_http.py does this in its broad POST-handler except).
|
||||||
|
2. Transport closes the read stream (terminate() in stateless mode).
|
||||||
|
3. _receive_loop exits its `async with read_stream, write_stream:` block,
|
||||||
|
closing the write stream.
|
||||||
|
4. Meanwhile _handle_message(exc) was spawned via tg.start_soon and runs
|
||||||
|
after the write stream is closed.
|
||||||
|
|
||||||
|
Before the fix, _handle_message tried to send_log_message through the
|
||||||
|
closed write stream, raising ClosedResourceError inside the TaskGroup
|
||||||
|
and crashing server.run(). After the fix, it only logs locally.
|
||||||
|
"""
|
||||||
|
server = Server("test-server")
|
||||||
|
|
||||||
|
read_send, read_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
|
||||||
|
# Zero-buffer on the write stream forces send() to block until received.
|
||||||
|
# With no receiver, a send() sits blocked until _receive_loop exits its
|
||||||
|
# `async with self._read_stream, self._write_stream:` block and closes the
|
||||||
|
# stream, at which point the blocked send raises ClosedResourceError.
|
||||||
|
# This deterministically reproduces the race without sleeps.
|
||||||
|
write_send, write_recv = anyio.create_memory_object_stream[SessionMessage](0)
|
||||||
|
|
||||||
|
# What the streamable HTTP transport does: push the exception, then close.
|
||||||
|
read_send.send_nowait(RuntimeError("simulated transport error"))
|
||||||
|
read_send.close()
|
||||||
|
|
||||||
|
with anyio.fail_after(5):
|
||||||
|
# stateless=True so server.run doesn't wait for initialize handshake.
|
||||||
|
# Before this fix, this raised ExceptionGroup(ClosedResourceError).
|
||||||
|
await server.run(read_recv, write_send, server.create_initialization_options(), stateless=True)
|
||||||
|
|
||||||
|
# write_send was closed inside _receive_loop's `async with`; receive_nowait
|
||||||
|
# raises EndOfStream iff the buffer is empty (i.e., server wrote nothing).
|
||||||
|
with pytest.raises(anyio.EndOfStream):
|
||||||
|
write_recv.receive_nowait()
|
||||||
|
write_recv.close()
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue