feat(UpdateContract): add a runtime generator

This will make it easier to create new runtime objects
and simplify implementation all uipath commands.

Author: akshaylive

src/uipath/_cli/_evals/_runtime.py

@@ -451,12 +451,12 @@ def _get_and_clear_execution_data(
     async def execute_runtime(
         self, eval_item: EvaluationItem, execution_id: str
     ) -> UiPathEvalRunExecutionOutput:
-        runtime_context: C = self.factory.new_context(
-            execution_id=execution_id,
-            input_json=eval_item.inputs,
-            is_eval_run=True,
-            log_handler=self._setup_execution_logging(execution_id),
-        )
+        context_args = self.context.model_dump()
+        context_args["execution_id"] = execution_id
+        context_args["input_json"] = eval_item.inputs
+        context_args["is_eval_run"] = True
+        context_args["log_handler"] = self._setup_execution_logging(execution_id)
+        runtime_context: C = self.factory.new_context(**context_args)
         if runtime_context.execution_id is None:
             raise ValueError("execution_id must be set for eval runs")
 

src/uipath/_cli/_runtime/_contracts.py

@@ -39,6 +39,7 @@
 from uipath.agent.conversation import UiPathConversationEvent, UiPathConversationMessage
 from uipath.tracing import TracingManager
 
+from ..models.runtime_schema import BindingResource, Entrypoint
 from ._logging import LogsInterceptor
 
 logger = logging.getLogger(__name__)
@@ -516,6 +517,22 @@ def from_context(cls, context: UiPathRuntimeContext):
         runtime = cls(context)
         return runtime
 
+    @property
+    def get_binding_resources(self) -> List[BindingResource]:
+        """Get binding resources for this runtime.
+
+        Returns: A list of binding resources.
+        """
+        raise NotImplementedError()
+
+    @property
+    def get_entrypoint(self) -> Entrypoint:
+        """Get entrypoint for this runtime.
+
+        Returns: A entrypoint for this runtime.
+        """
+        raise NotImplementedError()
+
     async def __aenter__(self):
         """Async enter method called when entering the 'async with' block.
 
@@ -806,9 +823,9 @@ def new_context(self, **kwargs) -> C:
             return self.context_generator(**kwargs)
         return self.context_class(**kwargs)
 
-    def new_runtime(self) -> T:
+    def new_runtime(self, **kwargs) -> T:
         """Create a new runtime instance."""
-        context = self.new_context()
+        context = self.new_context(**kwargs)
         if self.runtime_generator:
             return self.runtime_generator(context)
         return self.runtime_class.from_context(context)

src/uipath/_cli/_runtime/_runtime.py

@@ -1,8 +1,18 @@
 """Python script runtime implementation for executing and managing python scripts."""
 
 import logging
-from typing import Any, Awaitable, Callable, Optional, TypeVar
-
+import os
+import uuid
+from functools import cached_property
+from pathlib import Path
+from typing import Any, Awaitable, Callable, List, Optional, TypeVar
+
+from typing_extensions import override
+
+from .._utils._console import ConsoleLogger
+from .._utils._input_args import generate_args
+from .._utils._parse_ast import generate_bindings  # type: ignore[attr-defined]
+from ..models.runtime_schema import BindingResource, Entrypoint
 from ._contracts import (
     UiPathBaseRuntime,
     UiPathErrorCategory,
@@ -63,6 +73,36 @@ async def execute(self) -> Optional[UiPathRuntimeResult]:
             ) from e
 
 
+console = ConsoleLogger()
+
+
+def get_user_script(directory: str, entrypoint: Optional[str] = None) -> Optional[str]:
+    """Find the Python script to process."""
+    if entrypoint:
+        script_path = os.path.join(directory, entrypoint)
+        if not os.path.isfile(script_path):
+            console.error(
+                f"The {entrypoint} file does not exist in the current directory."
+            )
+            return None
+        return script_path
+
+    python_files = [f for f in os.listdir(directory) if f.endswith(".py")]
+
+    if not python_files:
+        console.error(
+            "No python files found in the current directory.\nPlease specify the entrypoint: `uipath init <entrypoint_path>`"
+        )
+        return None
+    elif len(python_files) == 1:
+        return os.path.join(directory, python_files[0])
+    else:
+        console.error(
+            "Multiple python files found in the current directory.\nPlease specify the entrypoint: `uipath init <entrypoint_path>`"
+        )
+        return None
+
+
 class UiPathScriptRuntime(UiPathRuntime):
     """Runtime for executing Python scripts."""
 
@@ -74,3 +114,32 @@ def __init__(self, context: UiPathRuntimeContext, entrypoint: str):
     def from_context(cls, context: UiPathRuntimeContext):
         """Create runtime instance from context."""
         return UiPathScriptRuntime(context, context.entrypoint or "")
+
+    @cached_property
+    @override
+    def get_binding_resources(self) -> List[BindingResource]:
+        """Get binding resources for script runtime.
+
+        Returns: A list of binding resources.
+        """
+        working_dir = self.context.runtime_dir or os.getcwd()
+        script_path = get_user_script(working_dir, entrypoint=self.context.entrypoint)
+        bindings = generate_bindings(script_path)
+        return bindings.resources
+
+    @cached_property
+    @override
+    def get_entrypoint(self) -> Entrypoint:
+        working_dir = self.context.runtime_dir or os.getcwd()
+        script_path = get_user_script(working_dir, entrypoint=self.context.entrypoint)
+        if not script_path:
+            raise ValueError("Entrypoint not found.")
+        relative_path = Path(script_path).relative_to(working_dir).as_posix()
+        args = generate_args(script_path)
+        return Entrypoint(
+            file_path=relative_path,  # type: ignore[call-arg] # This exists
+            unique_id=str(uuid.uuid4()),
+            type="agent",
+            input=args["input"],
+            output=args["output"],
+        )

src/uipath/_cli/_runtime/_runtime_factory.py

@@ -0,0 +1,21 @@
+from uipath._cli._runtime._contracts import (
+    UiPathBaseRuntime,
+    UiPathRuntimeContext,
+    UiPathRuntimeFactory,
+)
+from uipath._cli._runtime._runtime import UiPathScriptRuntime
+
+
+def generate_runtime_factory() -> UiPathRuntimeFactory[
+    UiPathBaseRuntime, UiPathRuntimeContext
+]:
+    runtime_factory: UiPathRuntimeFactory[UiPathBaseRuntime, UiPathRuntimeContext] = (
+        UiPathRuntimeFactory(
+            UiPathScriptRuntime,
+            UiPathRuntimeContext,
+            context_generator=lambda **kwargs: UiPathRuntimeContext.with_defaults(
+                **kwargs
+            ),
+        )
+    )
+    return runtime_factory

src/uipath/_cli/cli_eval.py

@@ -12,11 +12,7 @@
     UiPathEvalContext,
     UiPathEvalRuntime,
 )
-from uipath._cli._runtime._contracts import (
-    UiPathRuntimeContext,
-    UiPathRuntimeFactory,
-)
-from uipath._cli._runtime._runtime import UiPathScriptRuntime
+from uipath._cli._runtime._runtime_factory import generate_runtime_factory
 from uipath._cli._utils._constants import UIPATH_PROJECT_ID
 from uipath._cli._utils._folders import get_personal_workspace_key_async
 from uipath._cli.middlewares import Middlewares
@@ -96,6 +92,15 @@ def eval(
         workers: Number of parallel workers for running evaluations
         no_report: Do not report the evaluation results
     """
+    context_args = {
+        "entrypoint": entrypoint or auto_discover_entrypoint(),
+        "eval_set": eval_set,
+        "eval_ids": eval_ids,
+        "workers": workers,
+        "no_report": no_report,
+        "output_file": output_file,
+    }
+
     should_register_progress_reporter = setup_reporting_prereq(no_report)
 
     result = Middlewares.next(
@@ -119,16 +124,9 @@ def eval(
             progress_reporter = StudioWebProgressReporter(LlmOpsHttpExporter())
             asyncio.run(progress_reporter.subscribe_to_eval_runtime_events(event_bus))
 
-        def generate_runtime_context(**context_kwargs) -> UiPathRuntimeContext:
-            runtime_context = UiPathRuntimeContext.with_defaults(**context_kwargs)
-            runtime_context.entrypoint = runtime_entrypoint
-            return runtime_context
-
-        runtime_entrypoint = entrypoint or auto_discover_entrypoint()
-
         eval_context = UiPathEvalContext.with_defaults(
             execution_output_file=output_file,
-            entrypoint=runtime_entrypoint,
+            entrypoint=context_args["entrypoint"],
         )
 
         eval_context.no_report = no_report
@@ -140,11 +138,7 @@ def generate_runtime_context(**context_kwargs) -> UiPathRuntimeContext:
         asyncio.run(console_reporter.subscribe_to_eval_runtime_events(event_bus))
 
         try:
-            runtime_factory = UiPathRuntimeFactory(
-                UiPathScriptRuntime,
-                UiPathRuntimeContext,
-                context_generator=generate_runtime_context,
-            )
+            runtime_factory = generate_runtime_factory()
             if eval_context.job_id:
                 runtime_factory.add_span_exporter(LlmOpsHttpExporter())
 

src/uipath/_cli/cli_init.py

@@ -5,19 +5,18 @@
 import os
 import shutil
 import uuid
-from pathlib import Path
 from typing import Any, Dict, Optional
 
 import click
 
 from .._utils.constants import ENV_TELEMETRY_ENABLED
 from ..telemetry import track
 from ..telemetry._constants import _PROJECT_KEY, _TELEMETRY_CONFIG_FILE
+from ._runtime._runtime import get_user_script
+from ._runtime._runtime_factory import generate_runtime_factory
 from ._utils._console import ConsoleLogger
-from ._utils._input_args import generate_args
-from ._utils._parse_ast import generate_bindings
 from .middlewares import Middlewares
-from .models.runtime_schema import Bindings, Entrypoint, RuntimeSchema
+from .models.runtime_schema import Bindings, RuntimeSchema
 
 console = ConsoleLogger()
 logger = logging.getLogger(__name__)
@@ -126,33 +125,6 @@ def get_existing_settings(config_path: str) -> Optional[Dict[str, Any]]:
         return None
 
 
-def get_user_script(directory: str, entrypoint: Optional[str] = None) -> Optional[str]:
-    """Find the Python script to process."""
-    if entrypoint:
-        script_path = os.path.join(directory, entrypoint)
-        if not os.path.isfile(script_path):
-            console.error(
-                f"The {entrypoint} file does not exist in the current directory."
-            )
-            return None
-        return script_path
-
-    python_files = [f for f in os.listdir(directory) if f.endswith(".py")]
-
-    if not python_files:
-        console.error(
-            "No python files found in the current directory.\nPlease specify the entrypoint: `uipath init <entrypoint_path>`"
-        )
-        return None
-    elif len(python_files) == 1:
-        return os.path.join(directory, python_files[0])
-    else:
-        console.error(
-            "Multiple python files found in the current directory.\nPlease specify the entrypoint: `uipath init <entrypoint_path>`"
-        )
-        return None
-
-
 def write_config_file(config_data: Dict[str, Any] | RuntimeSchema) -> None:
     existing_settings = get_existing_settings(CONFIG_PATH)
     if existing_settings is not None:
@@ -205,39 +177,28 @@ def init(entrypoint: str, infer_bindings: bool) -> None:
 
         generate_agent_md_files(current_directory)
         script_path = get_user_script(current_directory, entrypoint=entrypoint)
-
         if not script_path:
             return
 
-        try:
-            args = generate_args(script_path)
-
-            relative_path = Path(script_path).relative_to(current_directory).as_posix()
-            bindings = None
-            if infer_bindings:
-                try:
-                    bindings = generate_bindings(script_path)
-                except Exception as e:
-                    console.warning(f"Warning: Could not generate bindings: {str(e)}")
-            if bindings is None:
+        context_args = {
+            "runtime_dir": os.getcwd(),
+            "entrypoint": script_path,
+        }
+
+        def initialize() -> None:
+            try:
+                runtime = generate_runtime_factory().new_runtime(**context_args)
                 bindings = Bindings(
                     version="2.0",
-                    resources=[],
+                    resources=runtime.get_binding_resources,
                 )
-            config_data = RuntimeSchema(
-                entrypoints=[
-                    Entrypoint(
-                        file_path=relative_path,
-                        unique_id=str(uuid.uuid4()),
-                        type="agent",
-                        input=args["input"],
-                        output=args["output"],
-                    )
-                ],
-                bindings=bindings,
-            )
+                config_data = RuntimeSchema(
+                    entryPoints=[runtime.get_entrypoint],
+                    bindings=bindings,
+                )
+                config_path = write_config_file(config_data)
+                console.success(f"Created '{config_path}' file.")
+            except Exception as e:
+                console.error(f"Error creating configuration file:\n {str(e)}")
 
-            config_path = write_config_file(config_data)
-            console.success(f"Created '{config_path}' file.")
-        except Exception as e:
-            console.error(f"Error creating configuration file:\n {str(e)}")
+        initialize()