StarRocks MCP Server

Instructions

Implementation Reference

• src/mcp_server_starrocks/server.py:456-513 (handler)

  The handler function implementing the 'db_overview' tool. It lists all tables in the specified database (or default), fetches overviews for each using cache or _get_table_details helper, and compiles them into a string response. Note: The @mcp.tool decorator is commented out (line 455), so it may not be actively registered.

  def db_overview(
          db: Annotated[str, Field(
              description="Database name. Optional: uses the default database if not provided.")] = None,
          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 overview for: {db_name}, refresh={refresh}")
          if not db_name:
              logger.error("Database overview called without database name")
              return "Error: Database name not provided and no default database is set."
  
          # List tables in the database
          query = f"SHOW TABLES FROM `{db_name}`"
          result = db_client.execute(query, db=db_name)
  
          if not result.success:
              logger.error(f"Failed to list tables in database {db_name}: {result.error_message}")
              return f"Database Error listing tables in '{db_name}': {result.error_message}"
  
          if not result.rows:
              logger.info(f"No tables found in database {db_name}")
              return f"No tables found in database '{db_name}'."
  
          tables = [row[0] for row in result.rows]
          logger.info(f"Found {len(tables)} tables in database {db_name}")
          all_overviews = [f"--- Overview for Database: `{db_name}` ({len(tables)} tables) ---"]
  
          total_length = 0
          limit_per_table = overview_length_limit * (math.log10(len(tables)) + 1) // len(tables)  # Limit per table
          for table_name in tables:
              cache_key = (db_name, table_name)
              overview_text = None
  
              # Check cache first
              if not refresh and cache_key in global_table_overview_cache:
                  logger.debug(f"Using cached overview for {db_name}.{table_name}")
                  overview_text = global_table_overview_cache[cache_key]
              else:
                  logger.debug(f"Fetching fresh overview for {db_name}.{table_name}")
                  # Fetch details for this table (will update cache via _get_table_details)
                  overview_text = _get_table_details(db_name, table_name, limit=limit_per_table)
  
              all_overviews.append(overview_text)
              all_overviews.append("\n")  # Add separator
              total_length += len(overview_text) + 1
  
          logger.info(f"Database overview completed for {db_name}, total length: {total_length}")
          return "\n".join(all_overviews)
  
      except Exception as e:
          # Catch any other unexpected errors during tool execution
          logger.exception(f"Unexpected error in db_overview for database {db}")
          reset_db_connections()
          stack_trace = traceback.format_exc()
          return f"Unexpected Error executing tool 'db_overview': {type(e).__name__}: {e}\nStack Trace:\n{stack_trace}"

• src/mcp_server_starrocks/server.py:125-187 (helper)

  Supporting helper function called by db_overview (and table_overview) to generate detailed overview for a single table: row count, DESCRIBE columns, and sample rows (LIMIT 3). 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:455-455 (registration)

  Commented-out registration decorator for the db_overview tool using @mcp.tool. Indicates it is not actively registered, preferring db_summary instead.

  #@mcp.tool(description="Get an overview (columns, sample rows, row count) for ALL tables in a database. Uses cache unless refresh=True" + description_suffix)