Elasticsearch MCP Server

Instructions

        Get a document by ID.
        
        Args:
            index: Name of the index
            id: Document ID
        

Implementation Reference

• src/tools/document.py:33-42 (handler)

  The MCP tool handler for 'get_document'. It retrieves a document from the specified index by ID using the search client.

  @mcp.tool()
  def get_document(index: str, id: str) -> Dict:
      """
      Get a document by ID.
      
      Args:
          index: Name of the index
          id: Document ID
      """
      return self.search_client.get_document(index=index, id=id)

• src/clients/common/document.py:26-28 (helper)

  The underlying helper method in DocumentClient that performs the actual document retrieval using the base client.

  def get_document(self, index: str, id: str) -> Dict:
      """Get a document by ID."""
      return self.client.get(index=index, id=id)

• src/server.py:38-53 (registration)

  Top-level registration point where DocumentTools is included in the list of tool classes passed to ToolsRegister.register_all_tools.

  def _register_tools(self):
      """Register all MCP tools."""
      # Create a tools register
      register = ToolsRegister(self.logger, self.search_client, self.mcp)
      
      # Define all tool classes to register
      tool_classes = [
          IndexTools,
          DocumentTools,
          ClusterTools,
          AliasTools,
          DataStreamTools,
          GeneralTools,
      ]        
      # Register all tools
      register.register_all_tools(tool_classes)

• src/tools/register.py:33-50 (registration)

  Logic in ToolsRegister.register_all_tools that instantiates each tool class (including DocumentTools) and registers their tools via with_exception_handling, which calls register_tools.

  for tool_class in tool_classes:
      self.logger.info(f"Registering tools from {tool_class.__name__}")
      tool_instance = tool_class(self.search_client)
      
      # Set logger and client attributes
      tool_instance.logger = self.logger
      tool_instance.search_client = self.search_client
      
      # Check if risk management is enabled (high-risk operations are disabled)
      if risk_manager.high_risk_ops_disabled:
          # Add risk manager attributes for filtering
          tool_instance.risk_manager = risk_manager
          tool_instance.tool_class_name = tool_class.__name__
          # Register tools with risk filtering
          self._register_with_risk_filter(tool_instance)
      else:
          # Register tools with just exception handling (original way)
          with_exception_handling(tool_instance, self.mcp)