feat: add wxO deployment adapter (#12079)

* checkout wxo deployment adapter implementation

* checkout wxo deps and flow reqs impl

* clean up / minor refactor with updated tests

* major refactor: split up the implementation into folders and files

* clean up logic

* refactor and clean up of execution logic, improve type safety of "retry_create" helper, and remove dead code

* remove unused wxo service file

* remove "provider_name" arg and references

* fix: harden watsonx orchestrate deployment adapter for PR readiness

- Fix rollback AttributeError when agent_create_response is None
- Fix NoneType access on params in list() when called without params
- Fix inconsistent error types: use DeploymentNotFoundError in get/update
- Fix typo "occured" -> "occurred" in all error prefix messages
- Fix variable shadowing of fastapi.status in get_status()
- Fix pre-existing test bugs (wrong exception types, stale method refs)
- Fix e2e monkey-patching of non-existent service methods
- Add structured logging to create, delete, retry, and rollback flows
- Add jitter to retry backoff to avoid thundering herd
- Add __repr__ to WxOCredentials that fully masks the API key
- Extract hard-coded LLM model string to DEFAULT_WXO_AGENT_LLM constant
- Remove commented-out snapshot update code
- Expand test coverage from 24 to 75 tests covering retry logic,
  service methods, client auth, utilities, execution/status helpers,
  and artifact roundtrip

* fix(deployment): fix bugs and harden watsonx orchestrate adapter

- Fix update method silently discarding changes instead of sending them
  to the WXO API
- Wrap all synchronous SDK calls in asyncio.to_thread to avoid blocking
  the event loop
- Add error handling to get_status (was propagating raw exceptions)
- Use DeploymentType enum values instead of raw strings for
  SUPPORTED_ADAPTER_DEPLOYMENT_TYPES
- Fix type annotations in list method (list -> set)
- Fix typo in comment (reccomend -> recommend)
- Remove dead code: extract_agent_connection_ids,
  require_exclusive_resource, require_non_empty_string,
  resolve_health_environment_id, fetch_agent_release_status,
  normalize_release_status, resolve_lfx_runner_requirement,
  _pin_requirement_name, sync_langflow_tool_connections,
  create_langflow_flow_tool, resolve_snapshot_connections, and unused constants
- Make assert_create_resources_available and validate_connection async
- Make create_agent_run and get_agent_run async
- Add tests for list_types, get_status error handling, and update
  side-effects
- Update existing tests for async function signatures

* actually remove the dead code from tools.py

* properly await "validate_connection"

* update e2e test file

* add more scenarios to e2e test runner

* refactor(watsonx-orchestrate): improve WxOClient encapsulation and error messages
Encapsulate SDK private method access and endpoint paths inside
WxOClient wrappers so callers never handle raw paths or touch internal
_get/_post methods. Route all private HTTP calls through a single
_base client auto-created in __post_init__, removing the externally
constructed `base` field.
Additionally fix double-period and redundant text in error messages
produced by the ErrorPrefix + handler pattern, and update E2E/unit
assertions to match the sanitised error output.
Changes:
- types.py: bake endpoint paths into wrapper signatures (post_run,
  get_run, upload_tool_artifact, get_agents_raw); auto-create _base
  via __post_init__; remove base from constructor
- client.py: drop BaseWXOClient import and base= constructor arg
- core/execution.py: pass run_id / query_suffix instead of raw paths
- core/tools.py: pass tool_id instead of raw path
- service.py: fix double-period in ErrorPrefix interpolation; remove
  redundant restatements in generic except handlers
- tests: update _with_wxo_wrappers helper and test doubles to use
  _base; update FakeBaseClient._get to accept params
- e2e: update rollback scenario detail_contains to match sanitised msg

* skip import and tests of wxo adapter if current env is running python 3.10

* clarify random prefix / retry  behavior

* implement comments

* fix: align watsonx adapter with updated deployment schema
- Add `deployment_type` parameter to `get`, `update`, `redeploy`,
  `duplicate`, `delete`, `get_status`, and `undeploy_deployment` in
  WatsonxOrchestrateDeploymentService to match the updated
  BaseDeploymentService ABC
- Update e2e script to use renamed `add_ids` field on
  SnapshotDeploymentBindingUpdate

* remove non-interface method undeploy_deployment

* implement listing configs and snapshots

* add update implementation and improve http->deployment error translation and add tests

* improve exception handling

* checkout payload slot work

* custom payload schema for update

* new update implementation

* stop passing client cache to helpers

* add docs for ordereduniquests

* improve import patterns and document future risks for wxo dependencies

* remove global-variable prefixing of flows

* ref: harden typing, DRY helpers, and correctness fixes
- Replace `db: Any` with `db: AsyncSession` across client, config, and
  update_helpers modules
- Extract `_ensure_dict` / `ensure_langflow_connections_binding` helpers
  to eliminate repeated nested-dict safety logic in tools.py and
  update_helpers.py, with documentation explaining why malformed API
  payloads are silently corrected rather than rejected
- Use `PayloadSlot.parse()` for update payload validation instead of
  standalone helper; remove `parse_provider_update_payload`
- Fix lambda late-binding with default-argument captures in service.py
- Use `zip(strict=True)` in update_helpers for defensive mismatch
  detection
- Replace O(n²) duplicate detection with `collections.Counter` in
  payloads.py
- Simplify `dedupe_list` to `list(dict.fromkeys(...))`
- Move type-annotation-only imports into TYPE_CHECKING blocks in
  types.py and config.py
- Introduce `UPDATE_MAX_RETRIES` as a distinct constant from
  `CREATE_MAX_RETRIES`
- Fix "watsonX" → "watsonx" casing in error messages
- Clarify `get_status` docstring and `validate_connection` error message
- Add TODO(deployments-cache) for client cache invalidation on
  credential updates

* ref: drop client cache and adopt typed deployment context across the wxo adapter path, add request-context memoization with strict mixed-context guards, and lazily initialize wxo SDK clients. Add safer credential handling by storing only authenticators in WxOCredentials. Also improve provider error-detail extraction/messages and make the direct wxo E2E conflict scenario diagnostics and expectations more resilient.

* use explicit naming for context class; providerId

* fix error handling, retry safety, and exception diagnostics in wxo adapter

- Raise DeploymentError on empty API responses instead of fabricating
  fake success in create_agent_run_result and get_agent_run
- Elevate rollback failure logging from WARNING to ERROR for alerting
- Apply retryable filter to retry_rollback to skip 401/403/409/422
- Preserve exception chains (from exc) instead of suppressing (from None)
  across service.py, utils.py, execution.py, and update_helpers.py
- Broaden credential resolution catch to handle arbitrary DB exceptions
- Separate status code dispatch from string heuristics in
  raise_for_status_and_detail to prevent misclassification
- Add tests for all behavioral changes

* [autofix.ci] apply automated fixes

* fix: harden wxO adapter safety, immutability, and error diagnostics

- Make WxOClient and WxOCredentials frozen dataclasses with eager SDK
  client initialization to eliminate thread-safety races from
  asyncio.to_thread workers and prevent post-construction mutation
- Move instance_url validation and normalization into type __post_init__
- Raise DeploymentError on missing run_id instead of returning partial
  success with execution_id=None
- Preserve exception chains (from exc) instead of suppressing (from None)
  across create, update, and delete service methods
- Add warning logs when _ensure_dict replaces non-dict binding values
  and when _resolve_lfx_requirement falls back to minimum version
- Initialize derived_spec before try block to prevent potential NameError
- Log all ToolUploadBatchError errors before re-raising the first
- Change SUPPORTED_ADAPTER_DEPLOYMENT_TYPES from mutable set to frozenset
- Add tests for 409/422 error mapping in create, unsupported deployment
  type rejection, empty update rejection, zip artifact extraction paths,
  validate_connection negative paths, missing run_id, multiple deployment
  ID rejection, exception chain preservation, and _ensure_dict warning

* [autofix.ci] apply automated fixes

* fix ruff errors and and todo for status method

* fix mypy errors

* [autofix.ci] apply automated fixes

---------

Co-authored-by: Jordan Frazier <jordan.frazier@datastax.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Hamza Rashid
2026-03-17 12:51:44 -04:00
committed by GitHub
parent 13908b2d31
commit 43253011ec
28 changed files with 9829 additions and 6 deletions

File diff suppressed because it is too large Load Diff

View File

@ -59,6 +59,15 @@ DeploymentIdQuery = Annotated[
# - Body ``provider_id`` is included on ``DeploymentCreateRequest`` and ``ExecutionCreateRequest``
# to allow provider routing without an extra DB lookup when the caller already has the context.
# - Deployment-scoped routes derive provider context from persisted Langflow relationships.
#
# TODO(deployments-routing): Before replacing 501 stubs with live routing:
# - Resolve provider_key from provider_id/deployment_id and fetch adapter via registry.
# - If adapter is unavailable in current runtime (e.g. optional SDK not installed),
# return a deterministic domain error/HTTP status instead of 500.
# - Add tests for:
# * provider account exists but adapter key not registered
# * adapter import was skipped at startup due to ModuleNotFoundError
# * direct WXO provider routes in unsupported runtimes (py3.10 scenarios)
# ---------------------------------------------------------------------------

View File

@ -0,0 +1 @@
"""Adapter namespaces for Langflow service-scoped plugin registries."""

View File

@ -0,0 +1 @@
"""Langflow deployment adapter implementations."""

View File

@ -0,0 +1,43 @@
from __future__ import annotations
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass
from typing import TYPE_CHECKING, ClassVar
if TYPE_CHECKING:
from contextvars import Token
from uuid import UUID
@dataclass(frozen=True, slots=True)
class DeploymentAdapterContext:
provider_id: UUID
class DeploymentProviderIDContext:
_current: ClassVar[ContextVar[DeploymentAdapterContext | None]] = ContextVar(
"langflow_current_deployment_context",
default=None,
)
@classmethod
def get_current(cls) -> DeploymentAdapterContext | None:
return cls._current.get()
@classmethod
def set_current(cls, context: DeploymentAdapterContext) -> Token[DeploymentAdapterContext | None]:
return cls._current.set(context)
@classmethod
def reset_current(cls, token: Token[DeploymentAdapterContext | None]) -> None:
cls._current.reset(token)
@classmethod
@contextmanager
def scope(cls, context: DeploymentAdapterContext):
token: Token[DeploymentAdapterContext | None] = cls.set_current(context)
try:
yield
finally:
cls.reset_current(token)

View File

@ -0,0 +1,20 @@
"""Watsonx Orchestrate deployment adapter."""
from lfx.services.adapters.registry import register_adapter
from lfx.services.adapters.schema import AdapterType
from langflow.services.adapters.deployment.watsonx_orchestrate.constants import (
WATSONX_ORCHESTRATE_DEPLOYMENT_ADAPTER_KEY,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.service import WatsonxOrchestrateDeploymentService
from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOCredentials
register_adapter(
AdapterType.DEPLOYMENT,
WATSONX_ORCHESTRATE_DEPLOYMENT_ADAPTER_KEY,
)(WatsonxOrchestrateDeploymentService)
__all__ = [
"WatsonxOrchestrateDeploymentService",
"WxOCredentials",
]

View File

@ -0,0 +1,275 @@
"""Client creation, authentication, and credential resolution for the Watsonx Orchestrate adapter.
This module uses request/execution-context memoization for provider clients:
- `get_provider_clients()` resolves provider context and prebuilt credentials/authenticator.
- The resulting `WxOClient` is memoized in a ContextVar for the active async execution context.
- Subsequent calls with the same `(provider_id, user_id)` in that context reuse the same
`WxOClient` instance and skip repeated DB/decryption work.
Important behavior notes:
- Memoization is execution-context scoped (not cross-request/global state).
- The context stores a single `(key, client)` entry because deployment routing enforces one
provider context per request path.
- If a different `(provider_id, user_id)` is requested in the same context, resolution fails.
"""
from __future__ import annotations
from contextvars import ContextVar
from dataclasses import dataclass
from typing import TYPE_CHECKING, ClassVar
from ibm_cloud_sdk_core.authenticators import Authenticator, IAMAuthenticator, MCSPAuthenticator
from ibm_watsonx_orchestrate_core.types.connections import KeyValueConnectionCredentials
from lfx.services.adapters.deployment.exceptions import AuthSchemeError, CredentialResolutionError
from lfx.services.adapters.deployment.schema import EnvVarSource, EnvVarValueSpec, IdLike
from langflow.services.adapters.deployment.context import DeploymentProviderIDContext
from langflow.services.adapters.deployment.watsonx_orchestrate.constants import WxOAuthURL
from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient, WxOCredentials
from langflow.services.auth import utils as auth_utils
from langflow.services.database.models.deployment_provider_account.crud import get_provider_account_by_id
from langflow.services.deps import get_variable_service
if TYPE_CHECKING:
from contextvars import Token
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
@dataclass(frozen=True, slots=True)
class WxOProviderClientsContext:
provider_id: str
user_id: str
clients: WxOClient
class WxOProviderClientsRequestContext:
_current: ClassVar[ContextVar[WxOProviderClientsContext | None]] = ContextVar(
"langflow_wxo_provider_clients_request_context",
default=None,
)
@classmethod
def get_current(cls) -> WxOProviderClientsContext | None:
return cls._current.get()
@classmethod
def set_current(cls, context: WxOProviderClientsContext) -> Token[WxOProviderClientsContext | None]:
return cls._current.set(context)
@classmethod
def reset_current(cls, token: Token[WxOProviderClientsContext | None]) -> None:
cls._current.reset(token)
@classmethod
def clear_current(cls) -> None:
cls._current.set(None)
def _provider_client_context_key(*, provider_id: UUID, user_id: UUID | str) -> tuple[str, str]:
return (str(provider_id), str(user_id))
def clear_provider_clients_request_context() -> None:
"""Clear execution-context memoized provider clients for the current async context.
This is mainly useful in tests and explicit context lifecycle control.
"""
WxOProviderClientsRequestContext.clear_current()
def get_request_context_provider_clients(*, provider_id: UUID, user_id: UUID | str) -> WxOClient | None:
"""Return memoized provider clients for the active execution context, if present.
Returns `None` when:
- no provider clients have been memoized in this context yet, or
- the memoized entry belongs to a different `(provider_id, user_id)` pair.
"""
request_context = WxOProviderClientsRequestContext.get_current()
if request_context is None:
return None
if (request_context.provider_id, request_context.user_id) == _provider_client_context_key(
provider_id=provider_id,
user_id=user_id,
):
return request_context.clients
return None
def _validate_request_context_provider_key(*, provider_id: UUID, user_id: UUID | str) -> None:
request_context = WxOProviderClientsRequestContext.get_current()
if request_context is None:
return
if (request_context.provider_id, request_context.user_id) != _provider_client_context_key(
provider_id=provider_id,
user_id=user_id,
):
msg = (
"A different deployment provider context was requested in the same execution context. "
"This indicates an invalid mixed provider resolution flow."
)
raise CredentialResolutionError(message=msg)
def set_request_context_provider_clients(*, provider_id: UUID, user_id: UUID | str, clients: WxOClient) -> None:
"""Memoize provider clients for the active execution context."""
_validate_request_context_provider_key(provider_id=provider_id, user_id=user_id)
context = WxOProviderClientsContext(
provider_id=str(provider_id),
user_id=str(user_id),
clients=clients,
)
WxOProviderClientsRequestContext.set_current(context)
def get_authenticator(instance_url: str, api_key: str) -> Authenticator:
"""Return the appropriate authenticator for the Watsonx Orchestrate API."""
if ".cloud.ibm.com" in instance_url:
return IAMAuthenticator(apikey=api_key, url=WxOAuthURL.IBM_IAM.value)
elif ".ibm.com" in instance_url: # noqa: RET505 - explicitness
return MCSPAuthenticator(apikey=api_key, url=WxOAuthURL.MCSP.value)
msg = f"Could not determine authentication scheme for instance URL: {instance_url}"
raise AuthSchemeError(message=msg)
async def resolve_wxo_client_credentials(
*,
user_id: UUID | str,
db: AsyncSession,
provider_id: UUID,
) -> WxOCredentials:
"""Resolve Watsonx Orchestrate client credentials from deployment provider account.
The decrypted API key is used only to instantiate the SDK authenticator and is not
retained in adapter credential objects.
"""
try:
provider_account = await get_provider_account_by_id(
db,
provider_id=provider_id,
user_id=user_id,
)
if provider_account is None:
msg = "Failed to find deployment provider account credentials."
raise CredentialResolutionError(message=msg)
instance_url = (provider_account.backend_url or "").strip()
api_key = auth_utils.decrypt_api_key((provider_account.api_key or "").strip())
if not instance_url or not api_key:
msg = "Watsonx Orchestrate backend URL and API key must be configured."
raise CredentialResolutionError(message=msg)
except CredentialResolutionError:
raise
except Exception as exc:
msg = "An unexpected error occurred while resolving Watsonx Orchestrate client credentials."
raise CredentialResolutionError(message=msg) from exc
authenticator = get_authenticator(instance_url=instance_url, api_key=api_key)
return WxOCredentials(instance_url=instance_url, authenticator=authenticator)
async def get_provider_clients(
*,
user_id: UUID | str,
db: AsyncSession,
) -> WxOClient:
"""Resolve and return provider clients for the active deployment provider context.
Fast-path: return execution-context memoized clients when `(provider_id, user_id)` matches.
Slow-path: resolve credentials from DB, build authenticator, construct `WxOClient`, then memoize.
"""
request_context = DeploymentProviderIDContext.get_current()
if request_context is None:
msg = "Deployment account context is not available for adapter resolution."
raise CredentialResolutionError(message=msg)
provider_id = request_context.provider_id
_validate_request_context_provider_key(provider_id=provider_id, user_id=user_id)
if context_clients := get_request_context_provider_clients(provider_id=provider_id, user_id=user_id):
return context_clients
credentials: WxOCredentials = await resolve_wxo_client_credentials(
user_id=user_id,
db=db,
provider_id=provider_id,
)
clients = WxOClient(
instance_url=credentials.instance_url,
authenticator=credentials.authenticator,
)
set_request_context_provider_clients(provider_id=provider_id, user_id=user_id, clients=clients)
return clients
async def resolve_runtime_credentials(
*,
user_id: IdLike,
environment_variables: dict[str, EnvVarValueSpec],
db: AsyncSession,
) -> KeyValueConnectionCredentials:
"""Resolve runtime credentials from environment variables."""
resolved: dict[str, str] = {}
for credential_key, env_var_value in environment_variables.items():
resolved[credential_key] = await resolve_env_var_value(
env_var_value,
user_id=user_id,
db=db,
)
return KeyValueConnectionCredentials(resolved)
async def resolve_env_var_value(
env_var_value: EnvVarValueSpec,
*,
user_id: IdLike,
db: AsyncSession,
) -> str:
if env_var_value.source == EnvVarSource.RAW:
return env_var_value.value
return await resolve_variable_value(
env_var_value.value,
user_id=user_id,
db=db,
)
async def resolve_variable_value(
variable_name: str,
*,
user_id: UUID | str,
db: AsyncSession,
optional: bool = False,
default_value: str | None = None,
) -> str:
variable_service = get_variable_service()
if variable_service is None:
msg = "Variable service is not available."
raise CredentialResolutionError(message=msg)
try:
value = await variable_service.get_variable(
user_id=user_id,
name=variable_name,
field="value",
session=db,
)
if value is not None:
return value
except CredentialResolutionError:
raise
except Exception as exc:
if not optional:
msg = "Failed to resolve a credential variable for the watsonx Orchestrate deployment provider."
raise CredentialResolutionError(message=msg) from exc
if optional:
return default_value or ""
msg = (
"Failed to find a necessary credential for the "
"watsonx Orchestrate deployment provider. "
"Please ensure all credentials are provided and valid."
)
raise CredentialResolutionError(message=msg)

View File

@ -0,0 +1,50 @@
"""Constants, enums, and configuration values for the Watsonx Orchestrate adapter."""
from __future__ import annotations
import re
from enum import Enum
from lfx.services.adapters.deployment.schema import DeploymentType
SUPPORTED_ADAPTER_DEPLOYMENT_TYPES: frozenset[DeploymentType] = frozenset({DeploymentType.AGENT})
CREATE_MAX_RETRIES = 3
UPDATE_MAX_RETRIES = 3
ROLLBACK_MAX_RETRIES = 5
RETRY_INITIAL_DELAY_SECONDS = 0.5
PROVIDER_SPEC_RESOURCE_NAME_PREFIX_KEY = "resource_name_prefix"
DEFAULT_WXO_AGENT_LLM = "groq/openai/gpt-oss-120b"
WXO_SANITIZE_RE = re.compile(r"[^a-zA-Z0-9_]")
WXO_TRANSLATE = str.maketrans({" ": "_", "-": "_"})
ERROR_PREFIX = "An error occurred while"
ERROR_SUFFIX_IN = "in Watsonx Orchestrate."
class WxOAuthURL(str, Enum):
MCSP = "https://iam.platform.saas.ibm.com"
IBM_IAM = "https://iam.cloud.ibm.com"
class ErrorPrefix(str, Enum):
CREATE = f"{ERROR_PREFIX} creating a deployment {ERROR_SUFFIX_IN}"
LIST = f"{ERROR_PREFIX} listing deployments {ERROR_SUFFIX_IN}"
GET = f"{ERROR_PREFIX} getting a deployment {ERROR_SUFFIX_IN}"
UPDATE = f"{ERROR_PREFIX} updating a deployment {ERROR_SUFFIX_IN}"
REDEPLOY = f"{ERROR_PREFIX} redeploying a deployment {ERROR_SUFFIX_IN}"
CLONE = f"{ERROR_PREFIX} cloning a deployment {ERROR_SUFFIX_IN}"
DELETE = f"{ERROR_PREFIX} deleting a deployment {ERROR_SUFFIX_IN}"
HEALTH = f"{ERROR_PREFIX} getting a deployment health {ERROR_SUFFIX_IN}"
CREATE_CONFIG = f"{ERROR_PREFIX} creating a deployment config {ERROR_SUFFIX_IN}"
LIST_CONFIGS = f"{ERROR_PREFIX} listing deployment configs {ERROR_SUFFIX_IN}"
GET_CONFIG = f"{ERROR_PREFIX} getting a deployment config {ERROR_SUFFIX_IN}"
UPDATE_CONFIG = f"{ERROR_PREFIX} updating a deployment config {ERROR_SUFFIX_IN}"
DELETE_CONFIG = f"{ERROR_PREFIX} deleting a deployment config {ERROR_SUFFIX_IN}"
CREATE_EXECUTION = f"{ERROR_PREFIX} creating a deployment execution {ERROR_SUFFIX_IN}"
GET_EXECUTION = f"{ERROR_PREFIX} getting a deployment execution {ERROR_SUFFIX_IN}"
# NOTE: this key must match the value of the provider_key column
# in the deployment_provider_account table.
WATSONX_ORCHESTRATE_DEPLOYMENT_ADAPTER_KEY = "watsonx-orchestrate"

View File

@ -0,0 +1 @@
"""Core internal modules for the Watsonx Orchestrate deployment adapter."""

View File

@ -0,0 +1,152 @@
"""Config management functions for the Watsonx Orchestrate adapter."""
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING
from ibm_watsonx_orchestrate_core.types.connections import (
ConnectionConfiguration,
ConnectionEnvironment,
ConnectionPreference,
ConnectionSecurityScheme,
)
from lfx.services.adapters.deployment.exceptions import (
InvalidContentError,
InvalidDeploymentOperationError,
)
from lfx.services.adapters.deployment.schema import ConfigItem, DeploymentConfig, IdLike
from langflow.services.adapters.deployment.watsonx_orchestrate.client import resolve_runtime_credentials
from langflow.services.adapters.deployment.watsonx_orchestrate.utils import validate_wxo_name
if TYPE_CHECKING:
from ibm_watsonx_orchestrate_clients.connections.connections_client import ConnectionsClient, GetConnectionResponse
from sqlalchemy.ext.asyncio import AsyncSession
from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient
async def create_config(
*,
clients: WxOClient,
config: DeploymentConfig,
user_id: IdLike,
db: AsyncSession,
) -> str:
"""Create/update a wxO draft key-value connection config plus runtime credentials."""
app_id = validate_wxo_name(config.name)
await asyncio.to_thread(clients.connections.create, payload={"app_id": app_id})
wxo_config = ConnectionConfiguration(
app_id=app_id,
environment=ConnectionEnvironment.DRAFT,
preference=ConnectionPreference.TEAM,
security_scheme=ConnectionSecurityScheme.KEY_VALUE,
)
await asyncio.to_thread(
clients.connections.create_config,
app_id=app_id,
payload=wxo_config.model_dump(exclude_unset=True, exclude_none=True),
)
runtime_credentials = await resolve_runtime_credentials(
environment_variables=config.environment_variables or {},
user_id=user_id,
db=db,
)
await asyncio.to_thread(
clients.connections.create_credentials,
app_id=app_id,
env=ConnectionEnvironment.DRAFT,
use_app_credentials=False,
payload={"runtime_credentials": runtime_credentials.model_dump()},
)
return app_id
async def process_config(
user_id: IdLike,
db: AsyncSession,
deployment_name: str,
config: ConfigItem | None,
*,
clients: WxOClient,
) -> str:
"""Create and bind deployment config using deployment name as app_id."""
validate_config_create_input(config)
environment_variables = None
description = ""
if config and config.raw_payload:
environment_variables = config.raw_payload.environment_variables
description = config.raw_payload.description or ""
config_payload = DeploymentConfig(
name=deployment_name,
description=description,
environment_variables=environment_variables,
)
app_id: str = await create_config(
clients=clients,
config=config_payload,
user_id=user_id,
db=db,
)
return app_id
def validate_config_create_input(config: ConfigItem | None) -> None:
if config and config.reference_id is not None:
msg = (
"Config reference binding is not supported for deployment creation in "
"watsonx Orchestrate. Provide raw config payload or omit config."
)
raise InvalidDeploymentOperationError(message=msg)
def resolve_create_app_id(
*,
prefixed_deployment_name: str,
config: ConfigItem | None,
) -> str:
validate_config_create_input(config)
if config is None or config.raw_payload is None:
return f"{prefixed_deployment_name}_app_id"
normalized_config_name = validate_wxo_name(config.raw_payload.name)
return f"{prefixed_deployment_name}_{normalized_config_name}_app_id"
async def validate_connection(connections_client: ConnectionsClient, *, app_id: str) -> GetConnectionResponse:
connection = await asyncio.to_thread(connections_client.get_draft_by_app_id, app_id=app_id)
if not connection:
msg = f"Connection '{app_id}' not found. Ensure the connection exists with a draft configuration."
raise InvalidContentError(message=msg)
config = await asyncio.to_thread(connections_client.get_config, app_id=app_id, env=ConnectionEnvironment.DRAFT)
if not config:
msg = f"Connection '{app_id}' is missing draft config. Deployments require draft mode."
raise InvalidContentError(message=msg)
if config.security_scheme != ConnectionSecurityScheme.KEY_VALUE:
msg = f"Connection '{app_id}' must use key-value credentials for Langflow flows."
raise InvalidContentError(message=msg)
runtime_credentials = await asyncio.to_thread(
connections_client.get_credentials,
app_id=app_id,
env=ConnectionEnvironment.DRAFT,
use_app_credentials=False,
)
if not runtime_credentials:
msg = f"Connection '{app_id}' is missing draft runtime credentials."
raise InvalidContentError(message=msg)
return connection

View File

@ -0,0 +1,161 @@
"""Execution creation, status, and output extraction for the Watsonx Orchestrate adapter."""
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING, Any
from fastapi import status
from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException
from lfx.services.adapters.deployment.exceptions import DeploymentError, DeploymentNotFoundError, InvalidContentError
from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail
if TYPE_CHECKING:
from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient
def build_orchestrate_runs_query(provider_input: dict[str, Any] | None) -> str:
if not provider_input:
return ""
query_segments: list[str] = []
for key in ("stream", "multiple_content", "stream_timeout"):
if key not in provider_input or provider_input[key] is None:
continue
value = provider_input[key]
normalized_value = str(value).lower() if isinstance(value, bool) else str(value)
query_segments.append(f"{key}={normalized_value}")
if not query_segments:
return ""
return f"?{'&'.join(query_segments)}"
def build_orchestrate_run_payload(
*,
provider_data: dict[str, Any],
deployment_id: str,
) -> dict[str, Any]:
message_payload = provider_data.get("message")
if message_payload is None:
message_payload = resolve_execution_message(provider_data.get("input"))
payload: dict[str, Any] = {
"message": message_payload,
"agent_id": str(provider_data.get("agent_id") or deployment_id),
}
extra_fields = [
"thread_id",
"llm_params",
"guardrails",
"context",
"additional_parameters",
"environment_id",
"version",
"context_variables",
]
payload.update({k: v for k in extra_fields if (v := provider_data.get(k)) is not None})
return payload
async def create_agent_run(
client: WxOClient,
*,
provider_data: dict[str, Any],
deployment_id: str,
) -> dict[str, Any]:
"""Create an orchestrate run through the WxOClient wrapper."""
query_suffix = build_orchestrate_runs_query(provider_data)
try:
run_payload = build_orchestrate_run_payload(
provider_data=provider_data,
deployment_id=deployment_id,
)
except ValueError as exc:
raise InvalidContentError(message=str(exc)) from exc
try:
response = await asyncio.to_thread(
client.post_run,
query_suffix=query_suffix,
data=run_payload,
)
except ClientAPIException as exc:
if exc.response.status_code == status.HTTP_404_NOT_FOUND:
msg = f"Agent Deployment '{deployment_id}' was not found in Watsonx Orchestrate."
raise DeploymentNotFoundError(message=msg) from exc
if exc.response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT:
msg = (
"Deployment execution request is unprocessable by Watsonx Orchestrate. "
f"{extract_error_detail(exc.response.text)}"
)
raise InvalidContentError(message=msg) from exc
raise
return create_agent_run_result(response or {})
def resolve_execution_message(execution_input: str | dict[str, Any] | None) -> dict[str, Any]:
if isinstance(execution_input, str):
if not execution_input.strip():
msg = "Agent execution input message must not be empty."
raise ValueError(msg)
return {"role": "user", "content": execution_input}
if isinstance(execution_input, dict):
if "role" in execution_input and "content" in execution_input:
return execution_input
if "message" in execution_input and isinstance(execution_input["message"], dict):
return execution_input["message"]
content = execution_input.get("content")
if isinstance(content, str) and content.strip():
return {"role": "user", "content": content}
msg = (
"Agent execution requires input content. Provide a non-empty string input "
"or a message payload with 'role' and 'content'."
)
raise ValueError(msg)
def create_agent_run_result(payload: dict[str, Any] | None) -> dict[str, Any]:
if not payload:
msg = "Watsonx Orchestrate returned an empty response for the execution request."
raise DeploymentError(message=msg, error_code="empty_provider_response")
result: dict[str, Any] = {"status": payload.get("status") or "accepted"}
run_id = str(payload.get("run_id") or payload.get("id") or "").strip()
if not run_id:
msg = "Watsonx Orchestrate accepted the execution but did not return a run_id."
raise DeploymentError(message=msg, error_code="missing_run_id")
result["run_id"] = run_id
return result
async def get_agent_run(client: WxOClient, *, run_id: str) -> dict[str, Any]:
payload = await asyncio.to_thread(client.get_run, run_id)
if not payload:
msg = f"Watsonx Orchestrate returned an empty response when fetching execution '{run_id}'."
raise DeploymentError(message=msg, error_code="empty_provider_response")
status_value = str(payload.get("status") or "unknown")
result: dict[str, Any] = {"status": status_value}
passthrough_fields = [
"agent_id",
"run_id",
"started_at",
"completed_at",
"failed_at",
"cancelled_at",
"last_error",
"result",
]
result.update({k: v for k in passthrough_fields if (v := payload.get(k)) is not None})
return result

View File

@ -0,0 +1,205 @@
"""Retry/backoff and rollback logic for the Watsonx Orchestrate adapter."""
from __future__ import annotations
import asyncio
import logging
import random
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING, Any, TypeVar
from fastapi import HTTPException, status
from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException
from lfx.services.adapters.deployment.exceptions import (
DeploymentConflictError,
InvalidContentError,
InvalidDeploymentOperationError,
InvalidDeploymentTypeError,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.constants import (
CREATE_MAX_RETRIES,
RETRY_INITIAL_DELAY_SECONDS,
ROLLBACK_MAX_RETRIES,
UPDATE_MAX_RETRIES,
)
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient
T = TypeVar("T")
Operation = Callable[..., Awaitable[T]]
ShouldRetry = Callable[[Exception], bool]
async def retry_with_backoff(
operation: Operation[T],
max_attempts: int,
*args: Any,
should_retry: ShouldRetry | None = None,
**kwargs: Any,
) -> T:
delay_seconds = RETRY_INITIAL_DELAY_SECONDS
for attempt in range(1, max_attempts + 1):
try:
return await operation(*args, **kwargs)
except Exception as exc:
retryable = True if should_retry is None else should_retry(exc)
if not retryable or attempt == max_attempts:
raise
jittered_delay = delay_seconds * (0.5 + random.random()) # noqa: S311
logger.info(
"Retry attempt %d/%d after %.2fs (%s)", attempt, max_attempts, jittered_delay, type(exc).__name__
)
await asyncio.sleep(jittered_delay)
delay_seconds *= 2
msg = "Retry helper exhausted attempts without result."
raise RuntimeError(msg)
async def retry_create(operation: Operation[T], *args: Any, **kwargs: Any) -> T:
return await retry_with_backoff(
operation,
CREATE_MAX_RETRIES,
*args,
should_retry=is_retryable_create_exception,
**kwargs,
)
async def retry_update(operation: Operation[T], *args: Any, **kwargs: Any) -> T:
"""Retry write/update operations with the standard provider retry policy."""
return await retry_with_backoff(
operation,
UPDATE_MAX_RETRIES,
*args,
should_retry=is_retryable_create_exception,
**kwargs,
)
async def retry_rollback(operation: Operation[T], *args: Any, **kwargs: Any) -> T:
return await retry_with_backoff(
operation,
ROLLBACK_MAX_RETRIES,
*args,
should_retry=is_retryable_create_exception,
**kwargs,
)
def is_retryable_create_exception(exc: Exception) -> bool:
non_retryable_status_codes = {
status.HTTP_400_BAD_REQUEST,
status.HTTP_401_UNAUTHORIZED,
status.HTTP_403_FORBIDDEN,
status.HTTP_404_NOT_FOUND,
status.HTTP_409_CONFLICT,
status.HTTP_422_UNPROCESSABLE_CONTENT,
}
if isinstance(exc, ClientAPIException):
return exc.response.status_code not in non_retryable_status_codes
if isinstance(exc, HTTPException):
return exc.status_code not in non_retryable_status_codes
return not isinstance(
exc,
(
DeploymentConflictError,
InvalidContentError,
InvalidDeploymentOperationError,
InvalidDeploymentTypeError,
),
)
async def rollback_created_resources(
*,
clients: WxOClient,
agent_id: str | None,
tool_ids: list[str],
app_id: str | None,
) -> None:
logger.info("Rolling back resources: agent_id=%s, tool_ids=%s, app_id=%s", agent_id, tool_ids, app_id)
if agent_id:
try:
await retry_rollback(delete_agent_if_exists, clients, agent_id=agent_id)
except Exception:
logger.exception("Rollback failed for agent_id=%s — resource may be orphaned", agent_id)
if tool_ids:
for tool_id in reversed(tool_ids):
try:
await retry_rollback(delete_tool_if_exists, clients, tool_id=tool_id)
except Exception:
logger.exception("Rollback failed for tool_id=%s — resource may be orphaned", tool_id)
if app_id:
try:
await retry_rollback(delete_config_if_exists, clients, app_id=app_id)
except Exception:
logger.exception("Rollback failed for app_id=%s — resource may be orphaned", app_id)
async def rollback_update_resources(
*,
clients: WxOClient,
created_tool_ids: list[str],
created_app_id: str | None,
original_tools: dict[str, dict],
) -> None:
"""Best-effort rollback for update operations.
Restores mutated tools first, then deletes newly created tools, then deletes
newly created config. Unlike ``rollback_created_resources`` this never
deletes the deployment/agent itself.
"""
logger.info(
"Rolling back update resources: created_tool_ids=%s, created_app_id=%s, mutated_tools=%s",
created_tool_ids,
created_app_id,
list(original_tools.keys()),
)
for tool_id, original_tool in reversed(list(original_tools.items())):
try:
await retry_rollback(asyncio.to_thread, clients.tool.update, tool_id, original_tool)
except Exception:
logger.exception(
"Rollback failed: could not restore tool payload for tool_id=%s — resource may be orphaned",
tool_id,
)
for tool_id in reversed(created_tool_ids):
try:
await retry_rollback(delete_tool_if_exists, clients, tool_id=tool_id)
except Exception:
logger.exception("Rollback failed for created tool_id=%s — resource may be orphaned", tool_id)
if created_app_id:
try:
await retry_rollback(delete_config_if_exists, clients, app_id=created_app_id)
except Exception:
logger.exception("Rollback failed for created app_id=%s — resource may be orphaned", created_app_id)
async def delete_agent_if_exists(clients: WxOClient, *, agent_id: str) -> None:
try:
await asyncio.to_thread(clients.agent.delete, agent_id)
except ClientAPIException as exc:
if exc.response.status_code != status.HTTP_404_NOT_FOUND:
raise
async def delete_tool_if_exists(clients: WxOClient, *, tool_id: str) -> None:
try:
await asyncio.to_thread(clients.tool.delete, tool_id)
except ClientAPIException as exc:
if exc.response.status_code != status.HTTP_404_NOT_FOUND:
raise
async def delete_config_if_exists(clients: WxOClient, *, app_id: str) -> None:
try:
await asyncio.to_thread(clients.connections.delete, app_id)
except ClientAPIException as exc:
if exc.response.status_code != status.HTTP_404_NOT_FOUND:
raise

View File

@ -0,0 +1,72 @@
"""Health/status helpers and metadata mapping for the Watsonx Orchestrate adapter."""
from __future__ import annotations
from typing import Any
from ibm_watsonx_orchestrate_core.types.connections import ConnectionEnvironment
from lfx.services.adapters.deployment.schema import DeploymentGetResult, DeploymentType, ItemResult
def get_deployment_metadata(
data: dict[str, Any],
deployment_type: DeploymentType,
provider_data: dict[str, Any] | None = None,
) -> ItemResult:
result: dict[str, Any] = {
"id": data.get("id"),
"type": deployment_type.value,
"name": data.get("name"),
"created_at": data.get("created_on"),
"updated_at": data.get("updated_at"),
}
if provider_data:
result["provider_data"] = provider_data
return ItemResult(**result)
def get_deployment_detail_metadata(
data: dict[str, Any],
deployment_type: DeploymentType,
provider_data: dict[str, Any] | None = None,
provider_raw: bool = False, # noqa: FBT001,FBT002
) -> DeploymentGetResult:
result: dict[str, Any] = {
"id": data.get("id"),
"type": deployment_type.value,
"name": data.get("name"),
"description": data.get("description"),
}
if provider_data:
result["provider_data"] = provider_data
if provider_raw:
result["provider_data"] = data if not provider_data else {**provider_data, "provider_raw": data}
return DeploymentGetResult(**result)
def derive_agent_environment(agent: dict[str, Any]) -> str:
environments = agent.get("environments", [])
if not isinstance(environments, list) or not environments:
return "unknown"
has_draft = False
has_live = False
for env in environments:
if not isinstance(env, dict):
continue
env_name = str(env.get("name", "")).strip().lower()
if env_name == ConnectionEnvironment.DRAFT.value:
has_draft = True
continue
if env_name:
has_live = True
if has_draft and has_live:
return "both"
if has_live:
return "live"
if has_draft:
return "draft"
return "unknown"

View File

@ -0,0 +1,447 @@
"""Snapshot/flow tool creation, artifact building, and upload for the Watsonx Orchestrate adapter."""
from __future__ import annotations
import asyncio
import copy
import importlib.metadata as md
import io
import json
import logging
import zipfile
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from cachetools import func
from ibm_watsonx_orchestrate_core.types.tools.langflow_tool import LangflowTool
from ibm_watsonx_orchestrate_core.types.tools.langflow_tool import create_langflow_tool as _create_langflow_tool
from lfx.services.adapters.deployment.exceptions import InvalidContentError, InvalidDeploymentOperationError
from lfx.utils.flow_requirements import generate_requirements_from_flow
from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import retry_create
from langflow.services.adapters.deployment.watsonx_orchestrate.utils import (
dedupe_list,
normalize_wxo_name,
require_tool_id,
)
from langflow.utils.version import get_version_info
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from lfx.services.adapters.deployment.schema import BaseFlowArtifact, SnapshotItems
from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient
# TODO: ensure all fields from here are used
# https://developer.watson-orchestrate.ibm.com/apis/tools/patch-a-tool
# as it is a PUT endpoint (don't want to lose any fields)
_WRITABLE_TOOL_FIELDS = (
"description",
"permission",
"name",
"display_name",
"input_schema",
"output_schema",
"binding",
"tags",
"is_async",
"restrictions",
"bundled_agent_id",
)
@dataclass(slots=True)
class FlowToolBindingSpec:
flow_payload: BaseFlowArtifact
connections: dict[str, str]
class ToolUploadBatchError(RuntimeError):
"""Raised when a concurrent tool-upload batch partially succeeds."""
def __init__(self, *, created_tool_ids: list[str], errors: list[Exception]) -> None:
self.created_tool_ids = created_tool_ids
self.errors = errors
super().__init__("One or more tool uploads failed.")
def to_writable_tool_payload(tool: dict[str, Any]) -> dict[str, Any]:
"""Build tool payload accepted by wxO tool update endpoint."""
return {field: copy.deepcopy(tool[field]) for field in _WRITABLE_TOOL_FIELDS if field in tool}
def _ensure_dict(parent: dict[str, Any], key: str) -> dict[str, Any]:
"""Return ``parent[key]`` as a dict, replacing non-dict values with ``{}``."""
value = parent.setdefault(key, {})
if not isinstance(value, dict):
logger.warning(
"Expected dict at key '%s' but found %s; replacing with empty dict",
key,
type(value).__name__,
)
value = {}
parent[key] = value
return value
def ensure_langflow_connections_binding(tool_payload: dict[str, Any]) -> dict[str, str]:
"""Ensure ``binding.langflow.connections`` exists in *tool_payload* and return the mutable dict.
Non-dict values at any nesting level are silently replaced with ``{}``.
We intentionally do *not* raise on a malformed shape because
callers of this function are *writing* connection bindings into
these payloads (and the pre-mutation snapshot is captured for rollback),
replacing an unexpected value is traded with explcitiness
to prevent a stubbornly failing update.
"""
binding = _ensure_dict(tool_payload, "binding")
langflow = _ensure_dict(binding, "langflow")
return _ensure_dict(langflow, "connections")
async def update_existing_tool_connection_bindings(
*,
clients: WxOClient,
existing_target_tool_ids: list[str],
resolved_connections: dict[str, str],
original_tools: dict[str, dict[str, Any]],
) -> None:
"""Apply resolved connection bindings to existing tools.
Captures original writable payloads for rollback before any update call.
Raises ``InvalidContentError`` when any expected tool id is missing.
"""
if not existing_target_tool_ids:
return
tools = await asyncio.to_thread(clients.tool.get_drafts_by_ids, existing_target_tool_ids)
tool_by_id = {str(tool.get("id")): tool for tool in tools if isinstance(tool, dict) and tool.get("id")}
missing_tool_ids = [tool_id for tool_id in existing_target_tool_ids if tool_id not in tool_by_id]
if missing_tool_ids:
missing_ids = ", ".join(missing_tool_ids)
msg = f"Snapshot tool(s) not found: {missing_ids}"
raise InvalidContentError(message=msg)
tool_updates: list[tuple[str, dict[str, Any]]] = []
for tool_id in existing_target_tool_ids:
original_tool = to_writable_tool_payload(tool_by_id[tool_id])
original_tools[tool_id] = original_tool
writable_tool = copy.deepcopy(original_tool)
connections = ensure_langflow_connections_binding(writable_tool)
connections.update(resolved_connections)
tool_updates.append((tool_id, writable_tool))
await asyncio.gather(
*(
retry_create(asyncio.to_thread, clients.tool.update, tool_id, writable_tool)
for tool_id, writable_tool in tool_updates
)
)
def extract_langflow_artifact_from_zip(artifact_zip_bytes: bytes, *, snapshot_id: str) -> dict[str, Any]:
"""Read and parse the Langflow flow JSON from a wxO snapshot artifact zip."""
try:
with zipfile.ZipFile(io.BytesIO(artifact_zip_bytes), "r") as zip_artifact:
json_members = [name for name in zip_artifact.namelist() if name.lower().endswith(".json")]
if not json_members:
msg = f"Snapshot '{snapshot_id}' artifact does not include a flow JSON file."
raise InvalidContentError(message=msg)
flow_json_member = json_members[0]
flow_json_raw = zip_artifact.read(flow_json_member)
except InvalidContentError:
raise
except zipfile.BadZipFile as exc:
msg = f"Snapshot '{snapshot_id}' artifact is not a valid zip archive."
raise InvalidContentError(message=msg) from exc
try:
return json.loads(flow_json_raw.decode("utf-8"))
except UnicodeDecodeError as exc:
msg = f"Snapshot '{snapshot_id}' flow artifact is not valid UTF-8 JSON."
raise InvalidContentError(message=msg) from exc
except json.JSONDecodeError as exc:
msg = f"Snapshot '{snapshot_id}' flow artifact contains invalid JSON."
raise InvalidContentError(message=msg) from exc
def build_langflow_artifact_bytes(
*,
tool: LangflowTool,
flow_definition: dict[str, Any],
flow_filename: str | None = None,
) -> bytes:
filename = flow_filename or f"{tool.__tool_spec__.name}.json"
lfx_requirement = _resolve_lfx_requirement()
requirements = generate_requirements_from_flow(
flow_definition,
include_lfx=False,
pin_versions=True,
)
requirements = [lfx_requirement, *requirements]
requirements = dedupe_list(requirements)
requirements_content = "\n".join(requirements) + "\n"
flow_content = json.dumps(flow_definition, indent=2)
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zip_tool_artifacts:
zip_tool_artifacts.writestr(filename, flow_content)
zip_tool_artifacts.writestr("requirements.txt", requirements_content)
zip_tool_artifacts.writestr("bundle-format", "2.0.0\n")
return buffer.getvalue()
def upload_tool_artifact_bytes(
clients: WxOClient,
*,
tool_id: str,
artifact_bytes: bytes,
) -> dict[str, Any]:
file_obj = io.BytesIO(artifact_bytes)
return clients.upload_tool_artifact(
tool_id,
files={"file": (f"{tool_id}.zip", file_obj, "application/zip", {"Expires": "0"})},
)
def create_wxo_flow_tool(
*,
flow_payload: BaseFlowArtifact,
connections: dict[str, str],
tool_name_prefix: str,
) -> tuple[dict[str, Any], bytes]:
"""Create a Watsonx Orchestrate flow tool specification.
Given a flow payload and connections dictionary,
create a Watsonx Orchestrate flow tool specification
and the supporting artifacts of the requirements.txt
and the flow json file.
Args:
flow_payload: The flow payload to create the tool specification for.
connections: The connections dictionary to create the tool specification for.
tool_name_prefix: Deterministic prefix for the resulting tool name.
Returns:
Tuple[dict[str, Any], bytes]: a tuple containing:
- tool_payload: The Watsonx Orchestrate flow tool specification.
- artifacts: The supporting artifacts (the requirements.txt
and the flow json file) for the tool.
"""
flow_definition = flow_payload.model_dump()
flow_provider_data = flow_definition.pop("provider_data", None)
if not isinstance(flow_provider_data, dict):
msg = "Flow payload must include provider_data with a non-empty project_id for Watsonx deployment."
raise InvalidContentError(message=msg)
project_id = str(flow_provider_data.get("project_id") or "").strip()
if not project_id:
msg = "Flow payload must include provider_data with a non-empty project_id for Watsonx deployment."
raise InvalidContentError(message=msg)
flow_definition.update(
{
"name": normalize_wxo_name(flow_definition.get("name") or ""),
"id": str(flow_definition.get("id")),
}
)
# Fallback for flows that don't include last_tested_version in payload
if not flow_definition.get("last_tested_version"):
detected_version = (get_version_info() or {}).get("version")
if not detected_version:
msg = "Unable to determine running Langflow version for snapshot creation."
raise InvalidContentError(message=msg)
flow_definition["last_tested_version"] = detected_version
tool: LangflowTool = create_langflow_tool(
tool_definition=flow_definition,
connections=connections,
show_details=False,
)
tool_payload = tool.__tool_spec__.model_dump(
mode="json",
exclude_unset=True,
exclude_none=True,
by_alias=True,
)
current_name = str(tool_payload.get("name") or "").strip()
if current_name:
normalized_current_name = normalize_wxo_name(current_name)
tool_payload["name"] = f"{tool_name_prefix}{normalized_current_name}"
(tool_payload.setdefault("binding", {}).setdefault("langflow", {})["project_id"]) = project_id
artifacts: bytes = build_langflow_artifact_bytes(
tool=tool,
flow_definition=flow_definition,
)
return tool_payload, artifacts
def create_langflow_tool(
*,
tool_definition: dict[str, Any],
connections: dict[str, str],
show_details: bool,
) -> LangflowTool:
"""Module-level wrapper to keep tool creation monkeypatchable in tests."""
return _create_langflow_tool(
tool_definition=tool_definition,
connections=connections,
show_details=show_details,
)
async def create_and_upload_wxo_flow_tools(
*,
clients: WxOClient,
flow_payloads: list[BaseFlowArtifact],
connections: dict[str, str],
tool_name_prefix: str,
) -> list[str]:
tool_bindings = [
FlowToolBindingSpec(
flow_payload=flow_payload,
connections=connections,
)
for flow_payload in flow_payloads
]
return await create_and_upload_wxo_flow_tools_with_bindings(
clients=clients,
tool_bindings=tool_bindings,
tool_name_prefix=tool_name_prefix,
)
async def create_and_upload_wxo_flow_tools_with_bindings(
*,
clients: WxOClient,
tool_bindings: list[FlowToolBindingSpec],
tool_name_prefix: str,
) -> list[str]:
specs = [
create_wxo_flow_tool(
flow_payload=tool_binding.flow_payload,
connections=tool_binding.connections,
tool_name_prefix=tool_name_prefix,
)
for tool_binding in tool_bindings
]
created_tool_ids_journal: list[str] = []
results = await asyncio.gather(
*(
upload_wxo_flow_tool(
clients=clients,
tool_payload=tool_payload,
artifact_bytes=artifact_bytes,
created_tool_ids_journal=created_tool_ids_journal,
)
for tool_payload, artifact_bytes in specs
),
return_exceptions=True,
)
errors: list[Exception] = []
created_tool_ids: list[str] = []
for result in results:
if isinstance(result, BaseException):
if isinstance(result, Exception):
errors.append(result)
else:
errors.append(RuntimeError(f"Tool upload failed with non-standard exception: {type(result).__name__}"))
continue
created_tool_ids.append(result)
if errors:
raise ToolUploadBatchError(created_tool_ids=dedupe_list(created_tool_ids_journal), errors=errors)
return created_tool_ids
async def upload_wxo_flow_tool(
*,
clients: WxOClient,
tool_payload: dict[str, Any],
artifact_bytes: bytes,
created_tool_ids_journal: list[str] | None = None,
) -> str:
tool_response = await retry_create(asyncio.to_thread, clients.tool.create, tool_payload)
tool_id = require_tool_id(tool_response)
if created_tool_ids_journal is not None:
created_tool_ids_journal.append(tool_id)
await retry_create(
asyncio.to_thread,
upload_tool_artifact_bytes,
clients,
tool_id=tool_id,
artifact_bytes=artifact_bytes,
)
return tool_id
def build_snapshot_tool_names(
*,
snapshots: SnapshotItems | None,
tool_name_prefix: str,
) -> list[str]:
if snapshots is None:
return []
tool_names: list[str] = []
for snapshot in snapshots.raw_payloads:
normalized_tool_name = normalize_wxo_name(str(snapshot.name))
if not normalized_tool_name:
msg = "Snapshot name must include at least one alphanumeric character."
raise InvalidContentError(message=msg)
tool_names.append(f"{tool_name_prefix}{normalized_tool_name}")
return tool_names
async def process_raw_flows_with_app_id(
clients: WxOClient,
app_id: str,
flows: list[BaseFlowArtifact],
tool_name_prefix: str,
) -> list[str]:
"""Create langflow tools in wxO and connect them to the given app_id."""
from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import validate_connection
if not tool_name_prefix.strip():
msg = "Snapshot creation requires a non-empty tool_name_prefix."
raise InvalidDeploymentOperationError(message=msg)
connection = await validate_connection(clients.connections, app_id=app_id)
return await create_and_upload_wxo_flow_tools(
clients=clients,
flow_payloads=flows,
connections={app_id: connection.connection_id},
tool_name_prefix=tool_name_prefix,
)
# TODO(WXO): find a way to make this fallback not hard-coded.
_LFX_MINIMUM_REQUIREMENT = "lfx>=0.3.0"
@func.ttl_cache(maxsize=1, ttl=60)
def _pin_requirement_name(package_name: str) -> str:
return f"{package_name}=={md.version(package_name)}"
def _resolve_lfx_requirement() -> str:
"""Pin lfx to the installed version, falling back to a minimum spec."""
try:
return _pin_requirement_name("lfx")
except (md.PackageNotFoundError, ValueError):
logger.warning(
"Could not determine installed lfx version; falling back to minimum requirement '%s'",
_LFX_MINIMUM_REQUIREMENT,
)
return _LFX_MINIMUM_REQUIREMENT

View File

@ -0,0 +1,248 @@
"""Watsonx Orchestrate custom deployment update payload contracts."""
from __future__ import annotations
from collections import Counter
from typing import Annotated, Literal
from lfx.services.adapters.deployment.schema import BaseFlowArtifact, EnvVarKey, EnvVarValueSpec, NormalizedId
from lfx.services.adapters.payload import AdapterPayload
from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_validator, model_validator
RawToolName = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
class WatsonxConnectionRawPayload(BaseModel):
"""Connection payload for creating a new watsonx connection/config."""
app_id: NormalizedId = Field(
description=(
"App id used for operation references. resource_name_prefix is applied only when creating resources."
)
)
environment_variables: dict[EnvVarKey, EnvVarValueSpec] | None = Field(None, description="Environment variables.")
provider_config: AdapterPayload | None = Field(None, description="Provider-specific connection configuration.")
class WatsonxUpdateTools(BaseModel):
"""Tool pool available to update operations."""
model_config = ConfigDict(extra="forbid")
existing_ids: list[NormalizedId] | None = Field(
default=None,
description="Known existing provider tool ids available for operation references.",
)
raw_payloads: list[BaseFlowArtifact] | None = Field(
default=None,
description="Raw tool payloads keyed by BaseFlowArtifact.name.",
)
@field_validator("existing_ids")
@classmethod
def dedupe_existing_ids(cls, value: list[str] | None) -> list[str] | None:
if value is None:
return None
return list(dict.fromkeys(value))
@model_validator(mode="after")
def validate_unique_raw_tool_names(self) -> WatsonxUpdateTools:
raw_payloads = self.raw_payloads or []
name_counts = Counter(payload.name for payload in raw_payloads)
duplicates = sorted(name for name, count in name_counts.items() if count > 1)
if duplicates:
msg = f"tools.raw_payloads contains duplicate names: {duplicates}"
raise ValueError(msg)
return self
class WatsonxUpdateConnections(BaseModel):
"""Connection pool available to update operations."""
model_config = ConfigDict(extra="forbid")
existing_app_ids: list[NormalizedId] | None = Field(
default=None,
description="Known existing app ids available for operation references.",
)
raw_payloads: list[WatsonxConnectionRawPayload] | None = Field(
default=None,
description=(
"Raw connection payloads keyed by app_id. resource_name_prefix is applied only when resources are created."
),
)
@field_validator("existing_app_ids")
@classmethod
def dedupe_existing_app_ids(cls, value: list[str] | None) -> list[str] | None:
if value is None:
return None
return list(dict.fromkeys(value))
@model_validator(mode="after")
def validate_unique_raw_app_ids(self) -> WatsonxUpdateConnections:
raw_payloads = self.raw_payloads or []
app_id_counts = Counter(payload.app_id for payload in raw_payloads)
duplicates = sorted(app_id for app_id, count in app_id_counts.items() if count > 1)
if duplicates:
msg = f"connections.raw_payloads contains duplicate app_id values: {duplicates}"
raise ValueError(msg)
return self
class WatsonxToolReference(BaseModel):
"""Tool selector for bind operations."""
model_config = ConfigDict(extra="forbid")
reference_id: NormalizedId | None = Field(
default=None,
description="Existing provider tool id.",
)
name_of_raw: RawToolName | None = Field(
default=None,
description="Name of a tool entry declared in tools.raw_payloads.",
)
@model_validator(mode="after")
def validate_exactly_one_selector(self) -> WatsonxToolReference:
has_reference_id = self.reference_id is not None
has_name_of_raw = self.name_of_raw is not None
if has_reference_id == has_name_of_raw:
msg = "Exactly one of 'tool.reference_id' or 'tool.name_of_raw' must be provided."
raise ValueError(msg)
return self
class WatsonxBindOperation(BaseModel):
"""Bind a selected tool to a selected app id."""
model_config = ConfigDict(extra="forbid")
op: Literal["bind"]
tool: WatsonxToolReference
app_ids: list[NormalizedId] = Field(
min_length=1,
description=(
"Operation app ids to bind. Must match declared connection pools "
"(connections.existing_app_ids or connections.raw_payloads[*].app_id)."
),
)
@field_validator("app_ids")
@classmethod
def dedupe_app_ids(cls, value: list[str]) -> list[str]:
return list(dict.fromkeys(value))
class WatsonxUnbindOperation(BaseModel):
"""Unbind app connection from a tool."""
model_config = ConfigDict(extra="forbid")
op: Literal["unbind"]
tool_id: NormalizedId = Field(description="Existing provider tool id.")
app_ids: list[NormalizedId] = Field(
min_length=1,
description=("Operation app ids to unbind. Must reference connections.existing_app_ids only."),
)
@field_validator("app_ids")
@classmethod
def dedupe_app_ids(cls, value: list[str]) -> list[str]:
return list(dict.fromkeys(value))
class WatsonxRemoveToolOperation(BaseModel):
"""Detach an existing tool from the deployment."""
model_config = ConfigDict(extra="forbid")
op: Literal["remove_tool"]
tool_id: NormalizedId = Field(description="Existing provider tool id to remove from deployment.")
WatsonxUpdateOperation = Annotated[
WatsonxBindOperation | WatsonxUnbindOperation | WatsonxRemoveToolOperation,
Field(discriminator="op"),
]
class WatsonxDeploymentUpdatePayload(BaseModel):
"""Watsonx provider_data contract for deployment update patch operations.
Notes:
- operations[*].app_ids are operation-side ids.
- resource_name_prefix is applied only when creating resources
(for raw connections and raw tools).
"""
model_config = ConfigDict(extra="forbid")
resource_name_prefix: str | None = Field(
default=None,
description=(
"Prefix applied only when creating resources: "
"derived app ids from connections.raw_payloads[*].app_id and created tool names."
),
)
tools: WatsonxUpdateTools = Field(default_factory=WatsonxUpdateTools)
connections: WatsonxUpdateConnections = Field(default_factory=WatsonxUpdateConnections)
operations: list[WatsonxUpdateOperation] = Field(min_length=1)
@model_validator(mode="after")
def validate_operation_references(self) -> WatsonxDeploymentUpdatePayload:
raw_tool_names = {payload.name for payload in (self.tools.raw_payloads or [])}
existing_tool_ids = set(self.tools.existing_ids or [])
existing_app_ids = set(self.connections.existing_app_ids or [])
raw_app_ids = {payload.app_id for payload in (self.connections.raw_payloads or [])}
collisions = sorted(existing_app_ids.intersection(raw_app_ids))
if collisions:
msg = f"connections.existing_app_ids collides with raw app ids from connections.raw_payloads: {collisions}"
raise ValueError(msg)
valid_app_ids = existing_app_ids.union(raw_app_ids)
referenced_app_ids: set[str] = set()
for operation in self.operations:
if isinstance(operation, WatsonxBindOperation):
if operation.tool.name_of_raw is not None and operation.tool.name_of_raw not in raw_tool_names:
msg = f"bind.tool.name_of_raw not found in tools.raw_payloads: [{operation.tool.name_of_raw!r}]"
raise ValueError(msg)
if operation.tool.reference_id is not None and operation.tool.reference_id not in existing_tool_ids:
msg = f"bind.tool.reference_id not found in tools.existing_ids: [{operation.tool.reference_id!r}]"
raise ValueError(msg)
for app_id in operation.app_ids:
referenced_app_ids.add(app_id)
if app_id not in valid_app_ids:
msg = (
"operation app_ids must be declared in "
"connections.existing_app_ids or connections.raw_payloads[*].app_id: "
f"[{app_id!r}]"
)
raise ValueError(msg)
if isinstance(operation, WatsonxUnbindOperation):
for app_id in operation.app_ids:
referenced_app_ids.add(app_id)
if app_id not in valid_app_ids:
msg = (
"operation app_ids must be declared in "
"connections.existing_app_ids or connections.raw_payloads[*].app_id: "
f"[{app_id!r}]"
)
raise ValueError(msg)
if app_id in raw_app_ids:
msg = f"unbind.operation app_ids must reference connections.existing_app_ids only: [{app_id!r}]"
raise ValueError(msg)
unused_existing_app_ids = sorted(existing_app_ids.difference(referenced_app_ids))
if unused_existing_app_ids:
msg = f"connections.existing_app_ids contains ids not referenced by operations: {unused_existing_app_ids}"
raise ValueError(msg)
unused_raw_app_ids = sorted(raw_app_ids.difference(referenced_app_ids))
if unused_raw_app_ids:
msg = f"connections.raw_payloads contains app_id values not referenced by operations: {unused_raw_app_ids}"
raise ValueError(msg)
return self

View File

@ -0,0 +1,799 @@
"""Slim WatsonxOrchestrateDeploymentService that delegates to submodules."""
from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING
from fastapi import HTTPException, status
from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException
from lfx.services.adapters.deployment.base import BaseDeploymentService
from lfx.services.adapters.deployment.exceptions import (
AuthenticationError,
AuthorizationError,
DeploymentConflictError,
DeploymentError,
DeploymentNotFoundError,
DeploymentSupportError,
InvalidContentError,
InvalidDeploymentOperationError,
InvalidDeploymentTypeError,
OperationNotSupportedError,
)
from lfx.services.adapters.deployment.payloads import DeploymentPayloadSchemas
from lfx.services.adapters.deployment.schema import (
BaseDeploymentData,
ConfigListItem,
ConfigListParams,
ConfigListResult,
DeploymentCreate,
DeploymentCreateResult,
DeploymentDeleteResult,
DeploymentDuplicateResult,
DeploymentGetResult,
DeploymentListParams,
DeploymentListResult,
DeploymentListTypesResult,
DeploymentStatusResult,
DeploymentType,
DeploymentUpdate,
DeploymentUpdateResult,
ExecutionCreate,
ExecutionCreateResult,
ExecutionStatusResult,
IdLike,
RedeployResult,
SnapshotItem,
SnapshotListParams,
SnapshotListResult,
_normalize_and_validate_id,
)
from lfx.services.adapters.payload import AdapterPayloadValidationError, PayloadSlot
from langflow.services.adapters.deployment.watsonx_orchestrate.client import get_provider_clients
from langflow.services.adapters.deployment.watsonx_orchestrate.constants import (
PROVIDER_SPEC_RESOURCE_NAME_PREFIX_KEY,
SUPPORTED_ADAPTER_DEPLOYMENT_TYPES,
ErrorPrefix,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import (
process_config,
resolve_create_app_id,
validate_config_create_input,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import (
create_agent_run,
get_agent_run,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import (
retry_create,
rollback_created_resources,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.core.status import (
derive_agent_environment,
get_deployment_detail_metadata,
get_deployment_metadata,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import (
build_snapshot_tool_names,
process_raw_flows_with_app_id,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import (
WatsonxDeploymentUpdatePayload,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.update_helpers import (
apply_provider_update_plan_with_rollback,
build_provider_update_plan,
build_update_payload_from_spec,
validate_provider_update_request_sections,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.utils import (
_require_single_deployment_id,
build_agent_payload,
dedupe_list,
extract_agent_tool_ids,
extract_error_detail,
raise_as_deployment_error,
resolve_resource_name_prefix,
validate_wxo_name,
)
from langflow.services.deps import get_settings_service
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from collections.abc import Sequence
from typing import Any
from ibm_watsonx_orchestrate_clients.agents.agent_client import AgentUpsertResponse
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."""
name = "deployment_service"
payload_schemas = DeploymentPayloadSchemas(
deployment_update=PayloadSlot(WatsonxDeploymentUpdatePayload),
)
def __init__(self, settings_service: SettingsService | None = None):
super().__init__()
if settings_service is None:
settings_service = get_settings_service()
if settings_service is None:
msg = "Settings service is not available."
raise RuntimeError(msg)
self.settings_service = settings_service
self.set_ready()
async def _get_provider_clients(self, *, user_id: IdLike, db: AsyncSession) -> WxOClient:
"""Resolve provider clients through a service-level seam.
A dedicated method keeps call sites patchable in unit/e2e tests and
centralizes provider-client resolution behavior for this service.
"""
return await get_provider_clients(user_id=user_id, db=db)
async def create(
self,
*,
user_id: IdLike,
payload: DeploymentCreate,
db: AsyncSession,
) -> DeploymentCreateResult:
"""Create a deployment in Watsonx Orchestrate."""
# The wxO API does not have an endpoint to create
# a connection, tool, and agent atomically.
# We have to make a separate api call for each resource.
# --
# If one of these resources is created successfully
# but the next one fails, then we end up with orphaned resources.
# - We thus use a best-effort creation and rollback strategy:
# Attempt to create each resource with retries.
# If the creation of one resource fails,
# then attempt to delete all previously created
# resources with retries.
# --
# The caller must supply a resource name prefix via
# provider_spec["resource_name_prefix"].
# Every created resource (connection, tool, agent) is
# prefixed with this value, which prevents name collisions
# and supports idempotent retries (re-use the same prefix
# across attempts). We recommend using a random prefix.
# --
logger.info("Creating wxO deployment for user_id=%s", user_id)
agent_create_response: AgentUpsertResponse | None = None
created_tool_ids: list[str] = []
created_app_id: str | None = None
clients: WxOClient | None = None
derived_spec: BaseDeploymentData | None = None
try:
deployment_spec: BaseDeploymentData = payload.spec
normalized_deployment_name = validate_wxo_name(deployment_spec.name)
validate_config_create_input(payload.config)
if deployment_spec.type != DeploymentType.AGENT:
msg = (
f"{ErrorPrefix.CREATE.value}"
f"Deployment type '{deployment_spec.type.value}' "
"is not supported by watsonx Orchestrate."
)
raise DeploymentSupportError(message=msg)
caller_prefix = (deployment_spec.provider_spec or {}).get(PROVIDER_SPEC_RESOURCE_NAME_PREFIX_KEY)
if not caller_prefix:
msg = (
f"{ErrorPrefix.CREATE.value} provider_spec must include '{PROVIDER_SPEC_RESOURCE_NAME_PREFIX_KEY}'."
)
raise InvalidContentError(message=msg)
clients = await self._get_provider_clients(user_id=user_id, db=db)
resource_prefix = resolve_resource_name_prefix(
caller_prefix=caller_prefix,
)
prefixed_deployment_name = f"{resource_prefix}{normalized_deployment_name}"
tool_names = build_snapshot_tool_names(
snapshots=payload.snapshot,
tool_name_prefix=resource_prefix,
)
if len(tool_names) != len(set(tool_names)):
msg = (
f"{ErrorPrefix.CREATE.value} "
"Duplicate snapshot names detected in the request. "
"Each snapshot must have a unique name."
)
raise DeploymentConflictError(message=msg)
prefixed_app_id = resolve_create_app_id(
prefixed_deployment_name=prefixed_deployment_name,
config=payload.config,
)
try:
created_app_id = await retry_create(
process_config,
clients=clients,
user_id=user_id,
db=db,
deployment_name=prefixed_app_id,
config=payload.config,
)
if payload.snapshot and (flow_payloads := payload.snapshot.raw_payloads):
created_tool_ids = await retry_create(
process_raw_flows_with_app_id,
clients=clients,
app_id=prefixed_app_id,
flows=flow_payloads,
tool_name_prefix=resource_prefix,
)
derived_spec = deployment_spec.model_copy(deep=True)
if derived_spec.provider_spec is None:
derived_spec.provider_spec = {}
derived_spec.provider_spec.update(
{
"name": prefixed_deployment_name,
"display_name": derived_spec.name,
}
)
agent_create_response = await retry_create(
self._create_agent_deployment,
clients=clients,
data=derived_spec,
tool_ids=created_tool_ids,
)
except Exception:
logger.warning(
"wxO create failed; rolling back agent_id=%s, tool_ids=%s, app_id=%s",
agent_create_response.id if agent_create_response else None,
created_tool_ids,
created_app_id,
)
await rollback_created_resources(
clients=clients,
agent_id=agent_create_response.id if agent_create_response else None,
tool_ids=created_tool_ids,
app_id=created_app_id,
)
raise
except (ClientAPIException, HTTPException) as exc:
if isinstance(exc, ClientAPIException):
status_code = exc.response.status_code
error_detail = extract_error_detail(exc.response.text)
else:
status_code = exc.status_code
error_detail = extract_error_detail(str(exc.detail))
is_conflict = status_code == status.HTTP_409_CONFLICT or "already exists" in error_detail.lower()
if is_conflict:
msg = (
f"{ErrorPrefix.CREATE.value} "
"One or more resources already exist. "
"Please ensure the names and/or ids of the "
"following resources to be unique: "
"(1) The deployment specification, "
"(2) The deployment configuration, "
"(3) The deployment snapshot. "
f"error details: {error_detail}"
)
raise DeploymentConflictError(message=msg) from exc
if status_code == status.HTTP_422_UNPROCESSABLE_CONTENT:
msg = (
f"{ErrorPrefix.CREATE.value} "
"The deployment request entity is unprocessable. "
"Please ensure the request entity is valid and complete. "
f"error details: {error_detail}"
)
raise InvalidContentError(message=msg) from exc
msg = f"{ErrorPrefix.CREATE.value} error details: {error_detail}"
raise DeploymentError(message=msg, error_code="deployment_error") from exc
except (
AuthenticationError,
DeploymentConflictError,
InvalidContentError,
InvalidDeploymentOperationError,
InvalidDeploymentTypeError,
DeploymentSupportError,
):
raise
except Exception as exc:
logger.exception("Unexpected error during wxO deployment creation")
msg = f"{ErrorPrefix.CREATE.value} Please check server logs for details."
raise DeploymentError(message=msg, error_code="deployment_error") from exc
if agent_create_response is None or derived_spec is None:
msg = f"{ErrorPrefix.CREATE.value} Deployment response was empty."
raise DeploymentError(message=msg, error_code="deployment_error")
derived_spec.name = deployment_spec.name # restore the original name
return DeploymentCreateResult(
id=agent_create_response.id,
config_id=created_app_id,
snapshot_ids=created_tool_ids,
**derived_spec.model_dump(exclude_unset=True),
)
async def list_types(
self,
*,
user_id: IdLike, # noqa: ARG002
db: AsyncSession, # noqa: ARG002
) -> DeploymentListTypesResult:
"""List deployment types supported by the provider."""
return DeploymentListTypesResult(deployment_types=list(SUPPORTED_ADAPTER_DEPLOYMENT_TYPES))
async def list(
self,
*,
user_id: IdLike,
db: AsyncSession,
params: DeploymentListParams | None = None,
) -> DeploymentListResult:
"""List deployments from Watsonx Orchestrate."""
client_manager = await self._get_provider_clients(user_id=user_id, db=db)
deployments: list = []
try:
deployment_types: set[DeploymentType] = set()
invalid_deployment_types: set[DeploymentType] = set()
if params and params.deployment_types:
deployment_types = set(params.deployment_types)
invalid_deployment_types = deployment_types.difference(SUPPORTED_ADAPTER_DEPLOYMENT_TYPES)
if invalid_deployment_types:
invalid_values = ", ".join([dtype.value for dtype in invalid_deployment_types])
msg = (
f"{ErrorPrefix.LIST.value} watsonx Orchestrate has no such deployment type(s): '{invalid_values}'."
)
raise InvalidDeploymentTypeError(message=msg)
query_params: dict[str, Any] = {}
if params and params.provider_params:
query_params = params.provider_params
if params and params.deployment_ids and "ids" not in query_params:
query_params["ids"] = [str(_id) for _id in params.deployment_ids]
# if different deployment types
# are distinct resources in wxO
# then we should probably raise an error if
# the ids query parameter is not empty or null
# this is not a problem today, but might be in the future
raw_agents = await asyncio.to_thread(
client_manager.get_agents_raw,
params=query_params or None,
)
deployments = [
get_deployment_metadata(
data=agent,
deployment_type=DeploymentType.AGENT,
provider_data={
"snapshot_ids": extract_agent_tool_ids(agent),
"environment": derive_agent_environment(agent),
},
)
for agent in raw_agents
]
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.LIST,
log_msg="Unexpected error while listing wxO deployments",
pass_through=(AuthenticationError, AuthorizationError, InvalidDeploymentTypeError),
)
return DeploymentListResult(
deployments=deployments,
)
async def get(
self,
*,
user_id: IdLike,
deployment_id: IdLike,
deployment_type: DeploymentType | None = None, # noqa: ARG002
db: AsyncSession,
) -> DeploymentGetResult:
"""Get a deployment (agent) from Watsonx Orchestrate."""
client_manager = await self._get_provider_clients(user_id=user_id, db=db)
try:
agent = await asyncio.to_thread(client_manager.agent.get_draft_by_id, deployment_id)
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.GET,
log_msg="Unexpected error fetching wxO deployment",
pass_through=(AuthenticationError, AuthorizationError, DeploymentNotFoundError),
)
if not agent:
msg = f"Deployment '{deployment_id}' not found."
raise DeploymentNotFoundError(msg)
return get_deployment_detail_metadata(
data=agent,
deployment_type=DeploymentType.AGENT,
)
async def update(
self,
*,
user_id: IdLike,
deployment_id: IdLike,
deployment_type: DeploymentType | None = None, # noqa: ARG002
payload: DeploymentUpdate,
db: AsyncSession,
) -> DeploymentUpdateResult:
"""Update deployment metadata and provider-driven tool/config operations."""
try:
clients = await self._get_provider_clients(user_id=user_id, db=db)
agent_id = _normalize_and_validate_id(str(deployment_id), field_name="deployment_id")
agent = await asyncio.to_thread(clients.agent.get_draft_by_id, agent_id)
if not agent:
msg = f"Deployment '{agent_id}' not found."
raise DeploymentNotFoundError(msg)
# base agent payload to build for final update call
update_payload: dict[str, Any] = build_update_payload_from_spec(payload.spec)
validate_provider_update_request_sections(payload)
if payload.provider_data is None:
if not update_payload:
msg = "provider_data is required when update operations do not include spec changes."
raise InvalidContentError(message=msg)
await retry_create(
asyncio.to_thread,
clients.agent.update,
agent_id,
update_payload,
)
return DeploymentUpdateResult(id=deployment_id, snapshot_ids=[])
try:
provider_update = self.payload_schemas.deployment_update.parse(payload.provider_data)
except AdapterPayloadValidationError as exc:
first_error = exc.error.errors()[0] if exc.error.errors() else {}
msg = str(first_error.get("msg") or exc)
raise InvalidContentError(message=msg) from None
provider_plan = build_provider_update_plan(
agent=agent,
provider_update=provider_update,
)
added_snapshot_ids: list[str] = await apply_provider_update_plan_with_rollback(
clients=clients,
user_id=user_id,
db=db,
agent_id=agent_id,
agent=agent,
update_payload=update_payload,
plan=provider_plan,
)
return DeploymentUpdateResult(
id=deployment_id,
snapshot_ids=added_snapshot_ids,
)
except (ClientAPIException, HTTPException) as exc:
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.UPDATE,
log_msg="Unexpected provider error during wxO deployment update",
)
except (
AuthenticationError,
AuthorizationError,
DeploymentNotFoundError,
InvalidContentError,
InvalidDeploymentOperationError,
DeploymentConflictError,
):
raise
except Exception as exc:
logger.exception("Unexpected error during wxO deployment update")
msg = f"{ErrorPrefix.UPDATE.value} Please check server logs for details."
raise DeploymentError(message=msg, error_code="deployment_error") from exc
async def redeploy(
self,
*,
user_id: IdLike, # noqa: ARG002
deployment_id: IdLike, # noqa: ARG002
deployment_type: DeploymentType | None = None, # noqa: ARG002
db: AsyncSession, # noqa: ARG002
) -> RedeployResult:
"""Trigger a deployment redeployment for the agent in draft environment."""
msg = "Redeployment is not supported by the watsonx Orchestrate adapter."
raise OperationNotSupportedError(message=msg)
async def duplicate(
self,
*,
user_id: IdLike, # noqa: ARG002
deployment_id: IdLike, # noqa: ARG002
deployment_type: DeploymentType | None = None, # noqa: ARG002
db: AsyncSession, # noqa: ARG002
) -> DeploymentDuplicateResult:
"""Duplicate an existing deployment."""
msg = "Deployment duplication is not supported by the watsonx Orchestrate adapter."
raise OperationNotSupportedError(message=msg)
async def delete(
self,
*,
user_id: IdLike,
deployment_id: IdLike,
deployment_type: DeploymentType | None = None, # noqa: ARG002
db: AsyncSession,
) -> DeploymentDeleteResult:
"""Delete only the deployment agent (keep tools/configs reusable)."""
logger.info("Deleting wxO deployment deployment_id=%s", deployment_id)
agent_id = _normalize_and_validate_id(str(deployment_id), field_name="deployment_id")
clients = await self._get_provider_clients(user_id=user_id, db=db)
try:
await asyncio.to_thread(clients.agent.delete, agent_id)
except ClientAPIException as e:
status_code = e.response.status_code
if status_code == status.HTTP_404_NOT_FOUND:
msg = f"{ErrorPrefix.DELETE.value} deployment id '{agent_id}' not found."
raise DeploymentNotFoundError(msg) from e
msg = f"{ErrorPrefix.DELETE.value} error details: {extract_error_detail(e.response.text)}"
raise DeploymentError(msg, error_code="deployment_error") from e
except (AuthenticationError, AuthorizationError, DeploymentNotFoundError):
raise
except Exception as exc:
logger.exception("Unexpected error while deleting wxO deployment %s", agent_id)
msg = f"{ErrorPrefix.DELETE.value} Please check server logs for details."
raise DeploymentError(msg, error_code="deployment_error") from exc
return DeploymentDeleteResult(id=agent_id)
# TODO: get status normally if its a live agent
# if its draft, use the current 'exists' or raise not found logic
async def get_status(
self,
*,
user_id: IdLike,
deployment_id: IdLike,
deployment_type: DeploymentType | None = None, # noqa: ARG002
db: AsyncSession,
) -> DeploymentStatusResult:
"""Get deployment status from wxO agent metadata.
Note: wxO does not expose a dedicated health endpoint for draft Agents. Status is
inferred from agent existence and environment metadata -- "connected"
means the agent draft was found, not that it is healthy at runtime.
"""
agent_id = _normalize_and_validate_id(str(deployment_id), field_name="deployment_id")
clients = await self._get_provider_clients(user_id=user_id, db=db)
try:
agent = await asyncio.to_thread(clients.agent.get_draft_by_id, agent_id)
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.HEALTH,
log_msg="Unexpected error fetching wxO deployment status",
)
if not agent or isinstance(agent, str): # the adk returns a string if not found
raise DeploymentNotFoundError(deployment_id=agent_id)
return DeploymentStatusResult(
id=agent_id,
provider_data={
"status": "connected",
"environment": derive_agent_environment(agent),
},
)
async def create_execution(
self,
*,
user_id: IdLike,
deployment_type: DeploymentType | None = None, # noqa: ARG002
payload: ExecutionCreate,
db: AsyncSession,
) -> ExecutionCreateResult:
"""Create a provider-agnostic deployment execution."""
agent_id = _normalize_and_validate_id(str(payload.deployment_id), field_name="deployment_id")
clients = await self._get_provider_clients(user_id=user_id, db=db)
provider_data: dict = payload.provider_data or {}
try:
agent_run_result = await create_agent_run(
clients,
provider_data=provider_data,
deployment_id=agent_id,
)
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.CREATE_EXECUTION,
log_msg="Unexpected error creating wxO deployment execution",
pass_through=(AuthenticationError, AuthorizationError, DeploymentNotFoundError, InvalidContentError),
)
return ExecutionCreateResult(
execution_id=agent_run_result.get("run_id"),
deployment_id=agent_id,
provider_result=agent_run_result,
)
async def get_execution(
self,
*,
user_id: IdLike,
execution_id: IdLike,
deployment_type: DeploymentType | None = None, # noqa: ARG002
db: AsyncSession,
) -> ExecutionStatusResult:
"""Get provider-agnostic deployment execution state/output."""
run_id = _normalize_and_validate_id(str(execution_id), field_name="execution_id")
clients = await self._get_provider_clients(user_id=user_id, db=db)
try:
agent_run_result = await get_agent_run(
clients,
run_id=run_id,
)
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.GET_EXECUTION,
log_msg="Unexpected error fetching wxO deployment execution",
pass_through=(AuthenticationError, AuthorizationError, DeploymentNotFoundError, InvalidContentError),
)
return ExecutionStatusResult(
execution_id=run_id,
deployment_id=agent_run_result.get("agent_id"),
provider_result=agent_run_result,
)
async def list_configs(
self,
*,
user_id: IdLike,
params: ConfigListParams | None = None,
db: AsyncSession,
) -> ConfigListResult:
"""List configs visible to this adapter."""
agent_id = _require_single_deployment_id(params, resource_label="config")
clients = await self._get_provider_clients(user_id=user_id, db=db)
try:
agent = await asyncio.to_thread(clients.agent.get_draft_by_id, agent_id)
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.LIST_CONFIGS,
log_msg="Unexpected error while listing wxO deployment configs",
)
if not agent:
msg = f"Deployment '{agent_id}' not found."
raise DeploymentNotFoundError(msg)
tool_ids = agent.get("tools", []) if isinstance(agent, dict) else []
tool_ids = dedupe_list(tool_ids)
if not tool_ids:
return ConfigListResult(
configs=[],
provider_result={"deployment_id": agent_id, "tool_ids": []},
)
tools: list[dict]
try:
tools = await asyncio.to_thread(clients.tool.get_drafts_by_ids, tool_ids)
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.LIST_CONFIGS,
log_msg="Unexpected error while listing wxO tools for config extraction",
)
app_ids: set[str] = set()
for tool in tools or []:
if not isinstance(tool, dict):
continue
connections: dict = tool.get("binding", {}).get("langflow", {}).get("connections", {})
if not connections:
continue
app_ids.update(connections.keys())
return ConfigListResult(
configs=[ConfigListItem(id=app_id, name=app_id) for app_id in app_ids],
provider_result={"deployment_id": agent_id},
)
async def list_snapshots(
self,
*,
user_id: IdLike,
params: SnapshotListParams | None = None,
db: AsyncSession,
) -> SnapshotListResult:
"""List snapshots visible to this adapter."""
agent_id = _require_single_deployment_id(params, resource_label="snapshot")
clients = await self._get_provider_clients(user_id=user_id, db=db)
try:
agent = await asyncio.to_thread(clients.agent.get_draft_by_id, agent_id)
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.LIST,
log_msg="Unexpected error while listing wxO deployment snapshots",
)
if not agent or not isinstance(agent, dict):
msg = f"Deployment '{agent_id}' not found."
raise DeploymentNotFoundError(msg)
tools: list[dict] = []
requested_tool_ids = dedupe_list(agent.get("tools", []))
if requested_tool_ids:
try:
tools = await asyncio.to_thread(clients.tool.get_drafts_by_ids, requested_tool_ids)
except Exception as exc: # noqa: BLE001
raise_as_deployment_error(
exc,
error_prefix=ErrorPrefix.LIST,
log_msg="Unexpected error while listing wxO tools for snapshot extraction",
)
snapshots = [
SnapshotItem(
id=tool["id"],
name=tool.get("name") or tool["id"],
)
for tool in (tools or [])
if tool.get("id")
]
if not snapshots and requested_tool_ids:
snapshots = [SnapshotItem(id=tool_id, name=tool_id) for tool_id in requested_tool_ids]
return SnapshotListResult(
snapshots=snapshots,
provider_result={"deployment_id": agent_id},
)
async def _create_agent_deployment(
self,
*,
clients: WxOClient,
tool_ids: Sequence[str],
data: BaseDeploymentData,
) -> AgentUpsertResponse:
"""Create an agent deployment."""
payload = build_agent_payload(
data=data,
tool_ids=tool_ids,
)
return await asyncio.to_thread(clients.agent.create, payload)
async def teardown(self) -> None:
"""Teardown provider-specific resources."""

View File

@ -0,0 +1,81 @@
"""Dataclasses for the Watsonx Orchestrate adapter.
`WxOClient` eagerly creates SDK clients (`AgentClient`, `ToolClient`,
`ConnectionsClient`, and `BaseWXOClient`) at construction time to
guarantee thread safety when accessed from ``asyncio.to_thread`` workers.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from ibm_watsonx_orchestrate_clients.agents.agent_client import AgentClient
from ibm_watsonx_orchestrate_clients.common.base_client import BaseWXOClient
from ibm_watsonx_orchestrate_clients.connections.connections_client import ConnectionsClient
from ibm_watsonx_orchestrate_clients.tools.tool_client import ToolClient
if TYPE_CHECKING:
from ibm_cloud_sdk_core.authenticators import Authenticator
@dataclass(frozen=True, slots=True)
class WxOClient:
"""Provider client facade with eager SDK client initialization.
All sub-clients are constructed in ``__post_init__`` from ``instance_url``
and ``authenticator`` so that they are guaranteed to share the same
URL and authentication context. The dataclass is frozen to prevent
post-construction mutation of credentials.
"""
instance_url: str
authenticator: Authenticator
base: BaseWXOClient = field(init=False, repr=False)
tool: ToolClient = field(init=False, repr=False)
connections: ConnectionsClient = field(init=False, repr=False)
agent: AgentClient = field(init=False, repr=False)
def __post_init__(self) -> None:
url = self.instance_url.rstrip("/")
if not url:
msg = "instance_url must be a non-empty string."
raise ValueError(msg)
# Use object.__setattr__ because the dataclass is frozen.
object.__setattr__(self, "instance_url", url)
object.__setattr__(self, "base", BaseWXOClient(base_url=url, authenticator=self.authenticator))
object.__setattr__(self, "tool", ToolClient(base_url=url, authenticator=self.authenticator))
object.__setattr__(self, "connections", ConnectionsClient(base_url=url, authenticator=self.authenticator))
object.__setattr__(self, "agent", AgentClient(base_url=url, authenticator=self.authenticator))
# -- SDK private-method wrappers ------------------------------------------
# Centralise access to SDK-internal _get/_post so breakage from SDK
# upgrades is confined to this single file.
def get_agents_raw(self, params: dict[str, Any] | None = None) -> Any:
return self.base._get("/agents", params=params) # noqa: SLF001
def post_run(self, *, query_suffix: str = "", data: dict[str, Any]) -> Any:
return self.base._post(f"/runs{query_suffix}", data=data) # noqa: SLF001
def get_run(self, run_id: str) -> Any:
return self.base._get(f"/runs/{run_id}") # noqa: SLF001
def upload_tool_artifact(self, tool_id: str, *, files: dict[str, Any]) -> Any:
return self.base._post(f"/tools/{tool_id}/upload", files=files) # noqa: SLF001
@dataclass(frozen=True, slots=True)
class WxOCredentials:
instance_url: str
authenticator: Authenticator = field(repr=False)
def __post_init__(self) -> None:
if not self.instance_url or not self.instance_url.strip():
msg = "instance_url must be a non-empty string."
raise ValueError(msg)
def __repr__(self) -> str:
return (
f"WxOCredentials(instance_url={self.instance_url!r}, authenticator={self.authenticator.__class__.__name__})"
)

View File

@ -0,0 +1,495 @@
"""Helpers used to flatten wxO deployment update control flow."""
from __future__ import annotations
import asyncio
import copy
import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from fastapi import HTTPException, status
from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException
from lfx.services.adapters.deployment.exceptions import (
DeploymentConflictError,
InvalidContentError,
InvalidDeploymentOperationError,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix
from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import create_config, validate_connection
from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import (
delete_config_if_exists,
retry_create,
retry_rollback,
retry_update,
rollback_update_resources,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import (
FlowToolBindingSpec,
ToolUploadBatchError,
create_and_upload_wxo_flow_tools_with_bindings,
ensure_langflow_connections_binding,
to_writable_tool_payload,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import (
WatsonxBindOperation,
WatsonxConnectionRawPayload,
WatsonxDeploymentUpdatePayload,
WatsonxRemoveToolOperation,
WatsonxUnbindOperation,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.utils import (
dedupe_list,
extract_agent_tool_ids,
extract_error_detail,
validate_wxo_name,
)
if TYPE_CHECKING:
from collections.abc import Iterator
from lfx.services.adapters.deployment.schema import (
BaseDeploymentDataUpdate,
BaseFlowArtifact,
DeploymentUpdate,
IdLike,
)
from sqlalchemy.ext.asyncio import AsyncSession
from langflow.services.adapters.deployment.watsonx_orchestrate.types import WxOClient
logger = logging.getLogger(__name__)
class OrderedUniqueStrs:
"""Ordered, de-duplicating string collection used for deterministic update plans.
We intentionally avoid a plain list to avoid repeated O(n) membership checks.
Using a dictionary keeps membership/update checks O(1) while preserving insertion order.
Deterministic order is primarily for rollback correctness
Stable ordering ensures we undo the exact resources we created
in the correct (reverse) sequence.
"""
def __init__(self, items: dict[str, None] | None = None) -> None:
self._items: dict[str, None] = items or {}
@classmethod
def from_values(cls, values: list[str]) -> OrderedUniqueStrs:
ordered = cls()
ordered.extend(values)
return ordered
def __iter__(self) -> Iterator[str]:
return iter(self._items)
def to_list(self) -> list[str]:
# Snapshot keys only at list-boundary call sites.
return list(self._items)
def add(self, value: str) -> None:
self._items.setdefault(value, None)
def extend(self, values: list[str]) -> None:
for value in values:
self.add(value)
def discard(self, value: str) -> None:
self._items.pop(value, None)
class ToolConnectionOps:
def __init__(
self,
*,
bind: OrderedUniqueStrs | None = None,
unbind: OrderedUniqueStrs | None = None,
) -> None:
self.bind = bind or OrderedUniqueStrs()
self.unbind = unbind or OrderedUniqueStrs()
def _get_or_create_tool_connection_ops(
deltas: dict[str, ToolConnectionOps],
*,
tool_id: str,
) -> ToolConnectionOps:
return deltas.setdefault(tool_id, ToolConnectionOps())
@dataclass(slots=True)
class RawConnectionCreatePlan:
operation_app_id: str
provider_app_id: str
payload: WatsonxConnectionRawPayload
@dataclass(slots=True)
class RawToolCreatePlan:
raw_name: str
payload: BaseFlowArtifact
app_ids: list[str]
@dataclass(slots=True)
class ProviderUpdatePlan:
resource_prefix: str
existing_app_ids: list[str]
raw_connections_to_create: list[RawConnectionCreatePlan]
existing_tool_deltas: dict[str, ToolConnectionOps]
raw_tools_to_create: list[RawToolCreatePlan]
final_existing_tool_ids: list[str]
bind_existing_tool_ids: list[str]
def validate_provider_update_request_sections(payload: DeploymentUpdate) -> None:
"""Reject legacy top-level update sections in watsonx clean-break mode."""
if payload.snapshot is not None or payload.config is not None:
msg = (
"Top-level 'snapshot' and 'config' update sections are no longer supported for "
"watsonx Orchestrate deployment updates. Use provider_data.operations instead."
)
raise InvalidDeploymentOperationError(message=msg)
def build_provider_update_plan(
*,
agent: dict[str, Any],
provider_update: WatsonxDeploymentUpdatePayload,
) -> ProviderUpdatePlan:
"""Build a deterministic CPU-only plan for provider_data update operations."""
resource_prefix = (provider_update.resource_name_prefix or "").strip()
agent_tool_ids = extract_agent_tool_ids(agent)
final_existing_tool_ids = OrderedUniqueStrs.from_values(agent_tool_ids)
# Per existing tool_id, track app_ids to bind/unbind during this update.
existing_tool_deltas: dict[str, ToolConnectionOps] = {}
# Existing tool_ids explicitly referenced by bind operations (for added snapshot reporting).
bind_existing_tool_ids = OrderedUniqueStrs()
# Per raw tool name, collect app_ids that should be bound when the raw tool is created.
raw_tool_app_ids: dict[str, OrderedUniqueStrs] = {}
for operation in provider_update.operations:
if isinstance(operation, WatsonxBindOperation):
if operation.tool.reference_id is not None:
tool_id = operation.tool.reference_id
bind_existing_tool_ids.add(tool_id)
final_existing_tool_ids.add(tool_id)
delta = _get_or_create_tool_connection_ops(existing_tool_deltas, tool_id=tool_id)
delta.bind.extend(operation.app_ids)
continue
raw_name = str(operation.tool.name_of_raw)
raw_apps = raw_tool_app_ids.setdefault(raw_name, OrderedUniqueStrs())
raw_apps.extend(operation.app_ids)
continue
if isinstance(operation, WatsonxUnbindOperation):
tool_id = operation.tool_id
delta = _get_or_create_tool_connection_ops(existing_tool_deltas, tool_id=tool_id)
delta.unbind.extend(operation.app_ids)
continue
if isinstance(operation, WatsonxRemoveToolOperation):
final_existing_tool_ids.discard(operation.tool_id)
continue
raw_connections_to_create = [
RawConnectionCreatePlan(
operation_app_id=raw_payload.app_id,
provider_app_id=f"{resource_prefix}{raw_payload.app_id}",
payload=raw_payload,
)
for raw_payload in (provider_update.connections.raw_payloads or [])
]
raw_tool_pool = {raw_payload.name: raw_payload for raw_payload in (provider_update.tools.raw_payloads or [])}
raw_tools_to_create = [
RawToolCreatePlan(raw_name=raw_name, payload=raw_tool_pool[raw_name], app_ids=app_ids.to_list())
for raw_name, app_ids in raw_tool_app_ids.items()
]
return ProviderUpdatePlan(
resource_prefix=resource_prefix,
existing_app_ids=list(provider_update.connections.existing_app_ids or []),
raw_connections_to_create=raw_connections_to_create,
existing_tool_deltas=existing_tool_deltas,
raw_tools_to_create=raw_tools_to_create,
final_existing_tool_ids=final_existing_tool_ids.to_list(),
bind_existing_tool_ids=bind_existing_tool_ids.to_list(),
)
async def _create_update_connection_with_conflict_mapping(
*,
clients: WxOClient,
app_id: str,
payload: WatsonxConnectionRawPayload,
user_id: IdLike,
db: AsyncSession,
) -> str:
from lfx.services.adapters.deployment.schema import DeploymentConfig
config_payload = DeploymentConfig(
name=app_id,
description=None,
environment_variables=payload.environment_variables,
provider_config=payload.provider_config,
)
try:
return await retry_create(
create_config,
clients=clients,
config=config_payload,
user_id=user_id,
db=db,
)
except (ClientAPIException, HTTPException) as exc:
if isinstance(exc, ClientAPIException):
status_code = exc.response.status_code
error_detail = str(extract_error_detail(exc.response.text))
else:
status_code = exc.status_code
error_detail = str(extract_error_detail(str(exc.detail)))
is_conflict = status_code == status.HTTP_409_CONFLICT or "already exists" in error_detail.lower()
if is_conflict:
msg = f"{ErrorPrefix.UPDATE.value} error details: {error_detail}"
raise DeploymentConflictError(message=msg) from exc
raise
async def _update_existing_tool_connection_deltas(
*,
clients: WxOClient,
existing_tool_deltas: dict[str, ToolConnectionOps],
resolved_connections: dict[str, str],
original_tools: dict[str, dict[str, Any]],
) -> None:
if not existing_tool_deltas:
return
tool_ids = list(existing_tool_deltas.keys())
tools = await asyncio.to_thread(clients.tool.get_drafts_by_ids, tool_ids)
tool_by_id = {str(tool.get("id")): tool for tool in tools if isinstance(tool, dict) and tool.get("id")}
missing_tool_ids = [tool_id for tool_id in tool_ids if tool_id not in tool_by_id]
if missing_tool_ids:
missing_ids = ", ".join(missing_tool_ids)
msg = f"Snapshot tool(s) not found: {missing_ids}"
raise InvalidContentError(message=msg)
tool_updates: list[tuple[str, dict[str, Any]]] = []
for tool_id in tool_ids:
delta = existing_tool_deltas[tool_id]
original_tool = to_writable_tool_payload(tool_by_id[tool_id])
original_tools[tool_id] = original_tool
writable_tool = copy.deepcopy(original_tool)
connections = ensure_langflow_connections_binding(writable_tool)
for app_id in delta.unbind:
connections.pop(app_id, None)
for app_id in delta.bind:
connection_id = resolved_connections.get(app_id)
if not connection_id:
msg = f"No resolved connection id available for app_id '{app_id}'."
raise InvalidContentError(message=msg)
connections[app_id] = connection_id
tool_updates.append((tool_id, writable_tool))
await asyncio.gather(
*(
retry_update(asyncio.to_thread, clients.tool.update, tool_id, writable_tool)
for tool_id, writable_tool in tool_updates
)
)
async def _rollback_created_app_ids(
*,
clients: WxOClient,
created_app_ids: list[str],
) -> None:
for app_id in reversed(created_app_ids):
try:
await retry_rollback(delete_config_if_exists, clients, app_id=app_id)
except Exception:
logger.exception("Rollback failed for created app_id=%s — resource may be orphaned", app_id)
def _build_agent_rollback_payload(*, agent: dict[str, Any], final_update_payload: dict[str, Any]) -> dict[str, Any]:
rollback_payload: dict[str, Any] = {}
if "tools" in final_update_payload:
rollback_payload["tools"] = extract_agent_tool_ids(agent)
for update_field in ("name", "display_name", "description"):
if update_field in final_update_payload and update_field in agent:
rollback_payload[update_field] = agent[update_field]
return rollback_payload
async def _rollback_agent_update(
*,
clients: WxOClient,
agent_id: str,
rollback_agent_payload: dict[str, Any],
) -> None:
if not rollback_agent_payload:
return
try:
await retry_rollback(asyncio.to_thread, clients.agent.update, agent_id, rollback_agent_payload)
except Exception:
logger.exception("Rollback failed for agent_id=%s — resource may be orphaned", agent_id)
async def apply_provider_update_plan_with_rollback(
*,
clients: WxOClient,
user_id: IdLike,
db: AsyncSession,
agent_id: str,
agent: dict[str, Any],
update_payload: dict[str, Any],
plan: ProviderUpdatePlan,
) -> list[str]:
"""Apply provider_data update operations with rollback protection."""
# Rollback journals:
# - created_tool_ids: provider tool ids created during this update.
# - created_app_ids: provider app ids created during this update.
# - original_tools: writable pre-update payloads for mutated existing tools.
created_tool_ids: list[str] = []
created_app_ids: list[str] = []
original_tools: dict[str, dict[str, Any]] = {}
# Working state:
# - resolved_connections: app_id -> connection_id map used for bind/update calls.
# - added_snapshot_ids: snapshot/tool ids to return in update result.
# - final_update_payload: outbound agent patch payload (spec + tools).
# - rollback_agent_payload: best-effort restore payload for agent update rollback.
resolved_connections: dict[str, str] = {}
added_snapshot_ids: list[str] = []
final_update_payload = dict(update_payload)
rollback_agent_payload: dict[str, Any] = {}
try:
if plan.existing_app_ids:
existing_connections = await asyncio.gather(
*(
retry_create(validate_connection, clients.connections, app_id=app_id)
for app_id in plan.existing_app_ids
)
)
for app_id, connection in zip(plan.existing_app_ids, existing_connections, strict=True):
resolved_connections[app_id] = connection.connection_id
if plan.raw_connections_to_create:
created_connections = await asyncio.gather(
*(
_create_update_connection_with_conflict_mapping(
clients=clients,
app_id=create_plan.provider_app_id,
payload=create_plan.payload,
user_id=user_id,
db=db,
)
for create_plan in plan.raw_connections_to_create
)
)
created_app_ids.extend(created_connections)
validated_created_connections = await asyncio.gather(
*(
retry_create(
validate_connection,
clients.connections,
app_id=create_plan.provider_app_id,
)
for create_plan in plan.raw_connections_to_create
)
)
for create_plan, connection in zip(
plan.raw_connections_to_create, validated_created_connections, strict=True
):
resolved_connections[create_plan.operation_app_id] = connection.connection_id
if plan.raw_tools_to_create:
tool_bindings = [
FlowToolBindingSpec(
flow_payload=raw_plan.payload,
connections={app_id: resolved_connections[app_id] for app_id in raw_plan.app_ids},
)
for raw_plan in plan.raw_tools_to_create
]
try:
raw_create_results = await create_and_upload_wxo_flow_tools_with_bindings(
clients=clients,
tool_bindings=tool_bindings,
tool_name_prefix=plan.resource_prefix,
)
except ToolUploadBatchError as exc:
created_tool_ids.extend(exc.created_tool_ids)
added_snapshot_ids.extend(exc.created_tool_ids)
for i, err in enumerate(exc.errors):
logger.exception("Tool upload batch error [%d/%d]: %s", i + 1, len(exc.errors), err)
raise exc.errors[0] from exc
for raw_plan, created_tool_id in zip(plan.raw_tools_to_create, raw_create_results, strict=True):
tool_id = str(created_tool_id).strip()
if not tool_id:
msg = f"Failed to create tool for raw payload '{raw_plan.raw_name}'."
raise InvalidContentError(message=msg)
created_tool_ids.append(tool_id)
added_snapshot_ids.append(tool_id)
if plan.existing_tool_deltas:
await _update_existing_tool_connection_deltas(
clients=clients,
existing_tool_deltas=plan.existing_tool_deltas,
resolved_connections=resolved_connections,
original_tools=original_tools,
)
added_snapshot_ids.extend(plan.bind_existing_tool_ids)
final_tools = dedupe_list([*plan.final_existing_tool_ids, *created_tool_ids])
final_update_payload["tools"] = final_tools
rollback_agent_payload = _build_agent_rollback_payload(
agent=agent,
final_update_payload=final_update_payload,
)
if final_update_payload:
await retry_update(asyncio.to_thread, clients.agent.update, agent_id, final_update_payload)
except Exception:
await _rollback_agent_update(
clients=clients,
agent_id=agent_id,
rollback_agent_payload=rollback_agent_payload,
)
await rollback_update_resources(
clients=clients,
created_tool_ids=created_tool_ids,
created_app_id=None,
original_tools=original_tools,
)
await _rollback_created_app_ids(
clients=clients,
created_app_ids=created_app_ids,
)
raise
return dedupe_list(added_snapshot_ids)
def build_update_payload_from_spec(spec: BaseDeploymentDataUpdate | None) -> dict[str, Any]:
"""Build agent update payload from deployment spec updates."""
update_payload: dict[str, Any] = {}
if not spec:
return update_payload
spec_updates = spec.model_dump(exclude_unset=True)
if "name" in spec_updates:
update_payload.update(
{
"name": validate_wxo_name(spec_updates["name"]),
"display_name": spec_updates["name"],
}
)
if "description" in spec_updates:
update_payload["description"] = spec_updates["description"]
return update_payload

View File

@ -0,0 +1,207 @@
"""Name validation, error helpers, and misc utilities for the Watsonx Orchestrate adapter."""
from __future__ import annotations
import json
import logging
from typing import TYPE_CHECKING, Any, NoReturn
from fastapi import HTTPException
from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException
from lfx.services.adapters.deployment.exceptions import (
DeploymentError,
DeploymentServiceError,
InvalidContentError,
OperationNotSupportedError,
raise_for_status_and_detail,
)
from lfx.services.adapters.deployment.schema import _normalize_and_validate_id
from langflow.services.adapters.deployment.watsonx_orchestrate.constants import (
DEFAULT_WXO_AGENT_LLM,
WXO_SANITIZE_RE,
WXO_TRANSLATE,
ErrorPrefix,
)
if TYPE_CHECKING:
from collections.abc import Sequence
from lfx.services.adapters.deployment.schema import (
BaseDeploymentData,
ConfigListParams,
SnapshotListParams,
)
logger = logging.getLogger(__name__)
def normalize_wxo_name(s: str) -> str:
return WXO_SANITIZE_RE.sub("", s.translate(WXO_TRANSLATE))
def validate_wxo_name(name: str) -> str:
"""Normalize and validate a wxO resource name."""
normalized_name = normalize_wxo_name(str(name))
if not normalized_name:
msg = "Deployment name must include at least one alphanumeric character."
raise InvalidContentError(message=msg)
if not normalized_name[0].isalpha():
msg = "Deployment name must start with a letter."
raise InvalidContentError(message=msg)
return normalized_name
def resolve_resource_name_prefix(
*,
caller_prefix: str,
) -> str:
"""Validate and return the caller-supplied resource name prefix for WxO resource creation."""
if not isinstance(caller_prefix, str) or not caller_prefix.strip():
msg = "resource_name_prefix must be a non-empty string."
raise InvalidContentError(message=msg)
validated = normalize_wxo_name(caller_prefix)
if not validated:
msg = "resource_name_prefix must contain at least one alphanumeric character."
raise InvalidContentError(message=msg)
if not validated[0].isalpha():
msg = "resource_name_prefix must start with a letter."
raise InvalidContentError(message=msg)
return validated
def require_tool_id(tool_response: dict[str, Any]) -> str:
tool_id = tool_response.get("id")
if not tool_id:
msg = "wxO did not return a tool id for snapshot creation."
raise InvalidContentError(message=msg)
return tool_id
def dedupe_list(items: list[str]) -> list[str]:
return list(dict.fromkeys(items))
def normalize_and_dedupe_ids(values: list[Any] | None, *, field_name: str) -> list[str]:
"""Normalize id values to non-empty strings and dedupe while preserving order."""
if not values:
return []
return dedupe_list([_normalize_and_validate_id(str(value), field_name=field_name) for value in values])
def _require_single_deployment_id(
params: ConfigListParams | SnapshotListParams | None,
*,
resource_label: str,
) -> str:
deployment_ids = params.deployment_ids if params else None
if not deployment_ids:
msg = (
f"watsonx Orchestrate {resource_label} listing requires exactly one "
"deployment_id. Global listing is not supported by this adapter."
)
raise OperationNotSupportedError(message=msg)
if len(deployment_ids) != 1:
msg = (
f"watsonx Orchestrate {resource_label} listing currently supports "
"exactly one deployment_id and only deployment-scoped listing."
)
raise InvalidContentError(message=msg)
return _normalize_and_validate_id(str(deployment_ids[0]), field_name="deployment_id")
def extract_error_detail(response_text: str) -> str:
"""Extract a human-readable error detail from a ClientAPIException response.
The response body may contain a ``detail`` value that is a string, a dict
with a ``msg`` key, or a list of such dicts. This helper normalises all
three shapes into a single value suitable for inclusion in an error message.
"""
fallback = response_text or "<empty response body>"
try:
payload = json.loads(response_text)
except (TypeError, ValueError, json.JSONDecodeError):
return fallback
if not isinstance(payload, dict):
return fallback
detail = payload.get("detail")
if detail in (None, "", [], {}):
for field in ("message", "details", "error"):
detail = payload.get(field)
if detail not in (None, "", [], {}):
break
else:
return fallback
if isinstance(detail, list):
detail = detail[0] if detail else None
if isinstance(detail, dict):
detail = detail.get("msg") or detail
return str(detail) if detail not in (None, "", [], {}) else fallback
def _resolve_exc_detail(exc: ClientAPIException | HTTPException) -> str:
if isinstance(exc, ClientAPIException):
raw_text = getattr(exc.response, "text", "")
return extract_error_detail(raw_text)
return str(extract_error_detail(str(exc.detail)))
def _resolve_exc_status_code(exc: ClientAPIException | HTTPException) -> int | None:
if isinstance(exc, ClientAPIException):
return int(getattr(exc.response, "status_code", 0) or 0) or None
return int(exc.status_code)
def raise_as_deployment_error(
exc: Exception,
*,
error_prefix: ErrorPrefix,
log_msg: str,
pass_through: tuple[type[DeploymentServiceError], ...] = (),
) -> NoReturn:
if isinstance(exc, pass_through):
raise exc
if isinstance(exc, DeploymentServiceError):
logger.exception(log_msg)
msg = f"{error_prefix.value} Please check server logs for details."
raise DeploymentError(message=msg, error_code="deployment_error") from exc
if isinstance(exc, (ClientAPIException, HTTPException)):
status_code = _resolve_exc_status_code(exc)
detail = _resolve_exc_detail(exc)
raise_for_status_and_detail(
status_code=status_code,
detail=detail,
message_prefix=error_prefix.value,
)
logger.exception(log_msg)
msg = f"{error_prefix.value} Please check server logs for details."
raise DeploymentError(message=msg, error_code="deployment_error") from exc
def build_agent_payload(
*,
data: BaseDeploymentData,
tool_ids: Sequence[str],
) -> dict[str, Any]:
if data.provider_spec is None:
msg = "Deployment data must include provider_spec with a non-empty name and display_name."
raise InvalidContentError(message=msg)
return {
"name": data.provider_spec["name"],
"display_name": data.provider_spec["display_name"],
"description": str(data.description or "").strip() or f"Langflow deployment {data.name}",
"tools": list(tool_ids),
"style": "default",
# TODO: make configurable; the llm field is required by the wxO api
# but retrieving available llms requires an extra api request.
"llm": DEFAULT_WXO_AGENT_LLM,
}
def extract_agent_tool_ids(agent: dict[str, Any]) -> list[str]:
# Shape source:
# - SDK/API agent payload uses "tools" as list[str] in this adapter flow.
return [str(tool_id) for tool_id in agent.get("tools", []) if tool_id]

View File

@ -267,6 +267,27 @@ def register_all_service_factories() -> None:
service_manager.set_factory_registered()
def register_builtin_adapters() -> None:
"""Import built-in adapter modules so ``@register_adapter`` decorators fire.
Mirrors ``register_all_service_factories()`` for the adapter registry system.
Each import triggers the ``@register_adapter`` decorator at module scope,
registering the adapter class on the AdapterRegistry singleton.
TODO: Watsonx risks are documented here because registration is runtime-optional:
missing ``ibm_*`` modules should skip adapter registration, but broad
``ModuleNotFoundError`` handling can also hide internal import regressions.
Future deployment API routing must treat "provider exists but adapter is not
registered in this runtime" as an explicit, deterministic error path.
Keep direct adapter imports limited to guarded paths and maintain CI
coverage that confirms Watsonx tests run (not skip) in eligible environments.
"""
try:
import langflow.services.adapters.deployment.watsonx_orchestrate # noqa: F401
except ModuleNotFoundError as exc:
logger.info("Skipping Watsonx Orchestrate adapter registration: %s", exc)
async def initialize_services(*, fix_migration: bool = False) -> None:
"""Initialize all the services needed."""
from langflow.helpers.windows_postgres_helper import configure_windows_postgres_event_loop
@ -275,6 +296,7 @@ async def initialize_services(*, fix_migration: bool = False) -> None:
# Register all service factories first
register_all_service_factories()
register_builtin_adapters()
cache_service = get_service(ServiceType.CACHE_SERVICE, default=CacheServiceFactory())
# Test external cache connection

View File

@ -163,7 +163,16 @@ ignore_missing_imports = true
requires = ["hatchling"]
build-backend = "hatchling.build"
# TODO(WXO): Remove testpypi index and uv.sources once the ibm-watsonx-orchestrate
# packages publish their official release to PyPI (expected late March 2026).
[[tool.uv.index]]
name = "testpypi"
url = "https://test.pypi.org/simple"
explicit = true
[tool.uv.sources]
ibm-watsonx-orchestrate-core = [{ index = "testpypi" }]
ibm-watsonx-orchestrate-clients = [{ index = "testpypi" }]
[project.urls]
Repository = "https://github.com/langflow-ai/langflow"
@ -353,6 +362,16 @@ astrapy = ["astrapy>=2.1.0,<3.0.0"]
# Windows-specific
gassist = ["gassist>=0.0.1; sys_platform == 'win32'"]
# SDKs for IBM watsonx Orchestrate deployment adapter.
# TODO(WXO): Pin to stable versions once official PyPI releases land (late March 2026).
ibm-watsonx = [
"ibm-cloud-sdk-core~=3.24.4",
"ibm-watsonx-orchestrate-clients==2.3.2.dev5117; python_version >= '3.11'",
"ibm-watsonx-orchestrate-core==2.3.2.dev5117; python_version >= '3.11'",
"rich",
"PyYAML"
]
# Complete installation
complete = [
"langflow-base[couchbase]",
@ -466,6 +485,8 @@ complete = [
"langflow-base[astrapy]",
# Windows-specific
"langflow-base[gassist]",
# wxO deployment adapter
"langflow-base[ibm-watsonx]",
]
all = [

View File

@ -0,0 +1 @@
# This file marks the deployment tests directory as a package for linting.

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,249 @@
"""Unit tests for Watsonx deployment update provider schema."""
from __future__ import annotations
from uuid import UUID
import pytest
try:
import langflow.services.adapters.deployment.watsonx_orchestrate # noqa: F401
except ModuleNotFoundError:
pytest.skip(
"Skipping Watsonx deployment tests: optional IBM SDK dependencies not available.",
allow_module_level=True,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import WatsonxDeploymentUpdatePayload
from langflow.services.adapters.deployment.watsonx_orchestrate.service import WatsonxOrchestrateDeploymentService
from lfx.services.adapters.payload import AdapterPayloadValidationError
def _raw_tool(name: str, suffix: int) -> dict:
return {
"id": str(UUID(int=suffix)),
"name": name,
"description": "desc",
"data": {"nodes": [], "edges": []},
"tags": [],
"provider_data": {"project_id": "project-1"},
}
def _raw_connection(app_id: str) -> dict:
return {
"app_id": app_id,
"environment_variables": {
"API_KEY": {"source": "raw", "value": "secret"},
},
}
def test_payload_schema_slot_registered_for_deployment_update() -> None:
slot = WatsonxOrchestrateDeploymentService.payload_schemas
assert slot is not None
assert slot.deployment_update is not None
assert slot.deployment_update.adapter_model is WatsonxDeploymentUpdatePayload
def test_update_schema_accepts_raw_tool_pool_and_shared_connection_refs() -> None:
slot = WatsonxOrchestrateDeploymentService.payload_schemas
assert slot is not None
assert slot.deployment_update is not None
payload = {
"resource_name_prefix": "lf_pref_",
"tools": {
"existing_ids": ["tool-existing-1"],
"raw_payloads": [_raw_tool("tool-new-1", 1)],
},
"connections": {
"existing_app_ids": ["app-existing-1"],
"raw_payloads": [_raw_connection("app-new-1")],
},
"operations": [
{
"op": "bind",
"tool": {"name_of_raw": "tool-new-1"},
"app_ids": ["app-new-1"],
},
{
"op": "bind",
"tool": {"reference_id": "tool-existing-1"},
"app_ids": ["app-existing-1"],
},
],
}
applied = slot.deployment_update.apply(payload)
assert applied["operations"][0]["tool"]["name_of_raw"] == "tool-new-1"
assert applied["operations"][1]["tool"]["reference_id"] == "tool-existing-1"
def test_update_schema_rejects_prefixed_app_id_collisions() -> None:
with pytest.raises(AdapterPayloadValidationError) as exc:
WatsonxOrchestrateDeploymentService.payload_schemas.deployment_update.apply( # type: ignore[union-attr]
{
"resource_name_prefix": "lf_",
"tools": {"existing_ids": ["tool-existing-1"]},
"connections": {
"existing_app_ids": ["dup"],
"raw_payloads": [_raw_connection("dup")],
},
"operations": [
{
"op": "bind",
"tool": {"reference_id": "tool-existing-1"},
"app_ids": ["dup"],
}
],
}
)
assert "collides with raw app ids" in str(exc.value.error)
def test_update_schema_rejects_missing_raw_tool_reference() -> None:
with pytest.raises(AdapterPayloadValidationError) as exc:
WatsonxOrchestrateDeploymentService.payload_schemas.deployment_update.apply( # type: ignore[union-attr]
{
"connections": {
"existing_app_ids": ["app-existing-1"],
},
"operations": [
{
"op": "bind",
"tool": {"name_of_raw": "missing-tool"},
"app_ids": ["app-existing-1"],
}
],
}
)
assert "name_of_raw not found" in str(exc.value.error)
def test_update_schema_rejects_reference_id_when_existing_ids_not_declared() -> None:
with pytest.raises(AdapterPayloadValidationError) as exc:
WatsonxOrchestrateDeploymentService.payload_schemas.deployment_update.apply( # type: ignore[union-attr]
{
"connections": {"existing_app_ids": ["app-existing-1"]},
"operations": [
{
"op": "bind",
"tool": {"reference_id": "tool-existing-1"},
"app_ids": ["app-existing-1"],
}
],
}
)
assert "reference_id not found in tools.existing_ids" in str(exc.value.error)
def test_update_schema_rejects_bind_app_id_not_in_declared_pools() -> None:
with pytest.raises(AdapterPayloadValidationError) as exc:
WatsonxOrchestrateDeploymentService.payload_schemas.deployment_update.apply( # type: ignore[union-attr]
{
"tools": {
"raw_payloads": [_raw_tool("tool-new-1", 2)],
},
"connections": {
"existing_app_ids": ["app-existing-1"],
},
"operations": [
{
"op": "bind",
"tool": {"name_of_raw": "tool-new-1"},
"app_ids": ["app-not-declared"],
}
],
}
)
assert "operation app_ids must be declared in" in str(exc.value.error)
def test_update_schema_rejects_unused_existing_app_ids() -> None:
with pytest.raises(AdapterPayloadValidationError) as exc:
WatsonxOrchestrateDeploymentService.payload_schemas.deployment_update.apply( # type: ignore[union-attr]
{
"tools": {"existing_ids": ["tool-existing-1"]},
"connections": {"existing_app_ids": ["app-existing-1", "app-unused"]},
"operations": [
{
"op": "bind",
"tool": {"reference_id": "tool-existing-1"},
"app_ids": ["app-existing-1"],
}
],
}
)
assert "existing_app_ids contains ids not referenced by operations" in str(exc.value.error)
def test_update_schema_rejects_unused_raw_connection_app_ids() -> None:
with pytest.raises(AdapterPayloadValidationError) as exc:
WatsonxOrchestrateDeploymentService.payload_schemas.deployment_update.apply( # type: ignore[union-attr]
{
"resource_name_prefix": "lf_pref_",
"tools": {"raw_payloads": [_raw_tool("tool-new-1", 3)]},
"connections": {"raw_payloads": [_raw_connection("app-new-1"), _raw_connection("app-unused")]},
"operations": [
{
"op": "bind",
"tool": {"name_of_raw": "tool-new-1"},
"app_ids": ["app-new-1"],
}
],
}
)
assert "raw_payloads contains app_id values not referenced by operations" in str(exc.value.error)
def test_update_schema_dedupes_bind_app_ids() -> None:
slot = WatsonxOrchestrateDeploymentService.payload_schemas
assert slot is not None
assert slot.deployment_update is not None
payload = {
"tools": {"existing_ids": ["tool-existing-1"]},
"connections": {"existing_app_ids": ["app-existing-1", "app-existing-2"]},
"operations": [
{
"op": "bind",
"tool": {"reference_id": "tool-existing-1"},
"app_ids": ["app-existing-1", "app-existing-1", "app-existing-2"],
}
],
}
applied = slot.deployment_update.apply(payload)
assert applied["operations"][0]["app_ids"] == ["app-existing-1", "app-existing-2"]
def test_update_schema_requires_unbind_app_ids() -> None:
with pytest.raises(AdapterPayloadValidationError):
WatsonxOrchestrateDeploymentService.payload_schemas.deployment_update.apply( # type: ignore[union-attr]
{
"operations": [
{
"op": "unbind",
"tool_id": "tool-1",
}
]
}
)
def test_update_schema_rejects_unbind_raw_app_ids() -> None:
with pytest.raises(AdapterPayloadValidationError) as exc:
WatsonxOrchestrateDeploymentService.payload_schemas.deployment_update.apply( # type: ignore[union-attr]
{
"connections": {"raw_payloads": [_raw_connection("app-new-1")]},
"operations": [
{
"op": "unbind",
"tool_id": "tool-1",
"app_ids": ["app-new-1"],
}
],
}
)
assert "must reference connections.existing_app_ids only" in str(exc.value.error)

View File

@ -118539,6 +118539,6 @@
"num_components": 359,
"num_modules": 97
},
"sha256": "13f65a29f6c2ff9456219241d124739b669f2d58199107d961e515d1da6a73e4",
"version": "0.3.1"
"sha256": "594e0427eb317645ab26b635d9910dab39775c531fe06fe1e225f60f18e20a5e",
"version": "0.4.0"
}

View File

@ -6,12 +6,17 @@ regardless of adapter.
from __future__ import annotations
from typing import NoReturn
from fastapi import status
class DeploymentServiceError(Exception):
"""Root exception for all deployment service failures.
Both operational errors (:class:`DeploymentError`) and authentication
errors (:class:`AuthenticationError`) inherit from this class.
or authorization errors (:class:`AuthenticationError`, :class:`AuthorizationError`)
inherit from this class.
Catch this only when you truly want to handle *every* deployment-related
failure uniformly.
"""
@ -41,6 +46,18 @@ class AuthenticationError(DeploymentServiceError):
"""
class AuthorizationError(DeploymentServiceError):
"""Base exception for authorization failures (authenticated but forbidden).
Intentionally a sibling of :class:`DeploymentError`, not a child.
This ensures ``except DeploymentError`` will not accidentally swallow
authorization failures, allowing API layers to return status 403
separately from deployment errors (400/404/409/422).
Please ensure that the message does not leak sensitive information.
"""
class CredentialResolutionError(AuthenticationError):
"""Raised when credentials resolution fails.
@ -84,6 +101,27 @@ class DeploymentSupportError(DeploymentError):
super().__init__(message, error_code="unsupported_deployment_type", cause=cause)
class RateLimitError(DeploymentError):
"""Raised when provider/API rate limits deployment operations."""
def __init__(self, message: str = "Deployment operation rate limited", *, cause: Exception | None = None):
super().__init__(message, error_code="deployment_rate_limited", cause=cause)
class DeploymentTimeoutError(DeploymentError):
"""Raised when deployment operation times out."""
def __init__(self, message: str = "Deployment operation timed out", *, cause: Exception | None = None):
super().__init__(message, error_code="deployment_timeout", cause=cause)
class ServiceUnavailableError(DeploymentError):
"""Raised when upstream provider is temporarily unavailable."""
def __init__(self, message: str = "Deployment provider is unavailable", *, cause: Exception | None = None):
super().__init__(message, error_code="deployment_provider_unavailable", cause=cause)
class AuthSchemeError(AuthenticationError):
"""Raised when no matching authentication scheme was found.
@ -101,7 +139,25 @@ class InvalidDeploymentTypeError(DeploymentError):
super().__init__(message, error_code="invalid_deployment_type", cause=cause)
class DeploymentNotFoundError(DeploymentError):
class ResourceNotFoundError(DeploymentError):
"""Raised when a requested deployment-related resource is not found."""
def __init__(
self,
message: str | None = None,
*,
resource_id: str | None = None,
cause: Exception | None = None,
):
self.resource_id = resource_id
if message is None:
message = "Resource not found"
if resource_id:
message = f"Resource not found: {resource_id}"
super().__init__(message, error_code="resource_not_found", cause=cause)
class DeploymentNotFoundError(ResourceNotFoundError):
"""Raised when a deployment is not found."""
def __init__(
@ -116,7 +172,25 @@ class DeploymentNotFoundError(DeploymentError):
message = "Deployment not found"
if deployment_id:
message = f"Deployment not found: {deployment_id}"
super().__init__(message, error_code="deployment_not_found", cause=cause)
super().__init__(message=message, resource_id=deployment_id, cause=cause)
self.error_code = "deployment_not_found"
class OperationNotSupportedError(DeploymentError):
"""Raised when a deployment operation is not supported by the current adapter.
Use this for operations that are valid in the abstract contract but
are not implemented by the active adapter (e.g. ``redeploy`` or
``duplicate`` on a provider that does not support them).
"""
def __init__(
self,
message: str = "This operation is not supported by the current deployment adapter.",
*,
cause: Exception | None = None,
):
super().__init__(message, error_code="operation_not_supported", cause=cause)
class DeploymentNotConfiguredError(DeploymentError):
@ -142,3 +216,74 @@ class DeploymentNotConfiguredError(DeploymentError):
"Register one to enable deployment operations."
)
super().__init__(message, error_code="deployment_not_configured", cause=cause)
# lru+ttl cache?
def raise_for_status_and_detail(
*,
status_code: int | None,
detail: str,
message_prefix: str | None = None,
) -> NoReturn:
"""Raise domain-specific deployment exceptions based on HTTP-like status/detail."""
detail_text = str(detail)
detail_lower = detail_text.lower()
prefix = str(message_prefix or "").strip()
message = f"{prefix} error details: {detail_text}" if prefix else detail_text
if status_code == status.HTTP_401_UNAUTHORIZED:
raise AuthenticationError(message=message, error_code="authentication_error") from None
if status_code == status.HTTP_403_FORBIDDEN:
raise AuthorizationError(message=message, error_code="authorization_error") from None
if status_code == status.HTTP_422_UNPROCESSABLE_CONTENT:
raise InvalidContentError(message=message) from None
if status_code == status.HTTP_400_BAD_REQUEST:
raise InvalidDeploymentOperationError(message=message) from None
if status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
raise InvalidDeploymentOperationError(message=message) from None
if status_code == status.HTTP_413_REQUEST_ENTITY_TOO_LARGE:
raise InvalidContentError(message=message) from None
if status_code == status.HTTP_415_UNSUPPORTED_MEDIA_TYPE:
raise InvalidContentError(message=message) from None
if status_code == status.HTTP_404_NOT_FOUND:
raise DeploymentNotFoundError(message) from None
if status_code == status.HTTP_410_GONE:
raise DeploymentNotFoundError(message) from None
if status_code == status.HTTP_409_CONFLICT:
raise DeploymentConflictError(message=message) from None
if status_code == status.HTTP_429_TOO_MANY_REQUESTS:
raise RateLimitError(message=message) from None
if status_code in {status.HTTP_408_REQUEST_TIMEOUT, status.HTTP_504_GATEWAY_TIMEOUT}:
raise DeploymentTimeoutError(message=message) from None
if status_code in {status.HTTP_502_BAD_GATEWAY, status.HTTP_503_SERVICE_UNAVAILABLE}:
raise ServiceUnavailableError(message=message) from None
if "not found" in detail_lower:
raise DeploymentNotFoundError(message) from None
if "already exists" in detail_lower or "conflict" in detail_lower:
raise DeploymentConflictError(message=message) from None
if "unprocessable" in detail_lower:
raise InvalidContentError(message=message) from None
if "too many requests" in detail_lower or "rate limit" in detail_lower:
raise RateLimitError(message=message) from None
if "timed out" in detail_lower or "timeout" in detail_lower:
raise DeploymentTimeoutError(message=message) from None
if (
"service unavailable" in detail_lower
or "temporarily unavailable" in detail_lower
or "bad gateway" in detail_lower
):
raise ServiceUnavailableError(message=message) from None
if "unauthorized" in detail_lower or "authentication" in detail_lower:
raise AuthenticationError(message=message, error_code="authentication_error") from None
if "forbidden" in detail_lower or "permission" in detail_lower or "not allowed" in detail_lower:
raise AuthorizationError(message=message, error_code="authorization_error") from None
if "bad request" in detail_lower:
raise InvalidDeploymentOperationError(message=message) from None
if (
"invalid" in detail_lower
or "missing" in detail_lower
or "required" in detail_lower
or "malformed" in detail_lower
):
raise InvalidContentError(message=message) from None
raise DeploymentError(message=message, error_code="deployment_error") from None

View File

@ -10,14 +10,21 @@ from lfx.services.adapters.deployment import (
)
from lfx.services.adapters.deployment.exceptions import (
AuthenticationError,
AuthorizationError,
AuthSchemeError,
CredentialResolutionError,
DeploymentConflictError,
DeploymentNotFoundError,
DeploymentSupportError,
DeploymentTimeoutError,
InvalidContentError,
InvalidDeploymentOperationError,
InvalidDeploymentTypeError,
OperationNotSupportedError,
RateLimitError,
ResourceNotFoundError,
ServiceUnavailableError,
raise_for_status_and_detail,
)
from lfx.services.interfaces import DeploymentServiceProtocol
@ -26,10 +33,13 @@ def test_exception_hierarchy_is_preserved() -> None:
# DeploymentServiceError is the common root
assert issubclass(DeploymentError, DeploymentServiceError)
assert issubclass(AuthenticationError, DeploymentServiceError)
assert issubclass(AuthorizationError, DeploymentServiceError)
# AuthenticationError is a sibling of DeploymentError, NOT a child
assert not issubclass(AuthenticationError, DeploymentError)
assert not issubclass(AuthorizationError, DeploymentError)
assert not issubclass(DeploymentError, AuthenticationError)
assert not issubclass(DeploymentError, AuthorizationError)
# Auth subtypes
assert issubclass(CredentialResolutionError, AuthenticationError)
@ -39,18 +49,27 @@ def test_exception_hierarchy_is_preserved() -> None:
assert issubclass(DeploymentConflictError, DeploymentError)
assert issubclass(InvalidContentError, DeploymentError)
assert issubclass(InvalidDeploymentOperationError, DeploymentError)
assert issubclass(ResourceNotFoundError, DeploymentError)
assert issubclass(DeploymentNotFoundError, ResourceNotFoundError)
assert issubclass(DeploymentNotConfiguredError, DeploymentError)
assert issubclass(OperationNotSupportedError, DeploymentError)
def test_exception_error_codes_are_set() -> None:
assert AuthorizationError("forbidden", error_code="authorization_error").error_code == "authorization_error"
assert CredentialResolutionError().error_code == "credentials_resolution_error"
assert DeploymentConflictError().error_code == "deployment_conflict"
assert RateLimitError().error_code == "deployment_rate_limited"
assert DeploymentTimeoutError().error_code == "deployment_timeout"
assert ServiceUnavailableError().error_code == "deployment_provider_unavailable"
assert DeploymentSupportError().error_code == "unsupported_deployment_type"
assert ResourceNotFoundError().error_code == "resource_not_found"
assert InvalidDeploymentTypeError().error_code == "invalid_deployment_type"
assert InvalidContentError().error_code == "unprocessable_content_error"
assert InvalidDeploymentOperationError().error_code == "invalid_deployment_operation"
assert AuthSchemeError().error_code == "unsupported_auth_type"
assert DeploymentNotConfiguredError().error_code == "deployment_not_configured"
assert OperationNotSupportedError().error_code == "operation_not_supported"
def test_deployment_type_exceptions_have_distinct_default_messages() -> None:
@ -71,6 +90,13 @@ def test_deployment_not_found_default_message() -> None:
assert err.deployment_id is None
def test_resource_not_found_defaults() -> None:
err = ResourceNotFoundError(resource_id="r1")
assert str(err) == "Resource not found: r1"
assert err.resource_id == "r1"
assert err.error_code == "resource_not_found"
def test_deployment_not_found_preserves_deployment_id_with_custom_message() -> None:
err = DeploymentNotFoundError(message="Custom error", deployment_id="dep_1")
assert str(err) == "Custom error"
@ -156,6 +182,20 @@ def test_auth_errors_not_caught_by_deployment_error() -> None:
raise_auth_error()
def test_authorization_errors_not_caught_by_deployment_error() -> None:
"""Ensure except DeploymentError does NOT catch authorization failures."""
def raise_authorization_error() -> None:
error_message = "forbidden"
try:
raise AuthorizationError(error_message, error_code="authorization_error")
except DeploymentError:
pytest.fail("DeploymentError should not catch AuthorizationError")
with pytest.raises(AuthorizationError):
raise_authorization_error()
def test_deployment_service_error_catches_both_hierarchies() -> None:
"""DeploymentServiceError catches both deployment and auth errors."""
with pytest.raises(DeploymentServiceError):
@ -164,6 +204,52 @@ def test_deployment_service_error_catches_both_hierarchies() -> None:
raise DeploymentNotFoundError
def test_raise_for_status_and_detail_maps_known_http_statuses() -> None:
with pytest.raises(AuthenticationError):
raise_for_status_and_detail(status_code=401, detail="unauthorized", message_prefix="x")
with pytest.raises(AuthorizationError):
raise_for_status_and_detail(status_code=403, detail="forbidden", message_prefix="x")
with pytest.raises(DeploymentNotFoundError):
raise_for_status_and_detail(status_code=404, detail="missing", message_prefix="x")
with pytest.raises(DeploymentConflictError):
raise_for_status_and_detail(status_code=409, detail="conflict", message_prefix="x")
with pytest.raises(InvalidContentError):
raise_for_status_and_detail(status_code=422, detail="unprocessable", message_prefix="x")
with pytest.raises(InvalidDeploymentOperationError):
raise_for_status_and_detail(status_code=400, detail="bad request", message_prefix="x")
with pytest.raises(InvalidDeploymentOperationError):
raise_for_status_and_detail(status_code=405, detail="method not allowed", message_prefix="x")
with pytest.raises(InvalidContentError):
raise_for_status_and_detail(status_code=413, detail="payload too large", message_prefix="x")
with pytest.raises(InvalidContentError):
raise_for_status_and_detail(status_code=415, detail="unsupported media type", message_prefix="x")
with pytest.raises(DeploymentNotFoundError):
raise_for_status_and_detail(status_code=410, detail="gone", message_prefix="x")
with pytest.raises(RateLimitError):
raise_for_status_and_detail(status_code=429, detail="too many requests", message_prefix="x")
with pytest.raises(DeploymentTimeoutError):
raise_for_status_and_detail(status_code=408, detail="timeout", message_prefix="x")
with pytest.raises(DeploymentTimeoutError):
raise_for_status_and_detail(status_code=504, detail="gateway timeout", message_prefix="x")
with pytest.raises(ServiceUnavailableError):
raise_for_status_and_detail(status_code=502, detail="bad gateway", message_prefix="x")
with pytest.raises(ServiceUnavailableError):
raise_for_status_and_detail(status_code=503, detail="service unavailable", message_prefix="x")
def test_raise_for_status_and_detail_uses_detail_heuristics_without_status() -> None:
with pytest.raises(AuthorizationError):
raise_for_status_and_detail(status_code=None, detail="permission denied")
with pytest.raises(InvalidContentError):
raise_for_status_and_detail(status_code=None, detail="invalid payload")
with pytest.raises(RateLimitError):
raise_for_status_and_detail(status_code=None, detail="rate limit exceeded")
with pytest.raises(DeploymentTimeoutError):
raise_for_status_and_detail(status_code=None, detail="request timed out")
with pytest.raises(ServiceUnavailableError):
raise_for_status_and_detail(status_code=None, detail="service unavailable")
def test_package_exports_base_and_error() -> None:
from lfx.services.adapters.deployment import BaseDeploymentService, DeploymentError, DeploymentService

65
uv.lock generated
View File

@ -4240,6 +4240,21 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/c4/f7/5cc291d701094754a1d327b44d80a44971e13962881d9a400235726171da/hypothesis-6.151.9-py3-none-any.whl", hash = "sha256:7b7220585c67759b1b1ef839b1e6e9e3d82ed468cfc1ece43c67184848d7edd9", size = 529307, upload-time = "2026-02-16T22:59:20.443Z" },
]
[[package]]
name = "ibm-cloud-sdk-core"
version = "3.24.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pyjwt" },
{ name = "python-dateutil" },
{ name = "requests" },
{ name = "urllib3" },
]
sdist = { url = "https://files.pythonhosted.org/packages/4a/0c/965f4f4b34f1a9a2210bec224ac489c631153f18ef75c6e694750efda654/ibm_cloud_sdk_core-3.24.4.tar.gz", hash = "sha256:20498959ce0177a58938e1855ee09f08b9712e43cc3209e95010c046e8c6d2a6", size = 81778, upload-time = "2026-02-03T08:44:42.217Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5e/8d/90cf967468fd81f9fd22bd42dbf3b1fbb1234485d1f81b6f17c8de138fff/ibm_cloud_sdk_core-3.24.4-py3-none-any.whl", hash = "sha256:4c3487ed5eb5180770ea2d07d95cfed1e69002ed249bbbc7d155226985c0b785", size = 75786, upload-time = "2026-02-03T08:44:41.094Z" },
]
[[package]]
name = "ibm-cos-sdk"
version = "2.14.3"
@ -4333,6 +4348,31 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/99/89/f6a113d08bd8796c8ed82b151009479fe7e8a7b4792a12c98918b806c0e6/ibm_watsonx_ai-1.5.3-py3-none-any.whl", hash = "sha256:a7a1af3bebd8271e0fb7a04b17792131ad7f3e697cd4aea0aae0687695cb884f", size = 1080316, upload-time = "2026-02-26T09:18:01.778Z" },
]
[[package]]
name = "ibm-watsonx-orchestrate-clients"
version = "2.3.2.dev5117"
source = { registry = "https://test.pypi.org/simple" }
dependencies = [
{ name = "ibm-cloud-sdk-core", marker = "python_full_version >= '3.11'" },
{ name = "ibm-watsonx-orchestrate-core", marker = "python_full_version >= '3.11'" },
]
sdist = { url = "https://test-files.pythonhosted.org/packages/77/95/f202c3578199261afa69d525c7c8212fa2d328fe8ef189deac426685c6a4/ibm_watsonx_orchestrate_clients-2.3.2.dev5117.tar.gz", hash = "sha256:c0d1ce2831dc54ad578afd8fd6f586b160fc5ee9e6bfdfc0b7938c8a1ec0aa16", size = 1255, upload-time = "2026-02-11T13:47:57.775Z" }
wheels = [
{ url = "https://test-files.pythonhosted.org/packages/33/11/278075261cdaee932d4844a7ab0a422cdccb055c205b6a9043830d17c43c/ibm_watsonx_orchestrate_clients-2.3.2.dev5117-py3-none-any.whl", hash = "sha256:1248b6f0118043e4021fc9b759adc23a376e8ad806cc0f13f6aecb51b312914b", size = 22247, upload-time = "2026-02-11T13:47:52.677Z" },
]
[[package]]
name = "ibm-watsonx-orchestrate-core"
version = "2.3.2.dev5117"
source = { registry = "https://test.pypi.org/simple" }
dependencies = [
{ name = "pydantic", marker = "python_full_version >= '3.11'" },
]
sdist = { url = "https://test-files.pythonhosted.org/packages/f0/93/37d2a3b8374e0a8c9550c7b22574659ed6f871ebd9db1719c6952894ffac/ibm_watsonx_orchestrate_core-2.3.2.dev5117.tar.gz", hash = "sha256:4a2cdce1fc762839ce3502bbae2e3236b80bf53341622056b307b8137a5da4b9", size = 1195, upload-time = "2026-02-11T13:47:58.316Z" }
wheels = [
{ url = "https://test-files.pythonhosted.org/packages/79/9e/e6cea9daf819637807d722f3c00513dd34d31efe50481669928615298167/ibm_watsonx_orchestrate_core-2.3.2.dev5117-py3-none-any.whl", hash = "sha256:5bd64699ae3dad05683bffe2438e38dcffd2b58d2270893d5a4ac3f177b58339", size = 25175, upload-time = "2026-02-11T13:47:53.72Z" },
]
[[package]]
name = "identify"
version = "2.6.17"
@ -5873,6 +5913,9 @@ all = [
{ name = "google-search-results" },
{ name = "graph-retriever" },
{ name = "huggingface-hub", extra = ["inference"] },
{ name = "ibm-cloud-sdk-core" },
{ name = "ibm-watsonx-orchestrate-clients", marker = "python_full_version >= '3.11'" },
{ name = "ibm-watsonx-orchestrate-core", marker = "python_full_version >= '3.11'" },
{ name = "jigsawstack" },
{ name = "json-repair" },
{ name = "kubernetes" },
@ -5927,10 +5970,12 @@ all = [
{ name = "pypdf" },
{ name = "python-docx" },
{ name = "pytube" },
{ name = "pyyaml" },
{ name = "qdrant-client" },
{ name = "qianfan" },
{ name = "ragstack-ai-knowledge-store" },
{ name = "redis" },
{ name = "rich" },
{ name = "scrapegraph-py" },
{ name = "smolagents" },
{ name = "spider-client" },
@ -6039,6 +6084,9 @@ complete = [
{ name = "google-search-results" },
{ name = "graph-retriever" },
{ name = "huggingface-hub", extra = ["inference"] },
{ name = "ibm-cloud-sdk-core" },
{ name = "ibm-watsonx-orchestrate-clients", marker = "python_full_version >= '3.11'" },
{ name = "ibm-watsonx-orchestrate-core", marker = "python_full_version >= '3.11'" },
{ name = "jigsawstack" },
{ name = "json-repair" },
{ name = "kubernetes" },
@ -6093,10 +6141,12 @@ complete = [
{ name = "pypdf" },
{ name = "python-docx" },
{ name = "pytube" },
{ name = "pyyaml" },
{ name = "qdrant-client" },
{ name = "qianfan" },
{ name = "ragstack-ai-knowledge-store" },
{ name = "redis" },
{ name = "rich" },
{ name = "scrapegraph-py" },
{ name = "smolagents" },
{ name = "spider-client" },
@ -6190,6 +6240,13 @@ groq = [
huggingface = [
{ name = "huggingface-hub", extra = ["inference"] },
]
ibm-watsonx = [
{ name = "ibm-cloud-sdk-core" },
{ name = "ibm-watsonx-orchestrate-clients", marker = "python_full_version >= '3.11'" },
{ name = "ibm-watsonx-orchestrate-core", marker = "python_full_version >= '3.11'" },
{ name = "pyyaml" },
{ name = "rich" },
]
jigsawstack = [
{ name = "jigsawstack" },
]
@ -6490,7 +6547,10 @@ requires-dist = [
{ name = "gunicorn", specifier = ">=22.0.0,<23.0.0" },
{ name = "httpx", extras = ["http2"], specifier = ">=0.27,<1.0.0" },
{ name = "huggingface-hub", extras = ["inference"], marker = "extra == 'huggingface'", specifier = ">=0.23.2,<1.0.0" },
{ name = "ibm-cloud-sdk-core", marker = "extra == 'ibm-watsonx'", specifier = "~=3.24.4" },
{ name = "ibm-watsonx-ai", specifier = ">=1.3.1,<2.0.0" },
{ name = "ibm-watsonx-orchestrate-clients", marker = "python_full_version >= '3.11' and extra == 'ibm-watsonx'", specifier = "==2.3.2.dev5117", index = "https://test.pypi.org/simple" },
{ name = "ibm-watsonx-orchestrate-core", marker = "python_full_version >= '3.11' and extra == 'ibm-watsonx'", specifier = "==2.3.2.dev5117", index = "https://test.pypi.org/simple" },
{ name = "jaraco-context", specifier = ">=6.1.0" },
{ name = "jigsawstack", marker = "extra == 'jigsawstack'", specifier = "==0.2.7" },
{ name = "jq", marker = "sys_platform != 'win32'", specifier = ">=1.7.0,<2.0.0" },
@ -6569,6 +6629,7 @@ requires-dist = [
{ name = "langflow-base", extras = ["graph-retriever"], marker = "extra == 'complete'", editable = "src/backend/base" },
{ name = "langflow-base", extras = ["groq"], marker = "extra == 'complete'", editable = "src/backend/base" },
{ name = "langflow-base", extras = ["huggingface"], marker = "extra == 'complete'", editable = "src/backend/base" },
{ name = "langflow-base", extras = ["ibm-watsonx"], marker = "extra == 'complete'", editable = "src/backend/base" },
{ name = "langflow-base", extras = ["jigsawstack"], marker = "extra == 'complete'", editable = "src/backend/base" },
{ name = "langflow-base", extras = ["json-repair"], marker = "extra == 'complete'", editable = "src/backend/base" },
{ name = "langflow-base", extras = ["kubernetes"], marker = "extra == 'complete'", editable = "src/backend/base" },
@ -6685,11 +6746,13 @@ requires-dist = [
{ name = "python-docx", marker = "extra == 'docx'", specifier = ">=1.1.0" },
{ name = "python-multipart", specifier = ">=0.0.12,<1.0.0" },
{ name = "pytube", marker = "extra == 'pytube'", specifier = "==15.0.0" },
{ name = "pyyaml", marker = "extra == 'ibm-watsonx'" },
{ name = "qdrant-client", marker = "extra == 'qdrant'", specifier = "==1.9.2" },
{ name = "qianfan", marker = "extra == 'qianfan'", specifier = "==0.3.5" },
{ name = "ragstack-ai-knowledge-store", marker = "extra == 'ragstack'", specifier = "==0.2.1" },
{ name = "redis", marker = "extra == 'redis'", specifier = ">=5.2.1,<6.0.0" },
{ name = "rich", specifier = ">=13.7.0,<14.0.0" },
{ name = "rich", marker = "extra == 'ibm-watsonx'" },
{ name = "scipy", specifier = ">=1.15.2,<2.0.0" },
{ name = "scrapegraph-py", marker = "extra == 'scrapegraph'", specifier = ">=1.12.0" },
{ name = "sentence-transformers", marker = "extra == 'local'", specifier = ">=2.0.0" },
@ -6724,7 +6787,7 @@ requires-dist = [
{ name = "youtube-transcript-api", marker = "extra == 'youtube'", specifier = ">=1.0.0,<2.0.0" },
{ name = "zep-python", marker = "extra == 'zep'", specifier = "==2.0.2" },
]
provides-extras = ["audio", "postgresql", "local", "couchbase", "cassandra", "clickhouse", "mongodb", "supabase", "redis", "elasticsearch", "faiss", "qdrant", "weaviate", "chroma", "upstash", "pinecone", "milvus", "mongodb-vectors", "astradb", "openai", "anthropic", "cohere", "google", "ollama", "nvidia", "mistral", "sambanova", "groq", "llama-cpp", "sentence-transformers", "ctransformers", "langfuse", "langwatch", "langsmith", "arize", "pypdf", "docx", "pytube", "dspy", "smolagents", "beautifulsoup", "serpapi", "google-api", "wikipedia", "fake-useragent", "duckduckgo", "yfinance", "wolframalpha", "pyarrow", "fastavro", "gitpython", "nltk", "lark", "huggingface", "metaphor", "metal", "markup", "aws", "numexpr", "qianfan", "pgvector", "litellm", "zep", "youtube", "markdown", "kubernetes", "json-repair", "astra", "composio", "ragstack", "opensearch", "atlassian", "mem0", "needle", "sseclient", "openinference", "mcp", "ag2", "scrapegraph", "pydantic-ai", "apify", "spider", "altk", "langchain-huggingface", "langchain-unstructured", "graph-retriever", "langchain-mcp-adapters", "opik", "traceloop", "openlayer", "docling", "easyocr", "cleanlab", "twelvelabs", "jigsawstack", "assemblyai", "datasets", "fastparquet", "vlmrun", "cuga", "mlx", "fastmcp", "aioboto3", "astrapy", "gassist", "complete", "all"]
provides-extras = ["audio", "postgresql", "local", "couchbase", "cassandra", "clickhouse", "mongodb", "supabase", "redis", "elasticsearch", "faiss", "qdrant", "weaviate", "chroma", "upstash", "pinecone", "milvus", "mongodb-vectors", "astradb", "openai", "anthropic", "cohere", "google", "ollama", "nvidia", "mistral", "sambanova", "groq", "llama-cpp", "sentence-transformers", "ctransformers", "langfuse", "langwatch", "langsmith", "arize", "pypdf", "docx", "pytube", "dspy", "smolagents", "beautifulsoup", "serpapi", "google-api", "wikipedia", "fake-useragent", "duckduckgo", "yfinance", "wolframalpha", "pyarrow", "fastavro", "gitpython", "nltk", "lark", "huggingface", "metaphor", "metal", "markup", "aws", "numexpr", "qianfan", "pgvector", "litellm", "zep", "youtube", "markdown", "kubernetes", "json-repair", "astra", "composio", "ragstack", "opensearch", "atlassian", "mem0", "needle", "sseclient", "openinference", "mcp", "ag2", "scrapegraph", "pydantic-ai", "apify", "spider", "altk", "langchain-huggingface", "langchain-unstructured", "graph-retriever", "langchain-mcp-adapters", "opik", "traceloop", "openlayer", "docling", "easyocr", "cleanlab", "twelvelabs", "jigsawstack", "assemblyai", "datasets", "fastparquet", "vlmrun", "cuga", "mlx", "fastmcp", "aioboto3", "astrapy", "gassist", "ibm-watsonx", "complete", "all"]
[package.metadata.requires-dev]
dev = [