Couchbase MCP Server

Instructions

Args:
    limit: Number of queries to return (default: 10)

Returns:
    List of queries with their frequency count

Implementation Reference

• src/tools/query.py:163-195 (handler)

  The primary handler function for the 'get_most_frequent_queries' tool. It executes a SQL++ query on system:completed_requests to find the most frequent user queries, excluding system and DDL queries, and handles empty results gracefully.

  def get_most_frequent_queries(ctx: Context, limit: int = 10) -> list[dict[str, Any]]:
      """Get the N most frequent queries from the system:completed_requests catalog.
  
      Args:
          limit: Number of queries to return (default: 10)
  
      Returns:
          List of queries with their frequency count
      """
      query = """
      SELECT statement,
          COUNT(1) AS queries
      FROM system:completed_requests
      WHERE UPPER(statement) NOT LIKE 'INFER %'
          AND UPPER(statement) NOT LIKE 'CREATE INDEX%'
          AND UPPER(statement) NOT LIKE 'CREATE PRIMARY INDEX%'
          AND UPPER(statement) NOT LIKE 'EXPLAIN %'
          AND UPPER(statement) NOT LIKE 'ADVISE %'
          AND UPPER(statement) NOT LIKE '% SYSTEM:%'
      GROUP BY statement
      LETTING queries = COUNT(1)
      ORDER BY queries DESC
      LIMIT $limit
      """
  
      return _run_query_tool_with_empty_message(
          ctx,
          query,
          limit=limit,
          empty_message=(
              "No completed queries were available to calculate most frequent queries."
          ),
      )

• src/tools/query.py:108-127 (helper)

  Helper utility function used by get_most_frequent_queries (and other query analysis tools) to run cluster-level queries and provide a standardized empty result message.

  def _run_query_tool_with_empty_message(
      ctx: Context,
      query: str,
      *,
      limit: int,
      empty_message: str,
      extra_payload: dict[str, Any] | None = None,
      **query_kwargs: Any,
  ) -> list[dict[str, Any]]:
      """Execute a cluster query with a consistent empty-result response."""
      results = run_cluster_query(ctx, query, limit=limit, **query_kwargs)
  
      if results:
          return results
  
      payload: dict[str, Any] = {"message": empty_message, "results": []}
      if extra_payload:
          payload.update(extra_payload)
      return [payload]

• src/tools/__init__.py:42-64 (registration)

  Registration of the get_most_frequent_queries tool in the ALL_TOOLS list, which is used to register all MCP tools with the server.

  ALL_TOOLS = [
      get_buckets_in_cluster,
      get_server_configuration_status,
      test_cluster_connection,
      get_scopes_and_collections_in_bucket,
      get_collections_in_scope,
      get_scopes_in_bucket,
      get_document_by_id,
      upsert_document_by_id,
      delete_document_by_id,
      get_schema_for_collection,
      run_sql_plus_plus_query,
      get_index_advisor_recommendations,
      list_indexes,
      get_cluster_health_and_services,
      get_queries_not_selective,
      get_queries_not_using_covering_index,
      get_queries_using_primary_index,
      get_queries_with_large_result_count,
      get_queries_with_largest_response_sizes,
      get_longest_running_queries,
      get_most_frequent_queries,
  ]

• src/tools/__init__.py:18-20 (registration)

  Import of the get_most_frequent_queries handler from the query module into the tools package.

  from .query import (
      get_longest_running_queries,
      get_most_frequent_queries,