Make Foundry auth overrides replace inherited auth

src/anthropic/lib/foundry.py

@@ -154,7 +154,10 @@ def __init__(
             resource: Your Foundry resource name, e.g. `example-resource` for `https://example-resource.services.ai.azure.com/anthropic/`
             azure_ad_token_provider: A function that returns an Azure Active Directory token, will be invoked on every request.
         """
-        api_key = api_key if api_key is not None else os.environ.get("ANTHROPIC_FOUNDRY_API_KEY")
+        if api_key is not None and azure_ad_token_provider is not None:
+            raise MutuallyExclusiveAuthError()
+        if api_key is None and azure_ad_token_provider is None:
+            api_key = os.environ.get("ANTHROPIC_FOUNDRY_API_KEY")
         resource = resource if resource is not None else os.environ.get("ANTHROPIC_FOUNDRY_RESOURCE")
         base_url = base_url if base_url is not None else os.environ.get("ANTHROPIC_FOUNDRY_BASE_URL")
 
@@ -184,6 +187,11 @@ def __init__(
             middleware=middleware,
             _strict_response_validation=_strict_response_validation,
         )
+        # The base client also understands first-party ANTHROPIC_API_KEY /
+        # ANTHROPIC_AUTH_TOKEN environment variables. Foundry has its own auth
+        # namespace, so keep only the credential resolved above.
+        self.api_key = api_key
+        self.auth_token = None
         self._azure_ad_token_provider = azure_ad_token_provider
 
     @cached_property
@@ -231,6 +239,18 @@ def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodO
         if default_query is not None and set_default_query is not None:
             raise ValueError("The `default_query` and `set_default_query` arguments are mutually exclusive")
 
+        if api_key is not None and azure_ad_token_provider is not None:
+            raise MutuallyExclusiveAuthError()
+        if api_key is not None:
+            resolved_api_key = api_key
+            resolved_token_provider = None
+        elif azure_ad_token_provider is not None:
+            resolved_api_key = None
+            resolved_token_provider = azure_ad_token_provider
+        else:
+            resolved_api_key = self.api_key
+            resolved_token_provider = self._azure_ad_token_provider
+
         headers = self._custom_headers
         if default_headers is not None:
             headers = merge_headers(headers, default_headers)
@@ -244,8 +264,8 @@ def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodO
             params = set_default_query
 
         return self.__class__(
-            api_key=api_key or self.api_key,
-            azure_ad_token_provider=azure_ad_token_provider or self._azure_ad_token_provider,
+            api_key=resolved_api_key,
+            azure_ad_token_provider=resolved_token_provider,
             webhook_key=webhook_key or self.webhook_key,
             base_url=str(base_url or self.base_url),
             timeout=self.timeout if isinstance(timeout, NotGiven) else timeout,
@@ -379,7 +399,10 @@ def __init__(
             resource: Your Foundry resource name, e.g. `example-resource` for `https://example-resource.services.ai.azure.com/anthropic/`
             azure_ad_token_provider: A function that returns an Azure Active Directory token, will be invoked on every request.
         """
-        api_key = api_key if api_key is not None else os.environ.get("ANTHROPIC_FOUNDRY_API_KEY")
+        if api_key is not None and azure_ad_token_provider is not None:
+            raise MutuallyExclusiveAuthError()
+        if api_key is None and azure_ad_token_provider is None:
+            api_key = os.environ.get("ANTHROPIC_FOUNDRY_API_KEY")
         resource = resource if resource is not None else os.environ.get("ANTHROPIC_FOUNDRY_RESOURCE")
         base_url = base_url if base_url is not None else os.environ.get("ANTHROPIC_FOUNDRY_BASE_URL")
 
@@ -409,6 +432,8 @@ def __init__(
             middleware=middleware,
             _strict_response_validation=_strict_response_validation,
         )
+        self.api_key = api_key
+        self.auth_token = None
         self._azure_ad_token_provider = azure_ad_token_provider
 
     @cached_property
@@ -456,6 +481,18 @@ def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodO
         if default_query is not None and set_default_query is not None:
             raise ValueError("The `default_query` and `set_default_query` arguments are mutually exclusive")
 
+        if api_key is not None and azure_ad_token_provider is not None:
+            raise MutuallyExclusiveAuthError()
+        if api_key is not None:
+            resolved_api_key = api_key
+            resolved_token_provider = None
+        elif azure_ad_token_provider is not None:
+            resolved_api_key = None
+            resolved_token_provider = azure_ad_token_provider
+        else:
+            resolved_api_key = self.api_key
+            resolved_token_provider = self._azure_ad_token_provider
+
         headers = self._custom_headers
         if default_headers is not None:
             headers = merge_headers(headers, default_headers)
@@ -469,8 +506,8 @@ def copy(  # type: ignore[override]  # pyright: ignore[reportIncompatibleMethodO
             params = set_default_query
 
         return self.__class__(
-            api_key=api_key or self.api_key,
-            azure_ad_token_provider=azure_ad_token_provider or self._azure_ad_token_provider,
+            api_key=resolved_api_key,
+            azure_ad_token_provider=resolved_token_provider,
             webhook_key=webhook_key or self.webhook_key,
             base_url=str(base_url or self.base_url),
             timeout=self.timeout if isinstance(timeout, NotGiven) else timeout,

tests/lib/test_foundry_auth_switch.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import pytest
+
+from anthropic._models import FinalRequestOptions
+from anthropic.lib.foundry import AnthropicFoundry, AsyncAnthropicFoundry, MutuallyExclusiveAuthError
+
+
+BASE_URL = "https://foundry.example.test/anthropic/"
+
+
+def _options() -> FinalRequestOptions:
+    return FinalRequestOptions.construct(method="post", url="v1/messages", headers={})
+
+
+def _azure_token() -> str:
+    return "azure-token"
+
+
+def test_sync_copy_api_key_replaces_inherited_azure_ad_provider() -> None:
+    client = AnthropicFoundry(base_url=BASE_URL, azure_ad_token_provider=_azure_token)
+    clone = client.copy(api_key="foundry-key")
+    try:
+        assert clone.api_key == "foundry-key"
+        assert clone._azure_ad_token_provider is None
+
+        options = clone._prepare_options(_options())
+        assert options.headers["x-api-key"] == "foundry-key"
+        assert options.headers["api-key"] == "foundry-key"
+        assert "Authorization" not in options.headers
+    finally:
+        clone.close()
+        client.close()
+
+
+def test_sync_copy_azure_ad_provider_replaces_inherited_api_key() -> None:
+    client = AnthropicFoundry(base_url=BASE_URL, api_key="foundry-key")
+    clone = client.copy(azure_ad_token_provider=_azure_token)
+    try:
+        assert clone.api_key is None
+        assert clone._azure_ad_token_provider is _azure_token
+
+        options = clone._prepare_options(_options())
+        assert options.headers["Authorization"] == "Bearer azure-token"
+        assert "x-api-key" not in options.headers
+        assert "api-key" not in options.headers
+    finally:
+        clone.close()
+        client.close()
+
+
+def test_sync_rejects_two_explicit_auth_methods() -> None:
+    with pytest.raises(MutuallyExclusiveAuthError):
+        AnthropicFoundry(
+            base_url=BASE_URL,
+            api_key="foundry-key",
+            azure_ad_token_provider=_azure_token,
+        )
+
+
+def test_azure_ad_auth_ignores_first_party_api_key_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "first-party-key")
+    client = AnthropicFoundry(base_url=BASE_URL, azure_ad_token_provider=_azure_token)
+    try:
+        assert client.api_key is None
+        assert client.auth_token is None
+    finally:
+        client.close()
+
+
+@pytest.mark.asyncio()
+async def test_async_copy_api_key_replaces_inherited_azure_ad_provider() -> None:
+    async def provider() -> str:
+        return "azure-token"
+
+    client = AsyncAnthropicFoundry(base_url=BASE_URL, azure_ad_token_provider=provider)
+    clone = client.copy(api_key="foundry-key")
+    try:
+        assert clone.api_key == "foundry-key"
+        assert clone._azure_ad_token_provider is None
+
+        options = await clone._prepare_options(_options())
+        assert options.headers["x-api-key"] == "foundry-key"
+        assert options.headers["api-key"] == "foundry-key"
+        assert "Authorization" not in options.headers
+    finally:
+        await clone.close()
+        await client.close()
+
+
+@pytest.mark.asyncio()
+async def test_async_copy_azure_ad_provider_replaces_inherited_api_key() -> None:
+    async def provider() -> str:
+        return "azure-token"
+
+    client = AsyncAnthropicFoundry(base_url=BASE_URL, api_key="foundry-key")
+    clone = client.copy(azure_ad_token_provider=provider)
+    try:
+        assert clone.api_key is None
+        assert clone._azure_ad_token_provider is provider
+
+        options = await clone._prepare_options(_options())
+        assert options.headers["Authorization"] == "Bearer azure-token"
+        assert "x-api-key" not in options.headers
+        assert "api-key" not in options.headers
+    finally:
+        await clone.close()
+        await client.close()