ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/watchlist_tools.py:294-379 (handler)

  The SentinelWatchlistItemGetTool class, which defines the tool and contains the 'run' method that implements the core logic for retrieving a specific watchlist item using the Azure SecurityInsights client's watchlist_items.get method.

  class SentinelWatchlistItemGetTool(MCPToolBase):
      """
      Tool for retrieving a specific item from a Microsoft Sentinel watchlist by alias
      and item ID.
      """
  
      name = "sentinel_watchlist_item_get"
      description = "Get a specific item from 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")
          watchlist_item_id = self._extract_param(kwargs, "watchlist_item_id")
  
          if not watchlist_alias:
              return {"error": "watchlist_alias parameter is required"}
          if not watchlist_item_id:
              return {"error": "watchlist_item_id parameter is required"}
  
          # Get Azure context and SecurityInsights client using MCPToolBase methods
          workspace_name, resource_group, subscription_id = self.get_azure_context(ctx)
          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:
              # Get the specific watchlist item
              item = await run_in_thread(
                  client.watchlist_items.get,
                  resource_group_name=resource_group,
                  workspace_name=workspace_name,
                  watchlist_alias=watchlist_alias,
                  watchlist_item_id=watchlist_item_id,
              )
  
              # Log the item object to understand its structure
              logger.debug("Watchlist item object: %s", item)
  
              # Create a basic info dictionary with guaranteed attributes
              item_details = {
                  "id": item.id if hasattr(item, "id") else None,
                  "name": item.name if hasattr(item, "name") else None,
                  "watchlistAlias": watchlist_alias,
              }
  
              # 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_details["itemsKeyValue"] = item.items_key_value
                  if hasattr(item, "properties") and isinstance(item.properties, dict):
                      item_details["properties"] = item.properties
                  
                  # If we couldn't find any direct properties, try the nested properties approach
                  if len(item_details) <= 3 and hasattr(item, "properties") and not isinstance(item.properties, dict):
                      props = item.properties
                      if hasattr(props, "items_key_value"):
                          item_details["itemsKeyValue"] = props.items_key_value
                      if hasattr(props, "properties"):
                          item_details["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)
  
              return {"watchlistItem": item_details, "valid": True}
          except Exception as e:
              logger.error(
                  "Error retrieving watchlist item for alias %s, item ID %s: %s",
                  watchlist_alias,
                  watchlist_item_id,
                  e,
              )
              return {
                  "error": "Error retrieving watchlist item for alias %s, item ID %s: %s"
                  % (watchlist_alias, watchlist_item_id, e)
              }

• tools/watchlist_tools.py:392-392 (registration)

  Registration of the SentinelWatchlistItemGetTool with the MCP server instance.

  SentinelWatchlistItemGetTool.register(mcp)

• tools/watchlist_tools.py:300-302 (schema)

  Tool name and description, which serve as the schema identifier and usage hint. Parameters (watchlist_alias, watchlist_item_id) are handled in the run method.

  name = "sentinel_watchlist_item_get"
  description = "Get a specific item from a Sentinel watchlist"