StarRocks MCP Server

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/mcp_server_starrocks/server.py:406-453 (handler)

  Handler function for the 'table_overview' tool. Includes registration via @mcp.tool decorator, input schema via Annotated Fields, caching logic, parsing of table parameter, error handling, and invocation of helper function to retrieve table details.
  @mcp.tool(description="Get an overview of a specific table: columns, sample rows (up to 5), and total row count. Uses cache unless refresh=true" + description_suffix) def table_overview( table: Annotated[str, Field( description="Table name, optionally prefixed with database name (e.g., 'db_name.table_name'). If database is omitted, uses the default database.")], refresh: Annotated[ bool, Field(description="Set to true to force refresh, ignoring cache. Defaults to false.")] = False ) -> str: try: logger.info(f"Getting table overview for: {table}, refresh={refresh}") if not table: logger.error("Table overview called without table name") return "Error: Missing 'table' argument." # Parse table argument: [db.]<table> parts = table.split('.', 1) db_name = None table_name = None if len(parts) == 2: db_name, table_name = parts[0], parts[1] elif len(parts) == 1: table_name = parts[0] db_name = db_client.default_database # Use default if only table name is given if not table_name: # Should not happen if table_arg exists, but check logger.error(f"Invalid table name format: {table}") return f"Error: Invalid table name format '{table}'." if not db_name: logger.error(f"No database specified for table {table_name}") return f"Error: Database name not specified for table '{table_name}' and no default database is set." cache_key = (db_name, table_name) # Check cache if not refresh and cache_key in global_table_overview_cache: logger.debug(f"Using cached overview for {db_name}.{table_name}") return global_table_overview_cache[cache_key] logger.debug(f"Fetching fresh overview for {db_name}.{table_name}") # Fetch details (will also update cache) overview_text = _get_table_details(db_name, table_name, limit=overview_length_limit) return overview_text except Exception as e: # Reset connections on unexpected errors logger.exception(f"Unexpected error in table_overview for {table}") reset_db_connections() stack_trace = traceback.format_exc() return f"Unexpected Error executing tool 'table_overview': {type(e).__name__}: {e}\nStack Trace:\n{stack_trace}"
• src/mcp_server_starrocks/server.py:407-412 (schema)

  Pydantic input schema for 'table_overview' tool parameters: 'table' (string, table name possibly with db prefix) and 'refresh' (boolean, optional, default False).
  def table_overview( table: Annotated[str, Field( description="Table name, optionally prefixed with database name (e.g., 'db_name.table_name'). If database is omitted, uses the default database.")], refresh: Annotated[ bool, Field(description="Set to true to force refresh, ignoring cache. Defaults to false.")] = False ) -> str:
• src/mcp_server_starrocks/server.py:125-187 (helper)

  Supporting helper function _get_table_details(db_name, table_name, limit) that performs the core queries (COUNT(*), DESCRIBE, SELECT LIMIT 3), formats the overview string, and updates the global cache.
  def _get_table_details(db_name, table_name, limit=None): """ Helper function to get description, sample rows, and count for a table. Returns a formatted string. Handles DB errors internally and returns error messages. """ global global_table_overview_cache logger.debug(f"Fetching table details for {db_name}.{table_name}") output_lines = [] full_table_name = f"`{table_name}`" if db_name: full_table_name = f"`{db_name}`.`{table_name}`" else: output_lines.append( f"Warning: Database name missing for table '{table_name}'. Using potentially incorrect context.") logger.warning(f"Database name missing for table '{table_name}'") count = 0 output_lines.append(f"--- Overview for {full_table_name} ---") # 1. Get Row Count query = f"SELECT COUNT(*) FROM {full_table_name}" count_result = db_client.execute(query, db=db_name) if count_result.success and count_result.rows: count = count_result.rows[0][0] output_lines.append(f"\nTotal rows: {count}") logger.debug(f"Table {full_table_name} has {count} rows") else: output_lines.append(f"\nCould not determine total row count.") if not count_result.success: output_lines.append(f"Error: {count_result.error_message}") logger.error(f"Failed to get row count for {full_table_name}: {count_result.error_message}") # 2. Get Columns (DESCRIBE) if count > 0: query = f"DESCRIBE {full_table_name}" desc_result = db_client.execute(query, db=db_name) if desc_result.success and desc_result.column_names and desc_result.rows: output_lines.append(f"\nColumns:") output_lines.append(desc_result.to_string(limit=limit)) else: output_lines.append("(Could not retrieve column information or table has no columns).") if not desc_result.success: output_lines.append(f"Error getting columns for {full_table_name}: {desc_result.error_message}") return "\n".join(output_lines) # 3. Get Sample Rows (LIMIT 3) query = f"SELECT * FROM {full_table_name} LIMIT 3" sample_result = db_client.execute(query, db=db_name) if sample_result.success and sample_result.column_names and sample_result.rows: output_lines.append(f"\nSample rows (limit 3):") output_lines.append(sample_result.to_string(limit=limit)) else: output_lines.append(f"(No rows found in {full_table_name}).") if not sample_result.success: output_lines.append(f"Error getting sample rows for {full_table_name}: {sample_result.error_message}") overview_string = "\n".join(output_lines) # Update cache even if there were partial errors, so we cache the error message too cache_key = (db_name, table_name) global_table_overview_cache[cache_key] = overview_string return overview_string
• src/mcp_server_starrocks/server.py:406-406 (registration)

  Registration of the 'table_overview' tool via FastMCP @mcp.tool decorator with description.
  @mcp.tool(description="Get an overview of a specific table: columns, sample rows (up to 5), and total row count. Uses cache unless refresh=true" + description_suffix)
• src/mcp_server_starrocks/server.py:56-56 (helper)

  Global cache dictionary for storing table overview strings, keyed by (db_name, table_name). Used by table_overview and db_overview tools.
  global_table_overview_cache = {}