fixes

Author: cosminacho

src/uipath/llm_client/settings/llmgateway/auth.py

@@ -1,18 +1,23 @@
+import logging
 from collections.abc import Generator
 
 from httpx import Auth, Client, Request, Response
 
 from uipath.llm_client.settings.llmgateway.settings import LLMGatewayBaseSettings
 from uipath.llm_client.settings.llmgateway.utils import LLMGatewayEndpoints
 from uipath.llm_client.settings.utils import SingletonMeta
-from uipath.llm_client.utils.exceptions import UiPathAPIError, UiPathAuthenticationError
+
+logger = logging.getLogger(__name__)
 
 
 class LLMGatewayS2SAuth(Auth, metaclass=SingletonMeta):
     """Bearer authentication handler with automatic token refresh.
 
     Singleton class that reuses the same token across all requests to minimize
     token generation overhead. Automatically refreshes the token on 401 responses.
+
+    Does not raise errors on token retrieval failures — the request is sent
+    without a valid token and the downstream client handles the error response.
     """
 
     def __init__(
@@ -21,45 +26,50 @@ def __init__(
     ):
         self.settings = settings
         if self.settings.access_token is None:
-            self.access_token = self.get_llmgw_token_header()
+            self.access_token = self.get_llmgw_token()
         else:
             self.access_token = self.settings.access_token.get_secret_value()
 
-    def get_llmgw_token_header(
+    def get_llmgw_token(
         self,
-    ) -> str:
-        """Retrieve a new access token from the LLM Gateway identity endpoint."""
-        url_get_token = f"{self.settings.base_url}/{LLMGatewayEndpoints.IDENTITY_ENDPOINT.value}"
+    ) -> str | None:
+        """Retrieve a new access token from the LLM Gateway identity endpoint.
+
+        Returns None on failure instead of raising, so the request proceeds
+        and the client receives the actual error response from the server.
+        """
         if self.settings.client_id is None or self.settings.client_secret is None:
-            raise ValueError("client_id and client_secret are required for S2S authentication")
+            logger.warning("client_id and client_secret are required for S2S authentication")
+            return None
+        url_get_token = f"{self.settings.base_url}/{LLMGatewayEndpoints.IDENTITY_ENDPOINT.value}"
         token_credentials = dict(
             client_id=self.settings.client_id.get_secret_value(),
             client_secret=self.settings.client_secret.get_secret_value(),
             grant_type="client_credentials",
         )
-        with Client() as http_client:
-            response = http_client.post(url_get_token, data=token_credentials)
-            if response.is_client_error:
-                try:
-                    body = response.json()
-                except Exception:
-                    body = response.text
-                raise UiPathAuthenticationError(
-                    message="Failed to authenticate with LLM Gateway, invalid credentials",
-                    request=response.request,
-                    response=response,
-                    body=body,
-                )
-            elif response.is_error:
-                raise UiPathAPIError.from_response(response)
-            llmgw_token_header = response.json().get("access_token")
-            return llmgw_token_header
+        try:
+            with Client() as http_client:
+                response = http_client.post(url_get_token, data=token_credentials)
+                if response.is_error:
+                    logger.warning(
+                        "Failed to retrieve LLM Gateway token: %s %s",
+                        response.status_code,
+                        response.reason_phrase,
+                    )
+                    return None
+                return response.json().get("access_token")
+        except Exception:
+            logger.warning("Failed to retrieve LLM Gateway token", exc_info=True)
+            return None
 
     def auth_flow(self, request: Request) -> Generator[Request, Response, None]:
         """HTTPX auth flow that handles token refresh on authentication failures."""
-        request.headers["Authorization"] = f"Bearer {self.access_token}"
+        if self.access_token:
+            request.headers["Authorization"] = f"Bearer {self.access_token}"
         response = yield request
         if response.status_code == 401:
-            self.access_token = self.get_llmgw_token_header()
-            request.headers["Authorization"] = f"Bearer {self.access_token}"
-            yield request
+            new_token = self.get_llmgw_token()
+            if new_token:
+                self.access_token = new_token
+                request.headers["Authorization"] = f"Bearer {self.access_token}"
+                yield request

src/uipath/llm_client/settings/platform/auth.py

@@ -37,22 +37,36 @@ def get_access_token(self, refresh: bool = False) -> str:
 
         assert self.settings.base_url is not None
         assert self.settings.client_id is not None
-        assert self.settings.refresh_token is not None
+        if self.settings.refresh_token is None:
+            return access_token
+
+        try:
+            identity_service = IdentityService(self.settings.base_url)
+            new_token_data = identity_service.refresh_access_token(
+                refresh_token=self.settings.refresh_token.get_secret_value(),
+                client_id=self.settings.client_id,
+            )
+        except Exception:
+            return access_token
 
-        identity_service = IdentityService(self.settings.base_url)
-        new_token_data = identity_service.refresh_access_token(
-            self.settings.refresh_token.get_secret_value(), self.settings.client_id
-        )
         self.settings.access_token = SecretStr(new_token_data.access_token)
         self.settings.refresh_token = SecretStr(
             new_token_data.refresh_token or self.settings.refresh_token.get_secret_value()
         )
         return new_token_data.access_token
 
     def auth_flow(self, request: Request) -> Generator[Request, Response, None]:
-        """HTTPX auth flow that handles token refresh on authentication failures."""
-        request.headers["Authorization"] = f"Bearer {self.get_access_token()}"
+        """HTTPX auth flow that handles token refresh on authentication failures.
+
+        On 401, attempts to refresh the token. If refresh succeeds (new token differs
+        from the old one), retries the request. Otherwise, returns the 401 response
+        as-is so the client receives the actual error.
+        """
+        old_token = self.get_access_token()
+        request.headers["Authorization"] = f"Bearer {old_token}"
         response = yield request
         if response.status_code == 401:
-            request.headers["Authorization"] = f"Bearer {self.get_access_token(refresh=True)}"
-            yield request
+            new_token = self.get_access_token(refresh=True)
+            if new_token != old_token:
+                request.headers["Authorization"] = f"Bearer {new_token}"
+                yield request

src/uipath/llm_client/settings/platform/settings.py

@@ -8,11 +8,7 @@
 from uipath.platform.common import EndpointManager
 
 from uipath.llm_client.settings.base import UiPathAPIConfig, UiPathBaseSettings
-from uipath.llm_client.settings.platform.utils import (
-    get_auth_data,
-    is_token_expired,
-    parse_access_token,
-)
+from uipath.llm_client.settings.platform.utils import is_token_expired, parse_access_token
 
 
 class PlatformBaseSettings(UiPathBaseSettings):
@@ -43,9 +39,7 @@ class PlatformBaseSettings(UiPathBaseSettings):
 
     # Credentials used for refreshing the access token
     client_id: str | None = Field(default=None)
-    refresh_token: SecretStr | None = Field(
-        default=None,
-    )
+    refresh_token: SecretStr | None = Field(default=None, validation_alias="UIPATH_REFRESH_TOKEN")
 
     # AgentHub configuration (used for discovery)
     agenthub_config: str | None = Field(
@@ -68,16 +62,14 @@ def validate_environment(self) -> Self:
             raise ValueError(
                 "Base URL, access token, tenant ID, and organization ID are required. Try running `uipath auth` to authenticate."
             )
-        auth_data = get_auth_data()
-        if auth_data.access_token != self.access_token.get_secret_value():
-            raise ValueError("Access token mismatch between .auth.json and environment variables")
-        if is_token_expired(auth_data.access_token):
+
+        access_token = self.access_token.get_secret_value()
+        if is_token_expired(access_token):
             raise ValueError(
                 "Access token is expired. Try running `uipath auth` to refresh the token."
             )
-        if auth_data.refresh_token is not None:
-            self.refresh_token = SecretStr(auth_data.refresh_token)
-        parsed_token_data = parse_access_token(auth_data.access_token)
+
+        parsed_token_data = parse_access_token(access_token)
         self.client_id = parsed_token_data.get("client_id", None)
         return self
 

src/uipath/llm_client/settings/platform/utils.py

@@ -1,16 +1,6 @@
 import base64
 import json
 import time
-from pathlib import Path
-
-from uipath.platform.common.auth import TokenData
-
-
-def get_auth_data() -> TokenData:
-    auth_file = Path.cwd() / ".uipath" / ".auth.json"
-    if not auth_file.exists():
-        raise FileNotFoundError("No authentication file found")
-    return TokenData.model_validate(json.load(open(auth_file)))
 
 
 def parse_access_token(access_token: str):

tests/core/test_base_client.py

@@ -11,7 +11,7 @@
 8. Header utilities (extract_matching_headers, context vars)
 9. Logging config (LoggingConfig)
 10. SSL config (expand_path, get_httpx_ssl_client_kwargs)
-11. LLMGateway S2S auth (get_llmgw_token_header)
+11. LLMGateway S2S auth (get_llmgw_token)
 12. LLMGateway BYOM validation (validate_byo_model)
 13. Wait strategy (wait_retry_after_with_fallback)
 14. HTTPX client send() behavior (streaming header, URL freezing, header capture)
@@ -105,24 +105,16 @@ def platform_env_vars():
 
 @pytest.fixture
 def mock_platform_auth():
-    """Context manager that patches get_auth_data and parse_access_token for PlatformSettings tests."""
-    mock_auth_data = MagicMock()
-    mock_auth_data.access_token = "test-access-token"
-    mock_auth_data.refresh_token = None
-
+    """Patches is_token_expired and parse_access_token for PlatformSettings tests."""
     with (
         patch(
-            "uipath.llm_client.settings.platform.settings.get_auth_data",
-            return_value=mock_auth_data,
+            "uipath.llm_client.settings.platform.settings.is_token_expired",
+            return_value=False,
         ),
         patch(
             "uipath.llm_client.settings.platform.settings.parse_access_token",
             return_value={"client_id": "test-client-id"},
         ),
-        patch(
-            "uipath.llm_client.settings.platform.settings.is_token_expired",
-            return_value=False,
-        ),
     ):
         yield
 
@@ -439,7 +431,7 @@ def test_auth_flow_refreshes_on_401(self, llmgw_s2s_env_vars):
 
             # Mock the token retrieval
             with patch.object(
-                LLMGatewayS2SAuth, "get_llmgw_token_header", return_value="new-token"
+                LLMGatewayS2SAuth, "get_llmgw_token", return_value="new-token"
             ) as mock_get_token:
                 auth = LLMGatewayS2SAuth(settings=settings)
                 # First call is during __init__
@@ -1702,7 +1694,7 @@ def test_response_patched_with_raise_for_status(self):
 
 
 class TestLLMGatewayS2STokenAcquisition:
-    """Tests for LLMGatewayS2SAuth.get_llmgw_token_header."""
+    """Tests for LLMGatewayS2SAuth.get_llmgw_token."""
 
     def test_s2s_token_success(self, llmgw_s2s_env_vars):
         from uipath.llm_client.settings.llmgateway.auth import LLMGatewayS2SAuth
@@ -1718,53 +1710,45 @@ def test_s2s_token_success(self, llmgw_s2s_env_vars):
                 auth = LLMGatewayS2SAuth(settings=settings)
                 assert auth.access_token == "s2s-token-value"
 
-    def test_s2s_token_client_error_raises_auth_error(self, llmgw_s2s_env_vars):
+    def test_s2s_token_client_error_returns_none(self, llmgw_s2s_env_vars):
         from uipath.llm_client.settings.llmgateway.auth import LLMGatewayS2SAuth
 
         mock_response = MagicMock()
-        mock_response.is_client_error = True
-        mock_response.json.return_value = {"error": "invalid_client"}
-        mock_response.request = MagicMock(spec=Request)
+        mock_response.is_error = True
         mock_response.status_code = 401
         mock_response.reason_phrase = "Unauthorized"
-        mock_response.headers = {}
 
         with patch.dict(os.environ, llmgw_s2s_env_vars, clear=True):
             settings = LLMGatewaySettings()
             with patch.object(Client, "post", return_value=mock_response):
-                with pytest.raises(UiPathAuthenticationError, match="invalid credentials"):
-                    LLMGatewayS2SAuth(settings=settings)
+                auth = LLMGatewayS2SAuth(settings=settings)
+                assert auth.access_token is None
 
-    def test_s2s_token_server_error_raises_api_error(self, llmgw_s2s_env_vars):
+    def test_s2s_token_server_error_returns_none(self, llmgw_s2s_env_vars):
         from uipath.llm_client.settings.llmgateway.auth import LLMGatewayS2SAuth
 
         mock_response = MagicMock()
-        mock_response.is_client_error = False
         mock_response.is_error = True
         mock_response.status_code = 500
         mock_response.reason_phrase = "Server Error"
-        mock_response.json.return_value = {"error": "internal"}
-        mock_response.request = MagicMock(spec=Request)
-        mock_response.headers = {}
 
         with patch.dict(os.environ, llmgw_s2s_env_vars, clear=True):
             settings = LLMGatewaySettings()
             with patch.object(Client, "post", return_value=mock_response):
-                with pytest.raises(UiPathAPIError):
-                    LLMGatewayS2SAuth(settings=settings)
+                auth = LLMGatewayS2SAuth(settings=settings)
+                assert auth.access_token is None
 
-    def test_s2s_missing_credentials_raises_value_error(self, llmgw_env_vars):
+    def test_s2s_missing_credentials_returns_none(self, llmgw_env_vars):
         from uipath.llm_client.settings.llmgateway.auth import LLMGatewayS2SAuth
 
         with patch.dict(os.environ, llmgw_env_vars, clear=True):
             settings = LLMGatewaySettings()
             # Clear access_token to force S2S flow, but credentials are missing
             settings.access_token = None
             settings.client_id = None
-            with pytest.raises(ValueError, match="client_id and client_secret are required"):
-                auth = LLMGatewayS2SAuth.__new__(LLMGatewayS2SAuth)
-                auth.settings = settings
-                auth.get_llmgw_token_header()
+            auth = LLMGatewayS2SAuth.__new__(LLMGatewayS2SAuth)
+            auth.settings = settings
+            assert auth.get_llmgw_token() is None
 
 
 # ============================================================================