Add gen_ai.caller.agent.type attribute for A2A tracing

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/invoke_agent_scope.py

@@ -11,6 +11,7 @@
     GEN_AI_CALLER_AGENT_ID_KEY,
     GEN_AI_CALLER_AGENT_NAME_KEY,
     GEN_AI_CALLER_AGENT_TENANT_ID_KEY,
+    GEN_AI_CALLER_AGENT_TYPE_KEY,
     GEN_AI_CALLER_AGENT_UPN_KEY,
     GEN_AI_CALLER_AGENT_USER_CLIENT_IP,
     GEN_AI_CALLER_AGENT_USER_ID_KEY,
@@ -138,6 +139,10 @@ def __init__(
         if caller_agent_details:
             self.set_tag_maybe(GEN_AI_CALLER_AGENT_NAME_KEY, caller_agent_details.agent_name)
             self.set_tag_maybe(GEN_AI_CALLER_AGENT_ID_KEY, caller_agent_details.agent_id)
+            self.set_tag_maybe(
+                GEN_AI_CALLER_AGENT_TYPE_KEY,
+                caller_agent_details.agent_type.value if caller_agent_details.agent_type else None,
+            )
             self.set_tag_maybe(
                 GEN_AI_CALLER_AGENT_APPLICATION_ID_KEY, caller_agent_details.agent_blueprint_id
             )

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/trace_processor/util.py

@@ -41,6 +41,7 @@
     # Caller Agent (A2A) attributes
     consts.GEN_AI_CALLER_AGENT_ID_KEY,  # gen_ai.caller.agent.id
     consts.GEN_AI_CALLER_AGENT_NAME_KEY,  # gen_ai.caller.agent.name
+    consts.GEN_AI_CALLER_AGENT_TYPE_KEY,  # gen_ai.caller.agent.type
     consts.GEN_AI_CALLER_AGENT_USER_ID_KEY,  # gen_ai.caller.agent.userid
     consts.GEN_AI_CALLER_AGENT_UPN_KEY,  # gen_ai.caller.agent.upn
     consts.GEN_AI_CALLER_AGENT_TENANT_ID_KEY,  # gen_ai.caller.agent.tenantid

tests/observability/core/test_invoke_agent_scope.py

@@ -21,12 +21,14 @@
 )
 from microsoft_agents_a365.observability.core.config import _telemetry_manager
 from microsoft_agents_a365.observability.core.constants import (
+    GEN_AI_CALLER_AGENT_TYPE_KEY,
     GEN_AI_CALLER_AGENT_USER_CLIENT_IP,
     GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY,
     GEN_AI_EXECUTION_SOURCE_NAME_KEY,
     GEN_AI_EXECUTION_TYPE_KEY,
     GEN_AI_INPUT_MESSAGES_KEY,
 )
+from microsoft_agents_a365.observability.core.models.agent_type import AgentType
 from microsoft_agents_a365.observability.core.models.caller_details import CallerDetails
 from microsoft_agents_a365.observability.core.opentelemetry_scope import OpenTelemetryScope
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
@@ -94,6 +96,7 @@ def setUpClass(cls):
             agent_upn="[email protected]",
             tenant_id="tenant-789",
             agent_client_ip="192.168.1.100",
+            agent_type=AgentType.DECLARATIVE_AGENT,
         )
 
     def setUp(self):
@@ -232,6 +235,39 @@ def test_caller_agent_client_ip_in_scope(self):
                     span_attributes[GEN_AI_CALLER_AGENT_USER_CLIENT_IP], "192.168.1.100"
                 )
 
+    def test_caller_agent_type_in_scope(self):
+        """Test that caller agent type is properly set when creating InvokeAgentScope."""
+        # Set up tracer to capture spans
+        span_exporter = InMemorySpanExporter()
+        tracer_provider = get_tracer_provider()
+        tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
+
+        # Create scope with caller agent details that include agent_type
+        scope = InvokeAgentScope.start(
+            invoke_agent_details=self.invoke_details,
+            tenant_details=self.tenant_details,
+            caller_agent_details=self.caller_agent_details,
+        )
+
+        # Verify scope was created and caller agent details contain the expected type
+        self.assertIsNotNone(scope)
+        self.assertEqual(self.caller_agent_details.agent_type, AgentType.DECLARATIVE_AGENT)
+        scope.dispose()
+
+        # Verify the agent type is set as a span attribute
+        finished_spans = span_exporter.get_finished_spans()
+        self.assertTrue(len(finished_spans) > 0, "Expected at least one span to be created")
+
+        span = finished_spans[-1]
+        span_attributes = getattr(span, "attributes", {}) or {}
+
+        # Verify the caller agent type is set as a span attribute
+        self.assertIn(GEN_AI_CALLER_AGENT_TYPE_KEY, span_attributes)
+        self.assertEqual(
+            span_attributes[GEN_AI_CALLER_AGENT_TYPE_KEY],
+            AgentType.DECLARATIVE_AGENT.value,
+        )
+
 
 if __name__ == "__main__":
     # Run pytest only on the current file