get_table_indexes
Retrieve index details for a specified table in Apache Doris databases using the Doris MCP Server. Provide the table name and optional database name to access relevant index information.
Instructions
[Function Description]: Get index information for the specified table. [Parameter Content]:
table_name (string) [Required] - Name of the table to query
db_name (string) [Optional] - Target database name, defaults to the current database
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| db_name | No | ||
| table_name | Yes |
Implementation Reference
- doris_mcp_server/tools/tools_manager.py:223-244 (registration)MCP tool registration decorator and wrapper function for 'get_table_indexes' that delegates to internal call_tool method.
@mcp.tool( "get_table_indexes", description="""[Function Description]: Get index information for the specified table. [Parameter Content]: - table_name (string) [Required] - Name of the table to query - db_name (string) [Optional] - Target database name, defaults to the current database - catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog """, ) async def get_table_indexes_tool( table_name: str, db_name: str = None, catalog_name: str = None ) -> str: """Get table indexes""" return await self.call_tool("get_table_indexes", { "table_name": table_name, "db_name": db_name, "catalog_name": catalog_name }) - Explicit input schema definition for the 'get_table_indexes' tool in stdio mode listing.
Tool( name="get_table_indexes", description="""[Function Description]: Get index information for the specified table. [Parameter Content]: - table_name (string) [Required] - Name of the table to query - db_name (string) [Optional] - Target database name, defaults to the current database - catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog """, inputSchema={ "type": "object", "properties": { "table_name": {"type": "string", "description": "Table name"}, "db_name": {"type": "string", "description": "Database name"}, "catalog_name": {"type": "string", "description": "Catalog name"}, }, "required": ["table_name"], }, ), - MCP-specific handler interface that invokes the async index retrieval and formats the standardized response.
async def get_table_indexes_for_mcp( self, table_name: str, db_name: str = None, catalog_name: str = None ) -> Dict[str, Any]: """Get index information for specified table - MCP interface""" logger.info(f"Getting table indexes: Table: {table_name}, DB: {db_name}, Catalog: {catalog_name}") if not table_name: return self._format_response(success=False, error="Missing table_name parameter") try: indexes = await self.get_table_indexes_async(table_name=table_name, db_name=db_name, catalog_name=catalog_name) return self._format_response(success=True, result=indexes) except Exception as e: logger.error(f"Failed to get table indexes: {str(e)}", exc_info=True) return self._format_response(success=False, error=str(e), message="Error occurred while getting table indexes") - Core handler logic: executes 'SHOW INDEX FROM ...' query asynchronously, groups results by index name, extracts columns, uniqueness, and type.
async def get_table_indexes_async(self, table_name: str, db_name: str = None, catalog_name: str = None) -> List[Dict[str, Any]]: """Async version: get index information for a table.""" try: effective_db = db_name or self.db_name effective_catalog = catalog_name or self.catalog_name # Build query with catalog prefix if specified if effective_catalog: query = f"SHOW INDEX FROM `{effective_catalog}`.`{effective_db}`.`{table_name}`" logger.info(f"Using three-part naming for async index query: {query}") else: query = f"SHOW INDEX FROM `{effective_db}`.`{table_name}`" rows = await self._execute_query_async(query, effective_db) indexes: List[Dict[str, Any]] = [] if rows: # Group by Key_name current_index: Dict[str, Any] | None = None for r in rows: try: index_name = r.get('Key_name') column_name = r.get('Column_name') if current_index is None or current_index.get('name') != index_name: if current_index is not None: indexes.append(current_index) current_index = { 'name': index_name, 'columns': [column_name] if column_name else [], 'unique': r.get('Non_unique', 1) == 0, 'type': r.get('Index_type', '') } else: if column_name: current_index['columns'].append(column_name) except Exception as row_error: logger.warning(f"Failed to process async index row data: {row_error}") continue if current_index is not None: indexes.append(current_index) return indexes except Exception as e: logger.error(f"Error getting index information asynchronously: {str(e)}") return [] - Tool dispatching router that extracts parameters and delegates to MetadataExtractor's MCP handler.
async def _get_table_indexes_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]: """Get table indexes tool routing""" table_name = arguments.get("table_name") db_name = arguments.get("db_name") catalog_name = arguments.get("catalog_name") # Delegate to metadata extractor for processing return await self.metadata_extractor.get_table_indexes_for_mcp( table_name, db_name, catalog_name )