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)