feat(client): add validate_structured_output option to ClientSession

src/mcp/client/client.py

@@ -95,6 +95,15 @@ async def main():
     elicitation_callback: ElicitationFnT | None = None
     """Callback for handling elicitation requests."""
 
+    validate_structured_output: bool = True
+    """Whether to validate structured tool output against the server's advertised output schema.
+
+    When True (the default), tool results whose structured_content does not match the tool's
+    output_schema cause a RuntimeError. Set to False to skip validation and return the
+    result unchanged, which is useful when interoperating with servers that ship buggy or
+    incomplete output schemas.
+    """
+
     _session: ClientSession | None = field(init=False, default=None)
     _exit_stack: AsyncExitStack | None = field(init=False, default=None)
     _transport: Transport = field(init=False)
@@ -126,6 +135,7 @@ async def __aenter__(self) -> Client:
                     message_handler=self.message_handler,
                     client_info=self.client_info,
                     elicitation_callback=self.elicitation_callback,
+                    validate_structured_output=self.validate_structured_output,
                 )
             )
 

src/mcp/client/session.py

@@ -121,6 +121,7 @@ def __init__(
         *,
         sampling_capabilities: types.SamplingCapability | None = None,
         experimental_task_handlers: ExperimentalTaskHandlers | None = None,
+        validate_structured_output: bool = True,
     ) -> None:
         super().__init__(read_stream, write_stream, read_timeout_seconds=read_timeout_seconds)
         self._client_info = client_info or DEFAULT_CLIENT_INFO
@@ -133,6 +134,7 @@ def __init__(
         self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
         self._initialize_result: types.InitializeResult | None = None
         self._experimental_features: ExperimentalClientFeatures | None = None
+        self._validate_structured_output = validate_structured_output
 
         # Experimental: Task handlers (use defaults if not provided)
         self._task_handlers = experimental_task_handlers or ExperimentalTaskHandlers()
@@ -323,6 +325,10 @@ async def call_tool(
 
     async def _validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
         """Validate the structured content of a tool result against its output schema."""
+        if not self._validate_structured_output:
+            logger.debug(f"Skipping structured output validation for tool {name}")
+            return
+
         if name not in self._tool_output_schemas:
             # refresh output schema cache
             await self.list_tools()

tests/client/test_output_schema_validation.py

@@ -163,3 +163,64 @@ async def on_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams)
         assert result.is_error is False
 
         assert "Tool mystery_tool not listed" in caplog.text
+
+
+@pytest.mark.anyio
+async def test_validate_structured_output_disabled_returns_invalid_result(caplog: pytest.LogCaptureFixture):
+    """When validate_structured_output is False, invalid structured_content is returned as-is."""
+    output_schema = {
+        "type": "object",
+        "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
+        "required": ["name", "age"],
+        "title": "UserOutput",
+    }
+
+    invalid_content = {"name": "John", "age": "not_an_int"}
+    server = _make_server(
+        tools=[
+            Tool(
+                name="get_user",
+                description="Get user data",
+                input_schema={"type": "object"},
+                output_schema=output_schema,
+            )
+        ],
+        structured_content=invalid_content,
+    )
+
+    caplog.set_level(logging.DEBUG, logger="client")
+
+    async with Client(server, validate_structured_output=False) as client:
+        result = await client.call_tool("get_user", {})
+        assert result.structured_content == invalid_content
+        assert result.is_error is False
+
+    assert "Skipping structured output validation for tool get_user" in caplog.text
+
+
+@pytest.mark.anyio
+async def test_validate_structured_output_default_still_raises():
+    """The default for validate_structured_output is True; invalid structured_content still raises."""
+    output_schema = {
+        "type": "object",
+        "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
+        "required": ["name", "age"],
+        "title": "UserOutput",
+    }
+
+    server = _make_server(
+        tools=[
+            Tool(
+                name="get_user",
+                description="Get user data",
+                input_schema={"type": "object"},
+                output_schema=output_schema,
+            )
+        ],
+        structured_content={"name": "John", "age": "not_an_int"},
+    )
+
+    async with Client(server) as client:
+        with pytest.raises(RuntimeError) as exc_info:
+            await client.call_tool("get_user", {})
+        assert "Invalid structured content returned by tool get_user" in str(exc_info.value)