list_indexes
Retrieve and filter database indexes in a Couchbase cluster to view their names, definitions, and status. Use optional parameters to narrow results by bucket, scope, collection, or index name.
Instructions
List all indexes in the cluster with optional filtering by bucket, scope, collection, and index name. Returns a list of indexes with their names and CREATE INDEX definitions. Uses the Index Service REST API (/getIndexStatus) to retrieve index information directly.
Args:
ctx: MCP context for cluster connection
bucket_name: Optional bucket name to filter indexes
scope_name: Optional scope name to filter indexes (requires bucket_name)
collection_name: Optional collection name to filter indexes (requires bucket_name and scope_name)
index_name: Optional index name to filter indexes (requires bucket_name, scope_name, and collection_name)
include_raw_index_stats: If True, include raw index stats (as-is from API) in addition
to cleaned-up version. Default is False.
Returns:
List of dictionaries with keys:
- name (str): Index name
- definition (str): Cleaned-up CREATE INDEX statement
- status (str): Current status of the index (e.g., "Ready", "Building", "Deferred")
- isPrimary (bool): Whether this is a primary index
- bucket (str): Bucket name where the index exists
- scope (str): Scope name where the index exists
- collection (str): Collection name where the index exists
- raw_index_stats (dict, optional): Complete raw index status object from API including metadata,
state, keyspace info, etc. (only if include_raw_index_stats=True)Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket_name | No | ||
| scope_name | No | ||
| collection_name | No | ||
| index_name | No | ||
| include_raw_index_stats | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools/index.py:98-173 (handler)Core handler function for the 'list_indexes' tool. Validates input parameters, retrieves connection settings, fetches raw index data using the Couchbase Index Service REST API (/getIndexStatus), processes the data into a clean format, and returns a list of index dictionaries.
def list_indexes( ctx: Context, bucket_name: str | None = None, scope_name: str | None = None, collection_name: str | None = None, index_name: str | None = None, include_raw_index_stats: bool = False, ) -> list[dict[str, Any]]: """List all indexes in the cluster with optional filtering by bucket, scope, collection, and index name. Returns a list of indexes with their names and CREATE INDEX definitions. Uses the Index Service REST API (/getIndexStatus) to retrieve index information directly. Args: ctx: MCP context for cluster connection bucket_name: Optional bucket name to filter indexes scope_name: Optional scope name to filter indexes (requires bucket_name) collection_name: Optional collection name to filter indexes (requires bucket_name and scope_name) index_name: Optional index name to filter indexes (requires bucket_name, scope_name, and collection_name) include_raw_index_stats: If True, include raw index stats (as-is from API) in addition to cleaned-up version. Default is False. Returns: List of dictionaries with keys: - name (str): Index name - definition (str): Cleaned-up CREATE INDEX statement - status (str): Current status of the index (e.g., "Ready", "Building", "Deferred") - isPrimary (bool): Whether this is a primary index - bucket (str): Bucket name where the index exists - scope (str): Scope name where the index exists - collection (str): Collection name where the index exists - raw_index_stats (dict, optional): Complete raw index status object from API including metadata, state, keyspace info, etc. (only if include_raw_index_stats=True) """ try: # Validate parameters validate_filter_params(bucket_name, scope_name, collection_name, index_name) # Get and validate connection settings settings = get_settings() validate_connection_settings(settings) # Fetch indexes from REST API logger.info( f"Fetching indexes from REST API for bucket={bucket_name}, " f"scope={scope_name}, collection={collection_name}, index={index_name}" ) raw_indexes = fetch_indexes_from_rest_api( settings["connection_string"], settings["username"], settings["password"], bucket_name=bucket_name, scope_name=scope_name, collection_name=collection_name, index_name=index_name, ca_cert_path=settings.get("ca_cert_path"), ) # Process and format the results indexes = [ processed for idx in raw_indexes if (processed := process_index_data(idx, include_raw_index_stats)) is not None ] logger.info( f"Found {len(indexes)} indexes from REST API " f"(include_raw_index_stats={include_raw_index_stats})" ) return indexes except Exception as e: logger.error(f"Error listing indexes: {e}", exc_info=True) raise - src/mcp_server.py:176-177 (registration)Registration loop in the main MCP server that adds all tools from ALL_TOOLS (including list_indexes) to the FastMCP server instance using mcp.add_tool(tool).
for tool in ALL_TOOLS: mcp.add_tool(tool) - src/tools/__init__.py:54-55 (registration)list_indexes function is imported from index.py and included in the ALL_TOOLS list used for server registration.
get_index_advisor_recommendations, list_indexes, - src/utils/index_utils.py:224-309 (helper)Key helper function called by list_indexes handler to fetch raw index data directly from the Couchbase Index Service REST API endpoint /getIndexStatus. Handles connection details, TLS, multiple hosts failover, filtering params, and SSL verification (including Capella root CA).
def fetch_indexes_from_rest_api( connection_string: str, username: str, password: str, bucket_name: str | None = None, scope_name: str | None = None, collection_name: str | None = None, index_name: str | None = None, ca_cert_path: str | None = None, timeout: int = 30, ) -> list[dict[str, Any]]: """Fetch indexes from Couchbase Index Service REST API. Uses the /getIndexStatus endpoint to retrieve index information. This endpoint returns indexes with their definitions directly from the Index Service. Args: connection_string: Couchbase connection string (may contain multiple hosts) username: Username for authentication password: Password for authentication bucket_name: Optional bucket name to filter indexes scope_name: Optional scope name to filter indexes collection_name: Optional collection name to filter indexes index_name: Optional index name to filter indexes ca_cert_path: Optional path to CA certificate for SSL verification. If not provided and using Capella, will use Capella root CA. timeout: Request timeout in seconds (default: 30) Returns: List of index status dictionaries containing name, definition, and other metadata """ # Extract all hosts from connection string hosts = _extract_hosts_from_connection_string(connection_string) # Determine protocol and port based on whether TLS is enabled is_tls_enabled = connection_string.lower().startswith("couchbases://") protocol = "https" if is_tls_enabled else "http" port = 19102 if is_tls_enabled else 9102 logger.info( f"TLS {'enabled' if is_tls_enabled else 'disabled'}, " f"using {protocol.upper()} with port {port}" ) # Build query parameters and determine SSL verification params = _build_query_params(bucket_name, scope_name, collection_name, index_name) verify_ssl = _determine_ssl_verification(connection_string, ca_cert_path) # Try each host one by one until we get a successful response last_error = None for host in hosts: try: url = f"{protocol}://{host}:{port}/getIndexStatus" logger.info( f"Attempting to fetch indexes from: {url} with params: {params}" ) response = httpx.get( url, params=params, auth=(username, password), verify=verify_ssl, timeout=timeout, ) response.raise_for_status() data = response.json() indexes = data.get("status", []) logger.info(f"Successfully fetched {len(indexes)} indexes from {host}") return indexes except httpx.HTTPError as e: logger.warning(f"Failed to fetch indexes from {host}: {e}") last_error = e except Exception as e: logger.warning(f"Unexpected error when fetching from {host}: {e}") last_error = e # If we get here, all hosts failed error_msg = f"Failed to fetch indexes from all hosts: {hosts}" if last_error: error_msg += f". Last error: {last_error}" logger.error(error_msg) raise RuntimeError(error_msg) - src/utils/index_utils.py:54-91 (helper)Helper function that processes each raw index dictionary from the API into the clean output format used by list_indexes, including cleaning definition, extracting key fields, and optionally adding raw stats.
def process_index_data( idx: dict[str, Any], include_raw_index_stats: bool ) -> dict[str, Any] | None: """Process raw index data into formatted index info. Args: idx: Raw index data from the API include_raw_index_stats: Whether to include complete raw stats in the output Returns: Formatted index info dictionary, or None if the index should be skipped (e.g., no name). """ name = idx.get("name", "") if not name: return None # Start with name and optional definition index_info: dict[str, Any] = {"name": name} clean_def = clean_index_definition(idx.get("definition", "")) if clean_def: index_info["definition"] = clean_def # Copy standard fields from raw index data standard_fields = ["status", "bucket", "scope", "collection"] for field in standard_fields: if field in idx: index_info[field] = idx[field] # Always include isPrimary as a boolean index_info["isPrimary"] = idx.get("isPrimary", False) # Optionally include complete raw stats if include_raw_index_stats: index_info["raw_index_stats"] = idx return index_info