feat: refactor runtime system for extensible multi-entrypoint support

src/uipath/_cli/_runtime/_contracts.py

@@ -836,6 +836,14 @@ def from_context(self, context: C) -> T:
             return self.runtime_generator(context)
         return self.runtime_class.from_context(context)
 
+    def discover_all_runtimes(self) -> List[T]:
+        """Get a list of all available runtimes."""
+        return []
+
+    def get_runtime(self, entrypoint: str) -> Optional[T]:
+        """Get a specific runtime by entrypoint."""
+        return None
+
     async def execute(self, context: C) -> Optional[UiPathRuntimeResult]:
         """Execute runtime with context."""
         async with self.from_context(context) as runtime:

src/uipath/_cli/_runtime/_runtime.py

@@ -76,31 +76,26 @@ async def execute(self) -> Optional[UiPathRuntimeResult]:
 console = ConsoleLogger()
 
 
-def get_user_script(directory: str, entrypoint: Optional[str] = None) -> Optional[str]:
-    """Find the Python script to process."""
+def get_user_scripts(directory: str, entrypoint: Optional[str] = None) -> List[str]:
+    """Find all Python scripts 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
+            return []
+        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
+        return []
+
+    return [os.path.join(directory, f) for f in python_files]
 
 
 class UiPathScriptRuntime(UiPathRuntime):
@@ -122,18 +117,26 @@ def get_binding_resources(self) -> List[BindingResource]:
 
         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)
+        working_dir = os.getcwd()
+        script_paths = get_user_scripts(working_dir, self.context.entrypoint)
+        if not script_paths:
+            raise ValueError(
+                "No Python scripts found. Please specify an entrypoint or ensure Python files exist in the directory."
+            )
+        script_path = script_paths[0]
         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.")
+        working_dir = os.getcwd()
+        script_paths = get_user_scripts(working_dir, self.context.entrypoint)
+        if not script_paths:
+            raise ValueError(
+                "No Python scripts found. Please specify an entrypoint or ensure Python files exist in the directory."
+            )
+        script_path = script_paths[0]
         relative_path = Path(script_path).relative_to(working_dir).as_posix()
         args = generate_args(script_path)
         return Entrypoint(

src/uipath/_cli/_runtime/_runtime_factory.py

@@ -1,21 +1,53 @@
+import os
+from typing import List, Optional
+
 from uipath._cli._runtime._contracts import (
-    UiPathBaseRuntime,
     UiPathRuntimeContext,
     UiPathRuntimeFactory,
 )
-from uipath._cli._runtime._runtime import UiPathScriptRuntime
+from uipath._cli._runtime._runtime import UiPathScriptRuntime, get_user_scripts
+
 
+class UiPathScriptRuntimeFactory(
+    UiPathRuntimeFactory[UiPathScriptRuntime, UiPathRuntimeContext]
+):
+    """Factory for Python script runtimes."""
 
-def generate_runtime_factory() -> UiPathRuntimeFactory[
-    UiPathBaseRuntime, UiPathRuntimeContext
-]:
-    runtime_factory: UiPathRuntimeFactory[UiPathBaseRuntime, UiPathRuntimeContext] = (
-        UiPathRuntimeFactory(
+    def __init__(self):
+        super().__init__(
             UiPathScriptRuntime,
             UiPathRuntimeContext,
             context_generator=lambda **kwargs: UiPathRuntimeContext.with_defaults(
                 **kwargs
             ),
         )
-    )
-    return runtime_factory
+
+    def discover_all_runtimes(self) -> List[UiPathScriptRuntime]:
+        """Get a list of all available Python script runtimes."""
+        scripts = get_user_scripts(os.getcwd())
+
+        runtimes = []
+        for script_path in scripts:
+            runtime = self._create_runtime(script_path)
+            if runtime:
+                runtimes.append(runtime)
+
+        return runtimes
+
+    def get_runtime(self, entrypoint: str) -> Optional[UiPathScriptRuntime]:
+        """Get runtime for a specific Python script."""
+        if not os.path.isabs(entrypoint):
+            script_path = os.path.abspath(entrypoint)
+        else:
+            script_path = entrypoint
+
+        if not os.path.isfile(script_path) or not script_path.endswith(".py"):
+            return None
+
+        return self._create_runtime(script_path)
+
+    def _create_runtime(self, script_path: str) -> Optional[UiPathScriptRuntime]:
+        """Create runtime instance for a script path."""
+        context = self.new_context(entrypoint=script_path)
+        runtime = self.from_context(context)
+        return runtime

src/uipath/_cli/cli_eval.py

@@ -12,7 +12,6 @@
     UiPathEvalContext,
     UiPathEvalRuntime,
 )
-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
@@ -138,7 +137,9 @@ def eval(
         asyncio.run(console_reporter.subscribe_to_eval_runtime_events(event_bus))
 
         try:
-            runtime_factory = generate_runtime_factory()
+            # Last one is the Python script runtime factory
+            runtime_factory = Middlewares._runtime_factories[-1]
+
             if eval_context.job_id:
                 runtime_factory.add_span_exporter(LlmOpsHttpExporter())
 

src/uipath/_cli/cli_init.py

@@ -12,8 +12,7 @@
 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 ._runtime._contracts import BindingResource
 from ._utils._console import ConsoleLogger
 from .middlewares import Middlewares
 from .models.runtime_schema import Bindings, RuntimeSchema
@@ -163,6 +162,19 @@ def write_config_file(config_data: Dict[str, Any] | RuntimeSchema) -> None:
     return CONFIG_PATH
 
 
+def deduplicate_binding_resources(
+    binding_resources: list[BindingResource],
+) -> list[BindingResource]:
+    """Deduplicate binding resources by key."""
+    unique_keys = set()
+    unique_binding_resources = []
+    for binding_resource in binding_resources:
+        if binding_resource.key not in unique_keys:
+            unique_keys.add(binding_resource.key)
+            unique_binding_resources.append(binding_resource)
+    return unique_binding_resources
+
+
 @click.command()
 @click.argument("entrypoint", required=False, default=None)
 @click.option(
@@ -209,24 +221,39 @@ def init(entrypoint: str, infer_bindings: bool, no_agents_md_override: bool) ->
             return
 
         generate_agent_md_files(current_directory, no_agents_md_override)
-        script_path = get_user_script(current_directory, entrypoint=entrypoint)
-        if not script_path:
-            return
-
-        context_args = {
-            "runtime_dir": os.getcwd(),
-            "entrypoint": script_path,
-        }
 
         def initialize() -> None:
             try:
-                runtime = generate_runtime_factory().new_runtime(**context_args)
+                # Discover all runtimes (or get specific one if entrypoint specified)
+                if entrypoint:
+                    runtime = Middlewares.get_runtime(entrypoint)
+                    if not runtime:
+                        raise ValueError(
+                            f"No factory can handle entrypoint: {entrypoint}"
+                        )
+                    runtimes = [runtime]
+                else:
+                    runtimes = Middlewares.discover_all_runtimes()
+                    if not runtimes:
+                        raise ValueError("No runtimes discovered in current directory")
+
+                all_entrypoints = []
+                binding_resources = []
+
+                # Generate entrypoint for each runtime
+                for runtime in runtimes:
+                    all_entrypoints.append(runtime.get_entrypoint)
+                    if infer_bindings:
+                        binding_resources.extend(runtime.get_binding_resources)
+
+                binding_resources = deduplicate_binding_resources(binding_resources)
                 bindings = Bindings(
                     version="2.0",
-                    resources=runtime.get_binding_resources,
+                    resources=binding_resources,
                 )
+
                 config_data = RuntimeSchema(
-                    entryPoints=[runtime.get_entrypoint],
+                    entryPoints=all_entrypoints,
                     bindings=bindings,
                 )
                 config_path = write_config_file(config_data)

src/uipath/_cli/cli_run.py

@@ -6,7 +6,6 @@
 
 import click
 
-from uipath._cli._runtime._runtime_factory import generate_runtime_factory
 from uipath._cli._utils._debug import setup_debugging
 from uipath.tracing import LlmOpsHttpExporter
 
@@ -106,7 +105,8 @@ def run(
         try:
 
             async def execute() -> None:
-                runtime_factory = generate_runtime_factory()
+                # Last one is the Python script runtime factory
+                runtime_factory = Middlewares._runtime_factories[-1]
                 context = runtime_factory.new_context(**context_args)
                 if context.job_id:
                     runtime_factory.add_span_exporter(LlmOpsHttpExporter())

src/uipath/_cli/middlewares.py

@@ -4,6 +4,8 @@
 from dataclasses import dataclass
 from typing import Any, Callable, Dict, List, Optional
 
+from ._runtime._contracts import UiPathBaseRuntime, UiPathRuntimeFactory
+from ._runtime._runtime_factory import UiPathScriptRuntimeFactory
 from ._utils._console import ConsoleLogger
 
 logger = logging.getLogger(__name__)
@@ -33,6 +35,7 @@ class Middlewares:
         "eval": [],
         "debug": [],
     }
+    _runtime_factories: List[UiPathRuntimeFactory[Any, Any]] = []
     _plugins_loaded = False
 
     @classmethod
@@ -45,6 +48,50 @@ def register(cls, command: str, middleware: MiddlewareFunc) -> None:
             f"Registered middleware for command '{command}': {middleware.__name__}"
         )
 
+    @classmethod
+    def register_runtime_factory(cls, factory: UiPathRuntimeFactory[Any, Any]) -> None:
+        """Register a runtime factory in the chain."""
+        cls._runtime_factories.append(factory)
+        logger.debug(f"Registered runtime factory: {factory.__class__.__name__}")
+
+    @classmethod
+    def discover_all_runtimes(cls) -> List[UiPathBaseRuntime]:
+        """Discover all runtimes in the current directory using registered runtime factories."""
+        if not cls._plugins_loaded:
+            cls.load_plugins()
+
+        for factory in cls._runtime_factories:
+            try:
+                runtimes = factory.discover_all_runtimes()
+                if runtimes:
+                    return runtimes
+            except Exception as e:
+                logger.error(
+                    f"Runtime factory {factory.__class__.__name__} discovery failed: {e}"
+                )
+                raise
+
+        return []
+
+    @classmethod
+    def get_runtime(cls, entrypoint: str) -> Optional[UiPathBaseRuntime]:
+        """Get runtime for a specific entrypoint."""
+        if not cls._plugins_loaded:
+            cls.load_plugins()
+
+        for factory in cls._runtime_factories:
+            try:
+                runtime = factory.get_runtime(entrypoint)
+                if runtime:
+                    return runtime
+            except Exception as e:
+                logger.error(
+                    f"Runtime factory {factory.__class__.__name__} failed for {entrypoint}: {e}"
+                )
+                raise
+
+        return None
+
     @classmethod
     def get(cls, command: str) -> List[MiddlewareFunc]:
         """Get all middlewares for a specific command."""
@@ -104,7 +151,7 @@ def clear(cls, command: Optional[str] = None) -> None:
 
     @classmethod
     def load_plugins(cls) -> None:
-        """Load all middlewares registered via entry points."""
+        """Load all plugins and register runtime factories."""
         if cls._plugins_loaded:
             return
 
@@ -137,6 +184,8 @@ def load_plugins(cls) -> None:
             else:
                 logger.debug("No middleware plugins found")
 
+            # Register the default runtime factory after all the plugin ones
+            cls.register_runtime_factory(UiPathScriptRuntimeFactory())
         except Exception as e:
             logger.error(f"No middleware plugins loaded: {str(e)}")
         finally:

tests/cli/test_init.py

@@ -66,19 +66,16 @@ def test_init_script_detection(self, runner: CliRunner, temp_dir: str) -> None:
                 f.write("def main(input): return input")
 
             result = runner.invoke(cli, ["init"])
-            assert result.exit_code == 1
-            assert (
-                "Multiple python files found in the current directory" in result.output
-            )
-            assert "Please specify the entrypoint" in result.output
+            assert result.exit_code == 0
+            assert os.path.exists("uipath.json")
 
     def test_init_with_entrypoint(self, runner: CliRunner, temp_dir: str) -> None:
         """Test init with specified entrypoint."""
         with runner.isolated_filesystem(temp_dir=temp_dir):
             # Test with non-existent file
             result = runner.invoke(cli, ["init", "nonexistent.py"])
             assert result.exit_code == 1
-            assert "does not exist in the current directory" in result.output
+            assert "error" in result.output.lower()
 
             # Test with valid Python file
             with open("script.py", "w") as f: