get_db_table_list

Instructions

Implementation Reference

• doris_mcp_server/tools/tools_manager.py:138-156 (registration)

  Registers the get_db_table_list tool with the MCP server, defining its parameters and description. The tool function delegates to the internal call_tool method.

          @mcp.tool(
              "get_db_table_list",
              description="""[Function Description]: Get a list of all table names in the specified database.
  
  [Parameter Content]:
  
  - 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_db_table_list_tool(
              db_name: str = None, catalog_name: str = None
          ) -> str:
              """Get database table list"""
              return await self.call_tool("get_db_table_list", {
                  "db_name": db_name,
                  "catalog_name": catalog_name
              })

• doris_mcp_server/tools/tools_manager.py:828-845 (schema)

  Defines the input schema for the get_db_table_list tool in list_tools method for stdio mode compatibility.

              Tool(
                  name="get_db_table_list",
                  description="""[Function Description]: Get a list of all table names in the specified database.
  
  [Parameter Content]:
  
  - 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": {
                          "db_name": {"type": "string", "description": "Database name"},
                          "catalog_name": {"type": "string", "description": "Catalog name"},
                      },
                  },
              ),

• doris_mcp_server/tools/tools_manager.py:1487-1493 (handler)

  Handler routing function in DorisToolsManager that extracts parameters and delegates to MetadataExtractor.get_db_table_list_for_mcp.

  async def _get_db_table_list_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
      """Get database table list tool routing"""
      db_name = arguments.get("db_name")
      catalog_name = arguments.get("catalog_name")
      
      # Delegate to metadata extractor for processing
      return await self.metadata_extractor.get_db_table_list_for_mcp(db_name, catalog_name)

• doris_mcp_server/utils/schema_extractor.py:1569-1583 (handler)

  MCP interface in MetadataExtractor that calls get_database_tables_async and formats the response.

  async def get_db_table_list_for_mcp(
      self, 
      db_name: str = None, 
      catalog_name: str = None
  ) -> Dict[str, Any]:
      """Get list of all table names in specified database - MCP interface"""
      logger.info(f"Getting database table list: DB: {db_name}, Catalog: {catalog_name}")
      
      try:
          tables = await self.get_database_tables_async(db_name=db_name, catalog_name=catalog_name)
          return self._format_response(success=True, result=tables)
      except Exception as e:
          logger.error(f"Failed to get database table list: {str(e)}", exc_info=True)
          return self._format_response(success=False, error=str(e), message="Error occurred while getting database table list")

• doris_mcp_server/utils/schema_extractor.py:1254-1284 (handler)

  Core implementation that executes 'SHOW TABLES FROM catalog.db' query via _execute_query_async and extracts table names from the result.

  async def get_database_tables_async(self, db_name: str = None, catalog_name: str = None) -> List[str]:
      """Asynchronously get table list in database"""
      try:
          effective_catalog = catalog_name or self.catalog_name
          effective_db = db_name or self.db_name
          
          if effective_catalog and effective_catalog != "internal":
              query = f"SHOW TABLES FROM `{effective_catalog}`.`{effective_db}`"
          else:
              query = f"SHOW TABLES FROM `{effective_db}`"
          
          result = await self._execute_query_async(query, effective_db)
          
          if not result:
              return []
          
          # Extract table names
          tables = []
          for row in result:
              if isinstance(row, dict):
                  # Get the value of the first field (usually Tables_in_xxx field)
                  table_name = list(row.values())[0] if row else None
                  if table_name:
                      tables.append(table_name)
          
          return tables
          
      except Exception as e:
          logger.error(f"Failed to get table list: {e}")
          return []