feat(guardrails): pass HITL recipient type in the node metadata for observability purposes

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-langchain"
-version = "0.5.33"
+version = "0.5.34"
 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"

src/uipath_langchain/agent/guardrails/actions/escalate_action.py

@@ -91,7 +91,11 @@ def action_node(
             "guardrail": guardrail,
             "scope": scope,
             "execution_stage": execution_stage,
-            "escalation_data": {},
+            "escalation_data": {
+                "recipient_type": self.recipient.type.value
+                if self.recipient and self.recipient.type
+                else None,
+            },
         }
 
         async def _node(

src/uipath_langchain/agent/guardrails/guardrail_nodes.py

@@ -149,6 +149,7 @@ def _create_guardrail_node(
     output_data_extractor: Callable[[AgentGuardrailsGraphState], dict[str, Any]]
     | None = None,
     tool_name: str | None = None,
+    tool_type: str | None = None,
 ) -> tuple[str, Callable[[AgentGuardrailsGraphState], Any]]:
     """Private factory for guardrail evaluation nodes.
 
@@ -161,6 +162,7 @@ def _create_guardrail_node(
 
     metadata: dict[str, Any] = {
         "tool_name": tool_name,
+        "tool_type": tool_type,
         "guardrail": guardrail,
         "scope": scope,
         "execution_stage": execution_stage,
@@ -309,6 +311,7 @@ def create_tool_guardrail_node(
     success_node: str,
     failure_node: str,
     tool_name: str,
+    tool_type: str | None = None,
 ) -> tuple[str, Callable[[AgentGuardrailsGraphState], Any]]:
     """Create a guardrail node for TOOL scope guardrails.
 
@@ -318,6 +321,7 @@ def create_tool_guardrail_node(
         success_node: Node to route to on validation pass.
         failure_node: Node to route to on validation fail.
         tool_name: Name of the tool to extract arguments from.
+        tool_type: Optional type of the tool (e.g., "process", "escalation", "mcp").
 
     Returns:
         A tuple of (node_name, node_function) for the guardrail evaluation node.
@@ -369,4 +373,5 @@ def _output_data_extractor(state: AgentGuardrailsGraphState) -> dict[str, Any]:
         _input_data_extractor,
         _output_data_extractor,
         tool_name,
+        tool_type,
     )

src/uipath_langchain/agent/react/guardrails/guardrails_subgraph.py

@@ -196,6 +196,7 @@ def _build_guardrail_node_chain(
         **fail_node_metadata,
         "action_type": action.action_type,
         "node_type": "guardrail_action",
+        "tool_type": guardrail_node_metadata.get("tool_type"),
     }
 
     subgraph.add_node(
@@ -254,6 +255,23 @@ def create_llm_guardrails_subgraph(
     )
 
 
+def _extract_tool_type(tool_node: RunnableCallable) -> str | None:
+    """Extract tool_type from a UiPathToolNode's underlying tool metadata.
+
+    Args:
+        tool_node: A RunnableCallable, potentially a UiPathToolNode with tool metadata.
+
+    Returns:
+        The tool_type string if available, otherwise None.
+    """
+    tool = getattr(tool_node, "tool", None)
+    if tool is not None:
+        metadata = getattr(tool, "metadata", None)
+        if isinstance(metadata, dict):
+            return metadata.get("tool_type")
+    return None
+
+
 def create_tools_guardrails_subgraph(
     tool_nodes: Mapping[str, RunnableCallable],
     guardrails: Sequence[tuple[BaseGuardrail, GuardrailAction]] | None,
@@ -271,10 +289,12 @@ def create_tools_guardrails_subgraph(
     """
     result: dict[str, RunnableCallable] = {}
     for tool_name, tool_node in tool_nodes.items():
+        tool_type = _extract_tool_type(tool_node)
         subgraph = create_tool_guardrails_subgraph(
             (tool_name, tool_node),
             guardrails,
             input_schema=input_schema,
+            tool_type=tool_type,
         )
         result[tool_name] = subgraph
 
@@ -378,13 +398,15 @@ def create_tool_guardrails_subgraph(
     tool_node: tuple[str, Any],
     guardrails: Sequence[tuple[BaseGuardrail, GuardrailAction]] | None,
     input_schema: type[BaseModel] | None = None,
+    tool_type: str | None = None,
 ):
     """Create a guarded tool node.
 
     Args:
         tool_node: Tuple of (tool_name, tool_node_callable).
         guardrails: Optional sequence of (guardrail, action) tuples.
         input_schema: Optional input schema to include in state.
+        tool_type: Optional type of the tool (e.g., "process", "escalation", "mcp").
 
     Returns:
         Either the original tool node callable (if no matching guardrails) or a compiled
@@ -406,6 +428,8 @@ def create_tool_guardrails_subgraph(
         guardrails=applicable_guardrails,
         scope=GuardrailScope.TOOL,
         execution_stages=[ExecutionStage.PRE_EXECUTION, ExecutionStage.POST_EXECUTION],
-        node_factory=partial(create_tool_guardrail_node, tool_name=tool_name),
+        node_factory=partial(
+            create_tool_guardrail_node, tool_name=tool_name, tool_type=tool_type
+        ),
         input_schema=input_schema,
     )

tests/agent/guardrails/test_guardrail_nodes.py

@@ -720,6 +720,7 @@ def test_llm_guardrail_node_metadata_fields(self):
         assert metadata["scope"] == "Llm"
         assert metadata["execution_stage"] == "preExecution"
         assert metadata["tool_name"] is None
+        assert metadata["tool_type"] is None
 
     def test_tool_guardrail_node_has_tool_name(self):
         """Test that TOOL scope guardrail has tool_name in metadata."""
@@ -739,6 +740,41 @@ def test_tool_guardrail_node_has_tool_name(self):
         assert metadata["scope"] == "Tool"
         assert metadata["tool_name"] == "my_tool"
 
+    def test_tool_guardrail_node_has_tool_type(self):
+        """Test that TOOL scope guardrail has tool_type in metadata."""
+        guardrail = MagicMock(spec=BuiltInValidatorGuardrail)
+        guardrail.name = "TestGuardrail"
+
+        _, node = create_tool_guardrail_node(
+            guardrail=guardrail,
+            execution_stage=ExecutionStage.PRE_EXECUTION,
+            success_node="ok",
+            failure_node="nope",
+            tool_name="my_tool",
+            tool_type="process",
+        )
+
+        metadata = getattr(node, "__metadata__", None)
+        assert metadata is not None
+        assert metadata["tool_type"] == "process"
+
+    def test_tool_guardrail_node_tool_type_defaults_to_none(self):
+        """Test that tool_type defaults to None when not provided."""
+        guardrail = MagicMock(spec=BuiltInValidatorGuardrail)
+        guardrail.name = "TestGuardrail"
+
+        _, node = create_tool_guardrail_node(
+            guardrail=guardrail,
+            execution_stage=ExecutionStage.PRE_EXECUTION,
+            success_node="ok",
+            failure_node="nope",
+            tool_name="my_tool",
+        )
+
+        metadata = getattr(node, "__metadata__", None)
+        assert metadata is not None
+        assert metadata["tool_type"] is None
+
     def test_agent_init_guardrail_node_metadata(self):
         """Test that AGENT init guardrail has correct scope in metadata."""
         guardrail = MagicMock(spec=BuiltInValidatorGuardrail)