es-search
Search documents in Elasticsearch 7.x indexes to retrieve data using queries, aggregations, and sorting capabilities.
Instructions
Search documents in Elasticsearch index
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| req | Yes |
Implementation Reference
- 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)