feat: add per-service URL overrides for local development

src/uipath/_utils/__init__.py

@@ -3,6 +3,7 @@
 from ._logs import setup_logging
 from ._request_override import header_folder
 from ._request_spec import RequestSpec
+from ._service_url_overrides import resolve_endpoint_override
 from ._url import UiPathUrl
 from ._user_agent import header_user_agent, user_agent_value
 from .validation import validate_pagination_params
@@ -12,6 +13,7 @@
     "setup_logging",
     "RequestSpec",
     "header_folder",
+    "resolve_endpoint_override",
     "resource_override",
     "header_user_agent",
     "user_agent_value",

src/uipath/_utils/_service_url_overrides.py

@@ -0,0 +1,101 @@
+"""Service URL override resolution for local development.
+
+Allows routing specific service requests to alternative URLs (e.g., localhost)
+via environment variables like ``UIPATH_ORCHESTRATOR_URL``.
+"""
+
+import os
+from functools import lru_cache
+
+# Maps environment variable suffix to the endpoint prefix used in request paths.
+# E.g. UIPATH_ORCHESTRATOR_URL -> orchestrator_/
+_SERVICE_PREFIX_MAP: dict[str, str] = {
+    "ORCHESTRATOR": "orchestrator_",
+    "AGENTHUB": "agenthub_",
+    "DU": "du_",
+    "ECS": "ecs_",
+    "CONNECTIONS": "connections_",
+    "DATAFABRIC": "datafabric_",
+    "RESOURCECATALOG": "resourcecatalog_",
+    "AGENTSRUNTIME": "agentsruntime_",
+    "APPS": "apps_",
+    "IDENTITY": "identity_",
+    "STUDIO": "studio_",
+}
+
+
+@lru_cache(maxsize=1)
+def _load_service_overrides() -> dict[str, str]:
+    """Scan environment for ``UIPATH_{SERVICE}_URL`` variables.
+
+    Returns:
+        Mapping of endpoint prefix (e.g. ``orchestrator_``) to override URL.
+    """
+    overrides: dict[str, str] = {}
+    for suffix, prefix in _SERVICE_PREFIX_MAP.items():
+        value = os.environ.get(f"UIPATH_{suffix}_URL")
+        if value:
+            overrides[prefix] = value.rstrip("/")
+    return overrides
+
+
+def get_service_override(endpoint_prefix: str) -> str | None:
+    """Look up a URL override for the given endpoint prefix.
+
+    Args:
+        endpoint_prefix: The service endpoint prefix (e.g. ``orchestrator_``).
+            Lookup is case-insensitive.
+
+    Returns:
+        The override URL if configured, otherwise ``None``.
+    """
+    overrides = _load_service_overrides()
+    return overrides.get(endpoint_prefix.lower())
+
+
+def resolve_endpoint_override(
+    endpoint: str,
+) -> tuple[str | None, dict[str, str]]:
+    """Resolve a full endpoint path against service URL overrides.
+
+    Given a path like ``agenthub_/llm/raw/vendor/openai/model/gpt-4o/completions``,
+    checks if the service prefix (``agenthub_``) has an override configured.
+    When an override is active, also returns routing headers that would normally
+    be injected by the platform routing layer.
+
+    Args:
+        endpoint: Full endpoint path starting with a service prefix
+            (e.g. ``agenthub_/llm/raw/...``).
+
+    Returns:
+        A tuple of (resolved_url, headers). If no override is configured,
+        resolved_url is ``None`` and headers is empty.
+    """
+    stripped = endpoint.strip("/")
+    if not stripped:
+        return None, {}
+
+    first_segment, _, rest = stripped.partition("/")
+    if not first_segment.endswith("_"):
+        return None, {}
+
+    override_base = get_service_override(first_segment)
+    if override_base is None:
+        return None, {}
+
+    resolved_url = f"{override_base}/{rest}" if rest else override_base
+
+    headers: dict[str, str] = {}
+    tenant_id = os.environ.get("UIPATH_TENANT_ID")
+    organization_id = os.environ.get("UIPATH_ORGANIZATION_ID")
+    if tenant_id:
+        headers["X-UiPath-Internal-TenantId"] = tenant_id
+    if organization_id:
+        headers["X-UiPath-Internal-AccountId"] = organization_id
+
+    return resolved_url, headers
+
+
+def clear_overrides_cache() -> None:
+    """Clear the cached overrides. Intended for tests."""
+    _load_service_overrides.cache_clear()

src/uipath/_utils/_url.py

@@ -1,6 +1,8 @@
 from typing import Literal
 from urllib.parse import urlparse
 
+from ._service_url_overrides import get_service_override
+
 
 class UiPathUrl:
     """A class that represents a UiPath URL.
@@ -61,13 +63,41 @@ def scope_url(self, url: str, scoped: Literal["org", "tenant"] = "tenant") -> st
         if not self._is_relative_url(url):
             return url
 
+        # Check for service-specific URL override
+        override_url = self._resolve_service_override(url)
+        if override_url is not None:
+            return override_url
+
         parts = [self.org_name]
         if scoped == "tenant":
             parts.append(self.tenant_name)
         parts.append(url.strip("/"))
 
         return "/".join(parts)
 
+    def _resolve_service_override(self, url: str) -> str | None:
+        """Return an absolute URL if a service override is configured for *url*.
+
+        Extracts the first path segment.  If it ends with ``_`` (the UiPath
+        service-prefix convention), look up an override.  When found, strip the
+        prefix and return an absolute URL pointing at the override host.
+        """
+        stripped = url.strip("/")
+        if not stripped:
+            return None
+
+        first_segment, _, rest = stripped.partition("/")
+        if not first_segment.endswith("_"):
+            return None
+
+        override_base = get_service_override(first_segment)
+        if override_base is None:
+            return None
+
+        if rest:
+            return f"{override_base}/{rest}"
+        return override_base
+
     @property
     def _org_tenant_names(self):
         parsed = urlparse(self._url)

src/uipath/platform/common/_base_service.py

@@ -23,7 +23,7 @@
 from ..._utils._ssl_context import get_httpx_client_kwargs
 from ..._utils.constants import HEADER_USER_AGENT
 from ..errors import EnrichedException
-from ._config import UiPathApiConfig
+from ._config import UiPathApiConfig, UiPathConfig
 from ._execution_context import UiPathExecutionContext
 
 
@@ -103,6 +103,9 @@ def request(
         kwargs["headers"][HEADER_USER_AGENT] = user_agent_value(specific_component)
 
         scoped_url = self._url.scope_url(str(url), scoped)
+        if scoped_url.startswith(("http://", "https://")):
+            self._logger.debug(f"Service URL override active: {scoped_url}")
+            self._inject_routing_headers(kwargs["headers"])
 
         response = self._client.request(method, scoped_url, **kwargs)
 
@@ -140,6 +143,9 @@ async def request_async(
         )
 
         scoped_url = self._url.scope_url(str(url), scoped)
+        if scoped_url.startswith(("http://", "https://")):
+            self._logger.debug(f"Service URL override active: {scoped_url}")
+            self._inject_routing_headers(kwargs["headers"])
 
         response = await self._client_async.request(method, scoped_url, **kwargs)
 
@@ -150,6 +156,17 @@ async def request_async(
             raise EnrichedException(e) from e
         return response
 
+    def _inject_routing_headers(self, headers: dict[str, str]) -> None:
+        """Add routing headers that would normally be set by the routing layer.
+
+        When using a local service URL override, the platform routing layer
+        is bypassed, so tenant and account identifiers must be sent explicitly.
+        """
+        if UiPathConfig.tenant_id:
+            headers["X-UiPath-Internal-TenantId"] = UiPathConfig.tenant_id
+        if UiPathConfig.organization_id:
+            headers["X-UiPath-Internal-AccountId"] = UiPathConfig.organization_id
+
     @property
     def default_headers(self) -> dict[str, str]:
         return {

src/uipath/platform/common/_config.py

@@ -54,6 +54,12 @@ def tenant_name(self) -> str | None:
 
         return os.getenv(ENV_TENANT_NAME, None)
 
+    @property
+    def tenant_id(self) -> str | None:
+        from uipath._utils.constants import ENV_TENANT_ID
+
+        return os.getenv(ENV_TENANT_ID, None)
+
     @property
     def organization_id(self) -> str | None:
         from uipath._utils.constants import ENV_ORGANIZATION_ID

tests/sdk/test_service_url_overrides.py

@@ -0,0 +1,57 @@
+"""Tests for service URL override resolution module."""
+
+from typing import TYPE_CHECKING
+
+from uipath._utils._service_url_overrides import (
+    clear_overrides_cache,
+    get_service_override,
+)
+
+if TYPE_CHECKING:
+    from _pytest.monkeypatch import MonkeyPatch
+
+
+class TestGetServiceOverride:
+    """Tests for get_service_override()."""
+
+    def setup_method(self) -> None:
+        clear_overrides_cache()
+
+    def teardown_method(self) -> None:
+        clear_overrides_cache()
+
+    def test_no_override_returns_none(self, monkeypatch: "MonkeyPatch") -> None:
+        """Returns None when no env var is set."""
+        monkeypatch.delenv("UIPATH_ORCHESTRATOR_URL", raising=False)
+        assert get_service_override("orchestrator_") is None
+
+    def test_orchestrator_override(self, monkeypatch: "MonkeyPatch") -> None:
+        """Returns the override URL for orchestrator."""
+        monkeypatch.setenv("UIPATH_ORCHESTRATOR_URL", "http://localhost:8080")
+        assert get_service_override("orchestrator_") == "http://localhost:8080"
+
+    def test_trailing_slash_stripped(self, monkeypatch: "MonkeyPatch") -> None:
+        """Trailing slash on the env var value is stripped."""
+        monkeypatch.setenv("UIPATH_ORCHESTRATOR_URL", "http://localhost:8080/")
+        assert get_service_override("orchestrator_") == "http://localhost:8080"
+
+    def test_case_insensitive_lookup(self, monkeypatch: "MonkeyPatch") -> None:
+        """Prefix lookup is case-insensitive."""
+        monkeypatch.setenv("UIPATH_AGENTHUB_URL", "http://localhost:9090")
+        assert get_service_override("agenthub_") == "http://localhost:9090"
+        assert get_service_override("AGENTHUB_") == "http://localhost:9090"
+
+    def test_multiple_overrides_independent(self, monkeypatch: "MonkeyPatch") -> None:
+        """Multiple services can be overridden independently."""
+        monkeypatch.setenv("UIPATH_ORCHESTRATOR_URL", "http://localhost:8080")
+        monkeypatch.setenv("UIPATH_DU_URL", "http://localhost:9090")
+        assert get_service_override("orchestrator_") == "http://localhost:8080"
+        assert get_service_override("du_") == "http://localhost:9090"
+        assert get_service_override("agenthub_") is None
+
+    def test_unrecognised_service_returns_none(
+        self, monkeypatch: "MonkeyPatch"
+    ) -> None:
+        """A prefix not in the service map returns None."""
+        monkeypatch.setenv("UIPATH_UNKNOWN_URL", "http://localhost:1111")
+        assert get_service_override("unknown_") is None

tests/sdk/test_url_service_overrides.py

@@ -0,0 +1,80 @@
+"""Integration tests for UiPathUrl.scope_url() with service URL overrides."""
+
+from typing import TYPE_CHECKING
+
+from uipath._utils._service_url_overrides import clear_overrides_cache
+from uipath._utils._url import UiPathUrl
+
+if TYPE_CHECKING:
+    from _pytest.monkeypatch import MonkeyPatch
+
+
+class TestScopeUrlWithOverrides:
+    """Tests for scope_url() interacting with service URL overrides."""
+
+    def setup_method(self) -> None:
+        clear_overrides_cache()
+
+    def teardown_method(self) -> None:
+        clear_overrides_cache()
+
+    def _make_url(self) -> UiPathUrl:
+        return UiPathUrl("https://cloud.uipath.com/myorg/mytenant")
+
+    def test_no_override_unchanged(self) -> None:
+        """Standard scoping is preserved when no override is set."""
+        url = self._make_url()
+        result = url.scope_url("/orchestrator_/odata/Jobs")
+        assert result == "myorg/mytenant/orchestrator_/odata/Jobs"
+
+    def test_localhost_override_strips_prefix(self, monkeypatch: "MonkeyPatch") -> None:
+        """Service prefix is stripped and request routes to localhost."""
+        monkeypatch.setenv("UIPATH_ORCHESTRATOR_URL", "http://localhost:8080")
+        url = self._make_url()
+        result = url.scope_url("/orchestrator_/odata/Jobs")
+        assert result == "http://localhost:8080/odata/Jobs"
+
+    def test_remote_override_strips_prefix(self, monkeypatch: "MonkeyPatch") -> None:
+        """Prefix stripping works for non-localhost override URLs too."""
+        monkeypatch.setenv("UIPATH_AGENTHUB_URL", "https://dev.internal.example.com")
+        url = self._make_url()
+        result = url.scope_url("/agenthub_/api/v1/agents")
+        assert result == "https://dev.internal.example.com/api/v1/agents"
+
+    def test_absolute_url_not_intercepted(self, monkeypatch: "MonkeyPatch") -> None:
+        """Already-absolute URLs pass through without override check."""
+        monkeypatch.setenv("UIPATH_ORCHESTRATOR_URL", "http://localhost:8080")
+        url = self._make_url()
+        result = url.scope_url("https://other.host/orchestrator_/odata/Jobs")
+        assert result == "https://other.host/orchestrator_/odata/Jobs"
+
+    def test_non_service_url_unchanged(self, monkeypatch: "MonkeyPatch") -> None:
+        """Paths without a service prefix (trailing _) are not intercepted."""
+        monkeypatch.setenv("UIPATH_ORCHESTRATOR_URL", "http://localhost:8080")
+        url = self._make_url()
+        # "api" doesn't end with _ so it should follow normal scoping
+        result = url.scope_url("/api/v1/status")
+        assert result == "myorg/mytenant/api/v1/status"
+
+    def test_multiple_services_independent(self, monkeypatch: "MonkeyPatch") -> None:
+        """Only configured services are overridden; others scope normally."""
+        monkeypatch.setenv("UIPATH_ORCHESTRATOR_URL", "http://localhost:8080")
+        url = self._make_url()
+
+        # Overridden service
+        assert (
+            url.scope_url("/orchestrator_/odata/Jobs")
+            == "http://localhost:8080/odata/Jobs"
+        )
+        # Non-overridden service follows normal scoping
+        assert (
+            url.scope_url("/agenthub_/api/v1/agents")
+            == "myorg/mytenant/agenthub_/api/v1/agents"
+        )
+
+    def test_override_prefix_only_path(self, monkeypatch: "MonkeyPatch") -> None:
+        """URL consisting of only the service prefix returns just the base."""
+        monkeypatch.setenv("UIPATH_ORCHESTRATOR_URL", "http://localhost:8080")
+        url = self._make_url()
+        result = url.scope_url("/orchestrator_/")
+        assert result == "http://localhost:8080"