feat(agent): add quickForm escalation tool

Author: cotovanu-cristian

pyproject.toml

@@ -1,11 +1,11 @@
 [project]
 name = "uipath-langchain"
-version = "0.11.14"
+version = "0.11.15"
 description = "Python SDK that enables developers to build and deploy LangGraph agents to the UiPath Cloud Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"
 dependencies = [
-    "uipath>=2.10.74, <2.11.0",
+    "uipath>=2.10.79, <2.11.0",
     "uipath-core>=0.5.17, <0.6.0",
     "uipath-platform>=0.1.61, <0.2.0",
     "uipath-runtime>=0.11.0, <0.12.0",

src/uipath_langchain/agent/tools/escalation_tool.py

@@ -14,9 +14,11 @@
     AgentEscalationRecipient,
     AgentEscalationRecipientType,
     AgentEscalationResourceConfig,
+    AgentQuickFormEscalationChannel,
     ArgumentEmailRecipient,
     ArgumentGroupNameRecipient,
     AssetRecipient,
+    EscalationChannel,
     LowCodeAgentDefinition,
     StandardRecipient,
 )
@@ -37,7 +39,12 @@
     StructuredToolWithArgumentProperties,
 )
 
-from ..exceptions import AgentRuntimeError, AgentRuntimeErrorCode
+from ..exceptions import (
+    AgentRuntimeError,
+    AgentRuntimeErrorCode,
+    AgentStartupError,
+    AgentStartupErrorCode,
+)
 from ..react.types import AgentGraphState
 from .escalation_memory import (
     EscalationMemorySettings,
@@ -250,14 +257,79 @@ def _get_exported_trace_id(trace_id: str | None) -> str | None:
     return trace_id
 
 
+def _try_get_channel_app_name(channel: EscalationChannel) -> str | None:
+    return (
+        channel.properties.app_name
+        if isinstance(channel, AgentEscalationChannel)
+        else None
+    )
+
+
+async def create_task_for_channel(
+    client: UiPath,
+    channel: EscalationChannel,
+    *,
+    title: str,
+    data: dict[str, Any],
+    recipient: TaskRecipient | None,
+    folder_path: str | None,
+) -> Task:
+    """Create the human task backing an escalation channel."""
+    if isinstance(channel, AgentQuickFormEscalationChannel):
+        schema_id = channel.properties.schema_id
+        assert schema_id is not None
+        return await client.tasks.create_quickform_async(
+            title=title,
+            task_schema_key=schema_id,
+            schema=channel.properties.form_schema,
+            data=data,
+            folder_path=folder_path,
+            recipient=recipient,
+            priority=channel.priority,
+            labels=channel.labels,
+            is_actionable_message_enabled=channel.properties.is_actionable_message_enabled,
+            actionable_message_metadata=channel.properties.actionable_message_meta_data,
+        )
+    return await client.tasks.create_async(
+        title=title,
+        data=data,
+        app_name=channel.properties.app_name,
+        app_folder_path=folder_path,
+        recipient=recipient,
+        priority=channel.priority,
+        labels=channel.labels,
+        is_actionable_message_enabled=channel.properties.is_actionable_message_enabled,
+        actionable_message_metadata=channel.properties.actionable_message_meta_data,
+    )
+
+
+def _resolve_channel(resource: AgentEscalationResourceConfig) -> EscalationChannel:
+    """Return the escalation's channel, validating quick-form configuration."""
+    channel = resource.channels[0]
+    if (
+        isinstance(channel, AgentQuickFormEscalationChannel)
+        and channel.properties.schema_id is None
+    ):
+        raise AgentStartupError(
+            code=AgentStartupErrorCode.INVALID_TOOL_CONFIG,
+            title="Quick form escalation is missing a schema id",
+            detail=(
+                f"Escalation '{channel.name}' has a quick form "
+                "schema without a schemaId."
+            ),
+            category=UiPathErrorCategory.USER,
+        )
+    return channel
+
+
 def create_escalation_tool(
     resource: AgentEscalationResourceConfig,
     agent: LowCodeAgentDefinition | None = None,
 ) -> StructuredTool:
-    """Uses interrupt() for Action Center human-in-the-loop."""
+    """Build the human-in-the-loop escalation tool for an escalation resource."""
 
     tool_name: str = f"escalate_{sanitize_tool_name(resource.name)}"
-    channel: AgentEscalationChannel = resource.channels[0]
+    channel: EscalationChannel = _resolve_channel(resource)
 
     input_model: Any = create_model(channel.input_schema)
     output_model: Any = create_model(channel.output_schema)
@@ -327,16 +399,13 @@ async def escalate(**_tool_kwargs: Any):
             @durable_interrupt
             async def create_escalation_task():
                 client = UiPath()
-                created_task = await client.tasks.create_async(
+                created_task = await create_task_for_channel(
+                    client,
+                    channel,
                     title=task_title,
                     data=serialized_data,
-                    app_name=channel.properties.app_name,
-                    app_folder_path=folder_path,
                     recipient=recipient,
-                    priority=channel.priority,
-                    labels=channel.labels,
-                    is_actionable_message_enabled=channel.properties.is_actionable_message_enabled,
-                    actionable_message_metadata=channel.properties.actionable_message_meta_data,
+                    folder_path=folder_path,
                 )
 
                 if created_task.id is not None:
@@ -345,7 +414,7 @@ async def create_escalation_task():
                 return WaitEscalation(
                     action=created_task,
                     app_folder_path=folder_path,
-                    app_name=channel.properties.app_name,
+                    app_name=_try_get_channel_app_name(channel),
                     recipient=recipient,
                 )
 
@@ -487,7 +556,7 @@ async def escalation_wrapper(
         argument_properties=channel.argument_properties,
         metadata={
             "tool_type": "escalation",
-            "display_name": channel.properties.app_name,
+            "display_name": _try_get_channel_app_name(channel) or channel.name,
             "channel_type": channel.type,
             "recipient": None,
             "args_schema": input_model,

src/uipath_langchain/agent/tools/ixp_escalation_tool.py

@@ -6,7 +6,10 @@
 from langchain_core.messages import ToolCall
 from langchain_core.tools import StructuredTool
 from pydantic import BaseModel
-from uipath.agent.models.agent import AgentIxpVsEscalationResourceConfig
+from uipath.agent.models.agent import (
+    AgentEscalationChannel,
+    AgentIxpVsEscalationResourceConfig,
+)
 from uipath.eval.mocks import mockable
 from uipath.platform import UiPath
 from uipath.platform.common import WaitDocumentExtractionValidation
@@ -25,7 +28,12 @@
     ToolWrapperReturnType,
 )
 
-from ..exceptions import AgentRuntimeError, AgentRuntimeErrorCode
+from ..exceptions import (
+    AgentRuntimeError,
+    AgentRuntimeErrorCode,
+    AgentStartupError,
+    AgentStartupErrorCode,
+)
 from .structured_tool_with_output_type import StructuredToolWithOutputType
 from .utils import (
     resolve_task_title,
@@ -42,6 +50,24 @@ class EmptyInput(BaseModel):
     pass
 
 
+def _resolve_action_center_channel(
+    resource: AgentIxpVsEscalationResourceConfig,
+) -> AgentEscalationChannel:
+    """Return the VS escalation's Action Center channel, rejecting quick forms."""
+    channel = resource.channels[0]
+    if not isinstance(channel, AgentEscalationChannel):
+        raise AgentStartupError(
+            code=AgentStartupErrorCode.INVALID_TOOL_CONFIG,
+            title="Unsupported VS escalation channel",
+            detail=(
+                f"VS escalation '{resource.name}' must use an Action Center channel "
+                f"but received channel type '{channel.type.value}'."
+            ),
+            category=UiPathErrorCategory.USER,
+        )
+    return channel
+
+
 def create_ixp_escalation_tool(
     resource: AgentIxpVsEscalationResourceConfig,
 ) -> StructuredTool:
@@ -51,7 +77,7 @@ def create_ixp_escalation_tool(
     storage_bucket_folder_path: str = (
         resource.vs_escalation_properties.storage_bucket_folder_path
     )
-    channel = resource.channels[0]
+    channel = _resolve_action_center_channel(resource)
     action_priority = ActionPriority.from_str(channel.priority)
     ixp_tool_name: str = resource.vs_escalation_properties.ixp_tool_id