diff --git a/src/mcp/server/_streamable_http_modern.py b/src/mcp/server/_streamable_http_modern.py index 1932a5d9d2..f2c7b1c1c0 100644 --- a/src/mcp/server/_streamable_http_modern.py +++ b/src/mcp/server/_streamable_http_modern.py @@ -54,7 +54,7 @@ from mcp.server.runner import modern_error_data, serve_one from mcp.server.streamable_http import check_accept_headers from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings -from mcp.shared.dispatcher import CallOptions +from mcp.shared.dispatcher import CallOptions, request_id_in from mcp.shared.exceptions import NoBackChannelError from mcp.shared.inbound import ( ERROR_CODE_HTTP_STATUS, @@ -413,7 +413,10 @@ async def handle_modern_request( except ValidationError: # A batch, a posted response (clients MUST NOT send those: streamable-http # §Sending Messages item 4), or a request whose envelope is malformed. - await _write(_INVALID_BODY, scope, receive, send) + # Echo the original request id so envelope-invalid failures correlate. + request_id = request_id_in(decoded) + rej = JSONRPCError(jsonrpc="2.0", id=request_id, error=_INVALID_BODY.error) + await _write(rej, scope, receive, send) return if req.method == "subscriptions/listen" and not has_sse: diff --git a/src/mcp/server/stdio.py b/src/mcp/server/stdio.py index de8bbae5f1..353d61dbf4 100644 --- a/src/mcp/server/stdio.py +++ b/src/mcp/server/stdio.py @@ -26,6 +26,7 @@ async def run_server(): from mcp.os.win32.utilities import rebind_std_handle_to_fd from mcp.shared._context_streams import create_context_streams +from mcp.shared.jsonrpc_dispatcher import UnparseableMessageError from mcp.shared.message import SessionMessage if sys.platform != "win32": # pragma: no branch @@ -188,7 +189,7 @@ async def stdin_reader(): try: message = types.jsonrpc_message_adapter.validate_json(line, by_name=False) except Exception as exc: - await read_stream_writer.send(exc) + await read_stream_writer.send(UnparseableMessageError(line, cause=exc)) continue session_message = SessionMessage(message) diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 1a4e9939a4..77d4df0b98 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -22,7 +22,6 @@ from mcp_types import ( DEFAULT_NEGOTIATED_VERSION, INTERNAL_ERROR, - INVALID_PARAMS, INVALID_REQUEST, PARSE_ERROR, ErrorData, @@ -43,6 +42,7 @@ from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams from mcp.shared._stream_protocols import ReadStream, WriteStream +from mcp.shared.dispatcher import request_id_in from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER from mcp.shared.message import CloseSSEStreamCallback, ServerMessageMetadata, SessionMessage @@ -369,6 +369,7 @@ def _create_error_response( status_code: HTTPStatus, error_code: int = INVALID_REQUEST, headers: dict[str, str] | None = None, + request_id: RequestId | None = None, ) -> Response: """Create an error response with a simple string message.""" response_headers = {"Content-Type": CONTENT_TYPE_JSON} @@ -381,7 +382,7 @@ def _create_error_response( # Return a properly formatted JSON error response error_response = JSONRPCError( jsonrpc="2.0", - id=None, + id=request_id, error=ErrorData(code=error_code, message=error_message), ) @@ -546,10 +547,13 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re try: message = jsonrpc_message_adapter.validate_python(raw_message, by_name=False) except ValidationError as e: + # Echo the original request id so envelope-invalid failures correlate (#2848). + request_id = request_id_in(raw_message) response = self._create_error_response( f"Validation error: {str(e)}", HTTPStatus.BAD_REQUEST, - INVALID_PARAMS, + INVALID_REQUEST, + request_id=request_id, ) await response(scope, receive, send) return diff --git a/src/mcp/shared/dispatcher.py b/src/mcp/shared/dispatcher.py index f2ff96e7d5..eac27da1bc 100644 --- a/src/mcp/shared/dispatcher.py +++ b/src/mcp/shared/dispatcher.py @@ -40,6 +40,7 @@ "ProgressFnT", "as_request_id", "coerce_request_id", + "request_id_in", "run_notify_intercept", ] @@ -53,6 +54,16 @@ def as_request_id(value: object) -> RequestId | None: return None +def request_id_in(message: Any) -> RequestId | None: + """A decoded JSON-RPC message's top-level request id, or None when the message is + not an object or carries no scalar string/int id.""" + try: + rid: Any = message.get("id") + except AttributeError: + return None + return as_request_id(rid) + + def coerce_request_id(request_id: RequestId) -> RequestId: """Coerce a stringified int request id back to int so a peer-echoed id still correlates (matches the TS SDK). diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index 87bdf31ceb..8fb6c22917 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/src/mcp/shared/jsonrpc_dispatcher.py @@ -8,6 +8,7 @@ from __future__ import annotations import contextvars +import json import logging from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, field @@ -22,6 +23,7 @@ CONNECTION_CLOSED, INTERNAL_ERROR, INVALID_PARAMS, + INVALID_REQUEST, REQUEST_TIMEOUT, ErrorData, JSONRPCError, @@ -49,6 +51,7 @@ ProgressFnT, as_request_id, coerce_request_id, + request_id_in, run_notify_intercept, ) from mcp.shared.exceptions import MCPError, NoBackChannelError @@ -62,6 +65,7 @@ __all__ = [ "JSONRPCDispatcher", + "UnparseableMessageError", "cancelled_request_id_from_params", "handler_exception_to_error_data", "progress_token_from_params", @@ -85,6 +89,33 @@ answered - the handler's eventual result or error is dropped, not written.""" +class UnparseableMessageError(Exception): + """A transport failed to decode an inbound frame as a JSON-RPC message. + + Transports send this instead of a bare parse exception so the receiving + side can still correlate an error response with the frame's JSON-RPC + request id, when one is recoverable from the raw payload. + """ + + def __init__(self, payload: str | bytes | None = None, *, cause: BaseException | None = None) -> None: + super().__init__( + f"failed to decode inbound frame: {cause!r}" if cause is not None else "failed to decode inbound frame" + ) + self.payload = payload + self.__cause__ = cause + + @property + def request_id(self) -> RequestId | None: + """Best-effort recovery of the frame's top-level request id, else None.""" + if self.payload is None: + return None + try: + decoded = json.loads(self.payload) + except (ValueError, RecursionError): + return None + return request_id_in(decoded) + + def handler_exception_to_error_data(exc: BaseException) -> ErrorData | None: """Map a handler-raised exception to its wire `ErrorData`. @@ -537,6 +568,17 @@ async def _dispatch( """ if isinstance(item, Exception): if self.on_stream_exception is None: + # Answer an envelope-invalid frame whose request id is still + # recoverable; bare exceptions stay dropped as before. + request_id = item.request_id if isinstance(item, UnparseableMessageError) else None + if request_id is not None: + self._spawn( + self._write_error, + request_id, + ErrorData(code=INVALID_REQUEST, message="Invalid Request"), + sender_ctx=sender_ctx, + ) + return logger.debug("transport yielded exception: %r", item) return try: diff --git a/tests/interaction/transports/test_hosting_http.py b/tests/interaction/transports/test_hosting_http.py index ff9ac2ed05..800f0b1c9b 100644 --- a/tests/interaction/transports/test_hosting_http.py +++ b/tests/interaction/transports/test_hosting_http.py @@ -15,6 +15,7 @@ CLIENT_CAPABILITIES_META_KEY, CLIENT_INFO_META_KEY, INVALID_PARAMS, + INVALID_REQUEST, PARSE_ERROR, PROTOCOL_VERSION_META_KEY, UNSUPPORTED_PROTOCOL_VERSION, @@ -134,7 +135,7 @@ async def test_non_json_content_type_is_rejected() -> None: @requirement("hosting:http:parse-error-400") @requirement("hosting:http:batch") async def test_malformed_and_batched_bodies_return_400() -> None: - """A non-JSON body returns 400 Parse error; a JSON array of requests returns 400 Invalid params.""" + """A non-JSON body returns 400 Parse error; a JSON array of requests returns 400 Invalid Request.""" async with mounted_app(_server()) as (http, _): session_id = await initialize_via_http(http) not_json = await http.post( @@ -154,7 +155,7 @@ async def test_malformed_and_batched_bodies_return_400() -> None: assert not_json.status_code == 400 assert JSONRPCError.model_validate_json(not_json.text).error.code == PARSE_ERROR assert batched.status_code == 400 - assert JSONRPCError.model_validate_json(batched.text).error.code == INVALID_PARAMS + assert JSONRPCError.model_validate_json(batched.text).error.code == INVALID_REQUEST @requirement("hosting:http:protocol-version-400") diff --git a/tests/interaction/transports/test_hosting_http_modern.py b/tests/interaction/transports/test_hosting_http_modern.py index e26c70f3bf..73d8181ba4 100644 --- a/tests/interaction/transports/test_hosting_http_modern.py +++ b/tests/interaction/transports/test_hosting_http_modern.py @@ -181,9 +181,10 @@ async def test_modern_notification_post_is_acknowledged_202_and_a_posted_respons assert (acknowledged.status_code, acknowledged.content) == (202, b"") assert "mcp-session-id" not in acknowledged.headers assert refused.status_code == 400 + # The posted response's own id is echoed so the client can correlate the refusal. assert JSONRPCError.model_validate(refused.json()) == JSONRPCError( jsonrpc="2.0", - id=None, + id=1, error=ErrorData(code=INVALID_REQUEST, message="Body must be a single JSON-RPC request or notification object"), ) diff --git a/tests/server/test_stdio.py b/tests/server/test_stdio.py index eafd1fca59..e2a2c44e84 100644 --- a/tests/server/test_stdio.py +++ b/tests/server/test_stdio.py @@ -24,6 +24,7 @@ from mcp.server.mcpserver import MCPServer from mcp.server.stdio import stdio_server +from mcp.shared.jsonrpc_dispatcher import UnparseableMessageError from mcp.shared.message import SessionMessage @@ -75,6 +76,39 @@ async def test_stdio_server_round_trips_messages_over_injected_streams() -> None assert received_responses[1] == JSONRPCResponse(jsonrpc="2.0", id=4, result={}) +@pytest.mark.anyio +@pytest.mark.parametrize( + ("line", "expected_id"), + [ + pytest.param('{"jsonrpc": "1.0", "id": 3, "method": "ping", "params": {}}', 3, id="wrong-jsonrpc-version"), + pytest.param('{"id": 4, "method": "ping", "params": {}}', 4, id="missing-jsonrpc-field"), + pytest.param('{"jsonrpc": "2.0", "id": 8, "method": 12345, "params": {}}', 8, id="non-string-method"), + ], +) +async def test_stdio_server_envelope_invalid_line_surfaces_with_recoverable_request_id( + line: str, expected_id: int +) -> None: + """A line that is valid JSON but not a valid JSON-RPC envelope surfaces as an + UnparseableMessageError carrying the raw payload, so the session can still + correlate an error response with the request's original id.""" + stdin = io.StringIO(line + "\n") + stdout = io.StringIO() + + with anyio.fail_after(5): + async with stdio_server(stdin=anyio.AsyncFile(stdin), stdout=anyio.AsyncFile(stdout)) as ( + read_stream, + write_stream, + ): + async with read_stream: + received = await read_stream.receive() + + assert isinstance(received, UnparseableMessageError) + assert received.request_id == expected_id + + # Closing write_stream ends stdout_writer so the server context can join. + await write_stream.aclose() + + @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.""" diff --git a/tests/server/test_streamable_http_modern.py b/tests/server/test_streamable_http_modern.py index 11e775840f..4e0049810c 100644 --- a/tests/server/test_streamable_http_modern.py +++ b/tests/server/test_streamable_http_modern.py @@ -171,30 +171,35 @@ async def test_handle_modern_request_rejects_a_notification_post_at_an_unserved_ @pytest.mark.parametrize( - "body", + ("body", "expected_id"), [ - pytest.param({"jsonrpc": "2.0", "id": 1, "result": {}}, id="posted-response"), - pytest.param({"jsonrpc": "2.0", "id": 1, "error": {"code": -1, "message": "x"}}, id="posted-error"), - pytest.param([{"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 1}}], id="batch"), - pytest.param({"jsonrpc": "2.0", "id": None, "method": "tools/list"}, id="null-id-request"), - pytest.param({"jsonrpc": "2.0", "id": [1], "method": "tools/list"}, id="non-scalar-id-request"), - pytest.param({"jsonrpc": "2.0", "method": 7}, id="non-string-method-notification"), - pytest.param({"jsonrpc": "1.0", "method": "notifications/cancelled"}, id="wrong-jsonrpc-version"), - pytest.param("just a string", id="scalar"), + pytest.param({"jsonrpc": "2.0", "id": 1, "result": {}}, 1, id="posted-response"), + pytest.param({"jsonrpc": "2.0", "id": 1, "error": {"code": -1, "message": "x"}}, 1, id="posted-error"), + pytest.param( + [{"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 1}}], None, id="batch" + ), + pytest.param({"jsonrpc": "2.0", "id": None, "method": "tools/list"}, None, id="null-id-request"), + pytest.param({"jsonrpc": "2.0", "id": [1], "method": "tools/list"}, None, id="non-scalar-id-request"), + pytest.param({"jsonrpc": "2.0", "method": 7}, None, id="non-string-method-notification"), + pytest.param({"jsonrpc": "1.0", "method": "notifications/cancelled"}, None, id="wrong-jsonrpc-version"), + pytest.param("just a string", None, id="scalar"), ], ) -async def test_handle_modern_request_rejects_a_body_that_is_neither_request_nor_notification(body: Any) -> None: +async def test_handle_modern_request_rejects_a_body_that_is_neither_request_nor_notification( + body: Any, expected_id: int | None +) -> None: """Spec-mandated (streamable-http §Sending Messages item 4): the body MUST be a single request or notification and clients MUST NOT post responses. SDK-defined: anything else -- a posted response, a batch, a request whose `id` is malformed, a scalar -- is `INVALID_REQUEST` at - HTTP 400 with `id: null`, distinct from `PARSE_ERROR` (malformed JSON). A malformed-`id` - request in particular must not be mistaken for a notification and silently 202'd.""" + HTTP 400 with the original id echoed when one is recoverable (else `id: null`), distinct + from `PARSE_ERROR` (malformed JSON). A malformed-`id` request in particular must not be + mistaken for a notification and silently 202'd.""" async with _asgi_client(Server("test")) as http: response = await http.post("/mcp", json=body) assert response.status_code == 400 assert response.json() == { "jsonrpc": "2.0", - "id": None, + "id": expected_id, "error": {"code": INVALID_REQUEST, "message": "Body must be a single JSON-RPC request or notification object"}, } @@ -216,6 +221,27 @@ async def test_handle_modern_request_rejects_malformed_body_with_parse_error() - } +@pytest.mark.parametrize( + ("body", "expected_id"), + [ + pytest.param({"jsonrpc": "1.0", "id": 3, "method": "ping", "params": {}}, 3, id="wrong-jsonrpc-version"), + pytest.param({"id": 4, "method": "ping", "params": {}}, 4, id="missing-jsonrpc-field"), + pytest.param({"jsonrpc": "2.0", "id": 8, "method": 12345, "params": {}}, 8, id="non-string-method"), + ], +) +async def test_handle_modern_request_envelope_invalid_request_echoes_the_original_id( + body: dict[str, Any], expected_id: int +) -> None: + """An envelope-invalid but id-bearing request is answered -32600 with the + original request id, so the client can correlate the failure.""" + async with _asgi_client(Server("test")) as http: + response = await http.post("/mcp", json=body) + assert response.status_code == 400 + error = response.json() + assert error["id"] == expected_id + assert error["error"]["code"] == INVALID_REQUEST + + async def test_handle_modern_request_returns_transport_security_error_response() -> None: """The transport-security middleware's error response is sent verbatim and short-circuits.""" settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["good.example"]) diff --git a/tests/shared/test_jsonrpc_dispatcher.py b/tests/shared/test_jsonrpc_dispatcher.py index 9bee8b2c3b..6677f27b47 100644 --- a/tests/shared/test_jsonrpc_dispatcher.py +++ b/tests/shared/test_jsonrpc_dispatcher.py @@ -14,6 +14,7 @@ CONNECTION_CLOSED, INTERNAL_ERROR, INVALID_PARAMS, + INVALID_REQUEST, REQUEST_TIMEOUT, CallToolRequest, CallToolRequestParams, @@ -40,6 +41,7 @@ from mcp.shared.jsonrpc_dispatcher import ( # pyright: ignore[reportPrivateUsage] JSONRPCDispatcher, PeerCancelMode, + UnparseableMessageError, _OutboundPlan, _Pending, _plan_outbound, @@ -1594,6 +1596,56 @@ async def test_transport_exception_in_read_stream_is_logged_and_dropped(): s.close() +@pytest.mark.anyio +async def test_unparseable_message_with_recoverable_id_is_answered_with_invalid_request(): + """An envelope-invalid frame whose request id is still recoverable gets an + INVALID_REQUEST answer carrying that id, and the loop stays healthy.""" + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](4) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](4) + server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send) + on_request, on_notify = echo_handlers(Recorder()) + frame = '{"jsonrpc": "1.0", "id": 3, "method": "ping", "params": {}}' + try: + async with anyio.create_task_group() as tg: + await tg.start(server.run, on_request, on_notify) + await c2s_send.send(UnparseableMessageError(frame, cause=ValueError("invalid envelope"))) + with anyio.fail_after(5): + resp = await s2c_recv.receive() + assert isinstance(resp, SessionMessage) + assert isinstance(resp.message, JSONRPCError) + assert resp.message.id == 3 + assert resp.message.error.code == INVALID_REQUEST + + await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=1, method="t", params=None))) + with anyio.fail_after(5): + resp = await s2c_recv.receive() + assert isinstance(resp, SessionMessage) + assert isinstance(resp.message, JSONRPCResponse) + tg.cancel_scope.cancel() + finally: + for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): + s.close() + + +@pytest.mark.parametrize( + ("payload", "recovered_id"), + [ + pytest.param('{"jsonrpc": "1.0", "id": 3, "method": "ping"}', 3, id="int-id"), + pytest.param('{"jsonrpc": "1.0", "id": "abc", "method": "ping"}', "abc", id="string-id"), + pytest.param('{"jsonrpc": "1.0", "id": null, "method": "ping"}', None, id="null-id"), + pytest.param('{"jsonrpc": "1.0", "method": "ping"}', None, id="no-id-member"), + pytest.param('{"id": [1], "method": "ping"}', None, id="list-id"), + pytest.param('{"id": {"x": 1}, "method": "ping"}', None, id="object-id"), + pytest.param("not json", None, id="unparseable-json"), + pytest.param('"just a string"', None, id="scalar-json"), + pytest.param(None, None, id="no-payload"), + ], +) +def test_unparseable_message_request_id_recovery(payload: str | None, recovered_id: RequestId | None): + err = UnparseableMessageError(payload) + assert err.request_id == recovered_id + + @pytest.mark.anyio async def test_on_stream_exception_observes_transport_exceptions(): """With an observer set, Exception items reach it instead of being dropped; the loop stays healthy.""" diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py index aeef25a278..ab2d75ca29 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -509,6 +509,35 @@ async def test_json_parsing(basic_app: Starlette) -> None: assert "Validation error" in response.text +@pytest.mark.anyio +@pytest.mark.parametrize( + "body", + [ + pytest.param({"jsonrpc": "1.0", "id": 3, "method": "ping", "params": {}}, id="wrong-jsonrpc-version"), + pytest.param({"id": 4, "method": "ping", "params": {}}, id="missing-jsonrpc-field"), + pytest.param({"jsonrpc": "2.0", "id": 8, "method": 12345, "params": {}}, id="non-string-method"), + ], +) +async def test_envelope_invalid_request_error_echoes_the_original_id( + basic_app: Starlette, body: dict[str, Any] +) -> None: + """An envelope-invalid but id-bearing request is answered -32600 with the + original request id, so the client can correlate the failure.""" + async with make_client(basic_app) as client: + response = await client.post( + "/mcp", + headers={ + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + }, + json=body, + ) + assert response.status_code == 400 + error = response.json() + assert error["id"] == body["id"] + assert error["error"]["code"] == INVALID_REQUEST + + @pytest.mark.anyio async def test_method_not_allowed(basic_app: Starlette) -> None: """Unsupported HTTP methods are rejected with 405."""