Protocols trait support

.changes/next-release/enhancement-Protocols-98789.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "Protocols",
+  "description": "Added support for multiple protocols within a service based on performance priority."
+}

botocore/args.py

@@ -70,6 +70,14 @@
     "when_required",
 )
 
+PRIORITY_ORDERED_SUPPORTED_PROTOCOLS = (
+    'json',
+    'rest-json',
+    'rest-xml',
+    'query',
+    'ec2',
+)
+
 
 class ClientArgsCreator:
     def __init__(
@@ -210,7 +218,7 @@ def compute_client_args(
         scoped_config,
     ):
         service_name = service_model.endpoint_prefix
-        protocol = service_model.metadata['protocol']
+        protocol = self._resolve_protocol(service_model)
         parameter_validation = True
         if client_config and not client_config.parameter_validation:
             parameter_validation = False
@@ -810,6 +818,23 @@ def _compute_checksum_config(self, config_kwargs):
             valid_options=VALID_RESPONSE_CHECKSUM_VALIDATION_CONFIG,
         )
 
+    def _resolve_protocol(self, service_model):
+        # We need to ensure `protocols` exists in the metadata before attempting to
+        # access it directly since referencing service_model.protocols directly will
+        # raise an UndefinedModelAttributeError if protocols is not defined
+        if service_model.metadata.get('protocols'):
+            for protocol in PRIORITY_ORDERED_SUPPORTED_PROTOCOLS:
+                if protocol in service_model.protocols:
+                    return protocol
+            raise botocore.exceptions.UnsupportedServiceProtocolsError(
+                botocore_supported_protocols=PRIORITY_ORDERED_SUPPORTED_PROTOCOLS,
+                service_supported_protocols=service_model.protocols,
+                service=service_model.service_name,
+            )
+        # If a service does not have a `protocols` trait, fall back to the legacy
+        # `protocol` trait
+        return service_model.protocol
+
     def _handle_checksum_config(
         self,
         config_kwargs,

botocore/exceptions.py

@@ -823,3 +823,12 @@ class InvalidChecksumConfigError(BotoCoreError):
         'Unsupported configuration value for {config_key}. '
         'Expected one of {valid_options} but got {config_value}.'
     )
+
+
+class UnsupportedServiceProtocolsError(BotoCoreError):
+    """Error when a service does not use any protocol supported by botocore."""
+
+    fmt = (
+        'Botocore supports {botocore_supported_protocols}, but service {service} only '
+        'supports {service_supported_protocols}.'
+    )

botocore/model.py

@@ -422,6 +422,10 @@ def api_version(self):
     def protocol(self):
         return self._get_metadata_property('protocol')
 
+    @CachedProperty
+    def protocols(self):
+        return self._get_metadata_property('protocols')
+
     @CachedProperty
     def endpoint_prefix(self):
         return self._get_metadata_property('endpointPrefix')

tests/functional/test_supported_protocols.py

@@ -0,0 +1,53 @@
+# Copyright 2025 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# 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 pytest
+
+from botocore.args import PRIORITY_ORDERED_SUPPORTED_PROTOCOLS
+from botocore.loaders import Loader
+from botocore.session import get_session
+
+
+def _get_services_models_by_protocols_trait(has_protocol_trait):
+    session = get_session()
+    service_list = Loader().list_available_services('service-2')
+    for service in service_list:
+        service_model = session.get_service_model(service)
+        if ('protocols' in service_model.metadata) == has_protocol_trait:
+            yield service_model
+
+
+@pytest.mark.validates_models
+@pytest.mark.parametrize(
+    "service",
+    _get_services_models_by_protocols_trait(True),
+)
+def test_services_with_protocols_trait_have_supported_protocol(service):
+    service_supported_protocols = service.metadata.get('protocols', [])
+    message = f"No protocols supported for service {service.service_name}"
+    assert any(
+        protocol in PRIORITY_ORDERED_SUPPORTED_PROTOCOLS
+        for protocol in service_supported_protocols
+    ), message
+
+
+@pytest.mark.validates_models
+@pytest.mark.parametrize(
+    "service",
+    _get_services_models_by_protocols_trait(False),
+)
+def test_services_without_protocols_trait_have_supported_protocol(service):
+    message = f"Service protocol not supported for {service.service_name}"
+    assert (
+        service.metadata.get('protocol')
+        in PRIORITY_ORDERED_SUPPORTED_PROTOCOLS
+    ), message

tests/unit/test_args.py

@@ -14,11 +14,15 @@
 import socket
 
 from botocore import args, exceptions
+from botocore.args import PRIORITY_ORDERED_SUPPORTED_PROTOCOLS
 from botocore.client import ClientEndpointBridge
 from botocore.config import Config
 from botocore.configprovider import ConfigValueStore
+from botocore.exceptions import UnsupportedServiceProtocolsError
 from botocore.hooks import HierarchicalEmitter
 from botocore.model import ServiceModel
+from botocore.parsers import PROTOCOL_PARSERS
+from botocore.serialize import SERIALIZERS
 from botocore.useragent import UserAgentString
 from tests import get_botocore_default_config_mapping, mock, unittest
 
@@ -63,9 +67,12 @@ def _get_service_model(self, service_name=None):
         service_model = mock.Mock(ServiceModel)
         service_model.service_name = service_name
         service_model.endpoint_prefix = service_name
+        service_model.protocol = 'query'
+        service_model.protocols = ['query']
         service_model.metadata = {
             'serviceFullName': 'MyService',
             'protocol': 'query',
+            'protocols': ['query'],
         }
         service_model.operation_names = []
         return service_model
@@ -106,6 +113,19 @@ def call_get_client_args(self, **override_kwargs):
         call_kwargs.update(**override_kwargs)
         return self.args_create.get_client_args(**call_kwargs)
 
+    def call_compute_client_args(self, **override_kwargs):
+        call_kwargs = {
+            'service_model': self.service_model,
+            'client_config': None,
+            'endpoint_bridge': self.bridge,
+            'region_name': self.region,
+            'is_secure': True,
+            'endpoint_url': self.endpoint_url,
+            'scoped_config': {},
+        }
+        call_kwargs.update(**override_kwargs)
+        return self.args_create.compute_client_args(**call_kwargs)
+
     def assert_create_endpoint_call(self, mock_endpoint, **override_kwargs):
         call_kwargs = {
             'endpoint_url': self.endpoint_url,
@@ -679,6 +699,25 @@ def test_response_checksum_validation_invalid_client_config(self):
         with self.assertRaises(exceptions.InvalidChecksumConfigError):
             self.call_get_client_args()
 
+    def test_protocol_resolution_without_protocols_trait(self):
+        del self.service_model.protocols
+        del self.service_model.metadata['protocols']
+        client_args = self.call_compute_client_args()
+        self.assertEqual(client_args['protocol'], 'query')
+
+    def test_protocol_resolution_picks_highest_supported(self):
+        self.service_model.protocol = 'query'
+        self.service_model.protocols = ['query', 'json']
+        client_args = self.call_compute_client_args()
+        self.assertEqual(client_args['protocol'], 'json')
+
+    def test_protocol_raises_error_for_unsupported_protocol(self):
+        self.service_model.protocols = ['wrongprotocol']
+        with self.assertRaisesRegex(
+            UnsupportedServiceProtocolsError, self.service_model.service_name
+        ):
+            self.call_compute_client_args()
+
 
 class TestEndpointResolverBuiltins(unittest.TestCase):
     def setUp(self):
@@ -907,3 +946,21 @@ def test_sdk_endpoint_legacy_set_without_builtin_data(self):
             legacy_endpoint_url='https://my.legacy.endpoint.com',
         )
         self.assertEqual(bins['SDK::Endpoint'], None)
+
+
+class TestProtocolPriorityList:
+    def test_all_parsers_accounted_for(self):
+        assert set(PRIORITY_ORDERED_SUPPORTED_PROTOCOLS) == set(
+            PROTOCOL_PARSERS.keys()
+        ), (
+            "The map of protocol names to parsers is out of sync with the priority "
+            "ordered list of protocols supported by botocore"
+        )
+
+    def test_all_serializers_accounted_for(self):
+        assert set(PRIORITY_ORDERED_SUPPORTED_PROTOCOLS) == set(
+            SERIALIZERS.keys()
+        ), (
+            "The map of protocol names to serializers is out of sync with the "
+            "priority ordered list of protocols supported by botocore"
+        )