feat: add telemetry for cli

.github/workflows/cd.yml

@@ -57,7 +57,7 @@ jobs:
           rsync -a --no-whole-file --ignore-existing "$originalfile" "$tmpfile"
           envsubst '$CONNECTION_STRING' < "$originalfile" > "$tmpfile" && mv "$tmpfile" "$originalfile"
         env:
-          CONNECTION_STRING: ${{ secrets.APPINS_CONNECTION_STRING }}
+          CONNECTION_STRING: ${{ secrets.APPLICATIONINSIGHTS_CONNECTION_STRING }}
 
       - name: Build
         run: uv build

.github/workflows/publish-dev.yml

@@ -40,7 +40,7 @@ jobs:
           rsync -a --no-whole-file --ignore-existing "$originalfile" "$tmpfile"
           envsubst '$CONNECTION_STRING' < "$originalfile" > "$tmpfile" && mv "$tmpfile" "$originalfile"
         env:
-          CONNECTION_STRING: ${{ secrets.APPINS_CONNECTION_STRING }}
+          CONNECTION_STRING: ${{ secrets.APPLICATIONINSIGHTS_CONNECTION_STRING }}
 
       - name: Set development version
         shell: pwsh

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.8.21"
+version = "2.8.22"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

src/uipath/_cli/_telemetry.py

@@ -0,0 +1,231 @@
+import logging
+import os
+import time
+import tomllib
+from functools import wraps
+from importlib.metadata import version
+from typing import Any, Callable, Dict, Optional
+
+from dotenv import dotenv_values
+
+from uipath._cli._utils._common import get_claim_from_token
+from uipath.telemetry._track import (
+    _get_project_key,
+    is_telemetry_enabled,
+    track_cli_event,
+)
+
+logger = logging.getLogger(__name__)
+
+# Telemetry event name template for Application Insights
+CLI_COMMAND_EVENT = "Cli.{command}"
+
+
+def _read_env_file_value(key: str) -> Optional[str]:
+    """Read a value from the .env file in the current directory."""
+    try:
+        env_path = os.path.join(os.getcwd(), ".env")
+        values = dotenv_values(env_path)
+        return values.get(key)
+    except Exception:
+        return None
+
+
+class CliTelemetryTracker:
+    """Tracks CLI command execution and sends telemetry to Application Insights.
+
+    Sends a single event per command execution at completion with:
+    - Status: "Completed" or "Failed"
+    - Success: Boolean indicating success/failure
+    - Error details (if failed)
+    """
+
+    def __init__(self) -> None:
+        self._start_times: Dict[str, float] = {}
+
+    @staticmethod
+    def _get_event_name(command: str) -> str:
+        return f"Cli.{command.capitalize()}"
+
+    def _enrich_properties(self, properties: Dict[str, Any]) -> None:
+        """Enrich properties with common context information.
+
+        Args:
+            properties: The properties dictionary to enrich.
+        """
+        # Add UiPath context
+        project_key = _get_project_key()
+        if project_key:
+            properties["AgentId"] = project_key
+
+        # Get organization ID
+        organization_id = os.getenv("UIPATH_ORGANIZATION_ID") or _read_env_file_value(
+            "UIPATH_ORGANIZATION_ID"
+        )
+        if organization_id:
+            properties["CloudOrganizationId"] = organization_id
+
+        # Get tenant ID
+        tenant_id = os.getenv("UIPATH_TENANT_ID") or _read_env_file_value(
+            "UIPATH_TENANT_ID"
+        )
+        if tenant_id:
+            properties["CloudTenantId"] = tenant_id
+
+        # Get CloudUserId from JWT token
+        try:
+            env_file_token = _read_env_file_value("UIPATH_ACCESS_TOKEN")
+            cloud_user_id = get_claim_from_token("sub", env_file_token)
+            if cloud_user_id:
+                properties["CloudUserId"] = cloud_user_id
+        except Exception:
+            pass
+
+        properties["SessionId"] = "nosession"  # Placeholder for session ID
+
+        try:
+            properties["SDKVersion"] = version("uipath")
+        except Exception:
+            pass
+
+        # Add agent version from pyproject.toml
+        try:
+            with open("pyproject.toml", "rb") as f:
+                pyproject = tomllib.load(f)
+            agent_version = pyproject.get("project", {}).get("version")
+            if agent_version:
+                properties["AgentVersion"] = agent_version
+        except Exception:
+            pass
+
+        properties["IsCI"] = bool(os.getenv("GITHUB_ACTIONS"))
+
+        # Add source identifier
+        properties["Source"] = "uipath-python-cli"
+        properties["ApplicationName"] = "UiPath.AgentCli"
+
+    def track_command_start(self, command: str) -> None:
+        """Record the start time for duration calculation."""
+        try:
+            self._start_times[command] = time.time()
+            logger.debug(f"Started tracking CLI command: {command}")
+
+        except Exception as e:
+            logger.debug(f"Error recording CLI command start time: {e}")
+
+    def track_command_end(
+        self,
+        command: str,
+        duration_ms: Optional[int] = None,
+    ) -> None:
+        try:
+            if duration_ms is None:
+                start_time = self._start_times.pop(command, None)
+                if start_time:
+                    duration_ms = int((time.time() - start_time) * 1000)
+
+            properties: Dict[str, Any] = {
+                "Command": command,
+                "Status": "Completed",
+                "Success": True,
+            }
+
+            if duration_ms is not None:
+                properties["DurationMs"] = duration_ms
+
+            self._enrich_properties(properties)
+
+            track_cli_event(self._get_event_name(command), properties)
+            logger.debug(f"Tracked CLI command completed: {command}")
+
+        except Exception as e:
+            logger.debug(f"Error tracking CLI command end: {e}")
+
+    def track_command_failed(
+        self,
+        command: str,
+        duration_ms: Optional[int] = None,
+        exception: Optional[Exception] = None,
+    ) -> None:
+        try:
+            if duration_ms is None:
+                start_time = self._start_times.pop(command, None)
+                if start_time:
+                    duration_ms = int((time.time() - start_time) * 1000)
+
+            properties: Dict[str, Any] = {
+                "Command": command,
+                "Status": "Failed",
+                "Success": False,
+            }
+
+            if duration_ms is not None:
+                properties["DurationMs"] = duration_ms
+
+            if exception is not None:
+                properties["ErrorType"] = type(exception).__name__
+                properties["ErrorMessage"] = str(exception)[:500]
+
+            self._enrich_properties(properties)
+
+            track_cli_event(self._get_event_name(command), properties)
+            logger.debug(f"Tracked CLI command failed: {command}")
+
+        except Exception as e:
+            logger.debug(f"Error tracking CLI command failed: {e}")
+
+
+def track_command(command: str) -> Callable[..., Any]:
+    """Decorator to track CLI command execution.
+
+    Sends an event (Cli.<Command>) to Application Insights at command
+    completion with the execution outcome.
+
+    Properties tracked include:
+    - Command: The command name
+    - Status: Execution outcome ("Completed" or "Failed")
+    - Success: Whether the command succeeded (true/false)
+    - DurationMs: Execution time in milliseconds
+    - ErrorType: Exception type name (on failure)
+    - ErrorMessage: Exception message (on failure, truncated to 500 chars)
+    - AgentId: Project key from .uipath/.telemetry.json (GUID)
+    - Version: Package version (uipath package)
+    - ProjectId, CloudOrganizationId, etc. (if available)
+
+    Telemetry failures are silently ignored to ensure CLI execution
+    is never blocked by telemetry issues.
+
+    Args:
+        command: The CLI command name (e.g., "pack", "publish", "run").
+
+    Returns:
+        A decorator function that wraps the CLI command.
+
+    Example:
+        @click.command()
+        @track_command("pack")
+        def pack(root, nolock):
+            ...
+    """
+
+    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+        @wraps(func)
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            if not is_telemetry_enabled():
+                return func(*args, **kwargs)
+
+            tracker = CliTelemetryTracker()
+            tracker.track_command_start(command)
+
+            try:
+                result = func(*args, **kwargs)
+                tracker.track_command_end(command)
+                return result
+
+            except Exception as e:
+                tracker.track_command_failed(command, exception=e)
+                raise
+
+        return wrapper
+
+    return decorator

src/uipath/_cli/_utils/_common.py

@@ -21,10 +21,10 @@
 )
 
 
-def get_claim_from_token(claim_name: str) -> str | None:
+def get_claim_from_token(claim_name: str, token: str | None = None) -> str | None:
     import jwt
 
-    token = os.getenv(ENV_UIPATH_ACCESS_TOKEN)
+    token = token or os.getenv(ENV_UIPATH_ACCESS_TOKEN)
     if not token:
         raise Exception("JWT token not available")
     decoded_token = jwt.decode(token, options={"verify_signature": False})

src/uipath/_cli/cli_add.py

@@ -7,6 +7,7 @@
 import click
 
 from .._utils.constants import EVALS_FOLDER
+from ._telemetry import track_command
 from ._utils._console import ConsoleLogger
 from ._utils._resources import Resources
 
@@ -84,6 +85,7 @@ def create_evaluator(evaluator_name):
 @click.command()
 @click.argument("resource", required=True)
 @click.argument("args", nargs=-1)
+@track_command("add")
 def add(resource: str, args: tuple[str]) -> None:
     """Create a local resource.
 

src/uipath/_cli/cli_auth.py

@@ -1,6 +1,7 @@
 import click
 
 from ._auth._auth_service import AuthService
+from ._telemetry import track_command
 from ._utils._common import environment_options
 from ._utils._console import ConsoleLogger
 
@@ -43,6 +44,7 @@
     default="OR.Execution",
     help="Space-separated list of OAuth scopes to request (e.g., 'OR.Execution OR.Queues'). Defaults to 'OR.Execution'",
 )
+@track_command("authenticate")
 def auth(
     environment: str,
     force: bool = False,

src/uipath/_cli/cli_debug.py

@@ -34,6 +34,7 @@
 from uipath.platform.common import UiPathConfig
 from uipath.tracing import LiveTrackingSpanProcessor, LlmOpsHttpExporter
 
+from ._telemetry import track_command
 from ._utils._console import ConsoleLogger
 from .middlewares import Middlewares
 
@@ -125,6 +126,7 @@ def load_simulation_config() -> MockingContext | None:
     default=5678,
     help="Port for the debug server (default: 5678)",
 )
+@track_command("debug")
 def debug(
     entrypoint: str | None,
     input: str | None,

src/uipath/_cli/cli_deploy.py

@@ -1,5 +1,6 @@
 import click
 
+from ._telemetry import track_command
 from .cli_pack import pack
 from .cli_publish import publish
 
@@ -27,6 +28,7 @@
     help="Folder name to publish to (skips interactive selection)",
 )
 @click.argument("root", type=str, default="./")
+@track_command("deploy")
 def deploy(root, feed, folder):
     """Pack and publish the project."""
     ctx = click.get_current_context()

src/uipath/_cli/cli_dev.py

@@ -9,6 +9,8 @@
 from uipath._cli._utils._debug import setup_debugging
 from uipath._cli.middlewares import Middlewares
 
+from ._telemetry import track_command
+
 console = ConsoleLogger()
 
 
@@ -57,6 +59,7 @@ def _check_dev_dependency(interface: str) -> None:
     default=5678,
     help="Port for the debug server (default: 5678)",
 )
+@track_command("dev")
 def dev(interface: str, debug: bool, debug_port: int) -> None:
     """Launch UiPath Developer Console.
 

src/uipath/_cli/cli_init.py

@@ -31,6 +31,7 @@
 
 from .._utils.constants import ENV_TELEMETRY_ENABLED
 from ..telemetry._constants import _PROJECT_KEY, _TELEMETRY_CONFIG_FILE
+from ._telemetry import track_command
 from ._utils._console import ConsoleLogger
 from .middlewares import Middlewares
 from .models.runtime_schema import Bindings
@@ -326,6 +327,7 @@ def _display_entrypoint_graphs(entry_point_schemas: list[UiPathRuntimeSchema]) -
     default=False,
     help="Won't override existing .agent files and AGENTS.md file.",
 )
+@track_command("initialize")
 def init(no_agents_md_override: bool) -> None:
     """Initialize the project."""
     with console.spinner("Initializing UiPath project ..."):

src/uipath/_cli/cli_invoke.py

@@ -7,6 +7,7 @@
 import httpx
 
 from .._utils._ssl_context import get_httpx_client_kwargs
+from ._telemetry import track_command
 from ._utils._common import get_env_vars
 from ._utils._console import ConsoleLogger
 from ._utils._folders import get_personal_workspace_info_async
@@ -43,6 +44,7 @@ def _read_project_details() -> tuple[str, str]:
     type=click.Path(exists=True),
     help="File path for the .json input",
 )
+@track_command("invoke")
 def invoke(entrypoint: str | None, input: str | None, file: str | None) -> None:
     """Invoke an agent published in my workspace."""
     if file:

src/uipath/_cli/cli_new.py

@@ -4,6 +4,7 @@
 
 import click
 
+from ._telemetry import track_command
 from ._utils._console import ConsoleLogger
 from .middlewares import Middlewares
 
@@ -46,6 +47,7 @@ def generate_uipath_json(target_directory):
 
 @click.command()
 @click.argument("name", type=str, default="")
+@track_command("new")
 def new(name: str):
     """Generate a quick-start project."""
     directory = os.getcwd()