refactor: cli auth

Author: radugheo

samples/asset-modifier-agent/README.md

@@ -0,0 +1,44 @@
+# UiPath Coded Agent: Asset Value Checker
+
+This project demonstrates how to create a Python-based UiPath Coded Agent that retrieves an asset from Orchestrator and validates its IntValue against custom rules.
+
+## Overview
+
+The agent uses the UiPath Python SDK to:
+
+* Connect to UiPath Orchestrator as an external application
+* Authenticate via the Client Credentials flow (Client ID + Client Secret)
+* Retrieve a specific asset from a given folder
+* Check whether the asset has an integer value and validate it against a range (100–1000)
+* Return a descriptive message with the validation result
+
+## How to Set Up
+
+### Step 1: Install UiPath Python SDK
+
+1. Open it with your prefered editor
+2. In terminal run:
+```bash
+uv init
+uv add uipath
+uv run uipath init
+```
+
+### Step 2: Configure Environment Variables
+```bash
+UIPATH_CLIENT_SECRET=your-client-secret
+```
+
+### Step 3: Understanding the Event Flow
+
+When this agent runs, it will:
+1. Pass the configured input values (`asset_name` and `folder_path`) into the agent
+2. Connect to Orchestrator using Client Credentials (Client ID + Client Secret)
+3. Retrieve the specified asset from the given folder
+4. Check whether the asset contains an IntValue and validates it against the allowed range (100–1000)
+5. Return a message with the validation result
+
+### Step 4: Publish Your Coded Agent
+
+1. Use `uipath pack` and `uipath publish` to create and publish the package
+2. Create an Orchestrator Automation from the published process

samples/asset-modifier-agent/main.py

@@ -0,0 +1,83 @@
+import dataclasses
+import dotenv
+import logging
+import os
+
+from typing import Optional
+from uipath import UiPath
+from uipath.tracing import traced
+
+dotenv.load_dotenv()
+logger = logging.getLogger(__name__)
+
+UIPATH_CLIENT_ID = "your-client-id"
+UIPATH_SCOPE = "OR.Administration OR.Folders OR.Assets"
+UIPATH_CLIENT_SECRET = os.getenv("UIPATH_CLIENT_SECRET")
+
+uipath = UiPath(
+    client_id=UIPATH_CLIENT_ID,
+    client_secret=UIPATH_CLIENT_SECRET,
+    scope=UIPATH_SCOPE,
+)
+
+@dataclasses.dataclass
+class AgentInput:
+    """Input data structure for the UiPath agent.
+
+    Attributes:
+        asset_name (str): The name of the UiPath asset.
+        folder_path (str): The folder path where the asset is located.
+    """
+    asset_name: str
+    folder_path: str
+
+def get_asset(name: str, folder_path: str) -> Optional[object]:
+    """Retrieve an asset from UiPath.
+
+    Args:
+        name (str): The asset name.
+        folder_path (str): The UiPath folder path.
+
+    Returns:
+        Optional[object]: The asset object if found, else None.
+    """
+    return uipath.assets.retrieve(name=name, folder_path=folder_path)
+
+def check_asset(asset: object) -> str:
+    """Check if an asset's IntValue is within a valid range.
+
+    Args:
+        asset (object): The asset object.
+
+    Returns:
+        str: Result message depending on asset state.
+    """
+    if asset is None:
+        return "Asset not found."
+
+    int_value = getattr(asset, "int_value", None)
+    if int_value is None:
+        return "Asset does not have an IntValue."
+
+    if 100 <= int_value <= 1000:
+        return f"Asset '{asset.name}' has a valid IntValue: {int_value}"
+    else:
+        return f"Asset '{asset.name}' has an out-of-range IntValue: {int_value}"
+
+@traced()
+def main(input: AgentInput) -> str:
+    """Main entry point for the agent.
+
+    Args:
+        input (AgentInput): The input containing asset details.
+
+    Returns:
+        str: Message with the result of the asset check.
+    """
+    asset = get_asset(input.asset_name, input.folder_path)
+    return check_asset(asset)
+
+if __name__ == "__main__":
+    input_data = AgentInput(asset_name="test-asset", folder_path="TestFolder")
+    result = main(input_data)
+    print(result)

samples/asset-modifier-agent/pyproject.toml

@@ -0,0 +1,10 @@
+[project]
+name = "asset-modifier-agent"
+version = "0.0.1"
+description = "asset-modifier-agent"
+authors = [{ name = "John Doe", email = "john.doe@myemail.com" }]
+dependencies = [
+    "ipykernel>=6.30.1",
+    "uipath>=2.1.62",
+]
+requires-python = ">=3.10"

samples/asset-modifier-agent/uv.lock

@@ -1,22 +1,16 @@
 import asyncio
-import json
 import os
 import webbrowser
-from socket import AF_INET, SOCK_STREAM, error, socket
 from typing import Optional
-from urllib.parse import urlparse
 
 from uipath._cli._auth._auth_server import HTTPServer
-from uipath._cli._auth._client_credentials import ClientCredentialsService
+from uipath._cli._auth._models import TokenData
 from uipath._cli._auth._oidc_utils import OidcUtils
-from uipath._cli._auth._portal_service import (
-    PortalService,
-    get_tenant_id,
-    select_tenant,
-)
-from uipath._cli._auth._url_utils import set_force_flag
-from uipath._cli._auth._utils import update_auth_file, update_env_file
+from uipath._cli._auth._portal_service import PortalService
+from uipath._cli._auth._url_utils import extract_org_tenant, resolve_domain
+from uipath._cli._auth._utils import update_env_file
 from uipath._cli._utils._console import ConsoleLogger
+from uipath._services import ExternalApplicationService
 
 
 class AuthService:
@@ -25,35 +19,20 @@ def __init__(
         environment: str,
         *,
         force: bool,
-        client_id: Optional[str],
-        client_secret: Optional[str],
-        base_url: Optional[str],
-        tenant: Optional[str],
-        scope: Optional[str],
+        client_id: Optional[str] = None,
+        client_secret: Optional[str] = None,
+        base_url: Optional[str] = None,
+        tenant: Optional[str] = None,
+        scope: Optional[str] = None,
     ):
         self._force = force
         self._console = ConsoleLogger()
-        self._domain = self._get_domain(environment)
         self._client_id = client_id
         self._client_secret = client_secret
         self._base_url = base_url
         self._tenant = tenant
+        self._domain = resolve_domain(self._base_url, environment, self._force)
         self._scope = scope
-        set_force_flag(self._force)
-
-    def _get_domain(self, environment: str) -> str:
-        # only search env var if not force authentication
-        if not self._force:
-            uipath_url = os.getenv("UIPATH_URL")
-            if uipath_url and environment == "cloud":  # "cloud" is the default
-                parsed_url = urlparse(uipath_url)
-                if parsed_url.scheme and parsed_url.netloc:
-                    environment = f"{parsed_url.scheme}://{parsed_url.netloc}"
-                else:
-                    self._console.error(
-                        f"Malformed UIPATH_URL: '{uipath_url}'. Please ensure it includes both scheme and netloc (e.g., 'https://cloud.uipath.com')."
-                    )
-        return environment
 
     def authenticate(self) -> None:
         if self._client_id and self._client_secret:
@@ -62,103 +41,103 @@ def authenticate(self) -> None:
 
         self._authenticate_authorization_code()
 
-    def _authenticate_client_credentials(self) -> None:
+    def _authenticate_client_credentials(self):
         if not self._base_url:
             self._console.error(
                 "--base-url is required when using client credentials authentication."
             )
             return
-        self._console.hint("Using client credentials authentication.")
-        credentials_service = ClientCredentialsService(self._base_url)
-        credentials_service.authenticate(
+
+        organization_name, tenant_name = extract_org_tenant(self._base_url)
+        if not (organization_name and tenant_name):
+            self._console.error(
+                "--base-url must include both organization and tenant, "
+                "e.g., 'https://cloud.uipath.com/{organization}/{tenant}'."
+            )
+            return
+
+        app_service = ExternalApplicationService(self._domain)
+        token_data = app_service.get_token_data(
             self._client_id,  # type: ignore
             self._client_secret,  # type: ignore
             self._scope,
         )
 
+        self._tenant = tenant_name
+        with PortalService(
+            self._domain, access_token=token_data["access_token"]
+        ) as portal_service:
+            self._finalize_auth(portal_service, token_data)
+
     def _authenticate_authorization_code(self) -> None:
         with PortalService(self._domain) as portal_service:
-            if not self._force:
-                # use existing env vars
-                if (
-                    os.getenv("UIPATH_URL")
-                    and os.getenv("UIPATH_TENANT_ID")
-                    and os.getenv("UIPATH_ORGANIZATION_ID")
-                ):
-                    try:
-                        portal_service.ensure_valid_token()
-                        return
-                    except Exception:
-                        self._console.error(
-                            "Authentication token is invalid. Please reauthenticate using the '--force' flag.",
-                        )
+            if not self._force and self._can_reuse_existing_token(portal_service):
+                return
+
             auth_url, code_verifier, state = OidcUtils.get_auth_url(self._domain)
-            webbrowser.open(auth_url, 1)
-            auth_config = OidcUtils.get_auth_config()
+            self._open_browser(auth_url)
 
-            self._console.link(
-                "If a browser window did not open, please open the following URL in your browser:",
-                auth_url,
-            )
+            auth_config = OidcUtils.get_auth_config()
             server = HTTPServer(port=auth_config["port"])
             token_data = asyncio.run(server.start(state, code_verifier, self._domain))
-
             if not token_data:
                 self._console.error(
                     "Authentication failed. Please try again.",
                 )
+                return
 
-            portal_service.update_token_data(token_data)
-            update_auth_file(token_data)
-            access_token = token_data["access_token"]
-            update_env_file({"UIPATH_ACCESS_TOKEN": access_token})
+            self._finalize_auth(portal_service, token_data)
 
-            tenants_and_organizations = portal_service.get_tenants_and_organizations()
+    def _finalize_auth(
+        self,
+        portal_service: PortalService,
+        token_data: TokenData,
+    ) -> None:
+        portal_service.update_token_data(token_data)
+
+        tenant_info = (
+            portal_service.retrieve_tenant(self._tenant)
+            if self._tenant
+            else portal_service.select_tenant()
+        )
 
-            if self._tenant:
-                base_url = get_tenant_id(
-                    self._domain, self._tenant, tenants_and_organizations
-                )
-            else:
-                base_url = select_tenant(self._domain, tenants_and_organizations)
+        tenant_id = tenant_info["tenant_id"]
+        organization_id = tenant_info["organization_id"]
+        uipath_url = portal_service.build_tenant_url()
+
+        update_env_file(
+            {
+                "UIPATH_ACCESS_TOKEN": token_data["access_token"],
+                "UIPATH_URL": uipath_url,
+                "UIPATH_TENANT_ID": tenant_id,
+                "UIPATH_ORGANIZATION_ID": organization_id,
+            }
+        )
 
+        try:
+            portal_service.enable_studio_web(uipath_url)
+        except Exception:
+            self._console.error("Could not prepare the environment. Please try again.")
+
+    def _can_reuse_existing_token(self, portal_service: PortalService) -> bool:
+        if (
+            os.getenv("UIPATH_URL")
+            and os.getenv("UIPATH_TENANT_ID")
+            and os.getenv("UIPATH_ORGANIZATION_ID")
+        ):
             try:
-                portal_service.post_auth(base_url)
+                portal_service.ensure_valid_token()
+                return True
             except Exception:
                 self._console.error(
-                    "Could not prepare the environment. Please try again.",
+                    "Authentication token is invalid. Please reauthenticate using the '--force' flag."
                 )
-
-    def set_port(self):
-        def is_port_in_use(target_port: int) -> bool:
-            with socket(AF_INET, SOCK_STREAM) as s:
-                try:
-                    s.bind(("localhost", target_port))
-                    s.close()
-                    return False
-                except error:
-                    return True
-
-        auth_config = OidcUtils.get_auth_config()
-        port = int(auth_config.get("port", 8104))
-        port_option_one = int(auth_config.get("portOptionOne", 8104))  # type: ignore
-        port_option_two = int(auth_config.get("portOptionTwo", 8055))  # type: ignore
-        port_option_three = int(auth_config.get("portOptionThree", 42042))  # type: ignore
-        if is_port_in_use(port):
-            if is_port_in_use(port_option_one):
-                if is_port_in_use(port_option_two):
-                    if is_port_in_use(port_option_three):
-                        self._console.error(
-                            "All configured ports are in use. Please close applications using ports or configure different ports."
-                        )
-                    else:
-                        port = port_option_three
-                else:
-                    port = port_option_two
-            else:
-                port = port_option_one
-        auth_config["port"] = port
-        with open(
-            os.path.join(os.path.dirname(__file__), "..", "auth_config.json"), "w"
-        ) as f:
-            json.dump(auth_config, f)
+        return False
+
+    def _open_browser(self, url: str) -> None:
+        # Try to open browser. Always print the fallback link.
+        webbrowser.open(url, new=1)
+        self._console.link(
+            "If a browser window did not open, please open the following URL in your browser:",
+            url,
+        )