get_document
Retrieve a specific document from an Elasticsearch index using its unique ID to access stored data.
Instructions
Get a document by ID.
Args:
index: Name of the index
id: Document ID
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| index | Yes | ||
| id | Yes |
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)