diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py
index 84d7fd8ac08..e306e361e41 100644
--- a/lib/galaxy/app.py
+++ b/lib/galaxy/app.py
@@ -5,6 +5,7 @@ import os
 from galaxy import config, jobs
 import galaxy.model
 import galaxy.security
+from galaxy import dataset_collections
 import galaxy.quota
 from galaxy.tags.tag_handler import GalaxyTagHandler
 from galaxy.visualization.genomes import Genomes
@@ -54,6 +55,8 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
         self._configure_security()
         # Tag handler
         self.tag_handler = GalaxyTagHandler()
+        # Dataset Collection Plugins
+        self.dataset_collections_service = dataset_collections.DatasetCollectionsService(self)
         # Genomes
         self.genomes = Genomes( self )
         # Data providers registry.
diff --git a/lib/galaxy/dataset_collections/__init__.py b/lib/galaxy/dataset_collections/__init__.py
new file mode 100644
index 00000000000..6d22cbe356c
--- /dev/null
+++ b/lib/galaxy/dataset_collections/__init__.py
@@ -0,0 +1,266 @@
+from .registry import DatasetCollectionTypesRegistry
+from .structure import get_structure
+
+from galaxy import model
+from galaxy.exceptions import MessageException
+from galaxy.exceptions import ItemAccessibilityException
+from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.web.base.controller import (
+    UsesHistoryDatasetAssociationMixin,
+    UsesLibraryMixinItems,
+    UsesTagsMixin,
+)
+from galaxy.managers import hdas  # TODO: Refactor all mixin use into managers.
+
+from galaxy.util import validation
+from galaxy.util import odict
+
+import logging
+log = logging.getLogger( __name__ )
+
+
+ERROR_INVALID_ELEMENTS_SPECIFICATION = "Create called with invalid parameters, must specify element identifiers."
+ERROR_NO_COLLECTION_TYPE = "Create called without specifing a collection type."
+
+
+class DatasetCollectionsService(
+    UsesHistoryDatasetAssociationMixin,
+    UsesLibraryMixinItems,
+    UsesTagsMixin,
+):
+    """
+    Abstraction for interfacing with dataset collections instance - ideally abstarcts
+    out model and plugin details.
+    """
+
+    def __init__( self, app ):
+        self.type_registry = DatasetCollectionTypesRegistry( app )
+        self.model = app.model
+        self.security = app.security
+        self.hda_manager = hdas.HDAManager()
+
+    def create(
+        self,
+        trans,
+        parent,  # PRECONDITION: security checks on ability to add to parent occurred during load.
+        name,
+        collection_type,
+        element_identifiers=None,
+        elements=None,
+        implicit_collection_info=None,
+    ):
+        """
+        """
+        dataset_collection = self.__create_dataset_collection(
+            trans=trans,
+            collection_type=collection_type,
+            element_identifiers=element_identifiers,
+            elements=elements,
+        )
+        if isinstance( parent, model.History ):
+            dataset_collection_instance = self.model.HistoryDatasetCollectionAssociation(
+                collection=dataset_collection,
+                name=name,
+            )
+            if implicit_collection_info:
+                for input_name, input_collection in implicit_collection_info[ "implicit_inputs" ]:
+                    dataset_collection_instance.add_implicit_input_collection( input_name, input_collection )
+                dataset_collection_instance.implicit_output_name = implicit_collection_info[ "implicit_output_name" ]
+            # Handle setting hid
+            parent.add_dataset_collection( dataset_collection_instance )
+        elif isinstance( parent, model.LibraryFolder ):
+            dataset_collection_instance = self.model.LibraryDatasetCollectionAssociation(
+                collection=dataset_collection,
+                folder=parent,
+                name=name,
+            )
+        else:
+            message = "Internal logic error - create called with unknown parent type %s" % type( parent )
+            log.exception( message )
+            raise MessageException( message )
+
+        return self.__persist( dataset_collection_instance )
+
+    def __create_dataset_collection(
+        self,
+        trans,
+        collection_type,
+        element_identifiers=None,
+        elements=None,
+    ):
+        if element_identifiers is None and elements is None:
+            raise RequestParameterInvalidException( ERROR_INVALID_ELEMENTS_SPECIFICATION )
+        if not collection_type:
+            raise RequestParameterInvalidException( ERROR_NO_COLLECTION_TYPE )
+        rank_collection_type = collection_type.split( ":" )[ 0 ]
+        if elements is None:
+            if rank_collection_type != collection_type:
+                # Nested collection - recursively create collections and update identifiers.
+                self.__recursively_create_collections( trans, element_identifiers )
+            elements = self.__load_elements( trans, element_identifiers )
+        # else if elements is set, it better be an ordered dict!
+
+        type_plugin = self.__type_plugin( rank_collection_type )
+        dataset_collection = type_plugin.build_collection( elements )
+        dataset_collection.collection_type = collection_type
+        return dataset_collection
+
+    def delete( self, trans, instance_type, id ):
+        dataset_collection_instance = self.get_dataset_collection_instance( trans, instance_type, id, check_ownership=True )
+        dataset_collection_instance.deleted = True
+        trans.sa_session.add( dataset_collection_instance )
+        trans.sa_session.flush( )
+
+    def update( self, trans, instance_type, id, payload ):
+        dataset_collection_instance = self.get_dataset_collection_instance( trans, instance_type, id, check_ownership=True )
+        if trans.user is None:
+            anon_allowed_payload = {}
+            if 'deleted' in payload:
+                anon_allowed_payload[ 'deleted' ] = payload[ 'deleted' ]
+            if 'visible' in payload:
+                anon_allowed_payload[ 'visible' ] = payload[ 'visible' ]
+            payload = self._validate_and_parse_update_payload( anon_allowed_payload )
+        else:
+            payload = self._validate_and_parse_update_payload( payload )
+        changed = self._set_from_dict( trans, dataset_collection_instance, payload )
+        return changed
+
+    def _set_from_dict( self, trans, dataset_collection_instance, new_data ):
+        # Blatantly stolen from UsesHistoryDatasetAssociationMixin.set_hda_from_dict.
+
+        # send what we can down into the model
+        changed = dataset_collection_instance.set_from_dict( new_data )
+        # the rest (often involving the trans) - do here
+        if 'annotation' in new_data.keys() and trans.get_user():
+            dataset_collection_instance.add_item_annotation( trans.sa_session, trans.get_user(), dataset_collection_instance.collection, new_data[ 'annotation' ] )
+            changed[ 'annotation' ] = new_data[ 'annotation' ]
+        if 'tags' in new_data.keys() and trans.get_user():
+            self.set_tags_from_list( trans, dataset_collection_instance.collection, new_data[ 'tags' ], user=trans.user )
+
+        if changed.keys():
+            trans.sa_session.flush()
+
+        return changed
+
+    def _validate_and_parse_update_payload( self, payload ):
+        validated_payload = {}
+        for key, val in payload.items():
+            if val is None:
+                continue
+            if key in ( 'name' ):
+                val = validation.validate_and_sanitize_basestring( key, val )
+                validated_payload[ key ] = val
+            if key in ( 'deleted', 'visible' ):
+                validated_payload[ key ] = validation.validate_boolean( key, val )
+            elif key == 'tags':
+                validated_payload[ key ] = validation.validate_and_sanitize_basestring_list( key, val )
+        return validated_payload
+
+    def history_dataset_collections(self, history, query):
+        collections = history.dataset_collections
+        collection_type = query.get( "collection_type", None )
+        if collection_type:
+            collections = filter( lambda c: c.collection.collection_type == collection_type, collections )
+        return collections
+
+    def __persist( self, dataset_collection_instance ):
+        context = self.model.context
+        context.add( dataset_collection_instance )
+        context.flush()
+        return dataset_collection_instance
+
+    def __recursively_create_collections( self, trans, element_identifiers ):
+        # TODO: Optimize - don't recheck parent, reload created model, just use as is.
+        for index, element_identifier in enumerate( element_identifiers ):
+            try:
+                if not element_identifier[ "src" ] == "new_collection":
+                    # not a new collection, keep moving...
+                    continue
+            except KeyError:
+                # Not a dictionary, just an id of an HDA - move along.
+                continue
+
+            # element identifier is a dict with src new_collection...
+            collection_type = element_identifier.get( "collection_type", None )
+            if not collection_type:
+                raise RequestParameterInvalidException( "No collection_type define for nested collection." )
+            collection = self.__create_dataset_collection(
+                trans=trans,
+                collection_type=collection_type,
+                element_identifiers=element_identifier[ "element_identifiers" ],
+            )
+            self.__persist( collection )
+            element_identifier[ "src" ] = "dc"
+            element_identifier[ "id" ] = trans.security.encode_id( collection.id )
+
+        return element_identifiers
+
+    def __load_elements( self, trans, element_identifiers ):
+        elements = odict.odict()
+        for element_identifier in element_identifiers:
+            elements[ element_identifier[ "name" ] ] = self.__load_element( trans, element_identifier )
+        return elements
+
+    def __load_element( self, trans, element_identifier ):
+        #if not isinstance( element_identifier, dict ):
+        #    # Is allowing this to just be the id of an hda too clever? Somewhat
+        #    # consistent with other API methods though.
+        #    element_identifier = dict( src='hda', id=str( element_identifier ) )
+
+        # dateset_identifier is dict {src=hda|ldda, id=<encoded_id>}
+        try:
+            src_type = element_identifier.get( 'src', 'hda' )
+        except AttributeError:
+            raise MessageException( "Dataset collection element definition (%s) not dictionary-like." % element_identifier )
+        encoded_id = element_identifier.get( 'id', None )
+        if not src_type or not encoded_id:
+            raise RequestParameterInvalidException( "Problem decoding element identifier %s" % element_identifier )
+
+        if src_type == 'hda':
+            decoded_id = int( trans.app.security.decode_id( encoded_id ) )
+            element = self.hda_manager.get( trans, decoded_id, check_ownership=False )
+        elif src_type == 'ldda':
+            element = self.get_library_dataset_dataset_association( trans, encoded_id )
+        elif src_type == 'hdca':
+            # TODO: Option to copy? Force copy? Copy or allow if not owned?
+            element = self.__get_history_collection_instance( trans, encoded_id ).collection
+        # TODO: ldca.
+        elif src_type == "dc":
+            # TODO: Force only used internally during nested creation so no
+            # need to recheck security.
+            element = self.get_dataset_collection( trans, encoded_id )
+        else:
+            raise RequestParameterInvalidException( "Unknown src_type parameter supplied '%s'." % src_type )
+        return element
+
+    def __type_plugin( self, collection_type ):
+        return self.type_registry.get( collection_type )
+
+    def get_dataset_collection_instance( self, trans, instance_type, id, **kwds ):
+        """
+        """
+        if instance_type == "history":
+            return self.__get_history_collection_instance( trans, id, **kwds )
+        elif instance_type == "library":
+            return self.__get_library_collection_instance( trans, id, **kwds )
+
+    def get_dataset_collection( self, trans, encoded_id ):
+        collection_id = int( trans.app.security.decode_id( encoded_id ) )
+        collection = trans.sa_session.query( trans.app.model.DatasetCollection ).get( collection_id )
+        return collection
+
+    def __get_history_collection_instance( self, trans, id, check_ownership=False, check_accessible=True ):
+        instance_id = int( trans.app.security.decode_id( id ) )
+        collection_instance = trans.sa_session.query( trans.app.model.HistoryDatasetCollectionAssociation ).get( instance_id )
+        self.security_check( trans, collection_instance.history, check_ownership=check_ownership, check_accessible=check_accessible )
+        return collection_instance
+
+    def __get_library_collection_instance( self, trans, id, check_ownership=False, check_accessible=True ):
+        if check_ownership:
+            raise NotImplemented( "Functionality (getting library dataset collection with ownership check) unimplemented." )
+        instance_id = int( trans.security.decode_id( id ) )
+        collection_instance = trans.sa_session.query( trans.app.model.LibraryDatasetCollectionAssociation ).get( instance_id )
+        if check_accessible:
+            if not trans.app.security_agent.can_access_library_item( trans.get_current_user_roles(), collection_instance, trans.user ):
+                raise ItemAccessibilityException( "LibraryDatasetCollectionAssociation is not accessible to the current user", type='error' )
+        return collection_instance
diff --git a/lib/galaxy/dataset_collections/registry.py b/lib/galaxy/dataset_collections/registry.py
new file mode 100644
index 00000000000..66155040e6f
--- /dev/null
+++ b/lib/galaxy/dataset_collections/registry.py
@@ -0,0 +1,14 @@
+from .types import list
+from .types import paired
+
+
+PLUGIN_CLASSES = [list.ListDatasetCollectionType, paired.PairedDatasetCollectionType]
+
+
+class DatasetCollectionTypesRegistry(object):
+
+    def __init__(self, app):
+        self.__plugins = dict( [ ( p.collection_type, p() ) for p in PLUGIN_CLASSES ] )
+
+    def get( self, plugin_type ):
+        return self.__plugins[ plugin_type ]
diff --git a/lib/galaxy/dataset_collections/structure.py b/lib/galaxy/dataset_collections/structure.py
new file mode 100644
index 00000000000..16c393a6fa9
--- /dev/null
+++ b/lib/galaxy/dataset_collections/structure.py
@@ -0,0 +1,81 @@
+""" Module for reasoning about structure of and matching hierarchical collections of data.
+"""
+import logging
+log = logging.getLogger( __name__ )
+
+
+class Leaf( object ):
+
+    def __len__( self ):
+        return 1
+
+    @property
+    def is_leaf( self ):
+        return True
+
+leaf = Leaf()
+
+
+class Tree( object ):
+
+    def __init__( self, dataset_collection ):
+        self.collection_type = dataset_collection.collection_type
+        children = []
+        for element in dataset_collection.elements:
+            child_collection = element.child_collection
+            if child_collection:
+                children.append( ( element.element_identifier, Tree( child_collection )  ) )
+            elif element.hda:
+                children.append( ( element.element_identifier, leaf ) )
+
+        self.children = children
+
+    @property
+    def is_leaf( self ):
+        return False
+
+    def can_match( self, other_structure ):
+        if self.collection_type != other_structure.collection_type:
+            # TODO: generalize
+            return False
+
+        if len( self.children ) != len( other_structure.children ):
+            return False
+
+        for my_child, other_child in zip( self.children, other_structure.children ):
+            if my_child[ 0 ] != other_child[ 0 ]:  # Different identifiers, TODO: generalize
+                return False
+
+            # At least one is nested collection...
+            if my_child[ 1 ].is_leaf != other_child[ 1 ].is_leaf:
+                return False
+
+            if not my_child[ 1 ].is_leaf and not my_child[ 1 ].can_match( other_child[ 1 ]):
+                return False
+
+        return True
+
+    def __len__( self ):
+        return sum( [ len( c[ 1 ] ) for c in self.children ] )
+
+    def element_identifiers_for_datasets( self, trans, datasets ):
+        element_identifiers = []
+        for identifier, child in self.children:
+            if isinstance( child, Tree ):
+                child_identifiers = child.element_identifiers_for_datasets( trans, datasets[ 0:len( child ) ] )
+                child_identifiers[ "name" ] = identifier
+                element_identifiers.append( child_identifiers )
+            else:
+                element_identifiers.append( dict( name=identifier, src="hda", id=trans.security.encode_id( datasets[ 0 ].id ) ) )
+
+            datasets = datasets[ len( child ): ]
+
+        return dict(
+            src="new_collection",
+            collection_type=self.collection_type,
+            element_identifiers=element_identifiers,
+        )
+
+
+def get_structure( dataset_collection_instance ):
+    return Tree( dataset_collection_instance.collection )
diff --git a/lib/galaxy/dataset_collections/types/__init__.py b/lib/galaxy/dataset_collections/types/__init__.py
new file mode 100644
index 00000000000..397993b565f
--- /dev/null
+++ b/lib/galaxy/dataset_collections/types/__init__.py
@@ -0,0 +1,34 @@
+from galaxy import exceptions
+from abc import ABCMeta
+from abc import abstractmethod
+
+from galaxy import model
+
+import logging
+log = logging.getLogger( __name__ )
+
+
+class DatasetCollectionType(object):
+    __metaclass__ = ABCMeta
+
+    @abstractmethod
+    def build_collection( self, dataset_instances ):
+        """
+        Build DatasetCollection with populated DatasetcollectionElement objects
+        corresponding to the supplied dataset instances or throw exception if
+        this is not a valid collection of the specified type.
+        """
+
+
+class BaseDatasetCollectionType( DatasetCollectionType ):
+
+    def _validation_failed( self, message ):
+        raise exceptions.ObjectAttributeInvalidException( message )
+
+    def _new_collection_for_elements( self, elements ):
+        dataset_collection = model.DatasetCollection( )
+        for index, element in enumerate( elements ):
+            element.element_index = index
+            element.collection = dataset_collection
+        dataset_collection.elements = elements
+        return dataset_collection
diff --git a/lib/galaxy/dataset_collections/types/list.py b/lib/galaxy/dataset_collections/types/list.py
new file mode 100644
index 00000000000..03754547180
--- /dev/null
+++ b/lib/galaxy/dataset_collections/types/list.py
@@ -0,0 +1,23 @@
+from ..types import BaseDatasetCollectionType
+
+from galaxy.model import DatasetCollectionElement
+
+
+class ListDatasetCollectionType( BaseDatasetCollectionType ):
+    """ A flat list of named elements.
+    """
+    collection_type = "list"
+
+    def __init__( self ):
+        pass
+
+    def build_collection( self, elements ):
+        associations = []
+        for identifier, element in elements.iteritems():
+            association = DatasetCollectionElement(
+                element=element,
+                element_identifier=identifier,
+            )
+            associations.append( association )
+
+        return self._new_collection_for_elements( associations )
diff --git a/lib/galaxy/dataset_collections/types/paired.py b/lib/galaxy/dataset_collections/types/paired.py
new file mode 100644
index 00000000000..7b3310fcc36
--- /dev/null
+++ b/lib/galaxy/dataset_collections/types/paired.py
@@ -0,0 +1,31 @@
+from ..types import BaseDatasetCollectionType
+
+from galaxy.model import DatasetCollectionElement
+
+LEFT_IDENTIFIER = "left"
+RIGHT_IDENTIFIER = "right"
+
+
+class PairedDatasetCollectionType( BaseDatasetCollectionType ):
+    """
+    Paired (left/right) datasets.
+    """
+    collection_type = "paired"
+
+    def __init__( self ):
+        pass
+
+    def build_collection( self, elements ):
+        left_dataset = elements.get("left", None)
+        right_dataset = elements.get("right", None)
+        if not left_dataset or not right_dataset:
+            self._validation_failed("Paired instance must define 'left' and 'right' datasets .")
+        left_association = DatasetCollectionElement(
+            element=left_dataset,
+            element_identifier=LEFT_IDENTIFIER,
+        )
+        right_association = DatasetCollectionElement(
+            element=right_dataset,
+            element_identifier=RIGHT_IDENTIFIER,
+        )
+        return self._new_collection_for_elements([left_association, right_association])
diff --git a/lib/galaxy/dataset_collections/util.py b/lib/galaxy/dataset_collections/util.py
new file mode 100644
index 00000000000..b2a7f05612c
--- /dev/null
+++ b/lib/galaxy/dataset_collections/util.py
@@ -0,0 +1,52 @@
+from galaxy import exceptions
+from galaxy import web
+from galaxy import model
+
+
+def api_payload_to_create_params( payload ):
+    """
+    Cleanup API payload to pass into dataset_collections.
+    """
+    required_parameters = [ "collection_type", "element_identifiers" ]
+    missing_parameters = [ p for p in required_parameters if p not in payload ]
+    if missing_parameters:
+        message = "Missing required parameters %s" % missing_parameters
+        raise exceptions.ObjectAttributeMissingException( message )
+
+    params = dict(
+        collection_type=payload.get( "collection_type" ),
+        element_identifiers=payload.get( "element_identifiers" ),
+        name=payload.get( "name", None ),
+    )
+
+    return params
+
+
+def dictify_dataset_collection_instance( dataset_colleciton_instance, parent, security, view="element" ):
+    dict_value = dataset_colleciton_instance.to_dict( view=view )
+    encoded_id = security.encode_id( dataset_colleciton_instance.id )
+    if isinstance( parent, model.History ):
+        encoded_history_id = security.encode_id( parent.id )
+        dict_value[ 'url' ] = web.url_for( 'history_content', history_id=encoded_history_id, id=encoded_id, type="dataset_collection" )
+    elif isinstance( parent, model.LibraryFolder ):
+        encoded_library_id = security.encode_id( parent.library.id )
+        encoded_folder_id = security.encode_id( parent.id )
+        # TODO: Work in progress - this end-point is not right yet...
+        dict_value[ 'url' ] = web.url_for( 'library_content', library_id=encoded_library_id, id=encoded_id, folder_id=encoded_folder_id )
+    if view == "element":
+        dict_value[ 'elements' ] = map( dictify_element, dataset_colleciton_instance.collection.elements )
+    security.encode_dict_ids( dict_value )  # TODO: Use Kyle's recusrive formulation of this.
+    return dict_value
+
+
+def dictify_element( element ):
+    dictified = element.to_dict( view="element" )
+    object_detials = element.element_object.to_dict()
+    if element.child_collection:
+        # Recursively yield elements for each nested collection...
+        object_detials[ "elements" ] = map( dictify_element, element.child_collection.elements )
+
+    dictified[ "object" ] = object_detials
+    return dictified
+
+__all__ = [ api_payload_to_create_params, dictify_dataset_collection_instance ]
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index eb2003e9a77..04c019393ff 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -1109,6 +1109,8 @@ class History( object, Dictifiable, UsesAnnotations, HasName ):
         iters = []
         if 'dataset' in types:
             iters.append( self.__dataset_contents_iter( **kwds ) )
+        if 'dataset_collection' in types:
+            iters.append( self.__collection_contents_iter( **kwds ) )
         return galaxy.util.merge_sorted_iterables( operator.attrgetter( "hid" ), *iters )
 
     def __dataset_contents_iter(self, **kwds):
@@ -1138,6 +1140,9 @@ class History( object, Dictifiable, UsesAnnotations, HasName ):
         else:
             return query
 
+    def __collection_contents_iter( self, **kwds ):
+        return self.__filter_contents( HistoryDatasetCollectionAssociation, **kwds )
+
     def copy_tags_from(self,target_user,source_history):
         for src_shta in source_history.tags:
             new_shta = src_shta.copy()
diff --git a/lib/galaxy/security/__init__.py b/lib/galaxy/security/__init__.py
index 141cdd60d13..4e3ef5b850a 100644
--- a/lib/galaxy/security/__init__.py
+++ b/lib/galaxy/security/__init__.py
@@ -563,6 +563,8 @@ class GalaxyRBACAgent( RBACAgent ):
             return self.can_access_library( roles, item.folder.parent_library ) and self.can_access_dataset( roles, item.library_dataset_dataset_association.dataset )
         elif type( item ) == self.model.LibraryDatasetDatasetAssociation:
             return self.can_access_library( roles, item.library_dataset.folder.parent_library ) and self.can_access_dataset( roles, item.dataset )
+        elif type( item ) == self.model.LibraryDatasetCollectionAssociation:
+            return self.can_access_library( roles, item.folder.parent_library )
         else:
             log.warning( 'Unknown library item type: %s' % type( item ) )
             return False
diff --git a/lib/galaxy/web/base/controller.py b/lib/galaxy/web/base/controller.py
index 41571fd433c..28fe73c8091 100644
--- a/lib/galaxy/web/base/controller.py
+++ b/lib/galaxy/web/base/controller.py
@@ -983,6 +983,27 @@ class UsesLibraryMixinItems( SharableItemSecurityMixin ):
         return (  ( trans.user_is_admin() )
                or ( trans.app.security_agent.can_add_library_item( trans.get_current_user_roles(), item ) ) )
 
+    def check_user_can_add_to_library_item( self, trans, item, check_accessible=True ):
+        """
+        Raise exception if user cannot add to the specified library item (i.e.
+        Folder). Can set check_accessible to False if folder was loaded with
+        this check.
+        """
+        if not trans.user:
+            return False
+
+        current_user_roles = trans.get_current_user_roles()
+        if trans.user_is_admin():
+            return True
+
+        if check_accessible:
+            if not trans.app.security_agent.can_access_library_item( current_user_roles, item, trans.user ):
+                raise ItemAccessibilityException( )
+
+        if not trans.app.security_agent.can_add_library_item( trans.get_current_user_roles(), item ):
+            # Slight misuse of ItemOwnershipException?
+            raise ItemOwnershipException( "User cannot add to library item." )
+
     def copy_hda_to_library_folder( self, trans, hda, library_folder, roles=None, ldda_message='' ):
         #PRECONDITION: permissions for this action on hda and library_folder have been checked
         roles = roles or []
diff --git a/lib/galaxy/webapps/galaxy/api/dataset_collections.py b/lib/galaxy/webapps/galaxy/api/dataset_collections.py
new file mode 100644
index 00000000000..0fa2dacd7ee
--- /dev/null
+++ b/lib/galaxy/webapps/galaxy/api/dataset_collections.py
@@ -0,0 +1,75 @@
+from galaxy.web import _future_expose_api as expose_api
+
+from galaxy.web.base.controller import BaseAPIController
+from galaxy.web.base.controller import UsesHistoryMixin
+from galaxy.web.base.controller import UsesLibraryMixinItems
+
+from galaxy.dataset_collections.util import api_payload_to_create_params
+from galaxy.dataset_collections.util import dictify_dataset_collection_instance
+
+from logging import getLogger
+log = getLogger( __name__ )
+
+
+class DatasetCollectionsController(
+    BaseAPIController,
+    UsesHistoryMixin,
+    UsesLibraryMixinItems,
+):
+
+    @expose_api
+    def index( self, trans, **kwd ):
+        trans.response.status = 501
+        return 'not implemented'
+
+    @expose_api
+    def create( self, trans, payload, **kwd ):
+        """
+        * POST /api/dataset_collections:
+            create a new dataset collection instance.
+
+        :type   payload: dict
+        :param  payload: (optional) dictionary structure containing:
+            * collection_type: dataset colltion type to create.
+            * instance_type:   Instance type - 'history' or 'library'.
+            * name:            the new dataset collections's name
+            * datasets:        object describing datasets for collection
+        :rtype:     dict
+        :returns:   element view of new dataset collection
+        """
+        # TODO: Error handling...
+        create_params = api_payload_to_create_params( payload )
+        instance_type = payload.pop( "instance_type", "history" )
+        if instance_type == "history":
+            history_id = payload.get( 'history_id' )
+            history = self.get_history( trans, history_id, check_ownership=True, check_accessible=False )
+            create_params[ "parent" ] = history
+        elif instance_type == "library":
+            folder_id = payload.get( 'folder_id' )
+            library_folder = self.get_library_folder( trans, folder_id, check_accessible=True )
+            self.check_user_can_add_to_library_item( trans, library_folder, check_accessible=False )
+            create_params[ "parent" ] = library_folder
+        else:
+            trans.status = 501
+            return
+        dataset_collection_instance = self.__service( trans ).create( trans=trans, **create_params )
+        return dictify_dataset_collection_instance( dataset_collection_instance, security=trans.security, parent=create_params[ "parent" ] )
+
+    @expose_api
+    def show( self, trans, instance_type, id, **kwds ):
+        dataset_collection_instance = self.__service( trans ).get(
+            id=id,
+            instance_type=instance_type,
+        )
+        if instance_type == 'history':
+            parent = dataset_collection_instance.history
+        elif instance_type == 'library':
+            parent = dataset_collection_instance.folder
+        else:
+            trans.status = 501
+            return
+        return dictify_dataset_collection_instance( trans, dataset_collection_instance, parent )
+
+    def __service( self, trans ):
+        service = trans.app.dataset_collections_service
+        return service
diff --git a/lib/galaxy/webapps/galaxy/api/history_contents.py b/lib/galaxy/webapps/galaxy/api/history_contents.py
index aafb2d5a8e2..a550376a383 100644
--- a/lib/galaxy/webapps/galaxy/api/history_contents.py
+++ b/lib/galaxy/webapps/galaxy/api/history_contents.py
@@ -15,6 +15,9 @@ from galaxy.web.base.controller import UsesLibraryMixin
 from galaxy.web.base.controller import UsesLibraryMixinItems
 from galaxy.web.base.controller import UsesTagsMixin
 
+from galaxy.dataset_collections.util import api_payload_to_create_params
+from galaxy.dataset_collections.util import dictify_dataset_collection_instance
+
 from galaxy.web.base.controller import url_for
 
 from galaxy.managers import histories
@@ -87,7 +90,7 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
         if types:
             types = util.listify(types)
         else:
-            types = [ 'dataset' ]
+            types = [ 'dataset', "dataset_collection" ]
 
         contents_kwds = {'types': types}
         if ids:
@@ -112,7 +115,8 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
                     rval.append( self._detailed_hda_dict( trans, content ) )
                 else:
                     rval.append( self._summary_hda_dict( trans, history_id, content ) )
-
+            elif isinstance(content, trans.app.model.HistoryDatasetCollectionAssociation):
+                rval.append( self.__collection_dict( trans, content ) )
         return rval
 
     #TODO: move to model or Mixin
@@ -142,6 +146,9 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
             'url'   : url_for( 'history_content', history_id=encoded_history_id, id=encoded_id, type="dataset" ),
         }
 
+    def __collection_dict( self, trans, dataset_collection_instance, view="collection" ):
+        return dictify_dataset_collection_instance( dataset_collection_instance, security=trans.security, parent=dataset_collection_instance.history, view=view )
+
     def _detailed_hda_dict( self, trans, hda ):
         """
         Detailed dictionary of hda values.
@@ -177,9 +184,26 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
         contents_type = kwd.get('type', 'dataset')
         if contents_type == 'dataset':
             return self.__show_dataset( trans, id, **kwd )
+        elif contents_type == 'dataset_collection':
+            return self.__show_dataset_collection( trans, id, history_id, **kwd )
         else:
             return self.__handle_unknown_contents_type( trans, contents_type )
 
+    def __show_dataset_collection( self, trans, id, history_id, **kwd ):
+        try:
+            service = trans.app.dataset_collections_service
+            dataset_collection_instance = service.get_dataset_collection_instance(
+                trans=trans,
+                instance_type='history',
+                id=id,
+            )
+            return self.__collection_dict( trans, dataset_collection_instance, view="element" )
+        except Exception, e:
+            msg = "Error in history API at listing dataset collection: %s" % ( str(e) )
+            log.error( msg, exc_info=True )
+            trans.response.status = 500
+            return msg
+
     def __show_dataset( self, trans, id, **kwd ):
         hda = self.mgrs.hdas.get( trans, self._decode_id( trans, id ), check_ownership=False, check_accessible=True )
         #if hda.history.id != self._decode_id( trans, history_id ):
@@ -214,13 +238,20 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
         :rtype:     dict
         :returns:   dictionary containing detailed information for the new HDA
         """
-        #TODO: convert existing, accessible hda - model.DatasetInstance(or hda.datatype).get_converter_types
-        history = self.mgrs.histories.get( trans, self._decode_id( trans, history_id ),
-            check_ownership=True, check_accessible=False )
+        # get the history, if anon user and requesting current history - allow it
+        if( ( trans.user == None )
+        and ( history_id == trans.security.encode_id( trans.history.id ) ) ):
+            history = trans.history
+        # otherwise, check permissions for the history first
+        else:
+            history = self.mgrs.histories.get( trans, self._decode_id( trans, history_id ),
+                check_ownership=True, check_accessible=True )
 
         type = payload.get('type', 'dataset')
         if type == 'dataset':
             return self.__create_dataset( trans, history, payload, **kwd )
+        elif type == 'dataset_collection':
+            return self.__create_dataset_collection( trans, history, payload, **kwd )
         else:
             return self.__handle_unknown_contents_type( trans, type )
 
@@ -260,6 +291,12 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
         hda_dict[ 'display_apps' ] = self.get_display_apps( trans, hda )
         return hda_dict
 
+    def __create_dataset_collection( self, trans, history, payload, **kwd ):
+        create_params = api_payload_to_create_params( payload )
+        service = trans.app.dataset_collections_service
+        dataset_collection_instance = service.create( trans, parent=history, **create_params )
+        return self.__collection_dict( trans, dataset_collection_instance )
+
     @expose_api_anonymous
     def update( self, trans, history_id, id, payload, **kwd ):
         """
@@ -286,6 +323,8 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
         contents_type = kwd.get('type', 'dataset')
         if contents_type == "dataset":
             return self.__update_dataset( trans, history_id, id, payload, **kwd )
+        elif contents_type == "dataset_collection":
+            return self.__update_dataset_collection( trans, history_id, id, payload, **kwd )
         else:
             return self.__handle_unknown_contents_type( trans, contents_type )
 
@@ -325,6 +364,9 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
 
         return changed
 
+    def __update_dataset_collection( self, trans, history_id, id, payload, **kwd ):
+        return trans.app.dataset_collections_service.update( trans, "history", id, payload )
+
     #TODO: allow anonymous del/purge and test security on this
     @expose_api
     def delete( self, trans, history_id, id, purge=False, **kwd ):
@@ -356,6 +398,9 @@ class HistoryContentsController( BaseAPIController, UsesHistoryDatasetAssociatio
         contents_type = kwd.get('type', 'dataset')
         if contents_type == "dataset":
             return self.__delete_dataset( trans, history_id, id, purge=purge, **kwd )
+        elif contents_type == "dataset_collection":
+            trans.app.dataset_collections_service.delete( trans, "history", id )
+            return { 'id' : id, "deleted": True }
         else:
             return self.__handle_unknown_contents_type( trans, contents_type )
 
diff --git a/lib/galaxy/webapps/galaxy/api/library_contents.py b/lib/galaxy/webapps/galaxy/api/library_contents.py
index 117008fb5c9..f5037bd4a7f 100644
--- a/lib/galaxy/webapps/galaxy/api/library_contents.py
+++ b/lib/galaxy/webapps/galaxy/api/library_contents.py
@@ -5,7 +5,8 @@ from galaxy import util
 from galaxy import web
 from galaxy import exceptions
 from galaxy.web import _future_expose_api as expose_api
-from galaxy.web import _future_expose_api_anonymous as expose_api_anonymous
+from galaxy.dataset_collections.util import api_payload_to_create_params
+from galaxy.dataset_collections.util import dictify_dataset_collection_instance
 from galaxy.web.base.controller import BaseAPIController, UsesLibraryMixin, UsesLibraryMixinItems
 from galaxy.web.base.controller import UsesHistoryDatasetAssociationMixin
 from galaxy.web.base.controller import HTTPBadRequest, url_for
@@ -167,7 +168,7 @@ class LibraryContentsController( BaseAPIController, UsesLibraryMixin, UsesLibrar
             return "Missing required 'create_type' parameter."
         else:
             create_type = payload.pop( 'create_type' )
-        if create_type not in ( 'file', 'folder' ):
+        if create_type not in ( 'file', 'folder', 'collection' ):
             trans.response.status = 400
             return "Invalid value for 'create_type' parameter ( %s ) specified." % create_type
 
@@ -202,6 +203,15 @@ class LibraryContentsController( BaseAPIController, UsesLibraryMixin, UsesLibrar
             status, output = trans.webapp.controllers['library_common'].upload_library_dataset( trans, 'api', library_id, real_folder_id, **payload )
         elif create_type == 'folder':
             status, output = trans.webapp.controllers['library_common'].create_folder( trans, 'api', real_folder_id, library_id, **payload )
+        elif create_type == 'collection':
+            # Not delegating to library_common, so need to check access to parent
+            # folder here.
+            self.check_user_can_add_to_library_item( trans, parent, check_accessible=True )
+            create_params = api_payload_to_create_params( payload )
+            create_params[ 'parent' ] = parent
+            service = trans.app.dataset_collections_service
+            dataset_collection_instance = service.create( **create_params )
+            return [ dictify_dataset_collection_instance( dataset_collection_instance, security=trans.security, parent=parent ) ]
         if status != 200:
             trans.response.status = status
             return output
@@ -280,6 +290,8 @@ class LibraryContentsController( BaseAPIController, UsesLibraryMixin, UsesLibrar
             library = self.get_library( trans, library_id, check_accessible=True )
             folder = self.get_library_folder( trans, folder_id, check_accessible=True )
 
+            # TOOD: refactor to use check_user_can_add_to_library_item, eliminate boolean
+            # can_current_user_add_to_library_item.
             if not self.can_current_user_add_to_library_item( trans, folder ):
                 trans.response.status = 403
                 return { 'error' : 'user has no permission to add to library folder (%s)' %( folder_id ) }
diff --git a/lib/galaxy/webapps/galaxy/buildapp.py b/lib/galaxy/webapps/galaxy/buildapp.py
index ae9a3d39051..0727b690ad9 100644
--- a/lib/galaxy/webapps/galaxy/buildapp.py
+++ b/lib/galaxy/webapps/galaxy/buildapp.py
@@ -78,6 +78,7 @@ def app_factory( global_conf, **kwargs ):
 
     valid_history_contents_types = [
         'dataset',
+        'dataset_collection',
     ]
     # This must come before history contents below.
     # Accesss HDA details via histories/:history_id/contents/datasets/:hda_id
@@ -135,6 +136,7 @@ def app_factory( global_conf, **kwargs ):
                                path_prefix='/api/histories/:history_id/contents/:history_content_id' )
 
     webapp.mapper.resource( 'dataset', 'datasets', path_prefix='/api' )
+    webapp.mapper.resource( 'dataset_collection', 'dataset_collections', path_prefix='/api/')
     webapp.mapper.resource( 'sample', 'samples', path_prefix='/api' )
     webapp.mapper.resource( 'request', 'requests', path_prefix='/api' )
     webapp.mapper.resource( 'form', 'forms', path_prefix='/api' )
diff --git a/test/api/test_history_contents.py b/test/api/test_history_contents.py
index 2e394468ebf..063835387a2 100644
--- a/test/api/test_history_contents.py
+++ b/test/api/test_history_contents.py
@@ -3,6 +3,7 @@ import json
 
 from .helpers import TestsDatasets
 from .helpers import LibraryPopulator
+from .test_dataset_collections import DatasetCollectionPopulator
 from base.interactor import (
     put_request,
     delete_request,
@@ -15,6 +16,7 @@ class HistoryContentsApiTestCase( api.ApiTestCase, TestsDatasets ):
     def setUp( self ):
         super( HistoryContentsApiTestCase, self ).setUp()
         self.history_id = self._new_history()
+        self.dataset_collection_populator = DatasetCollectionPopulator( self.galaxy_interactor )
 
     def test_index_hda_summary( self ):
         hda1 = self._new_dataset( self.history_id )
@@ -84,6 +86,48 @@ class HistoryContentsApiTestCase( api.ApiTestCase, TestsDatasets ):
         assert delete_response.status_code < 300  # Something in the 200s :).
         assert str( self.__show( hda1 ).json()[ "deleted" ] ).lower() == "true"
 
+    def test_dataset_collections( self ):
+        payload = self.dataset_collection_populator.create_pair_payload(
+            self.history_id,
+            type="dataset_collection"
+        )
+        pre_collection_count = self.__count_contents( type="dataset_collection" )
+        pre_dataset_count = self.__count_contents( type="dataset" )
+        pre_combined_count = self.__count_contents( type="dataset,dataset_collection" )
+
+        dataset_collection_response = self._post( "histories/%s/contents" % self.history_id, payload )
+
+        self._assert_status_code_is( dataset_collection_response, 200 )
+        dataset_collection = dataset_collection_response.json()
+        self._assert_has_keys( dataset_collection, "url", "name", "deleted" )
+
+        post_collection_count = self.__count_contents( type="dataset_collection" )
+        post_dataset_count = self.__count_contents( type="dataset" )
+        post_combined_count = self.__count_contents( type="dataset,dataset_collection" )
+
+        # Test filtering types with index.
+        assert pre_collection_count == 0
+        assert post_collection_count == 1
+        assert post_combined_count == pre_dataset_count + 1
+        assert post_combined_count == pre_combined_count + 1
+        assert pre_dataset_count == post_dataset_count
+
+        # Test show dataset colleciton.
+        collection_url = "histories/%s/contents/dataset_collections/%s" % ( self.history_id, dataset_collection[ "id" ] )
+        show_response = self._get( collection_url )
+        self._assert_status_code_is( show_response, 200 )
+        dataset_collection = show_response.json()
+        self._assert_has_keys( dataset_collection, "url", "name", "deleted" )
+
+        assert not dataset_collection[ "deleted" ]
+
+        delete_response = delete_request( self._api_url( collection_url, use_key=True ) )
+        self._assert_status_code_is( delete_response, 200 )
+
+        show_response = self._get( collection_url )
+        dataset_collection = show_response.json()
+        assert dataset_collection[ "deleted" ]
+
     def __show( self, hda ):
         show_response = self._get( "histories/%s/contents/%s" % ( self.history_id, hda[ "id" ] ) )
         return show_response
diff --git a/test/functional/api/test_dataset_collections.py b/test/functional/api/test_dataset_collections.py
new file mode 100644
index 00000000000..e1af6974ac8
--- /dev/null
+++ b/test/functional/api/test_dataset_collections.py
@@ -0,0 +1,189 @@
+from base import api
+import json
+from .helpers import DatasetPopulator
+
+
+# TODO: Move into helpers with rest of populators
+class DatasetCollectionPopulator( object ):
+
+    def __init__( self, galaxy_interactor ):
+        self.galaxy_interactor = galaxy_interactor
+        self.dataset_populator = DatasetPopulator( galaxy_interactor )
+
+    def create_pair_in_history( self, history_id, **kwds ):
+        payload = self.create_pair_payload(
+            history_id,
+            instance_type="history",
+            **kwds
+        )
+        return self.__create( payload )
+
+    def create_list_in_history( self, history_id, **kwds ):
+        payload = self.create_list_payload(
+            history_id,
+            instance_type="history",
+            **kwds
+        )
+        return self.__create( payload )
+
+    def create_list_payload( self, history_id, **kwds ):
+        return self.__create_payload( history_id, identifiers_func=self.list_identifiers, collection_type="list", **kwds )
+
+    def create_pair_payload( self, history_id, **kwds ):
+        return self.__create_payload( history_id, identifiers_func=self.pair_identifiers, collection_type="paired", **kwds )
+
+    def __create_payload( self, history_id, identifiers_func, collection_type, **kwds ):
+        contents = None
+        if "contents" in kwds:
+            contents = kwds[ "contents" ]
+            del kwds[ "contents" ]
+
+        if "element_identifiers" not in kwds:
+            kwds[ "element_identifiers" ] = json.dumps( identifiers_func( history_id, contents=contents ) )
+
+        payload = dict(
+            history_id=history_id,
+            collection_type=collection_type,
+            **kwds
+        )
+        return payload
+
+    def pair_identifiers( self, history_id, contents=None ):
+        hda1, hda2 = self.__datasets( history_id, count=2, contents=contents )
+
+        element_identifiers = [
+            dict( name="left", src="hda", id=hda1[ "id" ] ),
+            dict( name="right", src="hda", id=hda2[ "id" ] ),
+        ]
+        return element_identifiers
+
+    def list_identifiers( self, history_id, contents=None ):
+        hda1, hda2, hda3 = self.__datasets( history_id, count=3, contents=contents )
+        element_identifiers = [
+            dict( name="data1", src="hda", id=hda1[ "id" ] ),
+            dict( name="data2", src="hda", id=hda2[ "id" ] ),
+            dict( name="data3", src="hda", id=hda3[ "id" ] ),
+        ]
+        return element_identifiers
+
+    def __create( self, payload ):
+        create_response = self.galaxy_interactor.post( "dataset_collections", data=payload )
+        return create_response
+
+    def __datasets( self, history_id, count, contents=None ):
+        datasets = []
+        for i in xrange( count ):
+            new_kwds = {}
+            if contents:
+                new_kwds[ "content" ] = contents[ i ]
+            datasets.append( self.dataset_populator.new_dataset( history_id, **new_kwds ) )
+        return datasets
+
+
+class DatasetCollectionApiTestCase( api.ApiTestCase ):
+
+    def setUp( self ):
+        super( DatasetCollectionApiTestCase, self ).setUp()
+        self.dataset_populator = DatasetPopulator( self.galaxy_interactor )
+        self.dataset_collection_populator = DatasetCollectionPopulator( self.galaxy_interactor )
+        self.history_id = self.dataset_populator.new_history()
+
+    def test_create_pair_from_history( self ):
+        payload = self.dataset_collection_populator.create_pair_payload(
+            self.history_id,
+            instance_type="history",
+        )
+        create_response = self._post( "dataset_collections", payload )
+        dataset_collection = self._check_create_response( create_response )
+        returned_datasets = dataset_collection[ "elements" ]
+        assert len( returned_datasets ) == 2, dataset_collection
+
+    def test_create_list_from_history( self ):
+        element_identifiers = self.dataset_collection_populator.list_identifiers( self.history_id )
+
+        payload = dict(
+            instance_type="history",
+            history_id=self.history_id,
+            element_identifiers=json.dumps(element_identifiers),
+            collection_type="list",
+        )
+
+        create_response = self._post( "dataset_collections", payload )
+        dataset_collection = self._check_create_response( create_response )
+        returned_datasets = dataset_collection[ "elements" ]
+        assert len( returned_datasets ) == 3, dataset_collection
+
+    def test_create_list_of_existing_pairs( self ):
+        pair_payload = self.dataset_collection_populator.create_pair_payload(
+            self.history_id,
+            instance_type="history",
+        )
+        pair_create_response = self._post( "dataset_collections", pair_payload )
+        dataset_collection = self._check_create_response( pair_create_response )
+        hdca_id = dataset_collection[ "id" ]
+
+        element_identifiers = [
+            dict( name="test1", src="hdca", id=hdca_id )
+        ]
+
+        payload = dict(
+            instance_type="history",
+            history_id=self.history_id,
+            element_identifiers=json.dumps(element_identifiers),
+            collection_type="list",
+        )
+        create_response = self._post( "dataset_collections", payload )
+        dataset_collection = self._check_create_response( create_response )
+        returned_collections = dataset_collection[ "elements" ]
+        assert len( returned_collections ) == 1, dataset_collection
+
+    def test_create_list_of_new_pairs( self ):
+        pair_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id )
+        element_identifiers = [ dict(
+            src="new_collection",
+            name="test_pair",
+            collection_type="paired",
+            element_identifiers=pair_identifiers,
+        ) ]
+        payload = dict(
+            collection_type="list:paired",
+            instance_type="history",
+            history_id=self.history_id,
+            name="nested_collecion",
+            element_identifiers=json.dumps( element_identifiers ),
+        )
+        create_response = self._post( "dataset_collections", payload )
+        dataset_collection = self._check_create_response( create_response )
+        assert dataset_collection[ "collection_type" ] == "list:paired"
+        returned_collections = dataset_collection[ "elements" ]
+        assert len( returned_collections ) == 1, dataset_collection
+        pair_1_element = returned_collections[ 0 ]
+        self._assert_has_keys( pair_1_element, "element_index" )
+        pair_1_object = pair_1_element[ "object" ]
+        self._assert_has_keys( pair_1_object, "collection_type", "elements" )
+        self.assertEquals( pair_1_object[ "collection_type" ], "paired" )
+        pair_elements = pair_1_object[ "elements" ]
+        assert len( pair_elements ) == 2
+        pair_1_element_1 = pair_elements[ 0 ]
+        assert pair_1_element_1[ "element_index" ] == 0
+
+    def test_hda_security( self ):
+        element_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id )
+
+        with self._different_user( ):
+            history_id = self.dataset_populator.new_history()
+            payload = dict(
+                instance_type="history",
+                history_id=history_id,
+                element_identifiers=json.dumps(element_identifiers),
+                collection_type="paired",
+            )
+
+            create_response = self._post( "dataset_collections", payload )
+            self._assert_status_code_is( create_response, 403 )
+
+    def _check_create_response( self, create_response ):
+        self._assert_status_code_is( create_response, 200 )
+        dataset_collection = create_response.json()
+        self._assert_has_keys( dataset_collection, "elements", "url", "name", "collection_type" )
+        return dataset_collection