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/common/test_bindings.py

@@ -0,0 +1,68 @@
+import pytest
+
+from uipath.platform.common._bindings import (
+    GenericResourceOverwrite,
+    ResourceOverwrite,
+    _resource_overwrites,
+    resource_override,
+)
+
+
+@resource_override(
+    resource_type="mcpServer",
+    resource_identifier="name",
+    legacy_resource_identifier="legacy_slug",
+)
+def _retrieve_by_name(
+    name: str,
+    *,
+    legacy_slug: str | None,
+    folder_path: str,
+) -> tuple[str, str]:
+    return name, folder_path
+
+
+@pytest.mark.parametrize(
+    ("overwrites", "legacy_slug", "expected"),
+    [
+        (
+            {
+                "mcpServer.Friendly MCP.Original/Folder": GenericResourceOverwrite(
+                    resource_type="mcpServer",
+                    name="Folder replacement",
+                    folder_path="Replacement/Folder",
+                )
+            },
+            "legacy-slug",
+            ("Folder replacement", "Replacement/Folder"),
+        ),
+        (
+            {
+                "mcpServer.legacy-slug": GenericResourceOverwrite(
+                    resource_type="mcpServer",
+                    name="Legacy replacement",
+                    folder_path="Legacy/Folder",
+                )
+            },
+            "legacy-slug",
+            ("Legacy replacement", "Legacy/Folder"),
+        ),
+        ({}, None, ("Friendly MCP", "Original/Folder")),
+    ],
+)
+def test_resource_override_supports_primary_and_legacy_binding_keys(
+    overwrites: dict[str, ResourceOverwrite],
+    legacy_slug: str | None,
+    expected: tuple[str, str],
+) -> None:
+    token = _resource_overwrites.set(overwrites)
+    try:
+        result = _retrieve_by_name(
+            "Friendly MCP",
+            legacy_slug=legacy_slug,
+            folder_path="Original/Folder",
+        )
+    finally:
+        _resource_overwrites.reset(token)
+
+    assert result == expected

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

@@ -1,4 +1,4 @@
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 from pytest_httpx import HTTPXMock
@@ -363,6 +363,41 @@ async def test_retrieve_server_async(
                 == f"UiPath.Python.Sdk/UiPath.Python.Sdk.Activities.McpService.retrieve_async/{version}"
             )
 
+        def test_retrieve_server_by_name(self, service: McpService) -> None:
+            response = Mock()
+            response.json.return_value = {
+                "name": "Friendly MCP/Europe",
+                "slug": "friendly-mcp-europe",
+            }
+
+            with patch.object(service, "request", return_value=response) as request:
+                server = service.retrieve_by_name("Friendly MCP/Europe")
+
+            assert server.name == "Friendly MCP/Europe"
+            assert "api/servers/Friendly%20MCP%2FEurope" in str(
+                request.call_args.kwargs["url"]
+            )
+
+        @pytest.mark.anyio
+        async def test_retrieve_server_by_name_async(self, service: McpService) -> None:
+            response = Mock()
+            response.json.return_value = {
+                "name": "Friendly MCP/Europe",
+                "slug": "friendly-mcp-europe",
+            }
+
+            with patch.object(
+                service,
+                "request_async",
+                new=AsyncMock(return_value=response),
+            ) as request:
+                server = await service.retrieve_by_name_async("Friendly MCP/Europe")
+
+            assert server.name == "Friendly MCP/Europe"
+            assert "api/servers/Friendly%20MCP%2FEurope" in str(
+                request.call_args.kwargs["url"]
+            )
+
     class TestRequestKwargs:
         """Test that all methods pass the correct kwargs to request/request_async."""
 
@@ -569,3 +604,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)