Python: Prefer MCP structuredContent over duplicate content

python/packages/core/agent_framework/_mcp.py

@@ -390,6 +390,33 @@ def _should_propagate_cancelled_error(ex: BaseException) -> bool:
     return task is not None and task.cancelling() > 0
 
 
+def _structured_content_contains_text(structured: Any, text: str) -> bool:
+    """Return whether *text* appears as a string value anywhere in *structured*."""
+    if isinstance(structured, str):
+        return structured == text
+    if isinstance(structured, Mapping):
+        return any(_structured_content_contains_text(value, text) for value in structured.values())
+    if isinstance(structured, Sequence) and not isinstance(structured, (str, bytes, bytearray)):
+        return any(_structured_content_contains_text(item, text) for item in structured)
+    return False
+
+
+def _text_duplicates_structured_content(text: str, structured: Any, structured_json: str) -> bool:
+    """Return whether a text content block is an echo of ``structuredContent``.
+
+    MCP servers often return the same payload as both a text ``content`` block and
+    ``structuredContent`` (for example ``{"result": "<same text>"}``). Treat those as
+    duplicates so agents are not charged twice. Complementary text (a human-readable
+    summary that is not present in the structured payload) is kept.
+    """
+    if text == structured_json:
+        return True
+    with contextlib.suppress(json.JSONDecodeError, TypeError):
+        if json.loads(text) == structured:
+            return True
+    return _structured_content_contains_text(structured, text)
+
+
 # region: MCP Plugin
 
 
@@ -630,6 +657,12 @@ def _parse_tool_result_from_mcp(
     ) -> list[Content]:
         """Parse an MCP CallToolResult into a list of Content items.
 
+        When ``structuredContent`` is present it is emitted first. Text (or embedded
+        text) ``content`` blocks that merely echo that structured payload are skipped
+        so servers such as MS Learn / DeepWiki do not duplicate tokens (#7866).
+        Non-text blocks (images, audio, resources) and complementary text that is not
+        represented in ``structuredContent`` are retained.
+
         If the server attached a ``_meta`` payload to the tool result (e.g. for
         Information Flow Control labels under the ``ifc`` key), a copy of that
         payload is stamped onto each produced :class:`Content` instance under
@@ -647,10 +680,19 @@ def _parse_tool_result_from_mcp(
         # each newly constructed Content; empty when the server provided no meta.
         additional_kwargs: dict[str, Any] = {"additional_properties": {"_meta": meta}} if meta else {}
 
+        structured = mcp_type.structuredContent
+        structured_json: str | None = None
         result: list[Content] = []
+        if structured is not None:
+            structured_json = json.dumps(structured, default=str)
+            result.append(Content.from_text(structured_json, **additional_kwargs))
+
         for item in mcp_type.content:
             match item:
                 case types.TextContent():
+                    if structured is not None and structured_json is not None:
+                        if _text_duplicates_structured_content(item.text, structured, structured_json):
+                            continue
                     result.append(Content.from_text(item.text, **additional_kwargs))
                 case types.ImageContent() | types.AudioContent():
                     decoded = base64.b64decode(item.data)
@@ -672,6 +714,11 @@ def _parse_tool_result_from_mcp(
                 case types.EmbeddedResource():
                     match item.resource:
                         case types.TextResourceContents():
+                            if structured is not None and structured_json is not None:
+                                if _text_duplicates_structured_content(
+                                    item.resource.text, structured, structured_json
+                                ):
+                                    continue
                             result.append(Content.from_text(item.resource.text, **additional_kwargs))
                         case types.BlobResourceContents():
                             blob = item.resource.blob
@@ -686,10 +733,11 @@ def _parse_tool_result_from_mcp(
                                 )
                             )
                 case _:
-                    result.append(Content.from_text(str(item), **additional_kwargs))
-
-        if mcp_type.structuredContent is not None:
-            result.append(Content.from_text(json.dumps(mcp_type.structuredContent, default=str)))
+                    fallback = str(item)
+                    if structured is not None and structured_json is not None:
+                        if _text_duplicates_structured_content(fallback, structured, structured_json):
+                            continue
+                    result.append(Content.from_text(fallback, **additional_kwargs))
 
         if not result:
             result.append(Content.from_text("null", **additional_kwargs))

python/packages/core/tests/core/test_mcp.py

@@ -473,7 +473,7 @@ def test_parse_tool_result_from_mcp_structured_content_only():
 
 
 def test_parse_tool_result_from_mcp_structured_content_with_text():
-    """Test that structuredContent is appended alongside regular content items."""
+    """Complementary human-readable text is kept alongside structuredContent."""
     mcp_result = types.CallToolResult(
         content=[types.TextContent(type="text", text="Summary")],
         structuredContent={"data": [1, 2, 3]},
@@ -483,11 +483,68 @@ def test_parse_tool_result_from_mcp_structured_content_with_text():
     assert isinstance(result, list)
     assert len(result) == 2
     assert result[0].type == "text"
-    assert result[0].text == "Summary"
+    assert result[0].text is not None
+    assert json.loads(result[0].text) == {"data": [1, 2, 3]}
     assert result[1].type == "text"
-    assert result[1].text is not None
-    parsed = json.loads(result[1].text)
-    assert parsed == {"data": [1, 2, 3]}
+    assert result[1].text == "Summary"
+
+
+def test_parse_tool_result_from_mcp_does_not_duplicate_equivalent_structured_content():
+    """Regression for #7866: servers often echo the same payload in content and structuredContent."""
+    text = (
+        "This repository, `microsoft/agent-framework`, is a multi-language framework "
+        "designed for building, orchestrating, and deploying AI agents."
+    )
+    mcp_result = types.CallToolResult(
+        content=[types.TextContent(type="text", text=text)],
+        structuredContent={"result": text},
+    )
+    result = _HELPER_MCP_TOOL._parse_tool_result_from_mcp(mcp_result)
+
+    assert len(result) == 1
+    assert result[0].type == "text"
+    assert result[0].text is not None
+    assert json.loads(result[0].text) == {"result": text}
+
+
+def test_parse_tool_result_from_mcp_structured_content_stamps_meta():
+    """structuredContent results must still carry server ``_meta``."""
+    mcp_result = types.CallToolResult(
+        content=[],
+        structuredContent={"ok": True},
+        _meta={"ifc": {"integrity": "untrusted", "confidentiality": "public"}},
+    )
+    result = _HELPER_MCP_TOOL._parse_tool_result_from_mcp(mcp_result)
+
+    assert len(result) == 1
+    assert result[0].additional_properties.get("_meta") == {
+        "ifc": {"integrity": "untrusted", "confidentiality": "public"},
+    }
+    assert json.loads(result[0].text) == {"ok": True}
+
+
+def test_parse_tool_result_from_mcp_keeps_rich_content_with_structured():
+    """Non-text content blocks must not be dropped when structuredContent is present."""
+    mcp_result = types.CallToolResult(
+        content=[
+            types.ImageContent(
+                type="image",
+                data="ZmFrZS1pbWFnZS1ieXRlcw==",  # base64 for b"fake-image-bytes"
+                mimeType="image/png",
+            ),
+            types.TextContent(type="text", text="caption echoed in structured"),
+        ],
+        structuredContent={"caption": "caption echoed in structured", "width": 32},
+    )
+    result = _HELPER_MCP_TOOL._parse_tool_result_from_mcp(mcp_result)
+
+    assert len(result) == 2
+    assert result[0].type == "text"
+    assert result[0].text is not None
+    assert json.loads(result[0].text) == {"caption": "caption echoed in structured", "width": 32}
+    assert result[1].type == "data"
+    assert result[1].media_type == "image/png"
+    assert "ZmFrZS1pbWFnZS1ieXRlcw==" in result[1].uri  # type: ignore[operator]
 
 
 def test_parse_tool_result_from_mcp_structured_content_none():