OpenSearch MCP Server

Instructions

Implementation Reference

• src/opensearch_mcp_server/tools/es_admin/admin_index.py:49-81 (handler)

  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)