diff --git a/pubtools/pulplib/_impl/client/search.py b/pubtools/pulplib/_impl/client/search.py
index 4a9b6cd6..219e72a6 100644
--- a/pubtools/pulplib/_impl/client/search.py
+++ b/pubtools/pulplib/_impl/client/search.py
@@ -92,7 +92,7 @@ def accumulate_from_match(self, match_expr):
             raise ValueError(
                 (
                     "Can't serialize criteria for Pulp query; too complicated. "
-                    "Try simplifying the query with respect to content_type_id."
+                    "Try simplifying the query with respect to unit_type/content_type_id."
                 )
             )
 
diff --git a/pubtools/pulplib/_impl/criteria.py b/pubtools/pulplib/_impl/criteria.py
index bf7b0e07..8bea17c1 100644
--- a/pubtools/pulplib/_impl/criteria.py
+++ b/pubtools/pulplib/_impl/criteria.py
@@ -16,6 +16,8 @@
 
 from pubtools.pulplib._impl import compat_attr as attr
 
+from .model.unit import type_ids_for_class
+
 
 class Criteria(object):
     """Represents a Pulp search criteria.
@@ -24,7 +26,7 @@ class Criteria(object):
     or used directly. Instances of this class should be obtained and
     composed by calls to the documented class methods.
 
-    Example:
+    Example - searching a repository:
         .. code-block:: python
 
             # With Pulp 2.x / mongo, this is roughly equivalent
@@ -40,6 +42,18 @@ class Criteria(object):
 
             # criteria may now be used with client to execute a search
             repos = client.search_repository(crit)
+
+    Example - searching across all repos for a specific content type:
+        .. code-block:: python
+
+            crit = Criteria.and_(
+                Criteria.with_unit_type(RpmUnit),
+                Criteria.with_field("sha256sum", Matcher.in_([
+                    "49ae93732fcf8d63fe1cce759664982dbd5b23161f007dba8561862adc96d063",
+                    "6b30e91df993d96df0bef0f9d232d1068fa2f7055f13650208d77b43cd7c99f6"])))
+
+            # Will find RpmUnit instances with above sums
+            units = client.search_content(crit)
     """
 
     exists = object()
@@ -87,6 +101,27 @@ def with_field(cls, field_name, field_value):
         """
         return FieldMatchCriteria(field_name, field_value)
 
+    @classmethod
+    def with_unit_type(cls, unit_type):
+        """Args:
+            unit_type (class)
+                A subclass of :class:`~pubtools.pulplib.Unit`.
+
+        Returns:
+            Criteria
+                criteria for finding units of type ``unit_type``.
+
+        .. versionadded:: 2.14.0
+        """
+
+        # This is just a thin wrapper for searching on content_type_id which allows
+        # the caller to avoid having to handle the (unit class <=> type id) mapping.
+        type_ids = type_ids_for_class(unit_type)
+        if not type_ids:
+            raise TypeError("Expected a Unit type, got: %s" % repr(unit_type))
+
+        return FieldMatchCriteria("content_type_id", Matcher.in_(type_ids))
+
     @classmethod
     def with_field_in(cls, field_name, field_value):
         warnings.warn(
diff --git a/pubtools/pulplib/_impl/model/unit/__init__.py b/pubtools/pulplib/_impl/model/unit/__init__.py
index adbf66b4..c371021f 100644
--- a/pubtools/pulplib/_impl/model/unit/__init__.py
+++ b/pubtools/pulplib/_impl/model/unit/__init__.py
@@ -1,4 +1,4 @@
-from .base import Unit
+from .base import Unit, type_ids_for_class
 from .file import FileUnit
 from .rpm import RpmUnit
 from .modulemd import ModulemdUnit
diff --git a/pubtools/pulplib/_impl/model/unit/base.py b/pubtools/pulplib/_impl/model/unit/base.py
index 70623a21..75d7d5e5 100644
--- a/pubtools/pulplib/_impl/model/unit/base.py
+++ b/pubtools/pulplib/_impl/model/unit/base.py
@@ -19,6 +19,18 @@ def decorate(klass):
     return decorate
 
 
+def type_ids_for_class(unit_class):
+    # Given a concrete Unit subclass, returns those Pulp type id(s)
+    # which may be used to find/load an object of that class.
+    out = []
+
+    for pulp_type, klass in UNIT_CLASSES.items():
+        if klass is unit_class:
+            out.append(pulp_type)
+
+    return sorted(out)
+
+
 @attr.s(kw_only=True, frozen=True)
 class Unit(PulpObject):
     """Represents a Pulp unit (a single piece of content).
diff --git a/setup.py b/setup.py
index 8dc6ae4a..0fa6f036 100644
--- a/setup.py
+++ b/setup.py
@@ -21,7 +21,7 @@ def get_requirements():
 
 setup(
     name="pubtools-pulplib",
-    version="2.13.0",
+    version="2.14.0",
     packages=find_packages(exclude=["tests"]),
     package_data={"pubtools.pulplib._impl.schema": ["*.yaml"]},
     url="https://github.com/release-engineering/pubtools-pulplib",
diff --git a/tests/client/test_client.py b/tests/client/test_client.py
index f35d639a..683d8d8d 100644
--- a/tests/client/test_client.py
+++ b/tests/client/test_client.py
@@ -620,3 +620,60 @@ def test_can_search_content_pagination(client, requests_mocker):
     assert [h.url for h in requests_mocker.request_history].count(
         "https://pulp.example.com/pulp/api/v2/content/units/srpm/search/"
     ) == 2
+
+
+def test_can_search_content_by_type(client, requests_mocker):
+    """search_content can search for a specified unit type."""
+    requests_mocker.get(
+        "https://pulp.example.com/pulp/api/v2/plugins/types/",
+        json=[{"id": "rpm"}, {"id": "srpm"}, {"id": "iso"}],
+    )
+    # Note although "iso" is supported, we don't mock the search URL for
+    # that content type, thus proving that we don't query it if we search
+    # for a specific unit_type
+    requests_mocker.post(
+        "https://pulp.example.com/pulp/api/v2/content/units/rpm/search/",
+        json=RPM_TEST_UNITS,
+    )
+    requests_mocker.post(
+        "https://pulp.example.com/pulp/api/v2/content/units/srpm/search/",
+        json=SRPM_TEST_UNITS,
+    )
+
+    units = client.search_content(Criteria.with_unit_type(RpmUnit))
+
+    # It should have returned the expected units
+    assert sorted(units) == [
+        RpmUnit(
+            content_type_id="srpm",
+            sha256sum="4f5a3a0da6f404f6d9988987cd75f13982bd655a0a4f692406611afbbc597679",
+            arch="src",
+            epoch="0",
+            name="glibc",
+            release="2.57.el4.1",
+            repository_memberships=["fake-repository-id-3"],
+            sourcerpm=None,
+            version="2.3.4",
+        ),
+        RpmUnit(
+            sha256sum="4f5a3a0da6f404f6d9988987cd75f13982bd655a0a4f692406611afbbc597679",
+            arch="ia64",
+            epoch="0",
+            name="glibc-headers",
+            release="2.57.el4.1",
+            repository_memberships=["fake-repository-id-3"],
+            sourcerpm="glibc-2.3.4-2.57.el4.1.src.rpm",
+            version="2.3.4",
+        ),
+        RpmUnit(
+            sha1sum="ca995eb1a635c97393466f67aaec8e9e753b8ed5",
+            sha256sum="1c4baac658fd56e6ec9cca37f440a4bd8c9c0b02a21f41b30b8ea17b402a1907",
+            arch="i386",
+            epoch="0",
+            name="gnu-efi-debuginfo",
+            release="1.1",
+            repository_memberships=["fake-repository-id-1", "fake-repository-id-2"],
+            sourcerpm="gnu-efi-3.0c-1.1.src.rpm",
+            version="3.0c",
+        ),
+    ]
diff --git a/tests/criteria/test_criteria.py b/tests/criteria/test_criteria.py
index e1088530..cd16457a 100644
--- a/tests/criteria/test_criteria.py
+++ b/tests/criteria/test_criteria.py
@@ -13,3 +13,10 @@ def test_field_in_str_invalid():
     with pytest.raises(ValueError) as exc_info:
         Criteria.with_field_in("x", "someval")
     assert "Must be an iterable: 'someval'" in str(exc_info.value)
+
+
+def test_unit_type_invalid():
+    """Criteria.with_unit_type raises if provided value isn't a unit subclass."""
+    with pytest.raises(TypeError) as exc_info:
+        Criteria.with_unit_type([1, 2, 3])
+    assert "Expected a Unit type, got: [1, 2, 3]" in str(exc_info.value)
diff --git a/tests/fake/test_fake_search_content.py b/tests/fake/test_fake_search_content.py
index d41003e2..b85f3cb7 100644
--- a/tests/fake/test_fake_search_content.py
+++ b/tests/fake/test_fake_search_content.py
@@ -120,6 +120,21 @@ def test_search_content_by_unit_field(populated_repo):
     ]
 
 
+def test_search_content_by_unit_type(populated_repo):
+    """search_content on unit_type returns only units of that type"""
+
+    crit = Criteria.with_unit_type(ModulemdUnit)
+    units = list(populated_repo.search_content(crit))
+    assert sorted(units) == [
+        ModulemdUnit(
+            name="module1", stream="s1", version=1234, context="a1b2", arch="x86_64"
+        ),
+        ModulemdUnit(
+            name="module2", stream="s2", version=1234, context="a1b2", arch="x86_64"
+        ),
+    ]
+
+
 def test_search_content_mixed_fields(populated_repo):
     """search_content crossing multiple fields and types returns matching units"""