get_db_list
Retrieve a complete list of database names stored on the Doris MCP Server using a unique identifier for each tool call.
Instructions
[Function Description]: Get a list of all database names on the server.
[Parameter Content]:
random_string (string) [Required] - Unique identifier for the tool call
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
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 })
- 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"}, }, }, ),
- 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)
- 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")
- 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 []