fix(server): surface structured validation errors in CallToolResult._meta (#3351)

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(

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"
+