feat(mcp): promote content-to-tool-result method to public API

strands-py/src/strands/tools/mcp/mcp_client.py

@@ -749,7 +749,7 @@ def _handle_tool_result(self, tool_use_id: str, call_tool_result: MCPCallToolRes
         mapped_contents: list[ToolResultContent] = [
             mc
             for content in call_tool_result.content
-            if (mc := self._map_mcp_content_to_tool_result_content(content)) is not None
+            if (mc := self.map_mcp_content_to_tool_result_content(content)) is not None
         ]
 
         status: ToolResultStatus = "error" if call_tool_result.isError else "success"
@@ -864,14 +864,16 @@ def _background_task(self) -> None:
         asyncio.set_event_loop(self._background_thread_event_loop)
         self._background_thread_event_loop.run_until_complete(self._async_background_thread())
 
-    def _map_mcp_content_to_tool_result_content(
+    def map_mcp_content_to_tool_result_content(
         self,
         content: MCPTextContent | MCPImageContent | MCPEmbeddedResource | Any,
     ) -> ToolResultContent | None:
         """Maps MCP content types to tool result content types.
 
         This method converts MCP-specific content types to the generic
-        ToolResultContent format used by the agent framework.
+        ToolResultContent format used by the agent framework. Subclasses can
+        override this to intercept or transform specific content blocks before
+        they reach the model.
 
         Args:
             content: The MCP content to convert
@@ -946,6 +948,7 @@ def _map_mcp_content_to_tool_result_content(
             self._log_debug_with_thread("unhandled content type: %s - dropping content", content.__class__.__name__)
             return None
 
+
     def _log_debug_with_thread(self, msg: str, *args: Any, **kwargs: Any) -> None:
         """Logger helper to help differentiate logs coming from MCPClient background thread."""
         formatted_msg = msg % args if args else msg

strands-py/tests/strands/tools/mcp/test_mcp_client.py

@@ -1165,3 +1165,29 @@ async def mock_awaitable():
         assert "https://example.com/auth" in result["content"][0]["text"]
         assert "Please authorize the application" in result["content"][0]["text"]
         assert "elicit-123" in result["content"][0]["text"]
+
+
+def test_map_mcp_content_subclass_override(mock_transport, mock_session):
+    """Subclass override of map_mcp_content_to_tool_result_content is invoked by the call site."""
+
+    class CustomMCPClient(MCPClient):
+        def map_mcp_content_to_tool_result_content(self, content):
+            result = super().map_mcp_content_to_tool_result_content(content)
+            if result and "text" in result:
+                result = {"text": "[intercepted]"}
+            return result
+
+    embedded_resource = {
+        "type": "resource",
+        "resource": {
+            "uri": "mcp://resource/test",
+            "text": "original text",
+            "mimeType": "text/plain",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[embedded_resource])
+
+    with CustomMCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="override-test", name="test_tool", arguments={})
+
+    assert result["content"][0]["text"] == "[intercepted]"