Couchbase MCP Server

Instructions

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