fix(aws): let AnthropicAWS.copy() accept credentials=None so environment helpers work on Claude Platform on AWS

src/anthropic/lib/aws/_client.py

@@ -18,6 +18,7 @@
 from ..._exceptions import AnthropicError
 from ..._middleware import MiddlewareInput
 from ..._base_client import DEFAULT_MAX_RETRIES
+from ..credentials._types import AccessTokenProvider
 
 
 class AnthropicAWS(Anthropic):
@@ -164,7 +165,7 @@ def _prepare_request(self, request: httpx.Request) -> None:
         request.headers.update(headers)
 
     @override
-    def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodOverride] — subclass intentionally drops `credentials`
+    def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodOverride] — narrows `credentials` to None-only
         self,
         *,
         api_key: str | None = None,
@@ -176,6 +177,7 @@ def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodO
         workspace_id: str | None = None,
         skip_auth: bool | None = None,
         auth_token: str | None = None,
+        credentials: AccessTokenProvider | None = None,
         webhook_key: str | None = None,
         base_url: str | httpx.URL | None = None,
         timeout: float | Timeout | None | NotGiven = NOT_GIVEN,
@@ -188,6 +190,14 @@ def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodO
         middleware: Sequence[MiddlewareInput] | None | NotGiven = NOT_GIVEN,
         _extra_kwargs: Mapping[str, Any] = {},
     ) -> Self:
+        # The AWS client authenticates with SigV4 (or an API key), not a token
+        # provider, so it has no `credentials`. Accept the argument for signature
+        # compatibility with the base client — internal helpers such as
+        # `_copy_client_with_bearer_auth` call `copy(credentials=None, ...)` — but
+        # only as a no-op; reject a real provider rather than silently ignoring it.
+        if credentials is not None:
+            raise TypeError("AnthropicAWS does not support a `credentials` provider (it authenticates with AWS SigV4).")
+
         # If region is changing and no explicit base_url, let __init__ derive it
         resolved_base_url = base_url or (None if aws_region else self.base_url)
 
@@ -363,7 +373,7 @@ async def _prepare_request(self, request: httpx.Request) -> None:
         request.headers.update(headers)
 
     @override
-    def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodOverride] — subclass intentionally drops `credentials`
+    def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodOverride] — narrows `credentials` to None-only
         self,
         *,
         api_key: str | None = None,
@@ -375,6 +385,7 @@ def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodO
         workspace_id: str | None = None,
         skip_auth: bool | None = None,
         auth_token: str | None = None,
+        credentials: AccessTokenProvider | None = None,
         webhook_key: str | None = None,
         base_url: str | httpx.URL | None = None,
         timeout: float | Timeout | None | NotGiven = NOT_GIVEN,
@@ -387,6 +398,14 @@ def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodO
         middleware: Sequence[MiddlewareInput] | None | NotGiven = NOT_GIVEN,
         _extra_kwargs: Mapping[str, Any] = {},
     ) -> Self:
+        # The AWS client authenticates with SigV4 (or an API key), not a token
+        # provider, so it has no `credentials`. Accept the argument for signature
+        # compatibility with the base client — internal helpers such as
+        # `_copy_client_with_bearer_auth` call `copy(credentials=None, ...)` — but
+        # only as a no-op; reject a real provider rather than silently ignoring it.
+        if credentials is not None:
+            raise TypeError("AnthropicAWS does not support a `credentials` provider (it authenticates with AWS SigV4).")
+
         # If region is changing and no explicit base_url, let __init__ derive it
         resolved_base_url = base_url or (None if aws_region else self.base_url)
 

tests/lib/test_aws.py

@@ -8,6 +8,7 @@
 
 from anthropic import AnthropicAWS, AsyncAnthropicAWS
 from anthropic._exceptions import AnthropicError
+from anthropic.lib.credentials import StaticToken
 
 
 class MockRequestCall(Protocol):
@@ -370,3 +371,87 @@ def test_copy_overrides_aws_options() -> None:
         base_url="https://aws-external-anthropic.eu-west-1.api.aws",
     )
     assert str(copied2.base_url).rstrip("/") == "https://aws-external-anthropic.eu-west-1.api.aws"
+
+
+def test_copy_accepts_credentials_none_noop() -> None:
+    # `credentials=None` is accepted as a no-op: the AWS client authenticates with
+    # SigV4, not a token provider, so there is nothing to clear. The copy stays SigV4.
+    client = AnthropicAWS(
+        aws_access_key="AKID",
+        aws_secret_key="secret",
+        aws_region="us-east-1",
+        workspace_id="ws-123",
+    )
+    copied = client.copy(credentials=None)
+    assert copied._use_sigv4 is True
+    assert copied.workspace_id == "ws-123"
+    assert copied.aws_region == "us-east-1"
+
+
+def test_copy_accepts_credentials_none_noop_async() -> None:
+    client = AsyncAnthropicAWS(
+        aws_access_key="AKID",
+        aws_secret_key="secret",
+        aws_region="us-east-1",
+        workspace_id="ws-123",
+    )
+    copied = client.copy(credentials=None)
+    assert copied._use_sigv4 is True
+    assert copied.workspace_id == "ws-123"
+    assert copied.aws_region == "us-east-1"
+
+
+def test_copy_rejects_real_credentials_provider() -> None:
+    client = AnthropicAWS(
+        aws_access_key="AKID",
+        aws_secret_key="secret",
+        aws_region="us-east-1",
+        workspace_id="ws-123",
+    )
+    with pytest.raises(TypeError, match="does not support a `credentials` provider"):
+        client.copy(credentials=StaticToken("token"))
+
+
+def test_copy_rejects_real_credentials_provider_async() -> None:
+    client = AsyncAnthropicAWS(
+        aws_access_key="AKID",
+        aws_secret_key="secret",
+        aws_region="us-east-1",
+        workspace_id="ws-123",
+    )
+    with pytest.raises(TypeError, match="does not support a `credentials` provider"):
+        client.copy(credentials=StaticToken("token"))
+
+
+def test_scoped_bearer_client_helper_on_aws() -> None:
+    # Regression: the environment poller / worker / session-tool-runner build a
+    # scoped sub-client via `_copy_client_with_bearer_auth`, which calls
+    # `client.copy(credentials=None, ...)`. That must work on the AWS client and
+    # yield a client that still signs with SigV4 (the bearer token is unused).
+    from anthropic.lib._scoped_client import _copy_client_with_bearer_auth
+
+    client = AnthropicAWS(
+        aws_access_key="AKID",
+        aws_secret_key="secret",
+        aws_region="us-east-1",
+        workspace_id="ws-123",
+    )
+    scoped = _copy_client_with_bearer_auth(client, auth_token="unused-under-sigv4", helper="environments-work-poller")
+    assert isinstance(scoped, AnthropicAWS)
+    assert scoped._use_sigv4 is True
+    assert scoped.workspace_id == "ws-123"
+
+
+def test_scoped_bearer_client_helper_on_aws_async() -> None:
+    from anthropic.lib._scoped_client import _copy_client_with_bearer_auth
+
+    client = AsyncAnthropicAWS(
+        aws_access_key="AKID",
+        aws_secret_key="secret",
+        aws_region="us-east-1",
+        workspace_id="ws-123",
+    )
+    scoped = _copy_client_with_bearer_auth(client, auth_token="unused-under-sigv4", helper="environments-worker")
+    assert isinstance(scoped, AsyncAnthropicAWS)
+    assert scoped._use_sigv4 is True
+    assert scoped.workspace_id == "ws-123"