Skip to content
Closed
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
53 changes: 51 additions & 2 deletions src/mcp/server/stdio.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,18 +11,21 @@ async def run_server():
```
"""

import json
import os
import re
import sys
import threading
from collections.abc import Callable
from contextlib import asynccontextmanager, suppress
from dataclasses import dataclass
from io import TextIOWrapper
from typing import BinaryIO, Literal, TextIO
from typing import BinaryIO, Literal, TextIO, cast

import anyio
import anyio.lowlevel
import mcp_types as types
from pydantic import ValidationError

from mcp.os.win32.utilities import rebind_std_handle_to_fd
from mcp.shared._context_streams import create_context_streams
Expand Down Expand Up @@ -158,6 +161,49 @@ def release() -> None:
return os.fdopen(private_fd, mode, closefd=False), release


_RAW_ID_PATTERN = re.compile(r'"id"\s*:\s*(-?\d+|"(?:[^"\\]|\\.)*")')


def _request_id_from_raw_line(line: str) -> types.RequestId | None:
"""Best-effort recovery of a request id from a line that failed validation.

A full parse is tried first; lines too malformed for it fall back to a
pattern scan, since the id sits ahead of the payload that usually breaks.
"""
try:
parsed: object = json.loads(line)
except (json.JSONDecodeError, RecursionError):
match = _RAW_ID_PATTERN.search(line)
if match is None:
return None
try:
parsed = json.loads(match.group(1))
except json.JSONDecodeError:
return None
if isinstance(parsed, dict):
parsed = cast(dict[str, object], parsed).get("id")
if isinstance(parsed, str) or type(parsed) is int:
return parsed
return None


def _validation_error_response(line: str, exc: Exception) -> SessionMessage:
"""The JSON-RPC error frame for a line that failed message validation.

The message stays bare: the validation error text embeds the offending
input, which can be arbitrarily large.
"""
if isinstance(exc, ValidationError) and any(error["type"] == "json_invalid" for error in exc.errors()):
code, message = types.PARSE_ERROR, "Parse error"
else:
code, message = types.INVALID_REQUEST, "Invalid request"
return SessionMessage(
types.JSONRPCError(
jsonrpc="2.0", id=_request_id_from_raw_line(line), error=types.ErrorData(code=code, message=message)
)
)


@asynccontextmanager
async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.AsyncFile[str] | None = None):
"""Serve MCP over the process's stdin and stdout.
Expand Down Expand Up @@ -188,7 +234,10 @@ 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)
# A failed line must still complete its request: send the
# error frame on the wire instead of surfacing a bare
# exception the server dispatcher has no observer for.
await write_stream.send(_validation_error_response(line, exc))
continue

session_message = SessionMessage(message)
Expand Down
161 changes: 150 additions & 11 deletions tests/server/test_stdio.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import gc
import io
import json
import os
import sys
import threading
Expand All @@ -13,8 +14,12 @@
from mcp_types import (
CLIENT_CAPABILITIES_META_KEY,
CLIENT_INFO_META_KEY,
INVALID_REQUEST,
PARSE_ERROR,
PROTOCOL_VERSION_META_KEY,
SERVER_INFO_META_KEY,
ErrorData,
JSONRPCError,
JSONRPCMessage,
JSONRPCRequest,
JSONRPCResponse,
Expand Down Expand Up @@ -75,26 +80,130 @@ async def test_stdio_server_round_trips_messages_over_injected_streams() -> None
assert received_responses[1] == JSONRPCResponse(jsonrpc="2.0", id=4, result={})


async def _serve_raw_lines(lines: list[str]) -> tuple[list[SessionMessage | Exception], list[JSONRPCMessage]]:
"""Feed raw stdin lines through `stdio_server()`; return (read-stream items, parsed stdout frames).

Raw lines are the only way to exercise malformed input: the typed API cannot produce it.
"""
stdin = io.StringIO("".join(line + "\n" for line in lines))
stdout = io.StringIO()

received: list[SessionMessage | Exception] = []
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:
try:
while True:
received.append(await read_stream.receive())
except anyio.EndOfStream:
pass
await write_stream.aclose()

stdout.seek(0)
sent = [jsonrpc_message_adapter.validate_json(line.strip()) for line in stdout.readlines()]
return received, sent


@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."""
"""Non-UTF-8 stdin bytes are answered with a parse error frame without killing the stream.

Invalid bytes are replaced with U+FFFD, fail JSON parsing, and complete as a
JSON-RPC parse error with `id: null` (no id can be recovered); subsequent
valid messages are still processed.
"""
valid = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
raw_stdin = io.BytesIO(b"\xff\xfe\n" + valid.model_dump_json(by_alias=True, exclude_none=True).encode() + b"\n")
stdout = io.StringIO()

# stdio_server()'s default path wraps sys.stdin.buffer with errors='replace'.
monkeypatch.setattr(sys, "stdin", TextIOWrapper(raw_stdin, encoding="utf-8"))
monkeypatch.setattr(sys, "stdout", TextIOWrapper(io.BytesIO(), encoding="utf-8"))

with anyio.fail_after(5):
async with stdio_server() as (read_stream, write_stream):
await write_stream.aclose()
async with stdio_server(stdout=anyio.AsyncFile(stdout)) as (read_stream, write_stream):
async with read_stream: # pragma: no branch
first = await read_stream.receive()
assert isinstance(first, Exception)
received = await read_stream.receive()
assert isinstance(received, SessionMessage)
assert received.message == valid
await write_stream.aclose()

second = await read_stream.receive()
assert isinstance(second, SessionMessage)
assert second.message == valid
stdout.seek(0)
sent = [jsonrpc_message_adapter.validate_json(line.strip()) for line in stdout.readlines()]
assert sent == [JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error"))]


@pytest.mark.anyio
async def test_stdio_server_deeply_nested_request_completes_with_a_parse_error() -> None:
"""A request too deeply nested to validate completes with a parse error carrying its id.

Regression for #2751: the failure used to surface only as a bare exception on
the read stream, which the server dispatcher drops, leaving the request id
pending until the client timed out.
"""
params: object = {"leaf": True}
for index in reversed(range(256)):
params = {f"p{index}": params}
deep_line = json.dumps({"jsonrpc": "2.0", "id": 900256, "method": "ping", "params": params})
shallow = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")

received, sent = await _serve_raw_lines([deep_line, shallow.model_dump_json(by_alias=True, exclude_none=True)])

# The shallow request after the failed one still arrives; the failed one does not.
assert len(received) == 1
assert isinstance(received[0], SessionMessage)
assert received[0].message == shallow
assert sent == [JSONRPCError(jsonrpc="2.0", id=900256, error=ErrorData(code=PARSE_ERROR, message="Parse error"))]


@pytest.mark.anyio
@pytest.mark.parametrize(
("line", "expected"),
[
# Truncated JSON: the id is recovered from the raw line, ahead of the break.
(
'{"jsonrpc":"2.0","id":"bad-1","method":"ping","params":[',
JSONRPCError(jsonrpc="2.0", id="bad-1", error=ErrorData(code=PARSE_ERROR, message="Parse error")),
),
(
'{"jsonrpc":"2.0","id":42,"method":"ping","params":[',
JSONRPCError(jsonrpc="2.0", id=42, error=ErrorData(code=PARSE_ERROR, message="Parse error")),
),
# No id anywhere in the line: the parse error falls back to `id: null`.
(
"not json",
JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error")),
),
# An id whose string token is itself invalid JSON is not recoverable either.
(
'{"jsonrpc":"2.0","id":"a\\q","method":"ping"}',
JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error")),
),
# Valid JSON that is not a valid message: invalid request, id preserved where present.
(
'{"jsonrpc":"2.0","id":"bad-method","method":1}',
JSONRPCError(
jsonrpc="2.0", id="bad-method", error=ErrorData(code=INVALID_REQUEST, message="Invalid request")
),
),
(
'{"jsonrpc":"2.0","id":null,"method":1}',
JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=INVALID_REQUEST, message="Invalid request")),
),
(
"[]",
JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=INVALID_REQUEST, message="Invalid request")),
),
],
)
async def test_stdio_server_unparseable_line_completes_with_an_error_frame(line: str, expected: JSONRPCError) -> None:
"""A line that fails validation is answered on stdout with a single JSON-RPC error frame."""
received, sent = await _serve_raw_lines([line])

assert received == []
assert sent == [expected]


@contextmanager
Expand Down Expand Up @@ -608,14 +717,19 @@ def close(self) -> None:


def _serve_stdio_and_collect(
monkeypatch: pytest.MonkeyPatch, server: MCPServer, frames: list[JSONRPCRequest], responses: int
monkeypatch: pytest.MonkeyPatch,
server: MCPServer,
frames: list[JSONRPCRequest],
responses: int,
raw_prefix: bytes = b"",
) -> list[JSONRPCMessage]:
"""Serve `frames` over process stdio and return the parsed response lines.

Runs the blocking server.run("stdio") in a daemon thread and signals stdin EOF only after
`responses` lines arrive - as a real client would - so handlers never race EOF cancellation.
`raw_prefix` prepends untyped wire bytes (e.g. a malformed line) the typed frames cannot express.
"""
payload = "".join(f.model_dump_json(by_alias=True, exclude_none=True) + "\n" for f in frames).encode()
payload = raw_prefix + "".join(f.model_dump_json(by_alias=True, exclude_none=True) + "\n" for f in frames).encode()
stdin = _GatedStdin(payload)
stdout = _NotifyingStdout()
monkeypatch.setattr(sys, "stdin", TextIOWrapper(stdin, encoding="utf-8"))
Expand Down Expand Up @@ -643,6 +757,31 @@ def test_mcpserver_run_stdio_serves_until_stdin_closes(monkeypatch: pytest.Monke
assert responses == [JSONRPCResponse(jsonrpc="2.0", id=1, result={})]


def test_mcpserver_run_stdio_answers_a_deeply_nested_request_with_a_parse_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A deeply nested request over real stdio gets a parse error response instead of hanging.

End-to-end regression for #2751 through `MCPServer.run("stdio")`: the client
waiting on the nested request's id receives a response, and a later shallow
request on the same connection is still served.
"""
params: object = {"leaf": True}
for index in reversed(range(256)):
params = {f"p{index}": params}
deep_line = (json.dumps({"jsonrpc": "2.0", "id": 900256, "method": "ping", "params": params}) + "\n").encode()
ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")

responses = _serve_stdio_and_collect(
monkeypatch, MCPServer(name="DeepRequestStdioServer"), [ping], 2, raw_prefix=deep_line
)

assert responses == [
JSONRPCError(jsonrpc="2.0", id=900256, error=ErrorData(code=PARSE_ERROR, message="Parse error")),
JSONRPCResponse(jsonrpc="2.0", id=1, result={}),
]


def test_mcpserver_run_stdio_runs_lifespan_cleanup_after_stdin_closes(monkeypatch: pytest.MonkeyPatch) -> None:
"""Code after `yield` in a lifespan runs when stdin EOF ends `run("stdio")`.

Expand Down
Loading