get_db_table_list
Retrieve a list of all table names within a specified database using this tool, defaulting to the current database if none is provided. Simplifies database metadata management and query preparation.
Instructions
[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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| db_name | No |
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 })
- 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"}, }, }, ),
- 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)
- 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")
- 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 []