ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/saved_search_tools.py:166-283 (handler)

  The async run method that implements the core logic for retrieving a specific Log Analytics saved search by ID using the Azure LogAnalyticsManagementClient.

  async def run(self, ctx: Context, **kwargs):
      """
      Retrieve a specific saved search by ID from the specified Log Analytics workspace.
  
      Args:
          ctx (Context): The FastMCP context containing authentication and request information.
          **kwargs: Keyword arguments containing 'saved_search_id'.
  
      Returns:
          dict: Dictionary containing the saved search details and validity flag, or
          error information.
      """
      # Extract saved_search_id parameter using the
      # centralized parameter extraction from MCPToolBase
      saved_search_id = self._extract_param(kwargs, "saved_search_id")
  
      if not saved_search_id:
          return {"error": "saved_search_id parameter is required"}
  
      # Get Azure context
      workspace_name, resource_group, subscription_id = self.get_azure_context(ctx)
  
      # Validate Azure context
      sdk_available = True
      try:
          # Just check if the module is available
          import importlib.util  # pylint: disable=import-outside-toplevel
  
          sdk_available = (
              importlib.util.find_spec("azure.mgmt.loganalytics") is not None
          )
      except ImportError:
          sdk_available = False
  
      if not self.validate_azure_context(
          sdk_available, workspace_name, resource_group, subscription_id, self.logger
      ):
          return {"error": "Missing Azure SDK or workspace details."}
  
      # Get Log Analytics client
      client = None
      try:
          client = self.get_loganalytics_client(subscription_id)
      except Exception as e:
          self.logger.error("Error initializing Azure LogAnalytics client: %s", e)
          return {
              "error": "Azure LogAnalytics client initialization failed: %s" % str(e)
          }
  
      if client is None:
          return {"error": "Azure LogAnalytics client is not initialized"}
  
      try:
          # Get the specific saved search
          search = await run_in_thread(
              client.saved_searches.get,
              resource_group_name=resource_group,
              workspace_name=workspace_name,
              saved_search_id=saved_search_id,
          )
  
          # Log the search object to understand its structure
          self.logger.debug("Saved search object: %s", search)
  
          # Create a detailed info dictionary with all available attributes
          search_details = {
              "id": search.id if hasattr(search, "id") else None,
              "name": search.name if hasattr(search, "name") else None,
              "type": search.type if hasattr(search, "type") else None,
          }
  
          # Based on the log output, the properties are directly accessible
          # as attributes of the search object, not nested under properties
          properties_to_check = [
              "category",
              "display_name",
              "query",
              "function_alias",
              "function_parameters",
              "version",
              "tags",
              "etag",
              "time_created",
              "time_modified",
          ]
  
          # Check for each property and add it if it exists
          for prop_name in properties_to_check:
              if hasattr(search, prop_name):
                  value = getattr(search, prop_name)
                  if value is not None:
                      # Convert snake_case to camelCase for consistency in the output
                      key = "".join(
                          [
                              x.capitalize() if i > 0 else x
                              for i, x in enumerate(prop_name.split("_"))
                          ]
                      )
                      search_details[key] = value
  
          # Check for additional_properties if they exist
          if (
              hasattr(search, "additional_properties")
              and search.additional_properties
          ):
              for key, value in search.additional_properties.items():
                  if value is not None and key not in search_details:
                      search_details[key] = value
  
          return {"savedSearch": search_details, "valid": True}
      except Exception as e:
          self.logger.error(
              "Error retrieving saved search with ID %s: %s", saved_search_id, e
          )
          return {
              "error": "Error retrieving saved search ID %s: %s"
              % (saved_search_id, str(e))
          }

• tools/saved_search_tools.py:158-165 (schema)

  Tool class definition including name, description, and docstring which define the tool's schema and parameters (saved_search_id).

  class LogAnalyticsSavedSearchGetTool(MCPToolBase):
      """
      Tool to retrieve a specific saved search from a Log Analytics workspace.
      """
  
      name = "log_analytics_saved_search_get"
      description = "Get a specific saved search from a Log Analytics workspace"

• tools/saved_search_tools.py:286-295 (registration)

  The register_tools function that calls register on LogAnalyticsSavedSearchGetTool to register it with the FastMCP server.

  def register_tools(mcp: FastMCP):
      """
      Register Log Analytics saved search tools with the MCP server.
  
      Args:
          mcp (FastMCP): The FastMCP server instance to register tools with.
      """
      LogAnalyticsSavedSearchesListTool.register(mcp)
      LogAnalyticsSavedSearchGetTool.register(mcp)