added type hint

Author: KyMidd

src/strands/tools/mcp/mcp_client.py

@@ -23,7 +23,7 @@
 from mcp.types import GetPromptResult, ListPromptsResult
 from mcp.types import ImageContent as MCPImageContent
 from mcp.types import TextContent as MCPTextContent
-from mcp.types import EmbeddedResource as MCPEmbeddedResource  # <-- minimal add
+from mcp.types import EmbeddedResource as MCPEmbeddedResource
 
 from ...types import PaginatedList
 from ...types.exceptions import MCPClientInitializationError
@@ -429,7 +429,7 @@ def _background_task(self) -> None:
 
     def _map_mcp_content_to_tool_result_content(
         self,
-        content: MCPTextContent | MCPImageContent | Any,
+        content: MCPTextContent | MCPImageContent | MCPEmbeddedResource | Any,
     ) -> Union[ToolResultContent, None]:
         """Maps MCP content types to tool result content types.
 
@@ -453,7 +453,6 @@ def _map_mcp_content_to_tool_result_content(
                     "source": {"bytes": base64.b64decode(content.data)},
                 }
             }
-        # If EmbeddedResource
         elif isinstance(content, MCPEmbeddedResource):
             self._log_debug_with_thread("mapping MCP embedded resource content")
             res = getattr(content, "resource", None)
@@ -530,7 +529,12 @@ def _is_textual(mt: Optional[str]) -> bool:
             # Handle URI-only resources
             uri = _get("uri")
             if uri:
-                return {"text": f"[embedded resource] {uri} ({mime_type or 'unknown mime'})"}
+                return {
+                    "json": {
+                        "uri": uri,
+                        "mime_type": mime_type
+                    }
+                }
 
             # Make sure we return in all paths
             self._log_debug_with_thread("embedded resource had no usable text/blob/uri; dropping")

tests/strands/tools/mcp/test_mcp_client.py

@@ -590,4 +590,82 @@ def test_call_tool_sync_embedded_nested_base64_textual_mime(mock_transport, mock
         mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
         assert result["status"] == "success"
         assert len(result["content"]) == 1
-        assert result["content"][0]["text"] == '{"k":"v"}'
+        assert result["content"][0]["text"] == '{"k":"v"}'
+
+
+def test_call_tool_sync_embedded_image_blob(mock_transport, mock_session):
+    """EmbeddedResource.resource (blob with image MIME) should map to image content."""
+    import base64
+    # Create a simple 1x1 PNG image
+    png_data = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg==")
+    payload = base64.b64encode(png_data).decode()
+
+    er = {
+        "type": "resource",
+        "resource": {
+            "uri": "mcp://resource/embedded-image",
+            "blob": payload,
+            "mimeType": "image/png",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[er])
+
+    from strands.tools.mcp import MCPClient
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-image", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 1
+        assert "image" in result["content"][0]
+        assert result["content"][0]["image"]["format"] == "png"
+        assert "bytes" in result["content"][0]["image"]["source"]
+
+
+def test_call_tool_sync_embedded_non_textual_blob_dropped(mock_transport, mock_session):
+    """EmbeddedResource.resource (blob with non-textual/unknown MIME) should be dropped."""
+    payload = base64.b64encode(b'\x00\x01\x02\x03').decode()
+
+    er = {
+        "type": "resource",
+        "resource": {
+            "uri": "mcp://resource/embedded-binary",
+            "blob": payload,
+            "mimeType": "application/octet-stream",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[er])
+
+    from strands.tools.mcp import MCPClient
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-binary", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 0  # Content should be dropped
+
+
+def test_call_tool_sync_embedded_multiple_textual_mimes(mock_transport, mock_session):
+    """EmbeddedResource with different textual MIME types should decode to text."""
+    import base64
+    
+    # Test YAML content
+    yaml_content = base64.b64encode(b'key: value\nlist:\n  - item1\n  - item2').decode()
+    er = {
+        "type": "resource",
+        "resource": {
+            "uri": "mcp://resource/embedded-yaml",
+            "blob": yaml_content,
+            "mimeType": "application/yaml",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[er])
+
+    from strands.tools.mcp import MCPClient
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-yaml", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 1
+        assert "key: value" in result["content"][0]["text"]