Elasticsearch 7.x MCP Server

Instructions

Implementation Reference

• src/elasticsearch7_mcp_server/server.py:58-101 (handler)

  The _handle_search method that implements the execution logic for the 'es-search' MCP tool. It validates the required 'index' parameter, constructs the search body from request params, performs the search using the Elasticsearch client, and returns the results or error.

  def _handle_search(self, req: Dict[str, Any]) -> Dict[str, Any]:
      """Search documents in Elasticsearch index.
      
      Expected request format:
      {
          "index": "index_name",
          "query": {
              "match": {"field": "value"}
          },
          "size": 10,  # optional
          "from": 0,   # optional
          "aggs": {    # optional
              "my_agg": {
                  "terms": {"field": "category"}
              }
          },
          # 支持所有其他Elasticsearch搜索参数
          "sort": [...],
          "highlight": {...},
          "_source": [...],
          ...
      }
      """
      try:
          # 验证请求参数
          if not req.get("index"):
              return {"success": False, "error": "Missing required parameter: index"}
          
          # 提取索引名称
          index = req["index"]
          
          # 构建搜索参数 - 保留所有可能的Elasticsearch搜索参数
          # 除了"index"之外的所有参数都传递给Elasticsearch
          search_params = {k: v for k, v in req.items() if k != "index"}
          
          # 记录搜索请求
          self.logger.info(f"Searching index '{index}' with params: {search_params}")
          
          # 执行搜索
          results = self.es_client.search(index=index, body=search_params)
          return {"success": True, "results": results}
      except Exception as e:
          self.logger.error(f"Error searching Elasticsearch: {str(e)}")
          return {"success": False, "error": str(e)}

• src/elasticsearch7_mcp_server/server.py:33-36 (registration)

  Registration of the 'es-search' tool in the FastMCP server using the @tool decorator, specifying the name and description, and binding it to the _handle_search handler function.

  self.server.tool(
      name="es-search", 
      description="Search documents in Elasticsearch index"
  )(self._handle_search)