get_db_list

Instructions

Implementation Reference

• doris_mcp_server/tools/tools_manager.py:158-172 (registration)

  Registers the 'get_db_list' tool with the MCP server using the @mcp.tool decorator, including description and parameter schema implied by function signature.

          # Get database list tool
          @mcp.tool(
              "get_db_list",
              description="""[Function Description]: Get a list of all database names on the server.
  
  [Parameter Content]:
  
  - catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog
  """,
          )
          async def get_db_list_tool(catalog_name: str = None) -> str:
              """Get database list"""
              return await self.call_tool("get_db_list", {
                  "catalog_name": catalog_name
              })

• doris_mcp_server/tools/tools_manager.py:846-860 (schema)

  Defines the input schema and description for the 'get_db_list' tool in the list_tools method for stdio mode.

              Tool(
                  name="get_db_list",
                  description="""[Function Description]: Get a list of all database names on the server.
  
  [Parameter Content]:
  
  - catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog
  """,
                  inputSchema={
                      "type": "object",
                      "properties": {
                          "catalog_name": {"type": "string", "description": "Catalog name"},
                      },
                  },
              ),

• doris_mcp_server/tools/tools_manager.py:1495-1501 (handler)

  Tool dispatcher in call_tool method routes 'get_db_list' calls to MetadataExtractor.get_db_list_for_mcp.

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

• doris_mcp_server/utils/schema_extractor.py:1584-1594 (handler)

  MCP-specific handler for get_db_list: formats response and calls core get_all_databases_async method.

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

• doris_mcp_server/utils/schema_extractor.py:1224-1253 (handler)

  Core implementation: executes 'SHOW DATABASES' (or with catalog) query asynchronously and extracts database names.

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