Couchbase MCP Server

Instructions

Implementation Reference

• src/tools/query.py:236-269 (handler)

  The main handler function that implements the tool logic. It constructs and executes a SQL++ query on the system:completed_requests catalog to identify queries with the largest average result counts, returning the results or an empty message.
  def get_queries_with_large_result_count( ctx: Context, limit: int = 10 ) -> list[dict[str, Any]]: """Get queries with the largest result counts from the system:completed_requests catalog. Args: limit: Number of queries to return (default: 10) Returns: List of queries with their average result count """ query = """ SELECT statement, avgResultCount, 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 '% SYSTEM:%' GROUP BY statement LETTING avgResultCount = AVG(resultCount) ORDER BY avgResultCount DESC LIMIT $limit """ return _run_query_tool_with_empty_message( ctx, query, limit=limit, empty_message=( "No completed queries were available to calculate result counts." ), )
• src/tools/__init__.py:42-64 (registration)

  The tool is registered by being included in the ALL_TOOLS list, which is likely used by the MCP server to register all available tools.
  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/query.py:108-127 (helper)

  Supporting helper utility that runs cluster queries and provides a standardized empty result response, used by the handler.
  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]