fix(mcp-scan): preserve complete MCP tool results

mcp-scan/mcp_scan/utils/mcp_tools.py

@@ -27,6 +27,7 @@
 from mcp import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamable_http_client
+
 # MCP 2.0 exposes its recommended HTTP client factory from this private module.
 # requirements.txt pins 2.0.0; revisit this import when upgrading the SDK.
 from mcp.shared._httpx_utils import create_mcp_http_client
@@ -373,12 +374,101 @@ async def call_remote_tool(self, tool_name: str, **kw) -> Any:
                 result = await session.call_tool(tool_name, converted_kw)
                 if result is None:
                     return None
-                result = result.content[0]
-                # 判断TextContent or ImageContent or VideoContent
-                if hasattr(result, "text"):
-                    return result.text
-                elif hasattr(result, "data"):
-                    return result.data
+
+                # Empty content is valid; without other fields it formats to an empty string.
+                if not hasattr(result, "content"):
+                    raise RuntimeError(f"Unsupported MCP tool result type: {type(result).__name__}")
+
+                content_blocks = result.content or []
+                structured_content = getattr(result, "structured_content", None)
+                # Pydantic tracks whether JSON null was supplied or the field was absent.
+                structured_content_present = (
+                    structured_content is not None
+                    or "structured_content" in getattr(result, "model_fields_set", ())
+                )
+                result_meta = getattr(result, "meta", None)
+                is_error = bool(getattr(result, "is_error", False))
+                if (
+                    len(content_blocks) == 1
+                    and not structured_content_present
+                    and not is_error
+                    and result_meta is None
+                ):
+                    # Preserve the existing simple return when no result or
+                    # block metadata would be dropped.
+                    block = content_blocks[0]
+                    block_has_metadata = any(
+                        getattr(block, name, None) is not None for name in ("annotations", "meta")
+                    )
+                    if not block_has_metadata:
+                        if hasattr(block, "text"):
+                            return block.text
+                        elif hasattr(block, "data"):
+                            return block.data
+
+                # Preserve original block indexes while prioritizing readable
+                # content ahead of opaque media in the agent's bounded preview.
+                status_parts = ["[MCP tool result: isError=true]"] if is_error else []
+                text_parts = []
+                structured_parts = []
+                result_meta_parts = []
+                metadata_parts = []
+                media_parts = []
+                for index, block in enumerate(content_blocks):
+                    model_dump = getattr(block, "model_dump", None)
+                    block_data = model_dump(mode="json") if callable(model_dump) else None
+                    block_text = getattr(block, "text", None)
+                    if isinstance(block_text, str):
+                        text_parts.append(f"[content[{index}] text]\n{block_text}")
+                        if isinstance(block_data, dict):
+                            block_metadata = {
+                                key: block_data[key]
+                                for key in ("annotations", "meta")
+                                if block_data.get(key) is not None
+                            }
+                            if block_metadata:
+                                metadata_parts.append(
+                                    f"[content[{index}] metadata]\n"
+                                    + json.dumps(block_metadata, ensure_ascii=False, default=str)
+                                )
+                        continue
+
+                    if block_data is not None:
+                        serialized_block = json.dumps(block_data, ensure_ascii=False, default=str)
+                    elif hasattr(block, "data"):
+                        serialized_block = str(block.data)
+                    else:
+                        serialized_block = str(block)
+
+                    embedded_resource = getattr(block, "resource", None)
+                    block_part = f"[content[{index}]]\n{serialized_block}"
+                    if (
+                        hasattr(block, "data")
+                        or getattr(embedded_resource, "blob", None) is not None
+                    ):
+                        media_parts.append(block_part)
+                    else:
+                        metadata_parts.append(block_part)
+
+                if structured_content_present:
+                    structured_parts.append(
+                        "[structuredContent]\n"
+                        + json.dumps(structured_content, ensure_ascii=False, default=str)
+                    )
+                if result_meta is not None:
+                    result_meta_parts.append(
+                        "[result _meta]\n"
+                        + json.dumps(result_meta, ensure_ascii=False, default=str)
+                    )
+
+                return "\n".join(
+                    status_parts
+                    + text_parts
+                    + structured_parts
+                    + result_meta_parts
+                    + metadata_parts
+                    + media_parts
+                )
         except BaseExceptionGroup as eg:
             # 提取 TaskGroup 中的原始错误
             root_cause = self._extract_root_cause(eg)

mcp-scan/pytests/test_mcp_sdk_v2.py

@@ -1,7 +1,21 @@
 import unittest
 from contextlib import asynccontextmanager
+from types import SimpleNamespace
 from unittest.mock import patch
 
+from mcp.types import (
+    AudioContent,
+    CallToolResult,
+    EmbeddedResource,
+    ImageContent,
+    InputRequiredResult,
+    ResourceLink,
+    TextContent,
+    TextResourceContents,
+)
+
+from mcp_scan.agent.base_agent import BaseAgent
+from mcp_scan.tools.dispatcher import ToolDispatcher
 from mcp_scan.utils.mcp_tools import MCPTools
 
 
@@ -93,6 +107,169 @@ async def fake_sse_client(url, headers=None):
 
         self.assertEqual(captured["headers"], {"X-Key": "test"})
 
+    async def _call_remote_tool_with_result(self, result):
+        class _ToolResultSession:
+            async def call_tool(self, _tool_name, _arguments):
+                return result
+
+        manager = MCPTools("https://example.test/mcp", "streamable-http")
+
+        @asynccontextmanager
+        async def fake_session():
+            yield _ToolResultSession()
+
+        manager._session = fake_session
+        return await manager.call_remote_tool("inspect", value="test")
+
+    async def test_call_remote_tool_keeps_single_text_block_shape(self):
+        result = CallToolResult(content=[TextContent(type="text", text="single result")])
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertEqual(actual, "single result")
+
+    async def test_call_remote_tool_keeps_single_image_data_shape(self):
+        result = CallToolResult(
+            content=[ImageContent(type="image", data="aGVsbG8=", mime_type="image/png")]
+        )
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertEqual(actual, "aGVsbG8=")
+
+    async def test_call_remote_tool_preserves_multiple_text_blocks(self):
+        result = CallToolResult(
+            content=[
+                TextContent(type="text", text="first: harmless status"),
+                TextContent(type="text", text="second: security evidence"),
+            ]
+        )
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertLess(
+            actual.index("first: harmless status"), actual.index("second: security evidence")
+        )
+        self.assertIn("[content[0] text]", actual)
+        self.assertIn("[content[1] text]", actual)
+
+    async def test_call_remote_tool_keeps_text_after_image_block(self):
+        result = CallToolResult(
+            content=[
+                ImageContent(type="image", data="aGVsbG8=", mime_type="image/png"),
+                TextContent(type="text", text="text after image"),
+            ]
+        )
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertIn("image/png", actual)
+        self.assertIn("text after image", actual)
+
+    async def test_call_remote_tool_preserves_structured_content_without_blocks(self):
+        result = CallToolResult(content=[], structured_content={"finding": "structured evidence"})
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertIn("structured evidence", actual)
+
+    async def test_call_remote_tool_marks_errors_and_preserves_metadata(self):
+        result = CallToolResult(
+            content=[
+                TextContent(
+                    type="text",
+                    text="tool failure detail",
+                    meta={"block_source": "remote"},
+                    annotations={"priority": 0.7},
+                )
+            ],
+            meta={"trace_id": "trace-123"},
+            is_error=True,
+        )
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertIn("isError=true", actual)
+        self.assertIn("tool failure detail", actual)
+        self.assertIn("block_source", actual)
+        self.assertIn("priority", actual)
+        self.assertIn("trace_id", actual)
+
+    async def test_call_remote_tool_preserves_explicit_null_structured_content(self):
+        result = CallToolResult(content=[], structured_content=None)
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertEqual(actual, "[structuredContent]\nnull")
+
+    async def test_call_remote_tool_returns_empty_string_for_empty_success(self):
+        result = CallToolResult(content=[])
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertEqual(actual, "")
+
+    async def test_call_remote_tool_rejects_input_required_result_explicitly(self):
+        result = InputRequiredResult(request_state="pending")
+
+        with self.assertRaisesRegex(
+            RuntimeError, "Unsupported MCP tool result type: InputRequiredResult"
+        ):
+            await self._call_remote_tool_with_result(result)
+
+    async def test_call_remote_tool_serializes_audio_and_resource_blocks(self):
+        result = CallToolResult(
+            content=[
+                AudioContent(type="audio", data="YXVkaW8=", mime_type="audio/wav"),
+                ResourceLink(
+                    type="resource_link",
+                    name="risk report",
+                    uri="mcp://server/report",
+                    description="resource metadata evidence",
+                ),
+                EmbeddedResource(
+                    type="resource",
+                    resource=TextResourceContents(
+                        uri="mcp://server/evidence",
+                        mime_type="text/plain",
+                        text="embedded evidence",
+                    ),
+                ),
+            ]
+        )
+
+        actual = await self._call_remote_tool_with_result(result)
+
+        self.assertIn("audio/wav", actual)
+        self.assertIn("resource metadata evidence", actual)
+        self.assertIn("embedded evidence", actual)
+
+    async def test_call_remote_tool_keeps_late_text_in_large_result_history(self):
+        result = CallToolResult(
+            content=[
+                ImageContent(type="image", data="A" * 50_000, mime_type="image/png"),
+                TextContent(type="text", text="LATE SECURITY EVIDENCE"),
+            ]
+        )
+        tool_result = await self._call_remote_tool_with_result(result)
+        dispatcher = ToolDispatcher(mcp_server_url="https://example.test/mcp")
+        formatted_result = dispatcher._format_result(
+            {"tool_name": "inspect", "tool_result": tool_result}
+        )
+        agent = BaseAgent(
+            name="test",
+            instruction="",
+            llm=SimpleNamespace(context_window=8192),
+            dispatcher=dispatcher,
+        )
+
+        history_result = agent._build_history_tool_result("call_mcp_tool", {}, formatted_result)
+
+        self.assertTrue(history_result.startswith("[tool_result_summary]"))
+        self.assertLess(tool_result.index("LATE SECURITY EVIDENCE"), tool_result.index('"data":'))
+        preview = history_result.split("preview=\n", maxsplit=1)[1]
+        self.assertIn("LATE SECURITY EVIDENCE", preview)
+
 
 if __name__ == "__main__":
     unittest.main()