search_documents
Search documents in an OpenSearch index using custom queries to find specific information within your data.
Instructions
Search documents in an opensearch index with a custom query
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| index | Yes | ||
| body | Yes |
Implementation Reference
- The handler function that executes the search_documents tool. It queries the OpenSearch index with the given body, defaults size to 20 if not set, formats the response including hits and aggregations, and returns a list of TextContent.
async def search_documents(index: str, body: dict) -> list[TextContent]: """ Search documents in a specified opensearch index using a custom query. Args: index: Name of the index to search body: Opensearch query DSL. If size is not specified, defaults to 20 results. """ # Ensure reasonable default size limit is set if 'size' not in body: body['size'] = 20 self.logger.info(f"Searching in index: {index} with query: {body}") try: response = self.es_client.search(index=index, body=body) # Extract and format relevant information formatted_response = { 'total_hits': response['hits']['total']['value'], 'max_score': response['hits']['max_score'], 'hits': [] } # Process each hit for hit in response['hits']['hits']: hit_data = { '_id': hit['_id'], '_score': hit['_score'], 'source': hit['_source'] } formatted_response['hits'].append(hit_data) # Include aggregations if present if 'aggregations' in response: formatted_response['aggregations'] = response['aggregations'] return [TextContent(type="text", text=json.dumps(formatted_response, indent=2))] except Exception as e: self.logger.error(f"Error searching documents: {e}") return [TextContent(type="text", text=f"Error: {str(e)}")] - src/opensearch_mcp_server/server.py:37-37 (registration)Registers the DocumentTools instance (containing search_documents) with the MCP server by calling its register_tools method.
document_tools.register_tools(self.mcp) - src/opensearch_mcp_server/tools/document.py:11-11 (registration)The @mcp.tool decorator registers the search_documents function with the MCP server, providing the tool description.
@mcp.tool(description="Search documents in an opensearch index with a custom query") - Type hints (index: str, body: dict -> list[TextContent]) and docstring describing parameters serve as the tool schema for MCP.
async def search_documents(index: str, body: dict) -> list[TextContent]: """ Search documents in a specified opensearch index using a custom query. Args: index: Name of the index to search body: Opensearch query DSL. If size is not specified, defaults to 20 results. """