Couchbase MCP Server

Instructions

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

Returns:
    List of queries with their average result count

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]