Typing activity span definition and new zero-code OTEL sample

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/telemetry/spans.py

@@ -11,7 +11,6 @@
     SimpleSpanWrapper,
     get_conversation_id,
 )
-from microsoft_agents.hosting.core.app._routes import _Route
 from . import constants, metrics
 
 
@@ -116,6 +115,16 @@ def _get_attributes(self) -> AttributeMap:
 class TypingSendTyping(SimpleSpanWrapper):
     """Span for the logic related to sending typing indicators in the TypingIndicator."""
 
-    def __init__(self):
+    def __init__(self, turn_context: TurnContextProtocol):
         """Initializes the TypingSendTyping SpanWrapper."""
         super().__init__(constants.SPAN_SEND_TYPING)
+        self._turn_context = turn_context
+
+    def _get_attributes(self) -> AttributeMap:
+        return {
+            attributes.ACTIVITY_CHANNEL_ID: self._turn_context.activity.channel_id
+            or attributes.UNKNOWN,
+            attributes.CONVERSATION_ID: get_conversation_id(
+                self._turn_context.activity
+            ),
+        }

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/typing_indicator.py

@@ -135,7 +135,7 @@ async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
 
     async def _send_typing(self) -> None:
         """Sends a single typing activity via the adapter, bypassing middleware."""
-        with spans.TypingSendTyping():
+        with spans.TypingSendTyping(self._context):
             ref = self._context.activity.get_conversation_reference()
             typing_activity = TurnContext.apply_conversation_reference(
                 Activity(type=ActivityTypes.typing), ref

tests/hosting_core/telemetry/test_app_spans.py

@@ -225,9 +225,22 @@ def test_app_download_files_no_attachments(test_exporter):
 
 
 def test_typing_send_typing_creates_span(test_exporter):
-    with TypingSendTyping():
+    ctx = _make_context()
+
+    with TypingSendTyping(ctx):
         pass
 
     spans = test_exporter.get_finished_spans()
     assert len(spans) == 1
     assert spans[0].name == constants.SPAN_SEND_TYPING
+
+
+def test_typing_send_typing_span_attributes(test_exporter):
+    ctx = _make_context()
+
+    with TypingSendTyping(ctx):
+        pass
+
+    span = test_exporter.get_finished_spans()[0]
+    assert span.attributes[attributes.ACTIVITY_CHANNEL_ID] == "msteams"
+    assert span.attributes[attributes.CONVERSATION_ID] == "conv-1"