db_summary
Retrieve a summary of a database showing table schemas and sizes. Optionally specify database name or refresh cache.
Instructions
Quickly get summary of a database with tables' schema and size information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| db | No | Database name. Optional: uses current database by default. | |
| limit | No | Output length limit in characters. Defaults to 10000. Higher values show more tables and details. | |
| refresh | No | Set to true to force refresh, ignoring cache. Defaults to false. |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The actual tool definition/handler for 'db_summary'. This is the @mcp.tool decorated function that serves as the entry point. It takes db (optional), limit (default 10000), and refresh (default False) parameters, and delegates to db_summary_manager.get_database_summary().
@mcp.tool(description="Quickly get summary of a database with tables' schema and size information" + description_suffix) def db_summary( db: Annotated[str|None, Field( description="Database name. Optional: uses current database by default.")] = None, limit: Annotated[int, Field( description="Output length limit in characters. Defaults to 10000. Higher values show more tables and details.")] = 10000, refresh: Annotated[bool, Field( description="Set to true to force refresh, ignoring cache. Defaults to false.")] = False ) -> str: try: db_name = db if db else db_client.default_database logger.info(f"Getting database summary for: {db_name}, limit={limit}, refresh={refresh}") if not db_name: logger.error("Database summary called without database name") return "Error: Database name not provided and no default database is set." # Use the database summary manager summary = db_summary_manager.get_database_summary(db_name, limit=limit, refresh=refresh) logger.info(f"Database summary completed for {db_name}") return summary except Exception as e: # Reset connections on unexpected errors logger.exception(f"Unexpected error in db_summary for database {db}") reset_db_connections() stack_trace = traceback.format_exc() return f"Unexpected Error executing tool 'db_summary': {type(e).__name__}: {e}\nStack Trace:\n{stack_trace}" - The core business logic of the db_summary tool. Method get_database_summary() syncs table list, fetches column info, fetches CREATE statements for large tables, and formats the summary.
def get_database_summary(self, database: str, limit: int = 10000, refresh: bool = False) -> str: """Generate comprehensive database summary with intelligent prioritization""" if not database: return "Error: Database name is required" logger.info(f"Generating database summary for {database}, limit={limit}, refresh={refresh}") # Sync table list if not self._sync_table_list(database, force=refresh): return f"Error: Failed to sync table information for database '{database}'" # Get all tables for this database from cache tables_info = [] for (db, table_name), table_info in self.table_cache.items(): if db == database: tables_info.append(table_info) if not tables_info: return f"No tables found in database '{database}'" # Sort tables by priority (large tables first) tables_info.sort(key=lambda t: t.priority_score(), reverse=True) # Check if any table needs column information refresh need_column_refresh = refresh or any(not table_info.columns or table_info.is_expired() for table_info in tables_info) # If any table needs refresh, fetch ALL tables' columns in one query (more efficient) if need_column_refresh: all_table_names = [table_info.name for table_info in tables_info] table_columns = self._fetch_column_info(database, all_table_names) # Update cache with column information for all tables current_time = time.time() for table_info in tables_info: if table_info.name in table_columns: table_info.columns = table_columns[table_info.name] table_info.last_updated = current_time # Identify large tables that need CREATE statements large_tables = [t for t in tables_info if t.is_large_table()][:10] # Top 10 large tables for table_info in large_tables: if refresh or not table_info.create_statement: table_info.create_statement = self._fetch_create_statement(database, table_info.name) table_info.last_updated = time.time() # Generate summary output return self._format_database_summary(database, tables_info, limit) - _format_database_summary() - Formats the database summary output with intelligent truncation based on the limit parameter. Shows large tables first with full details, then remaining tables.
def _format_database_summary(self, database: str, tables_info: List[TableInfo], limit: int) -> str: """Format database summary with intelligent truncation""" lines = [] lines.append(f"=== Database Summary: '{database}' ===") lines.append(f"Total tables: {len(tables_info)}") # Calculate totals total_size = sum(t.size_bytes for t in tables_info) total_replicas = sum(t.replica_count for t in tables_info) large_tables = [t for t in tables_info if t.is_large_table()] lines.append(f"Total size: {self._format_bytes(total_size)}") current_length = len("\n".join(lines)) table_limit = min(len(tables_info), 50) # Show max 50 tables # Show large tables first with full details if large_tables: for i, table_info in enumerate(large_tables): if current_length > limit * 0.8: # Reserve 20% for smaller tables lines.append(f"... and {len(large_tables) - i} more large tables") break table_summary = self._format_table_info(table_info, detailed=True) lines.append(table_summary) lines.append("") current_length = len("\n".join(lines)) # Show remaining tables with basic info remaining_tables = [t for t in tables_info if not t.is_large_table()] if remaining_tables and current_length < limit: lines.append("--- Other Tables ---") for i, table_info in enumerate(remaining_tables): if current_length > limit: lines.append(f"... and {len(remaining_tables) - i} more tables (use higher limit to see all)") break table_summary = self._format_table_info(table_info, detailed=False) lines.append(table_summary) current_length = len("\n".join(lines)) return "\n".join(lines) - _format_table_info() - Formats individual table information including size, replicas, CREATE statement or column list.
def _format_table_info(self, table_info: TableInfo, detailed: bool = True) -> str: """Format individual table information""" lines = [] # Basic info line size_info = f"{table_info.size_str} ({table_info.replica_count} replicas)" lines.append(f"Table: {table_info.name} - {size_info}") if table_info.error_message: lines.append(f" Error: {table_info.error_message}") return "\n".join(lines) # Show CREATE statement if available, otherwise show column list if table_info.create_statement: lines.append(table_info.create_statement) elif table_info.columns: # Sort columns by ordinal position and show as list sorted_columns = sorted(table_info.columns, key=lambda c: c.ordinal_position) if detailed or len(sorted_columns) <= 20: for col in sorted_columns: lines.append(f" {col.name} {col.column_type}") else: lines.append(f" Columns ({len(sorted_columns)}): {', '.join(col.name for col in sorted_columns[:100])}...") return "\n".join(lines) - src/mcp_server_starrocks/server.py:515-516 (registration)The @mcp.tool decorator registers the db_summary function as an MCP tool. The description is 'Quickly get summary of a database with tables' schema and size information'.
@mcp.tool(description="Quickly get summary of a database with tables' schema and size information" + description_suffix) def db_summary(