diff --git a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/ClientGenerator.java b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/ClientGenerator.java
index e9f5d7a35..ceda9dc55 100644
--- a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/ClientGenerator.java
+++ b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/ClientGenerator.java
@@ -72,10 +72,23 @@ private void generateService(PythonWriter writer) {
             }
 
             writer.addDependency(SmithyPythonDependency.SMITHY_CORE);
+            var asyncConfigSymbol = CodegenUtils.getAsyncConfigSymbol(context.settings(), context.model());
             writer.write("""
-                    def __init__(self, config: $1T | None = None, plugins: list[$2T] | None = None):
+                    def __init__(
+                        self,
+                        config: $1T | $6T | None = None,
+                        plugins: list[$2T] | None = None,
+                    ):
                         $3C
-                        self._config = config or $1T()
+                        if isinstance(config, $6T):
+                            self._config: $1T = config  # type: ignore[assignment]
+                        elif isinstance(config, $1T) or config is None:
+                            self._config = config or $1T()
+                        else:
+                            raise $7T(
+                                f"config must be $6L or $1L, got {type(config).__name__}. "
+                                f"Use 'await $6L.resolve()' instead."
+                            )
 
                         client_plugins: list[$2T] = [
                             $4C
@@ -92,7 +105,9 @@ def __init__(self, config: $1T | None = None, plugins: list[$2T] | None = None):
                     pluginSymbol,
                     writer.consumer(w -> writeConstructorDocs(w, serviceSymbol.getName())),
                     writer.consumer(w -> writeDefaultPlugins(w, defaultPlugins)),
-                    RuntimeTypes.RETRY_STRATEGY_RESOLVER);
+                    RuntimeTypes.RETRY_STRATEGY_RESOLVER,
+                    asyncConfigSymbol,
+                    RuntimeTypes.EXPECTATION_NOT_MET_ERROR);
 
             var topDownIndex = TopDownIndex.of(model);
             var eventStreamIndex = EventStreamIndex.of(model);
@@ -249,7 +264,9 @@ private void writeSharedOperationInit(
                     raise $2T("protocol and transport MUST be set on the config to make calls.")
 
                 retry_strategy = await self._retry_strategy_resolver.resolve_retry_strategy(
-                    retry_strategy=config.retry_strategy
+                    retry_strategy=config.retry_strategy,
+                    retry_mode=getattr(config, "retry_mode", None),
+                    max_attempts=getattr(config, "max_attempts", None),
                 )
 
                 pipeline = $3T(
diff --git a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/CodegenUtils.java b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/CodegenUtils.java
index a6def8968..1e803f33d 100644
--- a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/CodegenUtils.java
+++ b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/CodegenUtils.java
@@ -24,6 +24,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.logging.Logger;
+import software.amazon.smithy.aws.traits.ServiceTrait;
 import software.amazon.smithy.codegen.core.CodegenException;
 import software.amazon.smithy.codegen.core.Symbol;
 import software.amazon.smithy.model.Model;
@@ -87,6 +88,48 @@ public static Symbol getPluginSymbol(PythonSettings settings) {
                 .build();
     }
 
+    /**
+     * Gets the async configuration object symbol for the service.
+     *
+     * <p>This is the new async-resolved config class that inherits from AsyncAwsConfig.
+     * Derives the name from the SDK ID (e.g., "Bedrock Runtime" becomes
+     * "AsyncBedrockRuntimeConfig"). Falls back to "AsyncConfig" for non-AWS services.
+     *
+     * @param settings The client settings.
+     * @param model The model containing the service shape.
+     * @return Returns the async config symbol.
+     */
+    public static Symbol getAsyncConfigSymbol(PythonSettings settings, Model model) {
+        var service = settings.service(model);
+        var name = service.getTrait(ServiceTrait.class)
+                .map(trait -> "Async" + StringUtils.capitalize(trait.getSdkId()).replace(" ", "") + "Config")
+                .orElse("AsyncConfig");
+        return Symbol.builder()
+                .name(name)
+                .namespace(String.format("%s.config", settings.moduleName()), ".")
+                .definitionFile(String.format("./src/%s/config.py", settings.moduleName()))
+                .build();
+    }
+
+    /**
+     * Gets the async plugin type hint symbol for the service.
+     *
+     * @param settings The client settings.
+     * @param model The model containing the service shape.
+     * @return Returns the async plugin type hint symbol.
+     */
+    public static Symbol getAsyncPluginSymbol(PythonSettings settings, Model model) {
+        var service = settings.service(model);
+        var name = service.getTrait(ServiceTrait.class)
+                .map(trait -> "Async" + StringUtils.capitalize(trait.getSdkId()).replace(" ", "") + "Plugin")
+                .orElse("AsyncPlugin");
+        return Symbol.builder()
+                .name(name)
+                .namespace(String.format("%s.config", settings.moduleName()), ".")
+                .definitionFile(String.format("./src/%s/config.py", settings.moduleName()))
+                .build();
+    }
+
     /**
      * Gets the service error symbol.
      *
diff --git a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java
index 30d2b07ef..359e77371 100644
--- a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java
+++ b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java
@@ -275,6 +275,24 @@ public void run() {
             writer.write("$L: TypeAlias = Callable[[$T], None]", plugin.getName(), config);
             writer.writeDocs("A callable that allows customizing the config object on each request.", context);
         });
+
+        // Generate the async config subclass and its plugin type
+        var model = context.model();
+        var asyncConfig = CodegenUtils.getAsyncConfigSymbol(context.settings(), model);
+        var asyncPlugin = CodegenUtils.getAsyncPluginSymbol(context.settings(), model);
+        context.writerDelegator().useFileWriter(asyncConfig.getDefinitionFile(), asyncConfig.getNamespace(), writer -> {
+            generateAsyncConfig(context, writer, asyncConfig);
+
+            // Generate the async plugin type alias
+            writer.addStdlibImport("typing", "Callable");
+            writer.addStdlibImport("typing", "TypeAlias");
+            writer.write("");
+            writer.write("");
+            writer.write("$L: TypeAlias = Callable[[$L], None]", asyncPlugin.getName(), asyncConfig.getName());
+            writer.writeDocs(
+                    "A callable that allows customizing the async config object on each request.",
+                    context);
+        });
     }
 
     private void writeInterceptorsType(PythonWriter writer) {
@@ -340,10 +358,16 @@ private void generateConfig(GenerationContext context, PythonWriter writer) {
         writer.pushState(new ConfigSection(finalProperties));
         writer.addLocallyDefinedSymbol(configSymbol);
         writer.addStdlibImport("dataclasses", "dataclass");
+        writer.addStdlibImport("warnings");
+        var asyncConfigName = CodegenUtils.getAsyncConfigSymbol(context.settings(), context.model()).getName();
         writer.write("""
                 @dataclass(init=False)
                 class $L:
-                    \"""Configuration for $L.\"""
+                    \"""Configuration for $L.
+
+                    .. deprecated::
+                        Use :class:`$L` with ``await $L.resolve()`` instead.
+                    \"""
 
                     ${C|}
 
@@ -352,12 +376,22 @@ def __init__(
                         *,
                         ${C|}
                     ):
+                        warnings.warn(
+                            "$L is deprecated, use $L.resolve() instead. "
+                            "This class will be removed in a future version.",
+                            DeprecationWarning,
+                            stacklevel=2,
+                        )
                         ${C|}
                 """,
                 configSymbol.getName(),
                 serviceId,
+                asyncConfigName,
+                asyncConfigName,
                 writer.consumer(w -> writePropertyDeclarations(w, finalProperties)),
                 writer.consumer(w -> writeInitParams(w, finalProperties)),
+                configSymbol.getName(),
+                asyncConfigName,
                 writer.consumer(w -> initializeProperties(w, finalProperties)));
         writer.popState();
     }
@@ -385,6 +419,162 @@ private void initializeProperties(PythonWriter writer, Collection<ConfigProperty
         }
     }
 
+    /**
+     * Generates the async config subclass that inherits from AsyncAwsConfig.
+     *
+     * <p>This class uses the FieldSpec-based resolution pipeline and adds
+     * service-specific fields (endpoint_resolver, protocol, auth_schemes,
+     * auth_scheme_resolver) with their defaults derived from the Smithy model.
+     */
+    private void generateAsyncConfig(GenerationContext context, PythonWriter writer, Symbol asyncConfigSymbol) {
+        var model = context.model();
+        var service = context.settings().service(model);
+        final String serviceId = service.getTrait(ServiceTrait.class)
+                .map(ServiceTrait::getSdkId)
+                .orElse(context.settings().service().getName());
+
+        // Import AsyncAwsConfig base class
+        writer.addDependency(SmithyPythonDependency.SMITHY_AWS_CORE);
+        var asyncAwsConfigSymbol = Symbol.builder()
+                .name("AsyncAwsConfig")
+                .namespace("smithy_aws_core.config.aws_config", ".")
+                .addDependency(SmithyPythonDependency.SMITHY_AWS_CORE)
+                .build();
+
+        // Import FieldSpec and ClassVar
+        var fieldSpecSymbol = Symbol.builder()
+                .name("FieldSpec")
+                .namespace("smithy_aws_core.config.types", ".")
+                .addDependency(SmithyPythonDependency.SMITHY_AWS_CORE)
+                .build();
+        writer.addStdlibImport("typing", "ClassVar");
+        writer.addStdlibImport("typing", "Any");
+        writer.addStdlibImport("dataclasses", "dataclass");
+
+        writer.write("");
+        writer.write("");
+        writer.write("@dataclass(kw_only=True)");
+        writer.openBlock("class $L($T):", asyncConfigSymbol.getName(), asyncAwsConfigSymbol);
+        writer.write("\"\"\"$L configuration (async-resolved).\"\"\"", serviceId);
+        writer.write("");
+
+        // Write service-specific field declarations
+        writer.write("endpoint_resolver: $T | None = None", RuntimeTypes.ENDPOINT_RESOLVER);
+        writer.write("protocol: $T | None = None",
+                Symbol.builder()
+                        .name("ClientProtocol[Any, Any]")
+                        .addReference(Symbol.builder()
+                                .name("ClientProtocol")
+                                .namespace("smithy_core.aio.interfaces", ".")
+                                .addDependency(SmithyPythonDependency.SMITHY_CORE)
+                                .build())
+                        .build());
+        writer.write("auth_schemes: dict[$T, $T] | None = None",
+                RuntimeTypes.SHAPE_ID,
+                Symbol.builder()
+                        .name("AuthScheme[Any, Any, Any, Any]")
+                        .addReference(Symbol.builder()
+                                .name("AuthScheme")
+                                .namespace("smithy_core.aio.interfaces.auth", ".")
+                                .addDependency(SmithyPythonDependency.SMITHY_CORE)
+                                .build())
+                        .build());
+        writer.write("auth_scheme_resolver: $T | None = None",
+                CodegenUtils.getHttpAuthSchemeResolverSymbol(context.settings()));
+        writer.write("");
+
+        // Write _FIELDS class variable with service-specific defaults
+        writer.openBlock("_FIELDS: ClassVar[dict[str, $T]] = {", fieldSpecSymbol);
+        writer.write("**$T._FIELDS,", asyncAwsConfigSymbol);
+
+        // endpoint_uri FieldSpec — overrides base class with service-aware resolver
+        var makeEndpointResolverSymbol = Symbol.builder()
+                .name("EndpointUriResolver")
+                .namespace("smithy_aws_core.config.resolvers", ".")
+                .addDependency(SmithyPythonDependency.SMITHY_AWS_CORE)
+                .build();
+        var snakeCaseServiceId = serviceId.replace(" ", "_").toLowerCase();
+        writer.write("\"endpoint_uri\": $T(", fieldSpecSymbol);
+        writer.indent();
+        writer.write("default=None,");
+        writer.write("resolver=$T($S),", makeEndpointResolverSymbol, snakeCaseServiceId);
+        writer.dedent();
+        writer.write("),");
+
+        // endpoint_resolver FieldSpec
+        var endpointPrefix = service.getTrait(ServiceTrait.class)
+                .map(ServiceTrait::getEndpointPrefix)
+                .orElse(context.settings().service().getName());
+        var standardRegionalResolverSymbol = Symbol.builder()
+                .name("StandardRegionalEndpointsResolver")
+                .namespace("smithy_aws_core.endpoints.standard_regional", ".")
+                .addDependency(SmithyPythonDependency.SMITHY_AWS_CORE)
+                .build();
+        writer.write("\"endpoint_resolver\": $T(", fieldSpecSymbol);
+        writer.indent();
+        writer.write("default_factory=lambda: $T(endpoint_prefix=$S),",
+                standardRegionalResolverSymbol,
+                endpointPrefix);
+        writer.dedent();
+        writer.write("),");
+
+        // protocol FieldSpec
+        writer.write("\"protocol\": $T(", fieldSpecSymbol);
+        writer.indent();
+        writer.write("default_factory=lambda: ${C|},",
+                writer.consumer(w -> context.protocolGenerator().initializeProtocol(context, w)));
+        writer.dedent();
+        writer.write("),");
+
+        // auth_schemes FieldSpec
+        writer.write("\"auth_schemes\": $T(", fieldSpecSymbol);
+        writer.indent();
+        writer.write("default_factory=lambda: ${C|},",
+                writer.consumer(w -> writeAsyncDefaultAuthSchemes(context, w)));
+        writer.dedent();
+        writer.write("),");
+
+        // auth_scheme_resolver FieldSpec
+        writer.write("\"auth_scheme_resolver\": $T(", fieldSpecSymbol);
+        writer.indent();
+        writer.write("default_factory=HTTPAuthSchemeResolver,");
+        writer.dedent();
+        writer.write("),");
+
+        // transport FieldSpec
+        writer.write("\"transport\": $T(", fieldSpecSymbol);
+        writer.indent();
+        if (usesHttp2(context)) {
+            writer.addDependency(SmithyPythonDependency.SMITHY_HTTP.withOptionalDependencies("awscrt"));
+            writer.write("default_factory=lambda: $T(),", RuntimeTypes.AWS_CRT_HTTP_CLIENT);
+        } else {
+            writer.addDependency(SmithyPythonDependency.SMITHY_HTTP.withOptionalDependencies("aiohttp"));
+            writer.write("default_factory=lambda: $T(),", RuntimeTypes.AIOHTTP_CLIENT);
+        }
+        writer.dedent();
+        writer.write("),");
+
+        writer.closeBlock("}");
+        writer.closeBlock("");
+    }
+
+    private static void writeAsyncDefaultAuthSchemes(GenerationContext context, PythonWriter writer) {
+        var service = context.settings().service(context.model());
+        writer.openBlock("{");
+        for (PythonIntegration integration : context.integrations()) {
+            for (RuntimeClientPlugin plugin : integration.getClientPlugins(context)) {
+                if (plugin.matchesService(context.model(), service) && plugin.getAuthScheme().isPresent()) {
+                    var scheme = plugin.getAuthScheme().get();
+                    writer.write("$T($S): ${C|},",
+                            RuntimeTypes.SHAPE_ID,
+                            scheme.getAuthTrait(),
+                            writer.consumer(w -> scheme.initializeScheme(context, writer, service)));
+                }
+            }
+        }
+        writer.closeBlock("}");
+    }
+
     private static final class AddAuthHelper implements CodeInterceptor<ConfigSection, PythonWriter> {
         @Override
         public Class<ConfigSection> sectionType() {
diff --git a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/EnumGenerator.java b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/EnumGenerator.java
index b38106b96..85ce09be0 100644
--- a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/EnumGenerator.java
+++ b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/EnumGenerator.java
@@ -41,20 +41,24 @@ public void run() {
             writer.addStdlibImport("enum", "StrEnum");
             writer.addDependency(SmithyPythonDependency.SMITHY_CORE);
             writer.addLocallyDefinedSymbol(enumSymbol);
-            writer.openBlock("class $L($T, StrEnum):", "", enumSymbol.getName(), RuntimeTypes.UNKNOWN_ENUM_MIXIN, () -> {
-                shape.getTrait(DocumentationTrait.class).ifPresent(trait -> {
-                    writer.writeDocs(trait.getValue(), context);
-                });
+            writer.openBlock("class $L($T, StrEnum):",
+                    "",
+                    enumSymbol.getName(),
+                    RuntimeTypes.UNKNOWN_ENUM_MIXIN,
+                    () -> {
+                        shape.getTrait(DocumentationTrait.class).ifPresent(trait -> {
+                            writer.writeDocs(trait.getValue(), context);
+                        });
 
-                for (MemberShape member : shape.members()) {
-                    var name = context.symbolProvider().toMemberName(member);
-                    var value = member.expectTrait(EnumValueTrait.class).expectStringValue();
-                    writer.write("$L = $S", name, value);
-                    member.getTrait(DocumentationTrait.class).ifPresent(trait -> {
-                        writer.writeDocs(trait.getValue(), context);
+                        for (MemberShape member : shape.members()) {
+                            var name = context.symbolProvider().toMemberName(member);
+                            var value = member.expectTrait(EnumValueTrait.class).expectStringValue();
+                            writer.write("$L = $S", name, value);
+                            member.getTrait(DocumentationTrait.class).ifPresent(trait -> {
+                                writer.writeDocs(trait.getValue(), context);
+                            });
+                        }
                     });
-                }
-            });
         });
     }
 }
diff --git a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/IntEnumGenerator.java b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/IntEnumGenerator.java
index e9fb98ecf..b29d17132 100644
--- a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/IntEnumGenerator.java
+++ b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/IntEnumGenerator.java
@@ -41,20 +41,24 @@ public void run() {
             writer.addStdlibImport("enum", "IntEnum");
             writer.addDependency(SmithyPythonDependency.SMITHY_CORE);
             writer.addLocallyDefinedSymbol(enumSymbol);
-            writer.openBlock("class $L($T, IntEnum):", "", enumSymbol.getName(), RuntimeTypes.UNKNOWN_ENUM_MIXIN, () -> {
-                directive.shape().getTrait(DocumentationTrait.class).ifPresent(trait -> {
-                    writer.writeDocs(trait.getValue(), directive.context());
-                });
+            writer.openBlock("class $L($T, IntEnum):",
+                    "",
+                    enumSymbol.getName(),
+                    RuntimeTypes.UNKNOWN_ENUM_MIXIN,
+                    () -> {
+                        directive.shape().getTrait(DocumentationTrait.class).ifPresent(trait -> {
+                            writer.writeDocs(trait.getValue(), directive.context());
+                        });
 
-                for (MemberShape member : directive.shape().members()) {
-                    var name = directive.symbolProvider().toMemberName(member);
-                    var value = member.expectTrait(EnumValueTrait.class).expectIntValue();
-                    writer.write("$L = $L", name, value);
-                    member.getTrait(DocumentationTrait.class).ifPresent(trait -> {
-                        writer.writeDocs(trait.getValue(), directive.context());
+                        for (MemberShape member : directive.shape().members()) {
+                            var name = directive.symbolProvider().toMemberName(member);
+                            var value = member.expectTrait(EnumValueTrait.class).expectIntValue();
+                            writer.write("$L = $L", name, value);
+                            member.getTrait(DocumentationTrait.class).ifPresent(trait -> {
+                                writer.writeDocs(trait.getValue(), directive.context());
+                            });
+                        }
                     });
-                }
-            });
         });
     }
 }
diff --git a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/UnionGenerator.java b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/UnionGenerator.java
index badf2ea60..1cd2c62ea 100644
--- a/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/UnionGenerator.java
+++ b/codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/UnionGenerator.java
@@ -157,30 +157,31 @@ private void generateDeserializer() {
         var schemaSymbol = symbol.expectProperty(SymbolProperties.SCHEMA);
         var unknownSymbol = symbol.expectProperty(SymbolProperties.UNION_UNKNOWN);
         writer.putContext("schema", schemaSymbol);
-        writer.write("""
-                class $1L:
-                    _result: $2T | None = None
-
-                    def deserialize(self, deserializer: ${shapeDeserializer:T}) -> $2T:
-                        self._result = None
-                        deserializer.read_struct($3T, self._consumer)
-
-                        if self._result is None:
-                            raise ${serializationError:T}("Unions must have exactly one value, but found none.")
-
-                        return self._result
-
-                    def _consumer(self, schema: $4T, de: ${shapeDeserializer:T}) -> None:
-                        match schema.expect_member_index():
-                            ${5C|}
-                            case _:
-                                self._set_result($6L(tag=schema.expect_member_name()))
-
-                    def _set_result(self, value: $2T) -> None:
-                        if self._result is not None:
-                            raise ${serializationError:T}("Unions must have exactly one value, but found more than one.")
-                        self._result = value
-                """,
+        writer.write(
+                """
+                        class $1L:
+                            _result: $2T | None = None
+
+                            def deserialize(self, deserializer: ${shapeDeserializer:T}) -> $2T:
+                                self._result = None
+                                deserializer.read_struct($3T, self._consumer)
+
+                                if self._result is None:
+                                    raise ${serializationError:T}("Unions must have exactly one value, but found none.")
+
+                                return self._result
+
+                            def _consumer(self, schema: $4T, de: ${shapeDeserializer:T}) -> None:
+                                match schema.expect_member_index():
+                                    ${5C|}
+                                    case _:
+                                        self._set_result($6L(tag=schema.expect_member_name()))
+
+                            def _set_result(self, value: $2T) -> None:
+                                if self._result is not None:
+                                    raise ${serializationError:T}("Unions must have exactly one value, but found more than one.")
+                                self._result = value
+                        """,
                 deserializerSymbol.getName(),
                 symbol,
                 schemaSymbol,
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py b/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
index fc5b8f61c..84fcf144c 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/aws_config.py
@@ -1,18 +1,29 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+import os
 from dataclasses import dataclass, field
-from typing import Any, ClassVar, Self
+from typing import TYPE_CHECKING, Any, ClassVar, Self
 
 from smithy_core.retries import RetryStrategyOptions
 
+if TYPE_CHECKING:
+    from smithy_core.aio.interfaces import ClientTransport
+    from smithy_core.aio.interfaces.identity import IdentityResolver
+    from smithy_core.interfaces import URI
+    from smithy_http.interfaces import HTTPRequestConfiguration
+
+    from smithy_aws_core.identity import AWSCredentialsIdentity, AWSIdentityProperties
+
 from .context import SharedConfigContext
 from .exceptions import ConfigError, ConfigValidationError
 from .filesystem import FileSystem
 from .resolvers import (
+    resolve_endpoint_uri,
     resolve_max_attempts,
     resolve_region,
     resolve_retry_mode,
+    resolve_sdk_ua_app_id,
 )
 from .types import UNSET, ConfigSource, FieldSpec, Resolved
 from .validators import (
@@ -22,6 +33,8 @@
     validate_retry_mode,
 )
 
+_CREDENTIAL_FIELDS = ("aws_access_key_id", "aws_secret_access_key", "aws_session_token")
+
 
 @dataclass(kw_only=True)
 class AsyncAwsConfig:
@@ -37,6 +50,17 @@ class AsyncAwsConfig:
     region: str | None = None
     retry_mode: str | None = None
     max_attempts: int | None = None
+    endpoint_uri: "str | URI | None" = None
+    aws_access_key_id: str | None = field(default=None, repr=False)
+    aws_secret_access_key: str | None = field(default=None, repr=False)
+    aws_session_token: str | None = field(default=None, repr=False)
+    aws_credentials_identity_resolver: "IdentityResolver[AWSCredentialsIdentity, AWSIdentityProperties] | None" = None
+    sdk_ua_app_id: str | None = None
+    user_agent_extra: str | None = None
+    interceptors: list[Any] = field(default_factory=list)  # type: ignore
+    http_request_config: "HTTPRequestConfiguration | None" = None
+    transport: "ClientTransport[Any, Any] | None" = None
+    retry_strategy: Any | None = None
 
     _ctx: SharedConfigContext | None = field(default=None, repr=False, compare=False)
     _sources: dict[str, ConfigSource] = field(  # type: ignore[assignment]
@@ -61,6 +85,41 @@ class AsyncAwsConfig:
             resolver=resolve_max_attempts,
             validator=validate_max_attempts,
         ),
+        "endpoint_uri": FieldSpec(
+            default=None,
+            resolver=resolve_endpoint_uri,
+        ),
+        "aws_access_key_id": FieldSpec(
+            default=None,
+        ),
+        "aws_secret_access_key": FieldSpec(
+            default=None,
+        ),
+        "aws_session_token": FieldSpec(
+            default=None,
+        ),
+        "aws_credentials_identity_resolver": FieldSpec(
+            default=None,
+        ),
+        "sdk_ua_app_id": FieldSpec(
+            default=None,
+            resolver=resolve_sdk_ua_app_id,
+        ),
+        "user_agent_extra": FieldSpec(
+            default=None,
+        ),
+        "interceptors": FieldSpec(
+            default_factory=list,
+        ),
+        "http_request_config": FieldSpec(
+            default=None,
+        ),
+        "transport": FieldSpec(
+            default=None,
+        ),
+        "retry_strategy": FieldSpec(
+            default=None,
+        ),
     }
 
     def __post_init__(self) -> None:
@@ -150,7 +209,15 @@ async def _resolve_fields(self, overrides: dict[str, Any]) -> None:
                 f"Valid fields are: {sorted(self._FIELDS)}"
             )
 
+        # Resolve credentials atomically before the field loop
+        await self._resolve_credentials(overrides)
+
         for field_name, spec in self._FIELDS.items():
+            # Skip credentials — already resolved atomically above
+            if field_name in _CREDENTIAL_FIELDS:
+                if field_name in self._sources:
+                    continue
+
             # check for overrides first
             if field_name in overrides:
                 value = overrides[field_name]
@@ -183,8 +250,88 @@ def _apply_default(self, field_name: str, spec: FieldSpec) -> None:
         setattr(self, field_name, value)
         self._sources[field_name] = ConfigSource.DEFAULT
 
+    async def _resolve_credentials(self, overrides: dict[str, Any]) -> None:
+        """Resolve credential fields atomically from a single source.
+
+        Rules:
+        - If both aws_access_key_id and aws_secret_access_key are overridden,
+          resolve normally
+        - If only one credential is overridden, raise ConfigValidationError.
+        - Otherwise, resolve atomically: if both key and secret are present in
+          env, take all three from env. If both are in the profile, take all
+          three from profile. Token may be None in either case.
+
+        This prevents mixing credentials from different sources.
+        """
+
+        required = {"aws_access_key_id", "aws_secret_access_key"}
+
+        cred_overrides = {f for f in _CREDENTIAL_FIELDS if f in overrides}
+        if cred_overrides:
+            if required <= cred_overrides:
+                return
+            else:
+                raise ConfigValidationError(
+                    f"Partial credential override: {sorted(cred_overrides)}. "
+                    "Both 'aws_access_key_id' and 'aws_secret_access_key' must be "
+                    "provided together when overriding credentials."
+                )
+
+        # Check env vars atomically
+        env_creds = (
+            (os.environ.get("AWS_ACCESS_KEY_ID") or "").strip() or None,
+            (os.environ.get("AWS_SECRET_ACCESS_KEY") or "").strip() or None,
+            (os.environ.get("AWS_SESSION_TOKEN") or "").strip() or None,
+        )
+        if env_creds[0] and env_creds[1]:
+            self._set_credentials(_CREDENTIAL_FIELDS, env_creds, ConfigSource.ENV)
+            return
+
+        # Check profile atomically
+        ctx = self._ctx
+        if ctx is None:
+            raise ConfigError("Resolution context not initialized")
+        config_file = await ctx.parsed_profiles()
+        profile_creds = (
+            config_file.get(ctx.profile_name, "aws_access_key_id"),
+            config_file.get(ctx.profile_name, "aws_secret_access_key"),
+            config_file.get(ctx.profile_name, "aws_session_token"),
+        )
+        if profile_creds[0] and profile_creds[1]:
+            self._set_credentials(
+                _CREDENTIAL_FIELDS, profile_creds, ConfigSource.PROFILE
+            )
+
+    def _set_credentials(
+        self,
+        fields: tuple[str, ...],
+        values: tuple[str | None, ...],
+        source: ConfigSource,
+    ) -> None:
+        """Set credential fields atomically, bypassing __setattr__ tracking."""
+        for field_name, value in zip(fields, values, strict=True):
+            object.__setattr__(self, field_name, value or None)
+            self._sources[field_name] = source
+
     def __setattr__(self, name: str, value: Any) -> None:
         """Track provenance when fields are set with plugins after construction"""
+        # Reject unknown fields
+        if not name.startswith("_") and name not in self.__class__._FIELDS:
+            raise AttributeError(
+                f"'{type(self).__name__}' has no config field '{name}'"
+            )
+
+        # Block override for credentials after resolution
+        if (
+            name in _CREDENTIAL_FIELDS
+            and hasattr(self, "_sources")
+            and name in self._sources
+        ):
+            raise AttributeError(
+                f"'{name}' cannot be modified after resolution. "
+                "Create a new config with the desired credentials instead."
+            )
+
         # Mark as override only if the field is in _FIELDS and was already resolved
         if (
             name in self.__class__._FIELDS
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/merged_config.py b/packages/smithy-aws-core/src/smithy_aws_core/config/merged_config.py
index 5e550732a..f496d9aca 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/merged_config.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/merged_config.py
@@ -113,3 +113,47 @@ def _merge_profiles(
             else:
                 merged[name] = Section(properties=dict(section.properties))
         return merged
+
+    def get_service_config(
+        self, profile_name: str, service_id: str, key: str
+    ) -> str | None:
+        """Get a config value from the services section for a specific service.
+
+        Looks up the services section referenced by the profile, then finds
+        the service-specific sub-property within it.
+
+        For a config file like:
+            [profile default]
+            services = my-services
+
+            [services my-services]
+            bedrock_runtime =
+              endpoint_url = http://localhost:5678
+
+        Usage: get_service_config("default", "bedrock_runtime", "endpoint_url")
+
+        :param profile_name: The profile name to look up.
+        :param service_id: The service identifier (lowercase, underscored).
+        :param key: The property key within the service section.
+
+        :returns: The value, or None if not found.
+        """
+        # Get the services section name from the profile
+        profile = self._profiles.get(profile_name)
+        if profile is None:
+            return None
+        services_name = profile.properties.get("services")
+        if not services_name or not isinstance(services_name, str):
+            return None
+
+        # Look up the services section
+        services_section = self._services.get(services_name)
+        if services_section is None:
+            return None
+
+        # Get the service-specific sub-property
+        service_props = services_section.properties.get(service_id)
+        if not isinstance(service_props, dict):
+            return None
+
+        return service_props.get(key.lower())
diff --git a/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py b/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
index 7f48ff99c..f09ecb54c 100644
--- a/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
+++ b/packages/smithy-aws-core/src/smithy_aws_core/config/resolvers.py
@@ -120,3 +120,82 @@ async def resolve_max_attempts(ctx: SharedConfigContext) -> Resolved[int | None]
         env_vars=("AWS_MAX_ATTEMPTS",),
         profile_keys=("max_attempts",),
     )
+
+
+async def resolve_endpoint_uri(ctx: SharedConfigContext) -> Resolved[str | None]:
+    """Resolve the endpoint URI from global environment or config file.
+
+    This is the base resolver that only checks global sources.
+    For service-specific resolution, use EndpointUriResolver().
+
+    :param ctx: The shared resolution context.
+    :returns: Resolved endpoint URI value with source.
+    """
+    return await _resolve_str(
+        ctx,
+        env_vars=("AWS_ENDPOINT_URL",),
+        profile_keys=("endpoint_url",),
+    )
+
+
+async def resolve_sdk_ua_app_id(ctx: SharedConfigContext) -> Resolved[str | None]:
+    """Resolve the SDK user-agent app ID from environment or config file.
+
+    :param ctx: The shared resolution context.
+    :returns: Resolved app ID value with source.
+    """
+    return await _resolve_str(
+        ctx,
+        env_vars=("AWS_SDK_UA_APP_ID",),
+        profile_keys=("sdk_ua_app_id",),
+    )
+
+
+class EndpointUriResolver:
+    """Service-aware endpoint URI resolver.
+
+    Resolution order (first match wins):
+    1. Service-specific env var (AWS_ENDPOINT_URL_<SERVICE_ID>)
+    2. Global env var (AWS_ENDPOINT_URL)
+    3. Service-specific config file (services section -> service_id -> endpoint_url)
+    4. Global config file (profile -> endpoint_url)
+    """
+
+    def __init__(self, service_id: str):
+        """Initialize with a service identifier.
+
+        :param service_id: The service identifier (e.g., "bedrock_runtime").
+            Used to construct the service-specific env var and config lookup key.
+        """
+        self._service_env_var = (
+            f"AWS_ENDPOINT_URL_{service_id.replace(' ', '_').replace('-', '_').upper()}"
+        )
+
+        self._service_key = service_id.replace(" ", "_").replace("-", "_").lower()
+
+    async def __call__(self, ctx: SharedConfigContext) -> Resolved[str | None]:
+        """Resolve the endpoint URI from all sources.
+
+        :param ctx: The shared resolution context.
+        :returns: Resolved endpoint URI value with source.
+        """
+        value = os.environ.get(self._service_env_var)
+        if value:
+            return Resolved(value=value, source=ConfigSource.ENV)
+
+        value = os.environ.get("AWS_ENDPOINT_URL")
+        if value:
+            return Resolved(value=value, source=ConfigSource.ENV)
+
+        config_file = await ctx.parsed_profiles()
+        value = config_file.get_service_config(
+            ctx.profile_name, self._service_key, "endpoint_url"
+        )
+        if value:
+            return Resolved(value=value, source=ConfigSource.PROFILE)
+
+        value = config_file.get(ctx.profile_name, "endpoint_url")
+        if value:
+            return Resolved(value=value, source=ConfigSource.PROFILE)
+
+        return Resolved(value=UNSET, source=ConfigSource.DEFAULT)  # type: ignore[arg-type]
diff --git a/packages/smithy-aws-core/tests/unit/config/test_merged_config.py b/packages/smithy-aws-core/tests/unit/config/test_merged_config.py
index 53322d90c..eca880cdf 100644
--- a/packages/smithy-aws-core/tests/unit/config/test_merged_config.py
+++ b/packages/smithy-aws-core/tests/unit/config/test_merged_config.py
@@ -227,3 +227,100 @@ def test_services_property(self):
         )
         assert "my-svc" in cf.services
         assert cf.services["my-svc"].properties == {"endpoint_url": "http://localhost"}
+
+
+class TestGetServiceConfig:
+    """Tests for MergedConfig.get_service_config()"""
+
+    def test_returns_service_specific_endpoint_url(self):
+        config_data = StandardizedOutput(
+            profiles={"default": Section(properties={"services": "my-services"})},
+            services={
+                "my-services": Section(
+                    properties={
+                        "bedrock_runtime": {"endpoint_url": "https://custom.com"}
+                    }
+                )
+            },
+        )
+        cf = MergedConfig(config_data, StandardizedOutput())
+        assert (
+            cf.get_service_config("default", "bedrock_runtime", "endpoint_url")
+            == "https://custom.com"
+        )
+
+    def test_returns_none_when_profile_missing(self):
+        config_data = StandardizedOutput()
+        cf = MergedConfig(config_data, StandardizedOutput())
+        assert (
+            cf.get_service_config("default", "bedrock_runtime", "endpoint_url") is None
+        )
+
+    def test_returns_none_when_no_services_key_in_profile(self):
+        config_data = StandardizedOutput(
+            profiles={"default": Section(properties={"region": "us-east-1"})},
+        )
+        cf = MergedConfig(config_data, StandardizedOutput())
+        assert (
+            cf.get_service_config("default", "bedrock_runtime", "endpoint_url") is None
+        )
+
+    def test_returns_none_when_services_section_not_found(self):
+        config_data = StandardizedOutput(
+            profiles={"default": Section(properties={"services": "nonexistent"})},
+            services={},
+        )
+        cf = MergedConfig(config_data, StandardizedOutput())
+        assert (
+            cf.get_service_config("default", "bedrock_runtime", "endpoint_url") is None
+        )
+
+    def test_returns_none_when_service_id_not_in_section(self):
+        config_data = StandardizedOutput(
+            profiles={"default": Section(properties={"services": "my-services"})},
+            services={
+                "my-services": Section(
+                    properties={"dynamodb": {"endpoint_url": "https://dynamo.local"}}
+                )
+            },
+        )
+        cf = MergedConfig(config_data, StandardizedOutput())
+        assert (
+            cf.get_service_config("default", "bedrock_runtime", "endpoint_url") is None
+        )
+
+    def test_returns_none_when_key_not_in_service(self):
+        config_data = StandardizedOutput(
+            profiles={"default": Section(properties={"services": "my-services"})},
+            services={
+                "my-services": Section(
+                    properties={"bedrock_runtime": {"some_other_key": "value"}}
+                )
+            },
+        )
+        cf = MergedConfig(config_data, StandardizedOutput())
+        assert (
+            cf.get_service_config("default", "bedrock_runtime", "endpoint_url") is None
+        )
+
+    def test_multiple_services_in_section(self):
+        config_data = StandardizedOutput(
+            profiles={"default": Section(properties={"services": "my-services"})},
+            services={
+                "my-services": Section(
+                    properties={
+                        "bedrock_runtime": {"endpoint_url": "https://bedrock.local"},
+                        "dynamodb": {"endpoint_url": "https://dynamo.local"},
+                    }
+                )
+            },
+        )
+        cf = MergedConfig(config_data, StandardizedOutput())
+        assert (
+            cf.get_service_config("default", "bedrock_runtime", "endpoint_url")
+            == "https://bedrock.local"
+        )
+        assert (
+            cf.get_service_config("default", "dynamodb", "endpoint_url")
+            == "https://dynamo.local"
+        )
diff --git a/packages/smithy-aws-core/tests/unit/config/test_resolver.py b/packages/smithy-aws-core/tests/unit/config/test_resolver.py
index 5559cc35d..6e8df112f 100644
--- a/packages/smithy-aws-core/tests/unit/config/test_resolver.py
+++ b/packages/smithy-aws-core/tests/unit/config/test_resolver.py
@@ -18,6 +18,7 @@
     ProfileNotFoundError,
 )
 from smithy_aws_core.config.resolvers import (
+    EndpointUriResolver,
     resolve_max_attempts,
     resolve_region,
     resolve_retry_mode,
@@ -177,7 +178,7 @@ async def test_invalid_override_triggers_validator(self):
             with pytest.raises(
                 ConfigValidationError, match="Must be a valid AWS region"
             ):
-                await AsyncAwsConfig.resolve(region="bad-value!")
+                await AsyncAwsConfig.resolve(region="bad-value!", fs=NullFileSystem())
 
     @pytest.mark.asyncio
     async def test_invalid_profile_raises_error(self):
@@ -257,6 +258,30 @@ async def test_explicit_default_profile_is_validated(self):
                     fs=NullFileSystem(),
                 )
 
+    @pytest.mark.asyncio
+    async def test_base_class_resolves_endpoint_uri_from_global_env(self):
+        with patch.dict(
+            os.environ,
+            {"AWS_REGION": "us-east-1", "AWS_ENDPOINT_URL": "https://localhost:4567"},
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.endpoint_uri == "https://localhost:4567"
+            assert config.source_of("endpoint_uri") == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_resolve_defaults_all_non_resolved_fields(self):
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+        assert config.interceptors == []
+        assert config.transport is None
+        assert config.retry_strategy is None
+        assert config.http_request_config is None
+        assert config.user_agent_extra is None
+        assert config.aws_credentials_identity_resolver is None
+        for name in ("interceptors", "transport", "user_agent_extra"):
+            assert config.source_of(name) == ConfigSource.DEFAULT
+
 
 class TestProvenanceTracking:
     @pytest.mark.asyncio
@@ -335,6 +360,13 @@ async def test_setattr_validates_during_override(
             with pytest.raises(ConfigValidationError, match=match):
                 setattr(config, field_name, invalid_value)
 
+    @pytest.mark.asyncio
+    async def test_typo_in_field_name_raises_attribute_error(self):
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            with pytest.raises(AttributeError, match="has no config field 'regoin'"):
+                config.regoin = "us-west-2"
+
 
 class TestSharedConfigContext:
     def test_default_profile_is_default(self):
@@ -430,6 +462,7 @@ async def test_returns_unset_when_not_found(self):
             ctx = SharedConfigContext(fs=NullFileSystem())
             result = await resolve_retry_mode(ctx)
             assert result.value is UNSET
+            assert result.source is ConfigSource.DEFAULT
 
     @pytest.mark.asyncio
     async def test_legacy_warns_and_maps_to_standard(self):
@@ -512,3 +545,464 @@ async def test_invalid_value_raises_error(self):
             ctx = SharedConfigContext()
             with pytest.raises(ConfigValidationError, match="Invalid integer value"):
                 await resolve_max_attempts(ctx)
+
+
+class TestEndpointUriResolver:
+    @pytest.fixture
+    def resolver(self):
+
+        return EndpointUriResolver("bedrock_runtime")
+
+    @pytest.mark.asyncio
+    async def test_service_specific_env_var_takes_precedence(
+        self, resolver: EndpointUriResolver
+    ):
+        fs = FakeFileSystem(
+            {
+                "/fake/config": "[profile default]\nendpoint_url = https://global-profile.com\n"
+            }
+        )
+        with patch.dict(
+            os.environ,
+            {"AWS_ENDPOINT_URL_BEDROCK_RUNTIME": "https://service-env.com"},
+            clear=True,
+        ):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value == "https://service-env.com"
+            assert result.source == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_global_env_var_when_no_service_specific(
+        self, resolver: EndpointUriResolver
+    ):
+        with patch.dict(
+            os.environ, {"AWS_ENDPOINT_URL": "https://global-env.com"}, clear=True
+        ):
+            ctx = SharedConfigContext(
+                fs=NullFileSystem(),
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value == "https://global-env.com"
+            assert result.source == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_service_env_beats_global_env(self, resolver: EndpointUriResolver):
+        with patch.dict(
+            os.environ,
+            {
+                "AWS_ENDPOINT_URL_BEDROCK_RUNTIME": "https://service-env.com",
+                "AWS_ENDPOINT_URL": "https://global-env.com",
+            },
+            clear=True,
+        ):
+            ctx = SharedConfigContext(
+                fs=NullFileSystem(),
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value == "https://service-env.com"
+
+    @pytest.mark.asyncio
+    async def test_service_specific_config_file(self, resolver: EndpointUriResolver):
+        fs = FakeFileSystem(
+            {
+                "/fake/config": (
+                    "[profile default]\n"
+                    "services = my-services\n"
+                    "\n"
+                    "[services my-services]\n"
+                    "bedrock_runtime =\n"
+                    "  endpoint_url = https://service-config.com\n"
+                )
+            }
+        )
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value == "https://service-config.com"
+            assert result.source == ConfigSource.PROFILE
+
+    @pytest.mark.asyncio
+    async def test_global_config_file_fallback(self, resolver: EndpointUriResolver):
+        fs = FakeFileSystem(
+            {
+                "/fake/config": "[profile default]\nendpoint_url = https://global-config.com\n"
+            }
+        )
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value == "https://global-config.com"
+            assert result.source == ConfigSource.PROFILE
+
+    @pytest.mark.asyncio
+    async def test_service_config_beats_global_config(
+        self, resolver: EndpointUriResolver
+    ):
+        fs = FakeFileSystem(
+            {
+                "/fake/config": (
+                    "[profile default]\n"
+                    "endpoint_url = https://global-config.com\n"
+                    "services = my-services\n"
+                    "\n"
+                    "[services my-services]\n"
+                    "bedrock_runtime =\n"
+                    "  endpoint_url = https://service-config.com\n"
+                )
+            }
+        )
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value == "https://service-config.com"
+
+    @pytest.mark.asyncio
+    async def test_env_beats_config_file(self, resolver: EndpointUriResolver):
+        fs = FakeFileSystem(
+            {
+                "/fake/config": (
+                    "[profile default]\n"
+                    "endpoint_url = https://global-config.com\n"
+                    "services = my-services\n"
+                    "\n"
+                    "[services my-services]\n"
+                    "bedrock_runtime =\n"
+                    "  endpoint_url = https://service-config.com\n"
+                )
+            }
+        )
+        with patch.dict(
+            os.environ, {"AWS_ENDPOINT_URL": "https://global-env.com"}, clear=True
+        ):
+            ctx = SharedConfigContext(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value == "https://global-env.com"
+
+    @pytest.mark.asyncio
+    async def test_returns_unset_when_nothing_found(
+        self, resolver: EndpointUriResolver
+    ):
+        with patch.dict(os.environ, {}, clear=True):
+            ctx = SharedConfigContext(
+                fs=NullFileSystem(),
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value is UNSET
+
+    @pytest.mark.asyncio
+    async def test_spaced_sdk_id_produces_valid_env_var_name(self):
+        """Passing a raw SDK ID with spaces (e.g., 'Bedrock Runtime') should
+        still resolve from the correctly normalized env var."""
+        resolver = EndpointUriResolver("Bedrock Runtime")
+        with patch.dict(
+            os.environ,
+            {"AWS_ENDPOINT_URL_BEDROCK_RUNTIME": "https://from-env.com"},
+            clear=True,
+        ):
+            ctx = SharedConfigContext(
+                fs=NullFileSystem(),
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/creds",
+            )
+            result = await resolver(ctx)
+            assert result.value == "https://from-env.com"
+            assert result.source == ConfigSource.ENV
+
+
+class TestReprDoesNotLeakSecrets:
+    @pytest.mark.asyncio
+    async def test_repr_does_not_leak_secrets(self):
+        with patch.dict(
+            os.environ,
+            {
+                "AWS_REGION": "us-east-1",
+                "AWS_ACCESS_KEY_ID": "AKIAIOSFODNN7EXAMPLE",
+                "AWS_SECRET_ACCESS_KEY": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+                "AWS_SESSION_TOKEN": "FwoGZXIvYXdzEBYaDHqa0AP",
+            },
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            config_repr = repr(config)
+
+            assert "AKIAIOSFODNN7EXAMPLE" not in config_repr
+            assert "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" not in config_repr
+            assert "FwoGZXIvYXdzEBYaDHqa0AP" not in config_repr
+
+
+class TestCredentialSetIsAtomic:
+    """Credentials must be resolved from a single source — never mixed."""
+
+    @pytest.mark.asyncio
+    async def test_env_credentials_do_not_mix_with_profile_token(self):
+        """If access_key and secret come from env, token must also come from env (or be None)."""
+        fs = FakeFileSystem(
+            {"/fake/credentials": "[default]\naws_session_token = TOKEN_FROM_PROFILE\n"}
+        )
+        with patch.dict(
+            os.environ,
+            {
+                "AWS_REGION": "us-east-1",
+                "AWS_ACCESS_KEY_ID": "AKID_FROM_ENV",
+                "AWS_SECRET_ACCESS_KEY": "SECRET_FROM_ENV",
+            },
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            assert config.aws_access_key_id == "AKID_FROM_ENV"
+            assert config.aws_secret_access_key == "SECRET_FROM_ENV"
+            assert config.aws_session_token is None  # NOT from profile
+            assert config.source_of("aws_access_key_id") == ConfigSource.ENV
+            assert config.source_of("aws_session_token") == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_all_three_from_env_when_all_set(self):
+        with patch.dict(
+            os.environ,
+            {
+                "AWS_REGION": "us-east-1",
+                "AWS_ACCESS_KEY_ID": "AKID",
+                "AWS_SECRET_ACCESS_KEY": "SECRET",
+                "AWS_SESSION_TOKEN": "TOKEN",
+            },
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.aws_access_key_id == "AKID"
+            assert config.aws_secret_access_key == "SECRET"
+            assert config.aws_session_token == "TOKEN"
+            assert config.source_of("aws_access_key_id") == ConfigSource.ENV
+            assert config.source_of("aws_secret_access_key") == ConfigSource.ENV
+            assert config.source_of("aws_session_token") == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_all_three_from_profile_when_no_env(self):
+        fs = FakeFileSystem(
+            {
+                "/fake/credentials": (
+                    "[default]\n"
+                    "aws_access_key_id = AKID_PROFILE\n"
+                    "aws_secret_access_key = SECRET_PROFILE\n"
+                    "aws_session_token = TOKEN_PROFILE\n"
+                )
+            }
+        )
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            config = await AsyncAwsConfig.resolve(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            assert config.aws_access_key_id == "AKID_PROFILE"
+            assert config.aws_secret_access_key == "SECRET_PROFILE"
+            assert config.aws_session_token == "TOKEN_PROFILE"
+            assert config.source_of("aws_access_key_id") == ConfigSource.PROFILE
+            assert config.source_of("aws_secret_access_key") == ConfigSource.PROFILE
+            assert config.source_of("aws_session_token") == ConfigSource.PROFILE
+
+    @pytest.mark.asyncio
+    async def test_profile_token_not_used_when_env_has_key_and_secret(self):
+        """Even if profile has all three, env key+secret means token comes from env too."""
+        fs = FakeFileSystem(
+            {
+                "/fake/credentials": (
+                    "[default]\n"
+                    "aws_access_key_id = AKID_PROFILE\n"
+                    "aws_secret_access_key = SECRET_PROFILE\n"
+                    "aws_session_token = TOKEN_PROFILE\n"
+                )
+            }
+        )
+        with patch.dict(
+            os.environ,
+            {
+                "AWS_REGION": "us-east-1",
+                "AWS_ACCESS_KEY_ID": "AKID_ENV",
+                "AWS_SECRET_ACCESS_KEY": "SECRET_ENV",
+            },
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            # Env wins for all three — token is None because env doesn't have it
+            assert config.aws_access_key_id == "AKID_ENV"
+            assert config.aws_secret_access_key == "SECRET_ENV"
+            assert config.aws_session_token is None
+            assert config.source_of("aws_session_token") == ConfigSource.ENV
+
+    @pytest.mark.asyncio
+    async def test_no_credentials_when_nothing_set(self):
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            assert config.aws_access_key_id is None
+            assert config.aws_secret_access_key is None
+            assert config.aws_session_token is None
+            assert config.source_of("aws_access_key_id") == ConfigSource.DEFAULT
+
+    @pytest.mark.asyncio
+    async def test_partial_credential_override_raises_error(self):
+        """Overriding only one credential raises an error."""
+        fs = FakeFileSystem(
+            {
+                "/fake/credentials": (
+                    "[default]\n"
+                    "aws_access_key_id = AKID_PROFILE\n"
+                    "aws_secret_access_key = SECRET_PROFILE\n"
+                )
+            }
+        )
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            with pytest.raises(
+                ConfigValidationError, match="Partial credential override"
+            ):
+                await AsyncAwsConfig.resolve(
+                    fs=fs,
+                    config_file_path="/fake/config",
+                    credentials_file_path="/fake/credentials",
+                    aws_access_key_id="OVERRIDE_KEY",
+                )
+
+    @pytest.mark.asyncio
+    async def test_credentials_cannot_be_overridden_after_resolution(self):
+        with patch.dict(
+            os.environ,
+            {
+                "AWS_REGION": "us-east-1",
+                "AWS_ACCESS_KEY_ID": "AKID",
+                "AWS_SECRET_ACCESS_KEY": "SECRET",
+            },
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(fs=NullFileSystem())
+            with pytest.raises(
+                AttributeError, match="cannot be modified after resolution"
+            ):
+                config.aws_access_key_id = "NEW_KEY"
+
+    @pytest.mark.asyncio
+    async def test_session_token_only_override_raises_error(self):
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            with pytest.raises(
+                ConfigValidationError, match="Partial credential override"
+            ):
+                await AsyncAwsConfig.resolve(
+                    fs=NullFileSystem(),
+                    aws_session_token="FRESH_TOKEN",
+                )
+
+    @pytest.mark.asyncio
+    async def test_env_session_token_only_falls_through_to_profile(self):
+        fs = FakeFileSystem(
+            {
+                "/fake/credentials": (
+                    "[default]\n"
+                    "aws_access_key_id = AKID_PROFILE\n"
+                    "aws_secret_access_key = SECRET_PROFILE\n"
+                    "aws_session_token = TOKEN_PROFILE\n"
+                )
+            }
+        )
+        with patch.dict(
+            os.environ,
+            {"AWS_REGION": "us-east-1", "AWS_SESSION_TOKEN": "TOKEN_ENV"},
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            # Token-only env doesn't trigger env path — all from profile
+            assert config.aws_access_key_id == "AKID_PROFILE"
+            assert config.aws_secret_access_key == "SECRET_PROFILE"
+            assert config.aws_session_token == "TOKEN_PROFILE"
+            assert config.source_of("aws_access_key_id") == ConfigSource.PROFILE
+
+    @pytest.mark.asyncio
+    async def test_env_key_only_without_secret_falls_through_to_profile(self):
+        fs = FakeFileSystem(
+            {
+                "/fake/credentials": (
+                    "[default]\n"
+                    "aws_access_key_id = AKID_PROFILE\n"
+                    "aws_secret_access_key = SECRET_PROFILE\n"
+                )
+            }
+        )
+        with patch.dict(
+            os.environ,
+            {"AWS_REGION": "us-east-1", "AWS_ACCESS_KEY_ID": "AKID_ENV"},
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+
+            assert config.aws_access_key_id == "AKID_PROFILE"
+            assert config.aws_secret_access_key == "SECRET_PROFILE"
+            assert config.source_of("aws_access_key_id") == ConfigSource.PROFILE
+
+    @pytest.mark.asyncio
+    async def test_empty_string_env_credentials_fall_through_to_profile(self):
+        fs = FakeFileSystem(
+            {
+                "/fake/credentials": (
+                    "[default]\n"
+                    "aws_access_key_id = AKID_PROFILE\n"
+                    "aws_secret_access_key = SECRET_PROFILE\n"
+                )
+            }
+        )
+        with patch.dict(
+            os.environ,
+            {
+                "AWS_REGION": "us-east-1",
+                "AWS_ACCESS_KEY_ID": "",
+                "AWS_SECRET_ACCESS_KEY": "",
+            },
+            clear=True,
+        ):
+            config = await AsyncAwsConfig.resolve(
+                fs=fs,
+                config_file_path="/fake/config",
+                credentials_file_path="/fake/credentials",
+            )
+            assert config.aws_access_key_id == "AKID_PROFILE"
+            assert config.aws_secret_access_key == "SECRET_PROFILE"
+            assert config.source_of("aws_access_key_id") == ConfigSource.PROFILE
diff --git a/packages/smithy-core/src/smithy_core/aio/retries.py b/packages/smithy-core/src/smithy_core/aio/retries.py
index e3fa6340e..1ebac9495 100644
--- a/packages/smithy-core/src/smithy_core/aio/retries.py
+++ b/packages/smithy-core/src/smithy_core/aio/retries.py
@@ -25,16 +25,29 @@ class RetryStrategyResolver:
     """
 
     async def resolve_retry_strategy(
-        self, *, retry_strategy: RetryStrategy | RetryStrategyOptions | None
+        self,
+        *,
+        retry_strategy: RetryStrategy | RetryStrategyOptions | None,
+        retry_mode: RetryStrategyType | None = None,
+        max_attempts: int | None = None,
     ) -> RetryStrategy:
         """Resolve a retry strategy from the provided options, using cache when possible.
 
-        :param retry_strategy: An explicitly configured retry strategy or options for creating one.
+        :param retry_strategy: An explicitly configured retry strategy or options for
+            creating one. Takes precedence over ``retry_mode``/``max_attempts``.
+        :param retry_mode: Retry mode to fall back on when ``retry_strategy`` is None,
+            typically resolved from the ``AWS_RETRY_MODE`` env var or a config profile.
+        :param max_attempts: Maximum attempts to fall back on when ``retry_strategy`` is
+            None, typically resolved from ``AWS_MAX_ATTEMPTS`` or a config profile.
         """
         if isinstance(retry_strategy, RetryStrategy):
             return retry_strategy
         elif retry_strategy is None:
-            retry_strategy = RetryStrategyOptions()
+            # Fall back to the separately-resolved config values.
+            retry_strategy = RetryStrategyOptions(
+                retry_mode=retry_mode if retry_mode is not None else "standard",
+                max_attempts=max_attempts,
+            )
         elif not isinstance(retry_strategy, RetryStrategyOptions):  # type: ignore[reportUnnecessaryIsInstance]
             raise TypeError(
                 f"retry_strategy must be RetryStrategy, RetryStrategyOptions, or None, "
diff --git a/packages/smithy-core/tests/unit/aio/test_retries.py b/packages/smithy-core/tests/unit/aio/test_retries.py
index f35c50750..a9710f313 100644
--- a/packages/smithy-core/tests/unit/aio/test_retries.py
+++ b/packages/smithy-core/tests/unit/aio/test_retries.py
@@ -166,3 +166,62 @@ async def test_retry_strategy_resolver_rejects_invalid_type() -> None:
         match="retry_strategy must be RetryStrategy, RetryStrategyOptions, or None",
     ):
         await resolver.resolve_retry_strategy(retry_strategy="invalid")  # type: ignore
+
+
+async def test_retry_strategy_resolver_uses_max_attempts_fallback() -> None:
+    resolver = RetryStrategyResolver()
+
+    strategy = await resolver.resolve_retry_strategy(
+        retry_strategy=None, max_attempts=9
+    )
+
+    assert isinstance(strategy, StandardRetryStrategy)
+    assert strategy.max_attempts == 9
+
+
+async def test_retry_strategy_resolver_uses_retry_mode_fallback() -> None:
+    resolver = RetryStrategyResolver()
+
+    strategy = await resolver.resolve_retry_strategy(
+        retry_strategy=None, retry_mode="simple", max_attempts=4
+    )
+
+    assert isinstance(strategy, SimpleRetryStrategy)
+    assert strategy.max_attempts == 4
+
+
+async def test_retry_strategy_resolver_fallback_defaults_when_unset() -> None:
+    """Omitting both fallbacks must match the prior no-argument behavior."""
+    resolver = RetryStrategyResolver()
+
+    explicit = await resolver.resolve_retry_strategy(
+        retry_strategy=None, retry_mode=None, max_attempts=None
+    )
+    baseline = await resolver.resolve_retry_strategy(retry_strategy=None)
+
+    assert explicit is baseline
+    assert isinstance(explicit, StandardRetryStrategy)
+    assert explicit.max_attempts == 3
+
+
+async def test_explicit_retry_strategy_options_beat_fallbacks() -> None:
+    resolver = RetryStrategyResolver()
+    retry_strategy = RetryStrategyOptions(max_attempts=2)
+
+    strategy = await resolver.resolve_retry_strategy(
+        retry_strategy=retry_strategy, max_attempts=9
+    )
+
+    assert strategy.max_attempts == 2
+
+
+async def test_explicit_retry_strategy_instance_beats_fallbacks() -> None:
+    resolver = RetryStrategyResolver()
+    provided = SimpleRetryStrategy(max_attempts=7)
+
+    strategy = await resolver.resolve_retry_strategy(
+        retry_strategy=provided, retry_mode="standard", max_attempts=9
+    )
+
+    assert strategy is provided
+    assert strategy.max_attempts == 7