fix: add name-based MCP and A2A lookup

Author: radugheo

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.2.13"
+version = "0.2.14"
 description = "HTTP client library for programmatic access to UiPath Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

packages/uipath-platform/src/uipath/platform/agenthub/_remote_a2a_service.py

@@ -7,6 +7,7 @@
 
 import warnings
 from typing import Any, List
+from urllib.parse import quote
 
 from ..common._base_service import BaseService
 from ..common._bindings import resource_override
@@ -183,7 +184,7 @@ def retrieve(
             "remote_a2a.retrieve is experimental and subject to change.",
             stacklevel=2,
         )
-        spec = self._retrieve_spec(slug=slug, folder_path=folder_path)
+        spec = self._retrieve_spec(identifier=slug, folder_path=folder_path)
         response = self.request(
             spec.method,
             url=spec.endpoint,
@@ -229,7 +230,67 @@ async def main():
             "remote_a2a.retrieve_async is experimental and subject to change.",
             stacklevel=2,
         )
-        spec = self._retrieve_spec(slug=slug, folder_path=folder_path)
+        spec = self._retrieve_spec(identifier=slug, folder_path=folder_path)
+        response = await self.request_async(
+            spec.method,
+            url=spec.endpoint,
+            params=spec.params,
+            headers=spec.headers,
+        )
+        return RemoteA2aAgent.model_validate(response.json())
+
+    @resource_override(
+        resource_type="remoteA2aAgent",
+        resource_identifier="name",
+        legacy_resource_identifier="legacy_slug",
+    )
+    def retrieve_by_name(
+        self,
+        name: str,
+        *,
+        legacy_slug: str | None = None,
+        folder_path: str | None = None,
+    ) -> RemoteA2aAgent:
+        """Retrieve a Remote A2A agent by its display name.
+
+        ``legacy_slug`` supports already-deployed slug-keyed bindings during
+        the name-identity transition.
+        """
+        warnings.warn(
+            "remote_a2a.retrieve_by_name is experimental and subject to change.",
+            stacklevel=2,
+        )
+        spec = self._retrieve_spec(identifier=name, folder_path=folder_path)
+        response = self.request(
+            spec.method,
+            url=spec.endpoint,
+            params=spec.params,
+            headers=spec.headers,
+        )
+        return RemoteA2aAgent.model_validate(response.json())
+
+    @resource_override(
+        resource_type="remoteA2aAgent",
+        resource_identifier="name",
+        legacy_resource_identifier="legacy_slug",
+    )
+    async def retrieve_by_name_async(
+        self,
+        name: str,
+        *,
+        legacy_slug: str | None = None,
+        folder_path: str | None = None,
+    ) -> RemoteA2aAgent:
+        """Retrieve a Remote A2A agent by its display name.
+
+        ``legacy_slug`` supports already-deployed slug-keyed bindings during
+        the name-identity transition.
+        """
+        warnings.warn(
+            "remote_a2a.retrieve_by_name_async is experimental and subject to change.",
+            stacklevel=2,
+        )
+        spec = self._retrieve_spec(identifier=name, folder_path=folder_path)
         response = await self.request_async(
             spec.method,
             url=spec.endpoint,
@@ -279,14 +340,16 @@ def _list_spec(
 
     def _retrieve_spec(
         self,
-        slug: str,
+        identifier: str,
         *,
         folder_path: str | None,
     ) -> RequestSpec:
         folder_key = self._resolve_folder_key(folder_path)
         return RequestSpec(
             method="GET",
-            endpoint=Endpoint(f"/agenthub_/api/remote-a2a-agents/{slug}"),
+            endpoint=Endpoint(
+                f"/agenthub_/api/remote-a2a-agents/{quote(identifier, safe='')}"
+            ),
             headers={
                 **header_folder(folder_key, None),
             },

packages/uipath-platform/src/uipath/platform/common/_bindings.py

@@ -219,6 +219,7 @@ def resource_override(
     resource_type: str,
     resource_identifier: str = "name",
     folder_identifier: str = "folder_path",
+    legacy_resource_identifier: str | None = None,
 ) -> Callable[..., Any]:
     """Decorator for applying resource overrides for an overridable resource.
 
@@ -229,6 +230,9 @@ 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")
+        legacy_resource_identifier: Optional argument containing the identifier used
+            by older binding keys. It is checked only when the primary identifier
+            does not match.
 
     Returns:
         Decorated function that receives overridden resource identifiers when applicable
@@ -257,19 +261,26 @@ def process_args(args, kwargs) -> dict[str, Any]:
             context_overwrites = _resource_overwrites.get()
 
             if context_overwrites is not None:
-                resource_identifier_value = all_args.get(resource_identifier)
                 folder_identifier_value = all_args.get(folder_identifier)
-
-                key = f"{resource_type}.{resource_identifier_value}"
-                # try to apply folder path, fallback to resource_type.resource_name
-                if folder_identifier_value:
-                    key = (
-                        f"{key}.{folder_identifier_value}"
-                        if f"{key}.{folder_identifier_value}" in context_overwrites
-                        else key
-                    )
-
-                matched_overwrite = context_overwrites.get(key)
+                identifier_values = [all_args.get(resource_identifier)]
+                if legacy_resource_identifier is not None:
+                    identifier_values.append(all_args.get(legacy_resource_identifier))
+
+                matched_overwrite = None
+                key = f"{resource_type}.{identifier_values[0]}"
+                for identifier_value in identifier_values:
+                    if identifier_value is None:
+                        continue
+                    candidate_key = f"{resource_type}.{identifier_value}"
+                    # Try the folder-qualified key first, then the bare key.
+                    if folder_identifier_value:
+                        folder_key = f"{candidate_key}.{folder_identifier_value}"
+                        if folder_key in context_overwrites:
+                            candidate_key = folder_key
+                    matched_overwrite = context_overwrites.get(candidate_key)
+                    key = candidate_key
+                    if matched_overwrite is not None:
+                        break
 
                 # Apply the matched overwrite
                 if matched_overwrite is not None:

packages/uipath-platform/src/uipath/platform/orchestrator/_mcp_service.py

@@ -1,4 +1,5 @@
 from typing import List
+from urllib.parse import quote
 
 from uipath.core.tracing import traced
 
@@ -137,7 +138,7 @@ def retrieve(
             ```
         """
         spec = self._retrieve_spec(
-            slug=slug,
+            identifier=slug,
             folder_path=folder_path,
         )
 
@@ -183,7 +184,7 @@ async def main():
             ```
         """
         spec = self._retrieve_spec(
-            slug=slug,
+            identifier=slug,
             folder_path=folder_path,
         )
 
@@ -196,6 +197,60 @@ async def main():
 
         return McpServer.model_validate(response.json())
 
+    @resource_override(
+        resource_type="mcpServer",
+        resource_identifier="name",
+        legacy_resource_identifier="legacy_slug",
+    )
+    @traced(name="mcp_retrieve_by_name", run_type="uipath")
+    def retrieve_by_name(
+        self,
+        name: str,
+        *,
+        legacy_slug: str | None = None,
+        folder_path: str | None = None,
+    ) -> McpServer:
+        """Retrieve an MCP server by its display name.
+
+        ``legacy_slug`` lets already-deployed slug-keyed bindings apply their
+        overwrite while packages transition to name-keyed bindings.
+        """
+        spec = self._retrieve_spec(identifier=name, folder_path=folder_path)
+        response = self.request(
+            spec.method,
+            url=spec.endpoint,
+            params=spec.params,
+            headers=spec.headers,
+        )
+        return McpServer.model_validate(response.json())
+
+    @resource_override(
+        resource_type="mcpServer",
+        resource_identifier="name",
+        legacy_resource_identifier="legacy_slug",
+    )
+    @traced(name="mcp_retrieve_by_name", run_type="uipath")
+    async def retrieve_by_name_async(
+        self,
+        name: str,
+        *,
+        legacy_slug: str | None = None,
+        folder_path: str | None = None,
+    ) -> McpServer:
+        """Retrieve an MCP server by its display name.
+
+        ``legacy_slug`` lets already-deployed slug-keyed bindings apply their
+        overwrite while packages transition to name-keyed bindings.
+        """
+        spec = self._retrieve_spec(identifier=name, folder_path=folder_path)
+        response = await self.request_async(
+            spec.method,
+            url=spec.endpoint,
+            params=spec.params,
+            headers=spec.headers,
+        )
+        return McpServer.model_validate(response.json())
+
     @property
     def custom_headers(self) -> dict[str, str]:
         return self.folder_headers
@@ -223,14 +278,14 @@ def _list_spec(
 
     def _retrieve_spec(
         self,
-        slug: str,
+        identifier: str,
         *,
         folder_path: str | None,
     ) -> RequestSpec:
         folder_key = self._resolve_folder_key(folder_path)
         return RequestSpec(
             method="GET",
-            endpoint=Endpoint(f"/agenthub_/api/servers/{slug}"),
+            endpoint=Endpoint(f"/agenthub_/api/servers/{quote(identifier, safe='')}"),
             headers={
                 **header_folder(folder_key, None),
             },

packages/uipath-platform/tests/services/test_mcp_service.py

@@ -569,3 +569,9 @@ def test_validate_swagger_server(self) -> None:
         )
         assert server.type == 7
         assert server.slug == "contoso-directory"
+
+
+def test_mcp_retrieve_spec_encodes_display_name(service: McpService) -> None:
+    spec = service._retrieve_spec(identifier="Friendly MCP/Europe", folder_path=None)
+
+    assert "api/servers/Friendly%20MCP%2FEurope" in str(spec.endpoint)

packages/uipath-platform/tests/services/test_remote_a2a_service.py

@@ -36,7 +36,7 @@ def test_falls_back_to_folder_context_when_folder_path_missing(
         self, service: RemoteA2aService
     ) -> None:
         """No folder_path (e.g. local debug) must not raise; it falls back to context."""
-        spec = service._retrieve_spec(slug="weather", folder_path=None)
+        spec = service._retrieve_spec(identifier="weather", folder_path=None)
 
         assert "remote-a2a-agents/weather" in str(spec.endpoint)
         assert spec.headers[HEADER_FOLDER_KEY] == "context-folder-key"
@@ -50,6 +50,15 @@ def test_resolves_explicit_folder_path(
             lambda folder_path: "resolved-folder-key",
         )
 
-        spec = service._retrieve_spec(slug="weather", folder_path="MyFolder")
+        spec = service._retrieve_spec(identifier="weather", folder_path="MyFolder")
 
         assert spec.headers[HEADER_FOLDER_KEY] == "resolved-folder-key"
+
+    def test_encodes_display_name_in_lookup_path(
+        self, service: RemoteA2aService
+    ) -> None:
+        spec = service._retrieve_spec(
+            identifier="Friendly Agent/Europe", folder_path=None
+        )
+
+        assert "remote-a2a-agents/Friendly%20Agent%2FEurope" in str(spec.endpoint)

packages/uipath-platform/uv.lock

@@ -328,6 +328,45 @@ def some_method(a: str, b: str, **kwargs):
         assert len(result) == 1
         assert result["kwargs_param"] == 123
 
+    @pytest.mark.anyio
+    async def test_overrides_decorator_supports_legacy_binding_key(self):
+        from uipath.platform.common import (
+            GenericResourceOverwrite,
+            ResourceOverwritesContext,
+            resource_override,
+        )
+
+        @resource_override(
+            resource_type="mcpServer",
+            resource_identifier="name",
+            legacy_resource_identifier="legacy_slug",
+        )
+        async def retrieve_by_name(
+            name: str,
+            *,
+            legacy_slug: str,
+            folder_path: str,
+        ):
+            return name, folder_path
+
+        async def get_overwrites():
+            return {
+                "mcpServer.legacy-slug": GenericResourceOverwrite(
+                    resource_type="mcpServer",
+                    name="Replacement MCP",
+                    folder_path="Replacement/Folder",
+                )
+            }
+
+        async with ResourceOverwritesContext(get_overwrites):
+            result = await retrieve_by_name(
+                "Friendly MCP",
+                legacy_slug="legacy-slug",
+                folder_path="Original/Folder",
+            )
+
+        assert result == ("Replacement MCP", "Replacement/Folder")
+
     def test_debug_with_resource_overwrites(
         self,
         runner: CliRunner,