FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

fix(server): surface structured validation errors in CallToolResult._meta (#3351) by linhongyu510 · Pull Request #3383 · modelcontextprotocol/python-sdk · GitHub

Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (2) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
65 changes: 61 additions & 4 deletions src/mcp/server/lowlevel/server.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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(
Expand Down
81 changes: 81 additions & 0 deletions tests/server/test_lowlevel_input_validation.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Loading

Back | FazBrowse Home | New Git URL