get_shard_allocation
Retrieve shard allocation distribution for an OpenSearch index to monitor data distribution and optimize cluster performance.
Instructions
Get index shard allocation distribution
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| latest_index | Yes |
Implementation Reference
- The core handler function for the 'get_shard_allocation' tool. It queries the OpenSearch cluster's shard allocation using the '_cat/shards' endpoint, processes the data to count shards per node, and returns a formatted response as TextContent.
@mcp.tool(description="Get index shard allocation distribution") async def get_shard_allocation(latest_index: str) -> list[TextContent]: """ Get the current index shard allocation distribution across nodes. Returns index name, shard number, primary/replica status, and node assignment. This helps understand how shards are distributed across the cluster. Args: latest_index: The most recent index of interest. """ self.logger.info("Fetching shard allocation...") try: response = self.es_client.transport.perform_request( 'GET', '/_cat/shards', params={'h': 'index,shard,prirep,node', 'format': 'json'} ) # Count shards per node shard_counts = {} for shard in response: if shard['node'] not in shard_counts: shard_counts[shard['node']] = 0 shard_counts[shard['node']] += 1 # Format the response with both raw data and counts formatted_response = { 'shard_distribution': response, 'shards_per_node': shard_counts } return [TextContent(type="text", text=str(formatted_response))] except Exception as e: self.logger.error(f"Error fetching shard allocation: {e}") return [TextContent(type="text", text=f"Error: {str(e)}")] - src/opensearch_mcp_server/server.py:40-40 (registration)Registers all tools from AdminIndexTools, including 'get_shard_allocation', by calling its register_tools method on the MCP instance.
admin_index_tools.register_tools(self.mcp) - src/opensearch_mcp_server/server.py:32-32 (registration)Instantiates the AdminIndexTools class instance which provides the 'get_shard_allocation' tool.
admin_index_tools = AdminIndexTools(self.logger)