introduce input/output spans and middleware registry

Author: nikhilc-microsoft

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/models/response.py

@@ -0,0 +1,12 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from dataclasses import dataclass
+
+
+@dataclass
+class Response:
+    """Response details from agent execution."""
+
+    messages: list[str]
+    """The list of response messages."""

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/spans_scopes/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/spans_scopes/input_scope.py

@@ -0,0 +1,80 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from ..agent_details import AgentDetails
+from ..constants import (
+    GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY,
+    GEN_AI_EXECUTION_SOURCE_NAME_KEY,
+    GEN_AI_EXECUTION_TYPE_KEY,
+    GEN_AI_INPUT_MESSAGES_KEY,
+)
+from ..opentelemetry_scope import OpenTelemetryScope
+from ..request import Request
+from ..tenant_details import TenantDetails
+from ..utils import safe_json_dumps
+
+INPUT_OPERATION_NAME = "input_messages"
+
+
+class InputScope(OpenTelemetryScope):
+    """Provides OpenTelemetry tracing scope for input messages."""
+
+    @staticmethod
+    def start(
+        agent_details: AgentDetails,
+        tenant_details: TenantDetails,
+        request: Request,
+    ) -> "InputScope":
+        """Creates and starts a new scope for input tracing.
+
+        Args:
+            agent_details: The details of the agent
+            tenant_details: The details of the tenant
+            request: The request details which invokes the agent
+
+        Returns:
+            A new InputScope instance
+        """
+        return InputScope(agent_details, tenant_details, request)
+
+    def __init__(
+        self,
+        agent_details: AgentDetails,
+        tenant_details: TenantDetails,
+        request: Request,
+    ):
+        """Initialize the input scope.
+
+        Args:
+            agent_details: The details of the agent
+            tenant_details: The details of the tenant
+            request: The request details which invokes the agent
+        """
+        super().__init__(
+            kind="Client",
+            operation_name=INPUT_OPERATION_NAME,
+            activity_name=(f"{INPUT_OPERATION_NAME} {agent_details.agent_id}"),
+            agent_details=agent_details,
+            tenant_details=tenant_details,
+        )
+
+        # Set request metadata
+        if request.source_metadata:
+            self.set_tag_maybe(GEN_AI_EXECUTION_SOURCE_NAME_KEY, request.source_metadata.name)
+            self.set_tag_maybe(
+                GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY, request.source_metadata.description
+            )
+
+        self.set_tag_maybe(
+            GEN_AI_EXECUTION_TYPE_KEY,
+            request.execution_type.value if request.execution_type else None,
+        )
+        self.set_tag_maybe(GEN_AI_INPUT_MESSAGES_KEY, safe_json_dumps([request.content]))
+
+    def record_input_messages(self, messages: list[str]) -> None:
+        """Records the input messages for telemetry tracking.
+
+        Args:
+            messages: List of input messages
+        """
+        self.set_tag_maybe(GEN_AI_INPUT_MESSAGES_KEY, safe_json_dumps(messages))

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/spans_scopes/output_scope.py

@@ -0,0 +1,65 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from ..agent_details import AgentDetails
+from ..constants import GEN_AI_OUTPUT_MESSAGES_KEY
+from ..models.response import Response
+from ..opentelemetry_scope import OpenTelemetryScope
+from ..tenant_details import TenantDetails
+from ..utils import safe_json_dumps
+
+OUTPUT_OPERATION_NAME = "output_messages"
+
+
+class OutputScope(OpenTelemetryScope):
+    """Provides OpenTelemetry tracing scope for output messages."""
+
+    @staticmethod
+    def start(
+        agent_details: AgentDetails,
+        tenant_details: TenantDetails,
+        response: Response,
+    ) -> "OutputScope":
+        """Creates and starts a new scope for output tracing.
+
+        Args:
+            agent_details: The details of the agent
+            tenant_details: The details of the tenant
+            response: The response details from the agent
+
+        Returns:
+            A new OutputScope instance
+        """
+        return OutputScope(agent_details, tenant_details, response)
+
+    def __init__(
+        self,
+        agent_details: AgentDetails,
+        tenant_details: TenantDetails,
+        response: Response,
+    ):
+        """Initialize the output scope.
+
+        Args:
+            agent_details: The details of the agent
+            tenant_details: The details of the tenant
+            response: The response details from the agent
+        """
+        super().__init__(
+            kind="Client",
+            operation_name=OUTPUT_OPERATION_NAME,
+            activity_name=(f"{OUTPUT_OPERATION_NAME} {agent_details.agent_id}"),
+            agent_details=agent_details,
+            tenant_details=tenant_details,
+        )
+
+        # Set response messages
+        self.set_tag_maybe(GEN_AI_OUTPUT_MESSAGES_KEY, safe_json_dumps(response.messages))
+
+    def record_output_messages(self, messages: list[str]) -> None:
+        """Records the output messages for telemetry tracking.
+
+        Args:
+            messages: List of output messages
+        """
+        self.set_tag_maybe(GEN_AI_OUTPUT_MESSAGES_KEY, safe_json_dumps(messages))

libraries/microsoft-agents-a365-observability-hosting/microsoft_agents_a365/observability/hosting/middleware/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.

libraries/microsoft-agents-a365-observability-hosting/microsoft_agents_a365/observability/hosting/middleware/message_logging_middleware.py

@@ -0,0 +1,100 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import logging
+from collections.abc import Awaitable, Callable
+
+from microsoft_agents.activity import Activity
+from microsoft_agents.hosting.core.middleware_set import Middleware, TurnContext
+from microsoft_agents_a365.observability.core.agent_details import AgentDetails
+from microsoft_agents_a365.observability.core.execution_type import ExecutionType
+from microsoft_agents_a365.observability.core.request import Request
+from microsoft_agents_a365.observability.core.spans_scopes.input_scope import InputScope
+from microsoft_agents_a365.observability.core.tenant_details import TenantDetails
+
+
+class MessageLoggingMiddleware(Middleware):
+    """
+    Lightweight middleware for logging input and output messages.
+    """
+
+    def __init__(
+        self,
+        logger: logging.Logger | None = None,
+        log_user_messages: bool = True,
+        log_bot_messages: bool = True,
+    ):
+        """
+        Initialize the message logger middleware.
+
+        Args:
+            logger: Custom logger instance (defaults to module logger)
+            log_user_messages: Whether to log incoming user messages
+            log_bot_messages: Whether to log outgoing bot messages
+        """
+        self.logger = logger or logging.getLogger("agents. observability")
+        self.log_user_messages = log_user_messages
+        self.log_bot_messages = log_bot_messages
+
+    async def on_turn(self, turn_context: TurnContext, logic: Callable[[TurnContext], Awaitable]):
+        input_scope = None
+
+        # Start InputScope for the entire turn if we have user message
+        if self.log_user_messages and turn_context.activity.text:
+            input_scope = self._create_input_scope(turn_context.activity)
+            input_scope.__enter__()
+            self.logger.info(f"📥 User:  {turn_context.activity.text}")
+
+        try:
+            # Hook into outgoing messages
+            if self.log_bot_messages:
+                turn_context.on_send_activities(self._create_send_handler())
+
+            # Execute bot logic
+            await logic()
+        except Exception as exc:
+            # Clean up and propagate exception (let __exit__ handle error recording)
+            if input_scope:
+                input_scope.__exit__(type(exc), exc, exc.__traceback__)
+                input_scope = None  # Prevent double cleanup
+            raise
+        finally:
+            # Clean up the input scope if not already done
+            if input_scope:
+                input_scope.__exit__(None, None, None)
+
+    def _create_input_scope(self, activity: Activity) -> InputScope:
+        """Create InputScope for tracing the entire turn"""
+        # Extract details from activity
+        agent_details = AgentDetails(
+            agent_id=activity.recipient.id if activity.recipient else "unknown",
+            agent_name=activity.recipient.name if activity.recipient else None,
+            conversation_id=activity.conversation.id if activity.conversation else None,
+        )
+
+        tenant_details = TenantDetails(
+            tenant_id=activity.conversation.tenant_id
+            if activity.conversation and hasattr(activity.conversation, "tenant_id")
+            else "unknown"
+        )
+
+        request = Request(
+            content=activity.text or "",
+            execution_type=ExecutionType.HUMAN_TO_AGENT,
+            session_id=activity.conversation.id if activity.conversation else None,
+        )
+
+        return InputScope.start(agent_details, tenant_details, request)
+
+    def _create_send_handler(self):
+        """Create handler for outgoing bot messages"""
+
+        async def send_handler(ctx, activities, next_send):
+            # Log each outgoing message
+            for activity in activities:
+                if activity.text:
+                    self.logger.info(f"📤 Bot: {activity.text}")
+
+            return await next_send()
+
+        return send_handler

libraries/microsoft-agents-a365-observability-hosting/microsoft_agents_a365/observability/hosting/middleware/observability_middleware_registrar.py

@@ -0,0 +1,54 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+
+from microsoft_agents.hosting.aiohttp import CloudAdapter
+
+from .message_logging_middleware import MessageLoggingMiddleware
+
+
+class ObservabilityMiddlewareRegistrar:
+    """
+    Registrar for configuring and registering observability middleware.
+
+    Usage:
+        # Quick start with defaults
+        ObservabilityMiddlewareRegistrar().with_message_logging().apply(adapter)
+
+    """
+
+    def __init__(self):
+        """Initialize the registrar."""
+        self._middleware_configs: list = []
+
+    def with_message_logging(
+        self,
+        log_user_messages: bool = True,
+        log_bot_messages: bool = True,
+    ) -> "ObservabilityMiddlewareRegistrar":
+        """Configure message logging middleware.
+
+        Args:
+            log_user_messages: Whether to log user messages (default: True)
+            log_bot_messages: Whether to log bot messages (default: True)
+
+        Returns:
+            The registrar instance for chaining
+        """
+        self._middleware_configs.append(
+            lambda: MessageLoggingMiddleware(
+                log_user_messages=log_user_messages,
+                log_bot_messages=log_bot_messages,
+            )
+        )
+        return self
+
+    def apply(self, adapter: CloudAdapter) -> None:
+        """Apply all configured middleware to the adapter.
+
+        Args:
+            adapter: CloudAdapter to register middleware with
+        """
+        for create_middleware in self._middleware_configs:
+            middleware = create_middleware()
+            adapter.use(middleware)