feat: add observability metadata to tools and guardrails

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-langchain"
-version = "0.2.3"
+version = "0.2.4"
 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/guardrail_nodes.py

@@ -118,10 +118,11 @@ def _create_guardrail_node(
     | None = None,
     output_data_extractor: Callable[[AgentGuardrailsGraphState], dict[str, Any]]
     | None = None,
+    tool_name: str | None = None,
 ) -> tuple[str, Callable[[AgentGuardrailsGraphState], Any]]:
     """Private factory for guardrail evaluation nodes.
 
-    Returns a node that evaluates the guardrail and routes via Command:
+    Returns a node with observability metadata attached as __metadata__ attribute:
     - goto success_node on validation pass
     - goto failure_node on validation fail
     """
@@ -177,6 +178,15 @@ async def node(
             )
             raise
 
+    # Attach observability metadata as function attribute
+    node.__metadata__ = {  # type: ignore[attr-defined]
+        "guardrail_name": guardrail.name,
+        "guardrail_description": getattr(guardrail, "description", None),
+        "guardrail_scope": scope.value,
+        "guardrail_stage": execution_stage.value,
+        "tool_name": tool_name,
+    }
+
     return node_name, node
 
 
@@ -310,4 +320,5 @@ def _output_data_extractor(state: AgentGuardrailsGraphState) -> dict[str, Any]:
         failure_node,
         _input_data_extractor,
         _output_data_extractor,
+        tool_name,
     )

src/uipath_langchain/agent/tools/escalation_tool.py

@@ -108,6 +108,12 @@ async def escalation_tool_fn(
         description=resource.description,
         args_schema=input_model,
         coroutine=escalation_tool_fn,
+        metadata={
+            "tool_type": "escalation",
+            "display_name": channel.properties.app_name,
+            "channel_type": channel.type,
+            "assignee": assignee,
+        },
     )
 
     return tool

src/uipath_langchain/agent/tools/process_tool.py

@@ -46,6 +46,11 @@ async def process_tool_fn(**kwargs: Any):
         args_schema=input_model,
         coroutine=process_tool_fn,
         output_type=output_model,
+        metadata={
+            "tool_type": "process",
+            "display_name": process_name,
+            "folder_path": folder_path,
+        },
     )
 
     return tool

tests/agent/guardrails/test_guardrail_nodes.py

@@ -509,3 +509,78 @@ async def test_unsupported_guardrail_type_raises_error(self):
         assert "MagicMock" in error_message
         assert "DeterministicGuardrail" in error_message
         assert "BuiltInValidatorGuardrail" in error_message
+
+
+class TestGuardrailNodeMetadata:
+    """Tests for guardrail node __metadata__ attribute for observability."""
+
+    def test_llm_guardrail_node_has_metadata(self):
+        """Test that LLM guardrail node has __metadata__ attribute."""
+        guardrail = MagicMock(spec=BuiltInValidatorGuardrail)
+        guardrail.name = "TestGuardrail"
+        guardrail.description = "Test description"
+
+        _, node = create_llm_guardrail_node(
+            guardrail=guardrail,
+            execution_stage=ExecutionStage.PRE_EXECUTION,
+            success_node="ok",
+            failure_node="nope",
+        )
+
+        assert hasattr(node, "__metadata__")
+        assert isinstance(node.__metadata__, dict)
+
+    def test_llm_guardrail_node_metadata_fields(self):
+        """Test that LLM guardrail node has correct metadata fields."""
+        guardrail = MagicMock(spec=BuiltInValidatorGuardrail)
+        guardrail.name = "TestGuardrail"
+        guardrail.description = "Test description"
+
+        _, node = create_llm_guardrail_node(
+            guardrail=guardrail,
+            execution_stage=ExecutionStage.PRE_EXECUTION,
+            success_node="ok",
+            failure_node="nope",
+        )
+
+        metadata = getattr(node, "__metadata__", None)
+        assert metadata is not None
+        assert metadata["guardrail_name"] == "TestGuardrail"
+        assert metadata["guardrail_scope"] == "Llm"
+        assert metadata["guardrail_stage"] == "preExecution"
+        assert metadata["tool_name"] is None
+
+    def test_tool_guardrail_node_has_tool_name(self):
+        """Test that TOOL scope guardrail has tool_name 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",
+        )
+
+        metadata = getattr(node, "__metadata__", None)
+        assert metadata is not None
+        assert metadata["guardrail_scope"] == "Tool"
+        assert metadata["tool_name"] == "my_tool"
+
+    def test_agent_init_guardrail_node_metadata(self):
+        """Test that AGENT init guardrail has correct scope in metadata."""
+        guardrail = MagicMock(spec=BuiltInValidatorGuardrail)
+        guardrail.name = "TestGuardrail"
+
+        _, node = create_agent_init_guardrail_node(
+            guardrail=guardrail,
+            execution_stage=ExecutionStage.POST_EXECUTION,
+            success_node="ok",
+            failure_node="nope",
+        )
+
+        metadata = getattr(node, "__metadata__", None)
+        assert metadata is not None
+        assert metadata["guardrail_scope"] == "Agent"
+        assert metadata["guardrail_stage"] == "postExecution"

tests/agent/tools/test_escalation_tool.py

@@ -0,0 +1,106 @@
+"""Tests for escalation_tool.py metadata."""
+
+import pytest
+from uipath.agent.models.agent import (
+    AgentEscalationChannel,
+    AgentEscalationChannelProperties,
+    AgentEscalationRecipientType,
+    AgentEscalationResourceConfig,
+    StandardRecipient,
+)
+
+from uipath_langchain.agent.tools.escalation_tool import create_escalation_tool
+
+
+class TestEscalationToolMetadata:
+    """Test that escalation tool has correct metadata for observability."""
+
+    @pytest.fixture
+    def escalation_resource(self):
+        """Create a minimal escalation tool resource config."""
+        return AgentEscalationResourceConfig(
+            name="approval",
+            description="Request approval",
+            channels=[
+                AgentEscalationChannel(
+                    name="action_center",
+                    type="actionCenter",
+                    description="Action Center channel",
+                    input_schema={"type": "object", "properties": {}},
+                    output_schema={"type": "object", "properties": {}},
+                    properties=AgentEscalationChannelProperties(
+                        app_name="ApprovalApp",
+                        app_version=1,
+                        resource_key="test-key",
+                    ),
+                    recipients=[
+                        StandardRecipient(
+                            type=AgentEscalationRecipientType.USER_EMAIL,
+                            value="user@example.com",
+                        )
+                    ],
+                )
+            ],
+        )
+
+    @pytest.fixture
+    def escalation_resource_no_recipient(self):
+        """Create escalation resource without recipients."""
+        return AgentEscalationResourceConfig(
+            name="approval",
+            description="Request approval",
+            channels=[
+                AgentEscalationChannel(
+                    name="action_center",
+                    type="actionCenter",
+                    description="Action Center channel",
+                    input_schema={"type": "object", "properties": {}},
+                    output_schema={"type": "object", "properties": {}},
+                    properties=AgentEscalationChannelProperties(
+                        app_name="ApprovalApp",
+                        app_version=1,
+                        resource_key="test-key",
+                    ),
+                    recipients=[],
+                )
+            ],
+        )
+
+    def test_escalation_tool_has_metadata(self, escalation_resource):
+        """Test that escalation tool has metadata dict."""
+        tool = create_escalation_tool(escalation_resource)
+
+        assert tool.metadata is not None
+        assert isinstance(tool.metadata, dict)
+
+    def test_escalation_tool_metadata_has_tool_type(self, escalation_resource):
+        """Test that metadata contains tool_type for span detection."""
+        tool = create_escalation_tool(escalation_resource)
+        assert tool.metadata is not None
+        assert tool.metadata["tool_type"] == "escalation"
+
+    def test_escalation_tool_metadata_has_display_name(self, escalation_resource):
+        """Test that metadata contains display_name from app_name."""
+        tool = create_escalation_tool(escalation_resource)
+        assert tool.metadata is not None
+        assert tool.metadata["display_name"] == "ApprovalApp"
+
+    def test_escalation_tool_metadata_has_channel_type(self, escalation_resource):
+        """Test that metadata contains channel_type for span attributes."""
+        tool = create_escalation_tool(escalation_resource)
+        assert tool.metadata is not None
+        assert tool.metadata["channel_type"] == "actionCenter"
+
+    def test_escalation_tool_metadata_has_assignee(self, escalation_resource):
+        """Test that metadata contains assignee when recipient is USER_EMAIL."""
+        tool = create_escalation_tool(escalation_resource)
+        assert tool.metadata is not None
+        assert tool.metadata["assignee"] == "user@example.com"
+
+    def test_escalation_tool_metadata_assignee_none_when_no_recipients(
+        self, escalation_resource_no_recipient
+    ):
+        """Test that assignee is None when no recipients configured."""
+        tool = create_escalation_tool(escalation_resource_no_recipient)
+        assert tool.metadata is not None
+        assert tool.metadata["assignee"] is None

tests/agent/tools/test_process_tool.py

@@ -0,0 +1,54 @@
+"""Tests for process_tool.py metadata."""
+
+import pytest
+from uipath.agent.models.agent import (
+    AgentProcessToolProperties,
+    AgentProcessToolResourceConfig,
+    AgentToolType,
+)
+
+from uipath_langchain.agent.tools.process_tool import create_process_tool
+
+
+class TestProcessToolMetadata:
+    """Test that process tool has correct metadata for observability."""
+
+    @pytest.fixture
+    def process_resource(self):
+        """Create a minimal process tool resource config."""
+        return AgentProcessToolResourceConfig(
+            type=AgentToolType.PROCESS,
+            name="test_process",
+            description="Test process description",
+            input_schema={"type": "object", "properties": {}},
+            output_schema={"type": "object", "properties": {}},
+            properties=AgentProcessToolProperties(
+                process_name="MyProcess",
+                folder_path="/Shared/MyFolder",
+            ),
+        )
+
+    def test_process_tool_has_metadata(self, process_resource):
+        """Test that process tool has metadata dict."""
+        tool = create_process_tool(process_resource)
+
+        assert tool.metadata is not None
+        assert isinstance(tool.metadata, dict)
+
+    def test_process_tool_metadata_has_tool_type(self, process_resource):
+        """Test that metadata contains tool_type for span detection."""
+        tool = create_process_tool(process_resource)
+        assert tool.metadata is not None
+        assert tool.metadata["tool_type"] == "process"
+
+    def test_process_tool_metadata_has_display_name(self, process_resource):
+        """Test that metadata contains display_name from process_name."""
+        tool = create_process_tool(process_resource)
+        assert tool.metadata is not None
+        assert tool.metadata["display_name"] == "MyProcess"
+
+    def test_process_tool_metadata_has_folder_path(self, process_resource):
+        """Test that metadata contains folder_path for span attributes."""
+        tool = create_process_tool(process_resource)
+        assert tool.metadata is not None
+        assert tool.metadata["folder_path"] == "/Shared/MyFolder"