Add regression test for invalid stdio UTF-8

Author: raashish1601

tests/issues/test_2328_stdio_invalid_utf8.py

@@ -0,0 +1,72 @@
+"""Regression test for issue #2328: raw invalid UTF-8 over stdio must not crash the server."""
+
+import io
+from io import TextIOWrapper
+
+import anyio
+import pytest
+from pydantic import AnyUrl, TypeAdapter
+
+from mcp import types
+from mcp.server import ServerRequestContext
+from mcp.server.lowlevel.server import Server
+from mcp.server.stdio import stdio_server
+from mcp.types import JSONRPCError, JSONRPCResponse, jsonrpc_message_adapter
+
+
+@pytest.mark.anyio
+async def test_stdio_server_returns_error_for_raw_invalid_utf8_tool_arguments():
+    """Invalid UTF-8 bytes in a request body should become a JSON-RPC error, not a crash."""
+
+    url_adapter = TypeAdapter(AnyUrl)
+
+    async def handle_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(
+            tools=[
+                types.Tool(
+                    name="fetch",
+                    description="Fetch a URL",
+                    input_schema={
+                        "type": "object",
+                        "required": ["url"],
+                        "properties": {"url": {"type": "string"}},
+                    },
+                )
+            ]
+        )
+
+    async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
+        arguments = params.arguments or {}
+        url_adapter.validate_python(arguments["url"])
+        return types.CallToolResult(content=[types.TextContent(type="text", text="ok")])
+
+    server = Server("test-server", on_list_tools=handle_list_tools, on_call_tool=handle_call_tool)
+
+    raw_stdin = io.BytesIO(
+        b'{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}\n'
+        b'{"jsonrpc":"2.0","method":"notifications/initialized"}\n'
+        b'{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"fetch","arguments":{"url":"http://x\xff\xfe"}}}\n'
+    )
+    raw_stdout = io.BytesIO()
+    stdout = TextIOWrapper(raw_stdout, encoding="utf-8")
+
+    async with stdio_server(
+        stdin=anyio.wrap_file(TextIOWrapper(raw_stdin, encoding="utf-8", errors="replace")),
+        stdout=anyio.wrap_file(stdout),
+    ) as (read_stream, write_stream):
+        await server.run(read_stream, write_stream, server.create_initialization_options())
+
+    stdout.flush()
+    responses = [
+        jsonrpc_message_adapter.validate_json(line)
+        for line in raw_stdout.getvalue().decode("utf-8").splitlines()
+    ]
+
+    assert len(responses) == 2
+    assert isinstance(responses[0], JSONRPCResponse)
+    assert responses[0].id == 1
+    assert isinstance(responses[1], JSONRPCError)
+    assert responses[1].id == 3
+    assert responses[1].error.message