Elasticsearch MCP Server

Instructions

        Creates or updates a document in the index.
        
        Args:
            index: Name of the index
            document: Document data
            id: Optional document ID
        

Implementation Reference

• src/tools/document.py:21-31 (handler)

  The core handler function for the 'index_document' MCP tool. It is decorated with @mcp.tool() and delegates the indexing operation to the search client.

  @mcp.tool()
  def index_document(index: str, document: Dict, id: Optional[str] = None) -> Dict:
      """
      Creates or updates a document in the index.
      
      Args:
          index: Name of the index
          document: Document data
          id: Optional document ID
      """
      return self.search_client.index_document(index=index, id=id, document=document)

• src/clients/common/document.py:10-24 (helper)

  Supporting method in DocumentClient that performs the actual document indexing, handling differences between Elasticsearch and OpenSearch clients.

  def index_document(self, index: str, document: Dict, id: Optional[str] = None) -> Dict:
      """Creates a new document in the index."""
      # Handle parameter name differences between Elasticsearch and OpenSearch
      if self.engine_type == "elasticsearch":
          # For Elasticsearch: index(index, document, id=None, ...)
          if id is not None:
              return self.client.index(index=index, document=document, id=id)
          else:
              return self.client.index(index=index, document=document)
      else:
          # For OpenSearch: index(index, body, id=None, ...)
          if id is not None:
              return self.client.index(index=index, body=document, id=id)
          else:
              return self.client.index(index=index, body=document)

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

  Registration point where DocumentTools (containing index_document) is included in the list of tool classes and registered via ToolsRegister.register_all_tools.

  # Define all tool classes to register
  tool_classes = [
      IndexTools,
      DocumentTools,
      ClusterTools,
      AliasTools,
      DataStreamTools,
      GeneralTools,
  ]        
  # Register all tools
  register.register_all_tools(tool_classes)