Structured Error Output

.changes/next-release/enhancement-Output-51826.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "Output",
+  "description": "Add support for ``--output off`` to suppress all stdout output while preserving stderr for errors and warnings."
+}

.changes/next-release/feature-Output-59989.json

@@ -0,0 +1,5 @@
+{
+  "type": "feature",
+  "category": "Output",
+  "description": "Add structured error output with configurable formats. AWS service errors now display additional fields in the configured format (legacy, json, yaml, text, table, or enhanced). Configure via ``--cli-error-format``, ``cli_error_format`` config variable, or ``AWS_CLI_ERROR_FORMAT`` environment variable. The new enhanced format is the default. Set ``cli_error_format=legacy`` to preserve the original error format."
+}

awscli/clidriver.py

@@ -30,6 +30,7 @@
     ScopedConfigProvider,
 )
 from botocore.context import start_as_current_context
+from botocore.exceptions import ClientError
 from botocore.history import get_global_history_recorder
 
 from awscli import __version__
@@ -119,7 +120,7 @@ def create_clidriver(args=None):
         session.full_config.get('plugins', {}),
         event_hooks=session.get_component('event_emitter'),
     )
-    error_handlers_chain = construct_cli_error_handlers_chain()
+    error_handlers_chain = construct_cli_error_handlers_chain(session)
     driver = CLIDriver(
         session=session, error_handler=error_handlers_chain, debug=debug
     )
@@ -246,7 +247,9 @@ def __init__(self, session=None, error_handler=None, debug=False):
             self.session = session
         self._error_handler = error_handler
         if self._error_handler is None:
-            self._error_handler = construct_cli_error_handlers_chain()
+            self._error_handler = construct_cli_error_handlers_chain(
+                self.session
+            )
         if debug:
             self._set_logging(debug)
         self._update_config_chain()
@@ -275,6 +278,9 @@ def _update_config_chain(self):
         config_store.set_config_provider(
             'cli_help_output', self._construct_cli_help_output_chain()
         )
+        config_store.set_config_provider(
+            'cli_error_format', self._construct_cli_error_format_chain()
+        )
 
     def _construct_cli_region_chain(self):
         providers = [
@@ -368,6 +374,20 @@ def _construct_cli_auto_prompt_chain(self):
         ]
         return ChainProvider(providers=providers)
 
+    def _construct_cli_error_format_chain(self):
+        providers = [
+            EnvironmentProvider(
+                name='AWS_CLI_ERROR_FORMAT',
+                env=os.environ,
+            ),
+            ScopedConfigProvider(
+                config_var_name='cli_error_format',
+                session=self.session,
+            ),
+            ConstantProvider(value='enhanced'),
+        ]
+        return ChainProvider(providers=providers)
+
     @property
     def subcommand_table(self):
         return self._get_command_table()
@@ -516,12 +536,14 @@ def main(self, args=None):
         command_table = self._get_command_table()
         parser = self.create_parser(command_table)
         self._add_aliases(command_table, parser)
+        parsed_globals = None
         try:
             # Because _handle_top_level_args emits events, it's possible
             # that exceptions can be raised, which should have the same
             # general exception handling logic as calling into the
             # command table.  This is why it's in the try/except clause.
             parsed_args, remaining = parser.parse_known_args(args)
+            parsed_globals = parsed_args
             self._handle_top_level_args(parsed_args)
             validate_preferred_output_encoding()
             self._emit_session_event(parsed_args)
@@ -538,6 +560,7 @@ def main(self, args=None):
                 e,
                 stdout=get_stdout_text_writer(),
                 stderr=get_stderr_text_writer(),
+                parsed_globals=parsed_globals,
             )
 
     def _emit_session_event(self, parsed_args):
@@ -818,7 +841,10 @@ def _parse_potential_subcommand(self, args, subcommand_table):
     def __call__(self, args, parsed_globals):
         # Once we know we're trying to call a particular operation
         # of a service we can go ahead and load the parameters.
-        event = f'before-building-argument-table-parser.{self._parent_name}.{self._name}'
+        event = (
+            f'before-building-argument-table-parser.'
+            f'{self._parent_name}.{self._name}'
+        )
         self._emit(
             event,
             argument_table=self.arg_table,
@@ -1028,9 +1054,7 @@ def _make_client_call(
             paginator = client.get_paginator(py_operation_name)
             response = paginator.paginate(**parameters)
         else:
-            response = getattr(client, xform_name(operation_name))(
-                **parameters
-            )
+            response = getattr(client, py_operation_name)(**parameters)
         return response
 
     def _display_response(self, command_name, response, parsed_globals):

awscli/data/cli.json

@@ -26,7 +26,8 @@
                 "text",
                 "table",
                 "yaml",
-                "yaml-stream"
+                "yaml-stream",
+                "off"
             ],
 	        "help": "<p>The formatting style for command output.</p>"
         },
@@ -85,6 +86,17 @@
         "no-cli-auto-prompt": {
             "action": "store_true",
             "help": "<p>Disable automatically prompt for CLI input parameters.</p>"
+        },
+        "cli-error-format": {
+            "choices": [
+                "legacy",
+                "json",
+                "yaml",
+                "text",
+                "table",
+                "enhanced"
+            ],
+            "help": "<p>The formatting style for error output. By default, errors are displayed in enhanced format.</p>"
         }
     }
 }

awscli/errorhandler.py

@@ -10,6 +10,7 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
+import argparse
 import logging
 import signal
 
@@ -36,10 +37,64 @@
     ConfigurationError,
     ParamValidationError,
 )
+from awscli.formatter import get_formatter
 from awscli.utils import PagerInitializationException
 
 LOG = logging.getLogger(__name__)
 
+VALID_ERROR_FORMATS = ['legacy', 'json', 'yaml', 'text', 'table', 'enhanced']
+# Maximum number of items to display inline for collections
+MAX_INLINE_ITEMS = 5
+
+
+class EnhancedErrorFormatter:
+    def format_error(self, error_info, formatted_message, stream):
+        stream.write(f'{formatted_message}\n')
+
+        additional_fields = self._get_additional_fields(error_info)
+
+        if not additional_fields:
+            return
+
+        stream.write('\nAdditional error details:\n')
+        for key, value in additional_fields.items():
+            if self._is_simple_value(value):
+                stream.write(f'{key}: {value}\n')
+            elif self._is_small_collection(value):
+                stream.write(f'{key}: {self._format_inline(value)}\n')
+            else:
+                stream.write(
+                    f'{key}: <complex value>\n'
+                    f'(Use --cli-error-format with json or yaml '
+                    f'to see full details)\n'
+                )
+
+    def _is_simple_value(self, value):
+        return isinstance(value, (str, int, float, bool, type(None)))
+
+    def _is_small_collection(self, value):
+        if isinstance(value, list):
+            return len(value) < MAX_INLINE_ITEMS and all(
+                self._is_simple_value(item) for item in value
+            )
+        elif isinstance(value, dict):
+            return len(value) < MAX_INLINE_ITEMS and all(
+                self._is_simple_value(v) for v in value.values()
+            )
+        return False
+
+    def _format_inline(self, value):
+        if isinstance(value, list):
+            return f"[{', '.join(str(item) for item in value)}]"
+        elif isinstance(value, dict):
+            items = ', '.join(f'{k}: {v}' for k, v in value.items())
+            return f'{{{items}}}'
+        return str(value)
+
+    def _get_additional_fields(self, error_info):
+        standard_keys = {'Code', 'Message'}
+        return {k: v for k, v in error_info.items() if k not in standard_keys}
+
 
 def construct_entry_point_handlers_chain():
     handlers = [
@@ -51,7 +106,7 @@ def construct_entry_point_handlers_chain():
     return ChainedExceptionHandler(exception_handlers=handlers)
 
 
-def construct_cli_error_handlers_chain():
+def construct_cli_error_handlers_chain(session=None):
     handlers = [
         ParamValidationErrorsHandler(),
         UnknownArgumentErrorHandler(),
@@ -60,7 +115,7 @@ def construct_cli_error_handlers_chain():
         NoCredentialsErrorHandler(),
         PagerErrorHandler(),
         InterruptExceptionHandler(),
-        ClientErrorHandler(),
+        ClientErrorHandler(session),
         GeneralExceptionHandler(),
     ]
     return ChainedExceptionHandler(exception_handlers=handlers)
@@ -75,13 +130,15 @@ class FilteredExceptionHandler(BaseExceptionHandler):
     EXCEPTIONS_TO_HANDLE = ()
     MESSAGE = '%s'
 
-    def handle_exception(self, exception, stdout, stderr):
+    def handle_exception(self, exception, stdout, stderr, **kwargs):
         if isinstance(exception, self.EXCEPTIONS_TO_HANDLE):
-            return_val = self._do_handle_exception(exception, stdout, stderr)
+            return_val = self._do_handle_exception(
+                exception, stdout, stderr, **kwargs
+            )
             if return_val is not None:
                 return return_val
 
-    def _do_handle_exception(self, exception, stdout, stderr):
+    def _do_handle_exception(self, exception, stdout, stderr, **kwargs):
         stderr.write("\n")
         stderr.write(self.MESSAGE % exception)
         stderr.write("\n")
@@ -108,6 +165,115 @@ class ClientErrorHandler(FilteredExceptionHandler):
     EXCEPTIONS_TO_HANDLE = ClientError
     RC = CLIENT_ERROR_RC
 
+    def __init__(self, session=None):
+        self._session = session
+        self._enhanced_formatter = None
+
+    def _do_handle_exception(self, exception, stdout, stderr, **kwargs):
+        parsed_globals = kwargs.get('parsed_globals')
+        displayed_structured = False
+        if self._session:
+            displayed_structured = self._try_display_structured_error(
+                exception, stderr, parsed_globals
+            )
+
+        if not displayed_structured:
+            return super()._do_handle_exception(
+                exception, stdout, stderr, **kwargs
+            )
+
+        return self.RC
+
+    def _resolve_error_format(self, parsed_globals):
+        if parsed_globals:
+            error_format = getattr(parsed_globals, 'cli_error_format', None)
+            if error_format:
+                return error_format.lower()
+        try:
+            error_format = self._session.get_config_variable(
+                'cli_error_format'
+            )
+            if error_format:
+                return error_format.lower()
+        except (KeyError, AttributeError) as e:
+            LOG.debug(
+                'Failed to get cli_error_format from config: %s', e
+            )
+
+        return 'enhanced'
+
+    def _try_display_structured_error(
+        self, exception, stderr, parsed_globals=None
+    ):
+        try:
+            error_response = self._extract_error_response(exception)
+            if not error_response or 'Error' not in error_response:
+                return False
+
+            error_info = error_response['Error']
+            error_format = self._resolve_error_format(parsed_globals)
+
+            if error_format not in VALID_ERROR_FORMATS:
+                LOG.warning(
+                    f"Invalid cli_error_format: '{error_format}'. "
+                    f"Using 'enhanced' format."
+                )
+                error_format = 'enhanced'
+
+            if error_format == 'legacy':
+                return False
+
+            formatted_message = str(exception)
+
+            if error_format == 'enhanced':
+                if self._enhanced_formatter is None:
+                    self._enhanced_formatter = EnhancedErrorFormatter()
+                self._enhanced_formatter.format_error(
+                    error_info, formatted_message, stderr
+                )
+                return True
+
+            temp_parsed_globals = argparse.Namespace()
+            temp_parsed_globals.output = error_format
+            temp_parsed_globals.query = None
+            temp_parsed_globals.color = (
+                getattr(parsed_globals, 'color', 'auto')
+                if parsed_globals
+                else 'auto'
+            )
+
+            formatter = get_formatter(error_format, temp_parsed_globals)
+            formatter('error', error_info, stderr)
+            return True
+
+        except Exception as e:
+            LOG.debug(
+                'Failed to display structured error: %s', e, exc_info=True
+            )
+            return False
+
+    @staticmethod
+    def _extract_error_response(exception):
+        if not isinstance(exception, ClientError):
+            return None
+
+        if hasattr(exception, 'response') and 'Error' in exception.response:
+            error_dict = dict(exception.response['Error'])
+
+            # AWS services return modeled error fields
+            # at the top level of the error response,
+            # not nested under an Error key. Botocore preserves this structure.
+            # Include these fields to provide complete error information.
+            # Exclude response metadata and avoid duplicates.
+            excluded_keys = {'Error', 'ResponseMetadata', 'Code', 'Message'}
+            for key, value in exception.response.items():
+                if key not in excluded_keys and key not in error_dict:
+                    error_dict[key] = value
+
+            return {'Error': error_dict}
+
+        return None
+
 
 class ConfigurationErrorHandler(FilteredExceptionHandler):
     EXCEPTIONS_TO_HANDLE = ConfigurationError
@@ -180,8 +346,10 @@ def __init__(self, exception_handlers):
     def inject_handler(self, position, handler):
         self._exception_handlers.insert(position, handler)
 
-    def handle_exception(self, exception, stdout, stderr):
+    def handle_exception(self, exception, stdout, stderr, **kwargs):
         for handler in self._exception_handlers:
-            return_value = handler.handle_exception(exception, stdout, stderr)
+            return_value = handler.handle_exception(
+                exception, stdout, stderr, **kwargs
+            )
             if return_value is not None:
                 return return_value