feat(deployments): verify wxO credentials against instance API

- Probe GET /v1/orchestrate/models after IAM token to validate URL+key
- Add POST /deployments/providers/verify-credentials for connection tests
- Extend unit tests for models probe and WxO client stub

src/backend/base/langflow/api/v1/schemas/deployments.py

@@ -39,9 +39,6 @@ from pydantic import AfterValidator, BaseModel, Field, ValidationInfo, model_val
 from langflow.services.database.models.deployment_provider_account.schemas import (
     DeploymentProviderKey,
 )
-from langflow.services.database.models.deployment_provider_account.utils import (
-    validate_provider_url,
-)
 
 # ---------------------------------------------------------------------------
 # Shared validation helpers
@@ -82,10 +79,6 @@ NonEmptyStr = Annotated[str, AfterValidator(_strip_nonempty)]
 """String type that strips whitespace and rejects empty/whitespace-only values."""
 
 
-ValidatedUrl = Annotated[str, AfterValidator(validate_provider_url)]
-"""URL type that enforces HTTPS and normalizes."""
-
-
 def _validate_flow_version_ids(values: list[UUID] | None) -> list[UUID] | None:
     """AfterValidator for optional flow_version_ids query parameter."""
     if values is None:

src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/models.py

@@ -0,0 +1,13 @@
+"""Model-catalog retrieval helpers for the wxO deployment adapter."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient
+
+
+def fetch_models_adapter(clients: WxOClient, params: dict[str, Any] | None = None) -> Any:
+    """Fetch raw provider models through the adapter client seam."""
+    return clients.get_models_raw(params=params)

src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/service.py

@@ -77,6 +77,9 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution im
     create_agent_run,
     get_agent_run,
 )
+from langflow.services.adapters.deployment.watsonx_orchestrate.core.models import (
+    fetch_models_adapter,
+)
 from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import (
     retry_create,
     rollback_created_resources,
@@ -107,6 +110,7 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import (
     WatsonxDeploymentUpdateResultData,
     WatsonxModelOut,
 )
+from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient
 from langflow.services.adapters.deployment.watsonx_orchestrate.utils import (
     dedupe_list,
     extract_agent_tool_ids,
@@ -125,8 +129,6 @@ if TYPE_CHECKING:
     from lfx.services.settings.service import SettingsService
     from sqlalchemy.ext.asyncio import AsyncSession
 
-    from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient
-
 
 class WatsonxOrchestrateDeploymentService(BaseDeploymentService):
     """Deployment adapter for Watsonx Orchestrate."""
@@ -307,8 +309,12 @@ class WatsonxOrchestrateDeploymentService(BaseDeploymentService):
         hardcoded_names = {m.model_name for m in raw_models}
 
         try:
-            api_models = await asyncio.to_thread(client_manager.get_models_raw)
-            raw_models.extend(m for m in api_models if m.model_name not in hardcoded_names)
+            api_models = await asyncio.to_thread(fetch_models_adapter, client_manager)
+            for model in api_models:
+                model_name = model.get("model_name") if isinstance(model, dict) else getattr(model, "model_name", None)
+                if model_name and model_name in hardcoded_names:
+                    continue
+                raw_models.append(model)
             parsed_models: WatsonxDeploymentLlmListResultData = self._parse_provider_payload(
                 slot=self.payload_schemas.deployment_llm_list_result,
                 slot_name="deployment_llm_list_result",
@@ -870,7 +876,13 @@ class WatsonxOrchestrateDeploymentService(BaseDeploymentService):
         user_id: IdLike,  # noqa: ARG002
         payload: VerifyCredentials,
     ) -> VerifyCredentialsResult:
-        """Verify WXO credentials by obtaining a token from the provider."""
+        """Verify WXO credentials for the target instance.
+
+        Obtains an IAM/MCSP token, then calls the wxO models listing API for the
+        configured instance URL. Token-only checks are insufficient because a
+        valid API key may authenticate while still lacking access to the tenant
+        represented by the instance URL.
+        """
         verify_slot = self.payload_schemas.verify_credentials
         if verify_slot is None:
             msg = "Required slot 'verify_credentials' is not configured."
@@ -923,6 +935,31 @@ class WatsonxOrchestrateDeploymentService(BaseDeploymentService):
                 cause=exc,
             ) from exc
 
+        def _probe_instance_models() -> None:
+            wxo_client = WxOClient(instance_url=payload.base_url, authenticator=authenticator)
+            fetch_models_adapter(wxo_client)
+
+        try:
+            await asyncio.to_thread(_probe_instance_models)
+        except ClientAPIException as exc:
+            status_code = exc.response.status_code if exc.response is not None else None
+            logger.error(  # noqa: TRY400
+                "Credential verification failed: wxO instance probe rejected request (status=%s)",
+                status_code,
+            )
+            raise_deployment_error_from_status(
+                status_code=status_code,
+                detail="Credential verification failed.",
+                message_prefix="Credential verification",
+                cause=None,
+            )
+        except Exception as exc:
+            raise DeploymentError(
+                message="Credential verification failed unexpectedly.",
+                error_code="deployment_error",
+                cause=exc,
+            ) from exc
+
         return VerifyCredentialsResult()
 
     async def update_snapshot(

src/backend/tests/unit/services/deployment/test_watsonx_orchestrate.py

@@ -7013,6 +7013,7 @@ async def test_verify_credentials_success(monkeypatch):
         "get_authenticator",
         lambda **_kwargs: FakeAuthenticator(),
     )
+    monkeypatch.setattr(service_module, "fetch_models_adapter", lambda *_args, **_kwargs: {})
 
     svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService())
     payload = VerifyCredentials(
@@ -7052,6 +7053,43 @@ async def test_verify_credentials_invalid_key_raises(monkeypatch):
         await svc.verify_credentials(user_id="u1", payload=payload)
 
 
+@pytest.mark.anyio
+async def test_verify_credentials_instance_probe_forbidden(monkeypatch):
+    """403 from wxO models probe maps to AuthorizationError (wrong instance for key)."""
+    from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException
+    from lfx.services.adapters.deployment.exceptions import AuthorizationError
+    from lfx.services.adapters.deployment.schema import VerifyCredentials
+    from requests import Response
+
+    class FakeTokenManager:
+        def get_token(self):
+            return "fake-token"
+
+    class FakeAuthenticator:
+        token_manager = FakeTokenManager()
+
+    monkeypatch.setattr(
+        service_module,
+        "get_authenticator",
+        lambda **_kwargs: FakeAuthenticator(),
+    )
+
+    def _fail_fetch_models(*_args, **_kwargs):
+        response = Response()
+        response.status_code = 403
+        raise ClientAPIException(response=response)
+
+    monkeypatch.setattr(service_module, "fetch_models_adapter", _fail_fetch_models)
+
+    svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService())
+    payload = VerifyCredentials(
+        base_url="https://api.us-south.wxo.cloud.ibm.com",
+        provider_data={"api_key": "valid-key"},  # pragma: allowlist secret
+    )
+    with pytest.raises(AuthorizationError, match="Credential verification"):
+        await svc.verify_credentials(user_id="u1", payload=payload)
+
+
 @pytest.mark.anyio
 async def test_verify_credentials_malformed_key_from_authenticator_constructor_raises():
     """verify_credentials raises InvalidContentError when authenticator creation fails validation."""