get_cluster_stats
Retrieve comprehensive statistics for OpenSearch clusters, including index health, shard allocation, and node performance, to monitor and manage cluster operations effectively.
Instructions
Get cluster statistics
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The handler function decorated with @mcp.tool that implements the logic for retrieving OpenSearch cluster statistics using es_client.cluster.stats() and returns formatted TextContent.@mcp.tool(description="Get cluster statistics") async def get_cluster_stats() -> list[TextContent]: """ Get statistics from a cluster wide perspective. The API returns basic index metrics (shard numbers, store size, memory usage) and information about the current nodes that form the cluster (number, roles, os, jvm versions, memory usage, cpu and installed plugins). https://opensearch.org/docs/latest/tuning-your-cluster/ """ self.logger.info("Getting cluster stats") try: response = self.es_client.cluster.stats() return [TextContent(type="text", text=str(response))] except Exception as e: self.logger.error(f"Error getting cluster stats: {e}") return [TextContent(type="text", text=f"Error: {str(e)}")]
- src/opensearch_mcp_server/server.py:25-41 (registration)The _register_tools method instantiates ClusterTools (line 30) and calls its register_tools method (line 38), which defines and registers the get_cluster_stats tool via the @mcp.tool decorator.def _register_tools(self): """Register all MCP tools.""" # Initialize tool classes index_tools = IndexTools(self.logger) document_tools = DocumentTools(self.logger) cluster_tools = ClusterTools(self.logger) dashboard_tools = DashboardTools(self.logger) admin_index_tools = AdminIndexTools(self.logger) admin_cluster_tools = AdminClusterTools(self.logger) # Register tools from each module index_tools.register_tools(self.mcp) document_tools.register_tools(self.mcp) cluster_tools.register_tools(self.mcp) dashboard_tools.register_tools(self.mcp) admin_index_tools.register_tools(self.mcp) admin_cluster_tools.register_tools(self.mcp)
- src/opensearch_mcp_server/server.py:6-6 (registration)Imports the ClusterTools class which contains the get_cluster_stats tool implementation.from .tools.cluster import ClusterTools
- The ClusterTools class that inherits from OpensearchClient and defines the register_tools method where get_cluster_stats is implemented and registered.class ClusterTools(OpensearchClient): def register_tools(self, mcp: Any): """Register cluster-related tools.""" @mcp.tool(description="Get cluster health status") async def get_cluster_health() -> list[TextContent]: """ Get health status of the Opensearch cluster. Returns information about the number of nodes, shards, etc. """ self.logger.info("Getting cluster health") try: response = self.es_client.cluster.health() return [TextContent(type="text", text=str(response))] except Exception as e: self.logger.error(f"Error getting cluster health: {e}") return [TextContent(type="text", text=f"Error: {str(e)}")] @mcp.tool(description="Get cluster statistics") async def get_cluster_stats() -> list[TextContent]: """ Get statistics from a cluster wide perspective. The API returns basic index metrics (shard numbers, store size, memory usage) and information about the current nodes that form the cluster (number, roles, os, jvm versions, memory usage, cpu and installed plugins). https://opensearch.org/docs/latest/tuning-your-cluster/ """ self.logger.info("Getting cluster stats") try: response = self.es_client.cluster.stats() return [TextContent(type="text", text=str(response))] except Exception as e: self.logger.error(f"Error getting cluster stats: {e}") return [TextContent(type="text", text=f"Error: {str(e)}")]