Doris MCP Server

Instructions

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
              })

• doris_mcp_server/tools/tools_manager.py:905-926 (schema)

  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"],
                  },
              ),

• doris_mcp_server/utils/schema_extractor.py:1633-1651 (handler)

  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")

• doris_mcp_server/utils/schema_extractor.py:1370-1414 (handler)

  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 []

• doris_mcp_server/tools/tools_manager.py:1524-1534 (handler)

  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
      )