ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/watchlist_tools.py:203-292 (handler)

  The SentinelWatchlistItemsListTool class defines and implements the 'sentinel_watchlist_items_list' tool. It includes the name, description, and the async run() method that handles the tool execution logic: extracts watchlist_alias, gets Azure client, lists watchlist items using SecurityInsights SDK, processes and returns item details.

  class SentinelWatchlistItemsListTool(MCPToolBase):
      """
      Tool for listing all items in a specified Microsoft Sentinel watchlist.
      """
  
      name = "sentinel_watchlist_items_list"
      description = "List all items in a Sentinel watchlist"
  
      async def run(self, ctx: Context, **kwargs):
          logger = self.logger
  
          # Extract parameters using the base class method
          watchlist_alias = self._extract_param(kwargs, "watchlist_alias")
          if not watchlist_alias:
              return {"error": "watchlist_alias parameter is required"}
  
          # Get Azure context
          workspace_name, resource_group, subscription_id = self.get_azure_context(ctx)
  
          # Get security insights client
          client = None
          try:
              client = self.get_securityinsight_client(subscription_id)
          except Exception as e:
              logger.error("Error initializing Azure SecurityInsights client: %s", e)
              return {
                  "error": (
                      "Azure SecurityInsights client initialization failed: %s" % str(e)
                  )
              }
  
          if client is None:
              return {"error": "Azure SecurityInsights client is not initialized"}
  
          try:
              # List all items in the watchlist
              watchlist_items = await run_in_thread(
                  client.watchlist_items.list,
                  resource_group_name=resource_group,
                  workspace_name=workspace_name,
                  watchlist_alias=watchlist_alias,
              )
  
              result = []
              for item in watchlist_items:
                  # Log the item object to understand its structure
                  logger.debug("Watchlist item object: %s", item)
  
                  # Create a basic info dictionary with guaranteed attributes
                  item_info = {
                      "id": item.id if hasattr(item, "id") else None,
                      "name": item.name if hasattr(item, "name") else None,
                  }
  
                  # Try to access properties directly from the item object first
                  try:
                      # Check for direct properties on the item object
                      if hasattr(item, "items_key_value"):
                          item_info["itemsKeyValue"] = item.items_key_value
                      if hasattr(item, "properties") and isinstance(item.properties, dict):
                          item_info["properties"] = item.properties
                      
                      # If we couldn't find any direct properties, try the nested properties approach
                      if len(item_info) <= 2 and hasattr(item, "properties") and not isinstance(item.properties, dict):
                          props = item.properties
                          if hasattr(props, "items_key_value"):
                              item_info["itemsKeyValue"] = props.items_key_value
                          if hasattr(props, "properties"):
                              item_info["properties"] = props.properties
                  except Exception as prop_error:
                      # Log the property access error but continue with basic details
                      logger.error("Error accessing watchlist item properties: %s", prop_error)
  
                  result.append(item_info)
  
              return {
                  "watchlistItems": result,
                  "count": len(result),
                  "watchlistAlias": watchlist_alias,
                  "valid": True,
              }
          except Exception as e:
              logger.error(
                  "Error retrieving watchlist items for alias %s: %s", watchlist_alias, e
              )
              return {
                  "error": "Error retrieving watchlist items for alias %s: %s"
                  % (watchlist_alias, e)
              }

• tools/watchlist_tools.py:382-393 (registration)

  The register_tools function registers all watchlist-related tools with the MCP server, including a call to SentinelWatchlistItemsListTool.register(mcp) at line 391.

  def register_tools(mcp: FastMCP):
      """
      Register all Sentinel watchlist tools with the MCP server instance.
  
      Args:
          mcp (FastMCP): The MCP server instance to register tools with.
      """
      SentinelWatchlistsListTool.register(mcp)
      SentinelWatchlistGetTool.register(mcp)
      SentinelWatchlistItemsListTool.register(mcp)
      SentinelWatchlistItemGetTool.register(mcp)