diff --git a/src/anthropic/lib/foundry.py b/src/anthropic/lib/foundry.py index 9a3ba47bb..889eefe96 100644 --- a/src/anthropic/lib/foundry.py +++ b/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, diff --git a/tests/lib/test_foundry_auth_switch.py b/tests/lib/test_foundry_auth_switch.py new file mode 100644 index 000000000..418d037e9 --- /dev/null +++ b/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()