feat: add system resources resolution support

src/uipath/_cli/_utils/_studio_project.py

@@ -6,8 +6,9 @@
 from typing import Any, Callable, List, Optional, Union
 
 import click
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, field_validator
 
+from uipath._cli.models.runtime_schema import BindingResource, Bindings
 from uipath._utils._bindings import ResourceOverwrite, ResourceOverwriteParser
 from uipath._utils.constants import (
     ENV_TENANT_ID,
@@ -550,6 +551,10 @@ async def get_resource_overwrites(self) -> dict[str, ResourceOverwrite]:
         with open(UiPathConfig.bindings_file_path, "rb") as f:
             file_content = f.read()
 
+        bindings = TypeAdapter(Bindings).validate_python(
+            json.loads(file_content.decode("utf-8"))
+        )
+
         solution_id = await self._get_solution_id()
         tenant_id = os.getenv(ENV_TENANT_ID, None)
 
@@ -577,6 +582,33 @@ async def get_resource_overwrites(self) -> dict[str, ResourceOverwrite]:
         for key, value in data.items():
             overwrites[key] = ResourceOverwriteParser.parse(key, value)
 
+        from functools import cache
+
+        @cache
+        def get_uipath():
+            from uipath.platform import UiPath
+
+            return UiPath()
+
+        async def try_add_system_index(binding_resource: BindingResource) -> None:
+            binding_resource_name = binding_resource.value["name"].default_value
+            if (
+                system_index
+                := await get_uipath().context_grounding._resolve_system_index_async(
+                    binding_resource_name
+                )
+            ):
+                binding_key = f"{binding_resource.resource}.{binding_resource_name}"
+                overwrites[binding_key] = system_index
+
+        # check any missing binding from overwrites list (and try to resolve it)
+        for binding_resource in bindings.resources:
+            if binding_resource.key not in overwrites:
+                match binding_resource.resource:
+                    case "index":
+                        await try_add_system_index(binding_resource)
+                    # can be extended for other system resources
+
         return overwrites
 
     async def create_virtual_resource(

src/uipath/_utils/_bindings.py

@@ -40,6 +40,20 @@ def folder_identifier(self) -> str:
         pass
 
 
+class SystemResourceOverwrite(ResourceOverwrite):
+    resource_type: Literal["index"]
+    name: str = Field(alias="name")
+    folder_key: str = Field(alias="folderKey")
+
+    @property
+    def resource_identifier(self) -> str:
+        return self.name
+
+    @property
+    def folder_identifier(self) -> str:
+        return self.folder_key
+
+
 class GenericResourceOverwrite(ResourceOverwrite):
     resource_type: Literal["process", "index", "app", "asset", "bucket", "mcpServer"]
     name: str = Field(alias="name")
@@ -142,6 +156,8 @@ def resource_override(
     resource_type: str,
     resource_identifier: str = "name",
     folder_identifier: str = "folder_path",
+    identifiers_resolver_callable: Callable[[ResourceOverwrite], tuple[str, str]]
+    | None = None,
 ) -> Callable[..., Any]:
     """Decorator for applying resource overrides for an overridable resource.
 
@@ -152,6 +168,8 @@ def resource_override(
         resource_type: Type of resource to check for overrides (e.g., "asset", "bucket")
         resource_identifier: Key name for the resource ID in override data (default: "name")
         folder_identifier: Key name for the folder path in override data (default: "folder_path")
+        identifiers_resolver_callable: Resolver func for identifier names, used for more complex logic if needed.
+                              Should return a tuple[str, str] containing (resource_identifier, folder_identifier)
 
     Returns:
         Decorated function that receives overridden resource identifiers when applicable
@@ -180,6 +198,8 @@ def process_args(args, kwargs) -> dict[str, Any]:
             context_overwrites = _resource_overwrites.get()
 
             if context_overwrites is not None:
+                resource_identifier_key = resource_identifier
+                folder_identifier_key = folder_identifier
                 resource_identifier_value = all_args.get(resource_identifier)
                 folder_identifier_value = all_args.get(folder_identifier)
 
@@ -196,12 +216,18 @@ def process_args(args, kwargs) -> dict[str, Any]:
 
                 # Apply the matched overwrite
                 if matched_overwrite is not None:
-                    if resource_identifier in sig.parameters:
-                        all_args[resource_identifier] = (
+                    if identifiers_resolver_callable:
+                        # if identifiers resolver is provided, use it to extract the argument names to be replaced when calling the func
+                        resource_identifier_key, folder_identifier_key = (
+                            identifiers_resolver_callable(matched_overwrite)
+                        )
+
+                    if resource_identifier_key in sig.parameters:
+                        all_args[resource_identifier_key] = (
                             matched_overwrite.resource_identifier
                         )
-                    if folder_identifier in sig.parameters:
-                        all_args[folder_identifier] = (
+                    if folder_identifier_key in sig.parameters:
+                        all_args[folder_identifier_key] = (
                             matched_overwrite.folder_identifier
                         )
 

src/uipath/platform/context_grounding/_context_grounding_service.py

@@ -1,10 +1,15 @@
 from pathlib import Path
-from typing import Annotated, Any, Dict, List, Optional, Tuple, Union
+from typing import Annotated, Any, Dict, List, Optional, Tuple, Union, cast
 
 import httpx
 from pydantic import Field, TypeAdapter
 
 from ..._utils import Endpoint, RequestSpec, header_folder, resource_override
+from ..._utils._bindings import (
+    GenericResourceOverwrite,
+    ResourceOverwrite,
+    SystemResourceOverwrite,
+)
 from ..._utils._ssl_context import get_httpx_client_kwargs
 from ..._utils.constants import (
     LLMV4_REQUEST,
@@ -964,7 +969,16 @@ async def start_deep_rag_async(
 
         return DeepRagCreationResponse.model_validate(response.json())
 
-    @resource_override(resource_type="index")
+    @resource_override(
+        resource_type="index",
+        identifiers_resolver_callable=lambda overwrite: cast(
+            dict[type[ResourceOverwrite], tuple[str, str]],
+            {
+                GenericResourceOverwrite: ("name", "folder_path"),
+                SystemResourceOverwrite: ("name", "folder_key"),
+            },
+        ).get(type(overwrite), ("name", "folder_path")),
+    )
     @traced(name="contextgrounding_search", run_type="uipath")
     def search(
         self,
@@ -990,10 +1004,6 @@ def search(
             List[ContextGroundingQueryResponse]: A list of search results, each containing
                 relevant contextual information and metadata.
         """
-        index = self.retrieve(name, folder_key=folder_key, folder_path=folder_path)
-        if index and index.in_progress_ingestion():
-            raise IngestionInProgressException(index_name=name)
-
         spec = self._search_spec(
             name,
             query,
@@ -1013,7 +1023,16 @@ def search(
             response.json()
         )
 
-    @resource_override(resource_type="index")
+    @resource_override(
+        resource_type="index",
+        identifiers_resolver_callable=lambda overwrite: cast(
+            dict[type[ResourceOverwrite], tuple[str, str]],
+            {
+                GenericResourceOverwrite: ("name", "folder_path"),
+                SystemResourceOverwrite: ("name", "folder_key"),
+            },
+        ).get(type(overwrite), ("name", "folder_path")),
+    )
     @traced(name="contextgrounding_search", run_type="uipath")
     async def search_async(
         self,
@@ -1039,13 +1058,6 @@ async def search_async(
             List[ContextGroundingQueryResponse]: A list of search results, each containing
                 relevant contextual information and metadata.
         """
-        index = self.retrieve(
-            name,
-            folder_key=folder_key,
-            folder_path=folder_path,
-        )
-        if index and index.in_progress_ingestion():
-            raise IngestionInProgressException(index_name=name)
         spec = self._search_spec(
             name,
             query,
@@ -1191,6 +1203,60 @@ async def delete_index_async(
             headers=spec.headers,
         )
 
+    async def _resolve_system_index_async(
+        self,
+        name: str,
+        **kwargs,
+    ) -> SystemResourceOverwrite | None:
+        """Asynchronously retrieve context grounding system index by its name.
+
+        Args:
+            name (str): The name of the context index to retrieve.
+
+        Returns:
+            SystemResourceOverwrite: The index name and folder key, if found.
+
+        """
+        spec = self._retrieve_system_index_spec(
+            name,
+        )
+
+        response = (
+            await self.request_async(
+                spec.method,
+                spec.endpoint,
+                params=spec.params,
+                headers=spec.headers,
+            )
+        ).json()
+        try:
+            context_grounding_index = next(
+                ContextGroundingIndex.model_validate(item)
+                for item in response["value"]
+                if item["name"] == name
+            )
+        except StopIteration:
+            return None
+
+        assert context_grounding_index.name is not None
+        assert context_grounding_index.folder_key is not None
+
+        return SystemResourceOverwrite(
+            resource_type="index",
+            name=context_grounding_index.name,
+            folder_key=context_grounding_index.folder_key,
+        )
+
+    def _retrieve_system_index_spec(
+        self,
+        name: str,
+    ) -> RequestSpec:
+        return RequestSpec(
+            method="GET",
+            endpoint=Endpoint("/ecs_/v2/indexes/AllSystemIndexes"),
+            params={"$filter": f"Name eq '{name}'"},
+        )
+
     def _ingest_spec(
         self,
         key: str,

tests/resource_overrides/test_resource_overrides.py

@@ -1,4 +1,5 @@
 # type: ignore
+import contextlib
 import json
 import os
 from pathlib import Path
@@ -12,6 +13,7 @@
 
 from uipath._cli import cli
 from uipath._utils._bindings import ResourceOverwriteParser
+from uipath.platform.common import UiPathConfig
 
 
 @pytest.fixture
@@ -393,6 +395,92 @@ async def mock_get_resource_overwrites():
 
                     self._assert(result, httpx_mock)
 
+    @pytest.mark.anyio
+    async def test_get_resource_overwrites_resolves_system_index_from_bindings(
+        self,
+        temp_dir: str,
+        httpx_mock: HTTPXMock,
+    ):
+        """Test that get_resource_overwrites resolves system index bindings without explicit overwrites."""
+        from uipath._cli._utils._studio_project import StudioClient
+        from uipath._utils._bindings import SystemResourceOverwrite
+
+        UiPathConfig.studio_solution_id = None
+
+        with patch.dict(
+            os.environ,
+            {
+                "UIPATH_URL": "https://example.com/org/tenant",
+                "UIPATH_ORGANIZATION_ID": "test-org",
+                "UIPATH_TENANT_ID": "test-tenant",
+                "UIPATH_PROJECT_ID": "test-project-123",
+                "UIPATH_ACCESS_TOKEN": "test-token",
+            },
+        ):
+            # Create bindings.json with a system index binding (no explicit overwrite)
+            bindings_data = {
+                "version": "1.0",
+                "resources": [
+                    {
+                        "resource": "index",
+                        "key": "index.MySystemIndex",
+                        "value": {
+                            "name": {
+                                "defaultValue": "MySystemIndex",
+                                "isExpression": False,
+                                "displayName": "My System Index",
+                            }
+                        },
+                    }
+                ],
+            }
+
+            bindings_path = os.path.join(temp_dir, "bindings.json")
+            with open(bindings_path, "w") as f:
+                json.dump(bindings_data, f)
+
+            # solution ID API call
+            httpx_mock.add_response(
+                url="https://example.com/org/studio_/backend/api/Project/test-project-123",
+                method="GET",
+                json={"solutionId": "test-solution-123"},
+            )
+
+            # binding-overwrites API call (returns empty - no explicit overwrites)
+            httpx_mock.add_response(
+                url="https://example.com/org/studio_/backend/api/resourcebuilder/test-solution-123/binding-overwrites",
+                method="POST",
+                json={},  # no explicit overwrites
+            )
+
+            #  system index API response
+            httpx_mock.add_response(
+                url="https://example.com/org/tenant/ecs_/v2/indexes/AllSystemIndexes?$filter=Name eq 'MySystemIndex'",
+                method="GET",
+                json={
+                    "value": [
+                        {
+                            "id": "system-index-123",
+                            "name": "MySystemIndex",
+                            "folderKey": "system-folder-key-789",
+                            "lastIngestionStatus": "Completed",
+                        }
+                    ]
+                },
+            )
+
+            with contextlib.chdir(temp_dir):
+                client = StudioClient(project_id="test-project-123")
+                overwrites = await client.get_resource_overwrites()
+
+                # verify the system index was resolved and added to overwrites
+                assert "index.MySystemIndex" in overwrites
+                system_index = overwrites["index.MySystemIndex"]
+
+                assert isinstance(system_index, SystemResourceOverwrite)
+                assert system_index.name == "MySystemIndex"
+                assert system_index.folder_key == "system-folder-key-789"
+
 
 class TestResourceOverrideWithTracing:
     """Tests for resource_override decorator integration with tracing."""