feat(tools): add safety OTel span attribute helper

Author: DNKYr

tests/tools/safety/test_telemetry.py

@@ -0,0 +1,56 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Tests for safety telemetry integration."""
+
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+from trpc_agent_sdk.tools.safety._telemetry import set_safety_span_attrs
+from trpc_agent_sdk.tools.safety._types import Decision
+from trpc_agent_sdk.tools.safety._types import RiskLevel
+from trpc_agent_sdk.tools.safety._types import RiskType
+from trpc_agent_sdk.tools.safety._types import RuleFinding
+from trpc_agent_sdk.tools.safety._types import ScanReport
+
+
+class TestSetSafetySpanAttrs:
+    def test_span_attributes_set_on_active_span(self):
+        mock_span = MagicMock()
+        mock_span.is_recording.return_value = True
+
+        report = ScanReport(
+            decision=Decision.DENY,
+            risk_level=RiskLevel.CRITICAL,
+            findings=[
+                RuleFinding(
+                    rule_id="DANGEROUS_DELETE_001",
+                    risk_type=RiskType.DANGEROUS_FILE_OP,
+                    risk_level=RiskLevel.CRITICAL,
+                    evidence="rm -rf /",
+                    message="dangerous delete",
+                    recommendation="don't",
+                ),
+            ],
+            scan_duration_ms=15.0,
+        )
+
+        with patch("trpc_agent_sdk.tools.safety._telemetry.trace.get_current_span", return_value=mock_span):
+            set_safety_span_attrs(report)
+
+        mock_span.set_attribute.assert_any_call("tool.safety.decision", "deny")
+        mock_span.set_attribute.assert_any_call("tool.safety.risk_level", "critical")
+        mock_span.set_attribute.assert_any_call("tool.safety.scan_duration_ms", 15.0)
+
+    def test_no_crash_when_no_span(self):
+        mock_span = MagicMock()
+        mock_span.is_recording.return_value = False
+
+        report = ScanReport(decision=Decision.ALLOW)
+
+        with patch("trpc_agent_sdk.tools.safety._telemetry.trace.get_current_span", return_value=mock_span):
+            set_safety_span_attrs(report)
+
+        mock_span.set_attribute.assert_not_called()

trpc_agent_sdk/tools/safety/_telemetry.py

@@ -0,0 +1,39 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""OpenTelemetry integration for the Tool Script Safety Guard.
+
+Decorates existing tool execution spans with safety metadata.
+"""
+
+from __future__ import annotations
+
+from opentelemetry import trace
+
+from ._types import ScanReport
+
+
+def set_safety_span_attrs(report: ScanReport):
+    """Set safety-related span attributes on the current active span.
+
+    If no span is active or the span is not recording, this is a no-op.
+    Intended to be called from ToolSafetyFilter._before() so the attributes
+    appear on the tool execution span created by ToolsProcessor._execute_tool().
+
+    Attributes set:
+        tool.safety.decision: allow | deny | needs_human_review
+        tool.safety.risk_level: low | medium | high | critical | None
+        tool.safety.rule_ids: JSON array of triggered rule IDs
+        tool.safety.scan_duration_ms: float
+    """
+    span = trace.get_current_span()
+    if span is None or not span.is_recording():
+        return
+
+    span.set_attribute("tool.safety.decision", report.decision.value)
+    if report.risk_level:
+        span.set_attribute("tool.safety.risk_level", report.risk_level.value)
+    span.set_attribute("tool.safety.rule_ids", [f.rule_id for f in report.findings])
+    span.set_attribute("tool.safety.scan_duration_ms", report.scan_duration_ms)