From f4cd9623f13602a926b1939725e2aa247826def3 Mon Sep 17 00:00:00 2001 From: hylin Date: Tue, 25 Aug 2026 11:01:21 +0800 Subject: [PATCH] feat(server): surface structured validation errors in CallToolResult._meta Fixes #3351. When a tool call fails schema validation, the low-level Server previously returned only the interpolated free-text message, forcing clients to regex-match brittle wording to classify failures. This change forwards jsonschema.ValidationError's stable machine-readable fields (validator, validator_value, schema_path, json_path, message) into CallToolResult._meta under the MCP-namespaced key 'io.modelcontextprotocol/schema-validation-error', for both input- and output-schema failures. The human-readable message and isError=True stay unchanged, so existing clients keep working. - _make_error_result now accepts optional structured_data - new _validation_error_data helper extracts jsonschema fields - new _jsonable helper coerces non-JSON schema fragments (deques, sets, callables) to JSON-safe values before they cross the transport - added tests covering required/type/enum classification via _meta --- src/mcp/server/lowlevel/server.py | 65 ++++++++++++++- .../server/test_lowlevel_input_validation.py | 81 +++++++++++++++++++ 2 files changed, 142 insertions(+), 4 deletions(-) diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index 25a8fde37c..c7a69c13bf 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -109,6 +109,23 @@ async def main(): request_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any, Any]] = contextvars.ContextVar("request_ctx") +def _jsonable(value: Any) -> Any: + """Best-effort coerce a `jsonschema` schema fragment to JSON-serializable form. + + `ValidationError.validator_value` and `schema_path` elements can be + arbitrary Python objects (deques, sets, custom validators). We keep + JSON-native values as-is and fall back to `str(...)` for anything else + so the result can safely cross the JSON-RPC transport in `_meta`. + """ + if isinstance(value, (str, int, float, bool)) or value is None: + return value + if isinstance(value, dict): + return {str(k): _jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple, set, frozenset)): + return [_jsonable(v) for v in value] + return str(value) + + class NotificationOptions: def __init__( self, @@ -470,15 +487,49 @@ async def handler(req: types.ListToolsRequest): return decorator - def _make_error_result(self, error_message: str) -> types.ServerResult: - """Create a ServerResult with an error CallToolResult.""" + def _make_error_result( + self, + error_message: str, + *, + structured_data: dict[str, Any] | None = None, + ) -> types.ServerResult: + """Create a ServerResult with an error CallToolResult. + + When `structured_data` is provided, it is exposed on the result's `_meta` + field under `io.modelcontextprotocol/schema-validation-error`, so clients + can programmatically distinguish failure kinds (e.g. `required` vs + `type` vs `enum`) without regex-matching the free-text message. + """ + meta: dict[str, Any] | None = None + if structured_data is not None: + meta = {"io.modelcontextprotocol/schema-validation-error": structured_data} return types.ServerResult( types.CallToolResult( content=[types.TextContent(type="text", text=error_message)], isError=True, + **({"_meta": meta} if meta is not None else {}), ) ) + @staticmethod + def _validation_error_data( + e: jsonschema.ValidationError, *, kind: str + ) -> dict[str, Any]: + """Extract machine-readable fields from a `jsonschema.ValidationError`. + + `kind` distinguishes input-schema vs output-schema failures, and the + remaining fields mirror `jsonschema.ValidationError`'s stable + attributes so a client can classify failures without parsing prose. + """ + return { + "kind": kind, + "validator": e.validator, + "validator_value": _jsonable(e.validator_value), + "schema_path": [_jsonable(p) for p in e.schema_path], + "json_path": e.json_path, + "message": e.message, + } + async def _get_cached_tool_definition(self, tool_name: str) -> types.Tool | None: """Get tool definition from cache, refreshing if necessary. @@ -535,7 +586,10 @@ async def handler(req: types.CallToolRequest): try: jsonschema.validate(instance=arguments, schema=tool.inputSchema) except jsonschema.ValidationError as e: - return self._make_error_result(f"Input validation error: {e.message}") + return self._make_error_result( + f"Input validation error: {e.message}", + structured_data=self._validation_error_data(e, kind="input"), + ) # tool call results = await func(tool_name, arguments) @@ -572,7 +626,10 @@ async def handler(req: types.CallToolRequest): try: jsonschema.validate(instance=maybe_structured_content, schema=tool.outputSchema) except jsonschema.ValidationError as e: - return self._make_error_result(f"Output validation error: {e.message}") + return self._make_error_result( + f"Output validation error: {e.message}", + structured_data=self._validation_error_data(e, kind="output"), + ) # result return types.ServerResult( diff --git a/tests/server/test_lowlevel_input_validation.py b/tests/server/test_lowlevel_input_validation.py index 0614ad7c46..cdc6c2c283 100644 --- a/tests/server/test_lowlevel_input_validation.py +++ b/tests/server/test_lowlevel_input_validation.py @@ -309,3 +309,84 @@ async def test_callback(client_session: ClientSession) -> CallToolResult: assert any( "Tool 'unknown_tool' not listed, no validation will be performed" in record.message for record in caplog.records ) + + +_META_KEY = "io.modelcontextprotocol/schema-validation-error" + + +@pytest.mark.anyio +async def test_input_validation_error_carries_structured_meta(): + """Missing-required and type-mismatch failures both attach structured + `_meta["io.modelcontextprotocol/schema-validation-error"]` with the + `jsonschema` validator name and JSON path, so clients can classify + them without regexing the free-text message.""" + + async def call_tool_handler(name: str, arguments: dict[str, Any]) -> list[TextContent]: # pragma: no cover + raise RuntimeError("Should not reach here") + + async def missing_required(client_session: ClientSession) -> CallToolResult: + return await client_session.call_tool("add", {"a": 5}) # missing 'b' + + result_missing = await run_tool_test([create_add_tool()], call_tool_handler, missing_required) + + assert result_missing is not None + assert result_missing.isError + assert result_missing.meta is not None + payload = result_missing.meta[_META_KEY] + assert payload["kind"] == "input" + assert payload["validator"] == "required" + assert payload["json_path"] == "$" + assert "b" in payload["message"] + + async def wrong_type(client_session: ClientSession) -> CallToolResult: + return await client_session.call_tool("add", {"a": "five", "b": 3}) # 'a' should be number + + result_wrong = await run_tool_test([create_add_tool()], call_tool_handler, wrong_type) + + assert result_wrong is not None + assert result_wrong.isError + assert result_wrong.meta is not None + payload = result_wrong.meta[_META_KEY] + assert payload["kind"] == "input" + assert payload["validator"] == "type" + assert payload["validator_value"] == "number" + assert payload["json_path"] == "$.a" + + +@pytest.mark.anyio +async def test_enum_validation_error_carries_structured_meta(): + """Enum mismatches surface `validator="enum"` with the allowed values + intact in `validator_value`, so a client can render the choice list + without re-fetching the tool schema.""" + tools = [ + Tool( + name="greet", + description="Greet someone", + inputSchema={ + "type": "object", + "properties": { + "name": {"type": "string"}, + "title": {"type": "string", "enum": ["Mr", "Ms", "Dr"]}, + }, + "required": ["name"], + }, + ) + ] + + async def call_tool_handler(name: str, arguments: dict[str, Any]) -> list[TextContent]: # pragma: no cover + raise RuntimeError("Should not reach here") + + async def test_callback(client_session: ClientSession) -> CallToolResult: + return await client_session.call_tool("greet", {"name": "Smith", "title": "Prof"}) + + result = await run_tool_test(tools, call_tool_handler, test_callback) + + assert result is not None + assert result.isError + assert result.meta is not None + payload = result.meta[_META_KEY] + assert payload["kind"] == "input" + assert payload["validator"] == "enum" + assert payload["validator_value"] == ["Mr", "Ms", "Dr"] + assert payload["json_path"] == "$.title" +