Add API to send chat history to MCP platform for threat protection

libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/__init__.py

@@ -1,6 +1,9 @@
-# Copyright (c) Microsoft. All rights reserved.
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
 
 from .environment_utils import get_observability_authentication_scope
+from .operation_error import OperationError
+from .operation_result import OperationResult
 from .power_platform_api_discovery import ClusterCategory, PowerPlatformApiDiscovery
 from .utility import Utility
 
@@ -9,6 +12,8 @@
     "PowerPlatformApiDiscovery",
     "ClusterCategory",
     "Utility",
+    "OperationError",
+    "OperationResult",
 ]
 
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)

libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/operation_error.py

@@ -0,0 +1,58 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+Encapsulates an error from an operation.
+"""
+
+
+class OperationError:
+    """
+    Represents an error that occurred during an operation.
+
+    This class wraps an exception and provides a consistent interface for
+    accessing error information.
+    """
+
+    def __init__(self, exception: Exception):
+        """
+        Initialize a new instance of the OperationError class.
+
+        Args:
+            exception: The exception associated with the error.
+
+        Raises:
+            ValueError: If exception is None.
+        """
+        if exception is None:
+            raise ValueError("exception cannot be None")
+        self._exception = exception
+
+    @property
+    def exception(self) -> Exception:
+        """
+        Get the exception associated with the error.
+
+        Returns:
+            Exception: The exception associated with the error.
+        """
+        return self._exception
+
+    @property
+    def message(self) -> str:
+        """
+        Get the message associated with the error.
+
+        Returns:
+            str: The error message from the exception.
+        """
+        return str(self._exception)
+
+    def __str__(self) -> str:
+        """
+        Return a string representation of the error.
+
+        Returns:
+            str: A string representation of the error.
+        """
+        return str(self._exception)

libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/operation_result.py

@@ -0,0 +1,98 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+Represents the result of an operation.
+"""
+
+from typing import List, Optional
+
+from .operation_error import OperationError
+
+
+class OperationResult:
+    """
+    Represents the result of an operation.
+
+    This class encapsulates the success or failure state of an operation along
+    with any associated errors.
+    """
+
+    _success_instance: Optional["OperationResult"] = None
+
+    def __init__(self, succeeded: bool, errors: Optional[List[OperationError]] = None):
+        """
+        Initialize a new instance of the OperationResult class.
+
+        Args:
+            succeeded: Flag indicating whether the operation succeeded.
+            errors: Optional list of errors that occurred during the operation.
+        """
+        self._succeeded = succeeded
+        self._errors = errors if errors is not None else []
+
+    @property
+    def succeeded(self) -> bool:
+        """
+        Get a flag indicating whether the operation succeeded.
+
+        Returns:
+            bool: True if the operation succeeded, otherwise False.
+        """
+        return self._succeeded
+
+    @property
+    def errors(self) -> List[OperationError]:
+        """
+        Get the list of errors that occurred during the operation.
+
+        Note:
+            This property returns a defensive copy of the internal error list
+            to prevent external modifications, which is especially important for
+            protecting the singleton instance returned by success().
+
+        Returns:
+            List[OperationError]: A copy of the list of operation errors.
+        """
+        return list(self._errors)
+
+    @staticmethod
+    def success() -> "OperationResult":
+        """
+        Return an OperationResult indicating a successful operation.
+
+        Returns:
+            OperationResult: An OperationResult indicating a successful operation.
+        """
+        return OperationResult._success_instance
+
+    @staticmethod
+    def failed(*errors: OperationError) -> "OperationResult":
+        """
+        Create an OperationResult indicating a failed operation.
+
+        Args:
+            *errors: Variable number of OperationError instances.
+
+        Returns:
+            OperationResult: An OperationResult indicating a failed operation.
+        """
+        error_list = list(errors) if errors else []
+        return OperationResult(succeeded=False, errors=error_list)
+
+    def __str__(self) -> str:
+        """
+        Convert the value of the current OperationResult object to its string representation.
+
+        Returns:
+            str: A string representation of the current OperationResult object.
+        """
+        if self._succeeded:
+            return "Succeeded"
+        else:
+            error_messages = ", ".join(str(error.message) for error in self._errors)
+            return f"Failed: {error_messages}" if error_messages else "Failed"
+
+
+# Module-level eager initialization (thread-safe by Python's import lock)
+OperationResult._success_instance = OperationResult(succeeded=True)

libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/models/__init__.py

@@ -1,12 +1,15 @@
-# Copyright (c) Microsoft. All rights reserved.
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
 
 """
 Common models for MCP tooling.
 
 This module defines data models used across the MCP tooling framework.
 """
 
+from .chat_history_message import ChatHistoryMessage
+from .chat_message_request import ChatMessageRequest
 from .mcp_server_config import MCPServerConfig
 from .tool_options import ToolOptions
 
-__all__ = ["MCPServerConfig", "ToolOptions"]
+__all__ = ["MCPServerConfig", "ToolOptions", "ChatHistoryMessage", "ChatMessageRequest"]

libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/models/chat_history_message.py

@@ -0,0 +1,65 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+Chat History Message model.
+"""
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Dict
+
+
+@dataclass
+class ChatHistoryMessage:
+    """
+    Represents a single message in the chat history.
+
+    This class is used to send chat history to the MCP platform for real-time
+    threat protection analysis.
+    """
+
+    #: The unique identifier for the chat message.
+    id: str
+
+    #: The role of the message sender (e.g., "user", "assistant", "system").
+    role: str
+
+    #: The content of the chat message.
+    content: str
+
+    #: The timestamp of when the message was sent.
+    timestamp: datetime
+
+    def __post_init__(self):
+        """
+        Validate the message after initialization.
+
+        Ensures that all required fields are present and non-empty.
+
+        Raises:
+            ValueError: If id, role, or content is empty or whitespace-only,
+                        or if timestamp is None.
+        """
+        if not self.id or not self.id.strip():
+            raise ValueError("id cannot be empty or whitespace-only")
+        if not self.role or not self.role.strip():
+            raise ValueError("role cannot be empty or whitespace-only")
+        if not self.content or not self.content.strip():
+            raise ValueError("content cannot be empty or whitespace-only")
+        if self.timestamp is None:
+            raise ValueError("timestamp cannot be None")
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Convert the message to a dictionary for JSON serialization.
+
+        Returns:
+            Dict[str, Any]: Dictionary representation of the message.
+        """
+        return {
+            "id": self.id,
+            "role": self.role,
+            "content": self.content,
+            "timestamp": self.timestamp.isoformat(),
+        }

libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/models/chat_message_request.py

@@ -0,0 +1,66 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+Chat Message Request model.
+"""
+
+from dataclasses import dataclass
+from typing import Any, Dict, List
+
+from .chat_history_message import ChatHistoryMessage
+
+
+@dataclass
+class ChatMessageRequest:
+    """
+    Represents the request payload for a real-time threat protection check on a chat message.
+
+    This class encapsulates the information needed to send chat history to the MCP platform
+    for threat analysis.
+    """
+
+    #: The unique identifier for the conversation.
+    conversation_id: str
+
+    #: The unique identifier for the message within the conversation.
+    message_id: str
+
+    #: The content of the user's message.
+    user_message: str
+
+    #: The chat history messages.
+    chat_history: List[ChatHistoryMessage]
+
+    def __post_init__(self):
+        """
+        Validate the request after initialization.
+
+        Ensures that all required fields are present and non-empty.
+
+        Raises:
+            ValueError: If conversation_id, message_id, or user_message is empty
+                        or whitespace-only, or if chat_history is None or empty.
+        """
+        if not self.conversation_id or not self.conversation_id.strip():
+            raise ValueError("conversation_id cannot be empty")
+        if not self.message_id or not self.message_id.strip():
+            raise ValueError("message_id cannot be empty")
+        if not self.user_message or not self.user_message.strip():
+            raise ValueError("user_message cannot be empty")
+        if self.chat_history is None or len(self.chat_history) == 0:
+            raise ValueError("chat_history cannot be empty")
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Convert the request to a dictionary for JSON serialization.
+
+        Returns:
+            Dict[str, Any]: Dictionary representation of the request.
+        """
+        return {
+            "conversationId": self.conversation_id,
+            "messageId": self.message_id,
+            "userMessage": self.user_message,
+            "chatHistory": [msg.to_dict() for msg in self.chat_history],
+        }