diff --git a/CHANGES/7887.feature b/CHANGES/7887.feature
new file mode 100644
index 00000000000..615b469d095
--- /dev/null
+++ b/CHANGES/7887.feature
@@ -0,0 +1 @@
+Added Accept-header content negotiation to the content app so clients requesting `application/json` receive a paginated JSON directory listing.
diff --git a/CHANGES/plugin_api/7887.feature b/CHANGES/plugin_api/7887.feature
new file mode 100644
index 00000000000..ff0f9b5ea2a
--- /dev/null
+++ b/CHANGES/plugin_api/7887.feature
@@ -0,0 +1 @@
+Added `Distribution.content_handler_json()` so plugins can serve JSON from the content app when the client prefers `application/json`.
diff --git a/docs/dev/reference/code-api/plugins-api/content-app.md b/docs/dev/reference/code-api/plugins-api/content-app.md
index fc7c2fffc7e..9debae4b098 100644
--- a/docs/dev/reference/code-api/plugins-api/content-app.md
+++ b/docs/dev/reference/code-api/plugins-api/content-app.md
@@ -13,12 +13,36 @@ Making a custom Handler is a two-step process:
 2. Add the Handler to a route using aiohttp.server's [add_route()](https://aiohttp.readthedocs.io/en/stable/web_reference.html#aiohttp.web.UrlDispatcher.add_route) interface.
 
 If content needs to be served from within the `Distribution`'s base_path,
-overriding the `pulpcore.plugin.models.Distribution.content_handler` and
-`pulpcore.plugin.models.Distribution.content_handler_directory_listing`
-methods in your Distribution is an easier way to serve this content. The
-`pulpcore.plugin.models.Distribution.content_handler` method should
-return an instance of `aiohttp.web_response.Response` or a
-`pulpcore.plugin.models.ContentArtifact`.
+overriding `pulpcore.plugin.models.Distribution.content_handler`,
+`content_handler_json`, and `content_handler_list_directory` is an easier
+way to serve this content.
+
+`content_handler` should return an instance of `aiohttp.web_response.Response`
+or a `pulpcore.plugin.models.ContentArtifact`. It is used for the default
+HTML/binary representation.
+
+`content_handler_json` is invoked when the client's `Accept` header prefers
+JSON (see `pulpcore.cache.accept_prefers_json`). Return `None` (the default)
+to use pulpcore's generic paginated JSON directory listing, a JSON-serializable
+dict/list, or an `aiohttp.web.StreamResponse` for full control over
+headers/status. Concrete artifact paths stay binary unless this method returns
+JSON. Missing/`*/*`/`text/html` Accept headers keep today's HTML/binary
+responses.
+
+The generic JSON listing envelope is:
+
+```json
+{
+  "path": "/pulp/content/my-distro/",
+  "packages": [{"path": "subdir/file.iso", "size": 1024, "date": "..."}],
+  "count": 1,
+  "limit": 1000,
+  "offset": 0
+}
+```
+
+Pagination uses `?limit=` and `?offset=` (default limit 1000, max 10000).
+When more pages exist the body also includes `next_offset`.
 
 ## Creating your Handler
 
diff --git a/pulpcore/app/models/publication.py b/pulpcore/app/models/publication.py
index 7de105a50b2..980f8c1f1a8 100644
--- a/pulpcore/app/models/publication.py
+++ b/pulpcore/app/models/publication.py
@@ -740,6 +740,31 @@ def content_handler_list_directory(self, rel_path):
         """
         return set()
 
+    def content_handler_json(self, path):
+        """
+        Handler to serve a JSON representation of the content at ``path`` for this Distribution.
+
+        This is the JSON counterpart to :meth:`content_handler`. It is invoked instead of (and
+        checked before) the generic, plugin-agnostic JSON directory listing whenever the
+        client's ``Accept`` header indicates a preference for JSON over HTML. Plugins override
+        this to provide type-specific JSON (e.g. package metadata, a de-duplicated "package"
+        listing, etc.) rather than falling back to the generic file/size/date listing that
+        pulpcore builds automatically for every Distribution.
+
+        The default implementation returns ``None`` for every path, which is safe for any
+        Distribution subclass that doesn't override it: pulpcore's generic JSON directory
+        listing (or the normal HTML/binary behavior) is used instead.
+
+        Args:
+            path (str): The path being requested
+        Returns:
+            None if there is no JSON representation to serve at path. Otherwise, a
+            JSON-serializable object (dict/list) to be returned to the client, or an
+            aiohttp.web.StreamResponse (e.g. built via aiohttp.web.json_response) for full
+            control over headers/status.
+        """
+        return None
+
     def content_headers_for(self, path):
         """
         Opportunity for Distribution to specify response-headers for a specific path
diff --git a/pulpcore/cache/__init__.py b/pulpcore/cache/__init__.py
index a5a4b9d01ed..29248de33e7 100644
--- a/pulpcore/cache/__init__.py
+++ b/pulpcore/cache/__init__.py
@@ -1,9 +1,13 @@
 # ruff: noqa: F401
 from .cache import (
+    JSON_LIST_DEFAULT_LIMIT,
+    JSON_LIST_MAX_LIMIT,
     AsyncCache,
     AsyncContentCache,
     Cache,
     CacheKeys,
     ConnectionError,
     SyncContentCache,
+    accept_prefers_json,
+    json_listing_pagination,
 )
diff --git a/pulpcore/cache/cache.py b/pulpcore/cache/cache.py
index 6fcf470e14a..34abb4a8583 100644
--- a/pulpcore/cache/cache.py
+++ b/pulpcore/cache/cache.py
@@ -29,6 +29,93 @@ class CacheKeys(enum.Enum):
     path = "path"
     host = "host"
     method = "method"
+    format = "format"
+    query = "query"
+
+
+def accept_prefers_json(accept_header):
+    """
+    Determine whether an HTTP Accept header value prefers application/json over other types.
+
+    A missing/empty header, or one whose highest-quality (per RFC 9110 q-values) entry isn't
+    "application/json" or a "+json" subtype, is treated as "does not prefer JSON". This is the
+    single source of truth for JSON content negotiation in the content app; it lives here (a
+    dependency-free leaf module) rather than in ``pulpcore.content.handler`` so that both the
+    content app's response logic and its cache key (see ``AsyncContentCache.make_key``) can use
+    the exact same decision, avoiding any risk of a JSON response being cached/served for an
+    HTML request or vice versa.
+
+    Args:
+        accept_header (str): The raw value of the request's Accept header, or None.
+
+    Returns:
+        bool: True if the client's top choice is JSON, False otherwise.
+    """
+    if not isinstance(accept_header, str) or not accept_header:
+        return False
+
+    best_type = None
+    best_q = -1.0
+    for part in accept_header.split(","):
+        part = part.strip()
+        if not part:
+            continue
+        media_type, _, params_str = part.partition(";")
+        media_type = media_type.strip().lower()
+        q = 1.0
+        for param in params_str.split(";"):
+            param = param.strip()
+            if param.startswith("q="):
+                try:
+                    q = float(param[2:])
+                except ValueError:
+                    q = 1.0
+        if q > best_q:
+            best_q = q
+            best_type = media_type
+
+    if not best_type or best_q <= 0:
+        return False
+
+    return best_type == "application/json" or best_type.endswith("+json")
+
+
+JSON_LIST_DEFAULT_LIMIT = 1000
+JSON_LIST_MAX_LIMIT = 10000
+
+
+def json_listing_pagination(query):
+    """
+    Parse and bound ``limit``/``offset`` from a request query mapping.
+
+    Invalid or missing values fall back to defaults rather than raising. This is shared by
+    the content app's JSON listing and its cache key so paginated pages cannot collide, and
+    unrecognized query params cannot fragment the cache.
+
+    Args:
+        query: A mapping with ``.get()`` (e.g. aiohttp ``request.query``), or None.
+
+    Returns:
+        tuple: ``(limit, offset)`` integers.
+    """
+
+    def parse_int(name, default, minimum, maximum):
+        if query is None:
+            raw = default
+        else:
+            try:
+                raw = query.get(name, default)
+            except (AttributeError, TypeError):
+                raw = default
+        try:
+            value = int(raw)
+        except (TypeError, ValueError):
+            value = default
+        return max(minimum, min(value, maximum))
+
+    limit = parse_int("limit", JSON_LIST_DEFAULT_LIMIT, 1, JSON_LIST_MAX_LIMIT)
+    offset = parse_int("offset", 0, 0, 2**31 - 1)
+    return limit, offset
 
 
 def connection_error_wrapper(func):
@@ -323,7 +410,10 @@ def __init__(self, base_key=None, expires_ttl=None, keys=None, auth=None):
                       can be a callable taking the request and cache instance as arguments
             expires_ttl: length in seconds entries should live in the cache, EXPIRES_TTL is default
             keys: a list of CacheKeys to use for key creation upon entry placement,
-                (path, method) is default
+                (path, method) is default. Pass CacheKeys.format if responses for the same
+                path/method can differ based on the request's Accept header (e.g. JSON vs.
+                HTML). Pass CacheKeys.query to include normalized JSON ``limit``/``offset``
+                (other query params and HTML requests are ignored).
             auth: a callable to check authorization of the request; takes the request, cache
                   instance, and base_key as arguments.
         """
@@ -444,10 +534,23 @@ async def make_entry(self, key, base_key, handler, args, kwargs, expires=DEFAULT
     def make_key(self, request):
         """Makes the key based off the request"""
         # Might potentially have to make this async if keys require async data from request
+        wants_json = accept_prefers_json(request.headers.get("Accept"))
+        if wants_json:
+            limit, offset = json_listing_pagination(getattr(request, "query", None))
+            query_key = f"{limit}:{offset}"
+        else:
+            query_key = ""
         all_keys = {
             CacheKeys.path: request.path,
             CacheKeys.method: request.method,
             CacheKeys.host: request.url.host,
+            CacheKeys.format: "json" if wants_json else "other",
+            CacheKeys.query: query_key,
         }
-        key = ":".join(all_keys[k] for k in self.keys)
-        return key
+        parts = []
+        for key_name in self.keys:
+            value = all_keys[key_name]
+            if key_name is CacheKeys.query and value == "":
+                continue
+            parts.append(value)
+        return ":".join(parts)
diff --git a/pulpcore/content/handler.py b/pulpcore/content/handler.py
index dcaef910600..5b2728a39ca 100644
--- a/pulpcore/content/handler.py
+++ b/pulpcore/content/handler.py
@@ -1,4 +1,5 @@
 import asyncio
+import json
 import logging
 import os
 import re
@@ -58,7 +59,14 @@
     cache_key,
     get_domain,
 )
-from pulpcore.cache import AsyncContentCache  # noqa: E402
+from pulpcore.cache import (  # noqa: E402
+    JSON_LIST_DEFAULT_LIMIT,
+    JSON_LIST_MAX_LIMIT,
+    AsyncContentCache,
+    CacheKeys,
+    accept_prefers_json,
+    json_listing_pagination,
+)
 from pulpcore.exceptions import (  # noqa: E402
     DigestValidationError,
     UnsupportedDigestValidationError,
@@ -185,6 +193,10 @@ class Handler:
 
     distribution_model = None
 
+    # Defaults/bounds for the ?limit=&offset= pagination of the generic JSON directory listing.
+    DEFAULT_JSON_LIST_LIMIT = JSON_LIST_DEFAULT_LIMIT
+    MAX_JSON_LIST_LIMIT = JSON_LIST_MAX_LIMIT
+
     @staticmethod
     def _reset_db_connection():
         """
@@ -272,6 +284,13 @@ async def auth_cached(cls, request, cached, base_key):
     @AsyncContentCache(
         base_key=lambda req, cac: Handler.find_base_path_cached(req, cac),
         auth=lambda req, cac, bk: Handler.auth_cached(req, cac, bk),
+        # A response for a given path/method can now differ based on the client's Accept
+        # header (JSON vs. HTML/binary), so CacheKeys.format must be part of the cache key.
+        # Without it, a JSON response could be cached and served back for an HTML request
+        # (or vice versa). See pulpcore.cache.accept_prefers_json.
+        # JSON listings are paginated via ?limit=&offset=. CacheKeys.query stores only those
+        # normalized values (and only for JSON), or page N would be served from page 0.
+        keys=(CacheKeys.path, CacheKeys.method, CacheKeys.format, CacheKeys.query),
     )
     async def stream_content(self, request):
         """
@@ -582,6 +601,227 @@ def render_html(directory_list, path="", dates=None, sizes=None):
             sizes=sizes,
         )
 
+    @staticmethod
+    def negotiate_json(request):
+        """
+        Determine whether the client's Accept header prefers application/json over text/html.
+
+        A missing Accept header, or one whose highest-priority entry is "*/*" or some other
+        non-JSON type, is treated as "not asking for JSON" so that existing clients (browsers,
+        package managers, etc.) keep receiving today's HTML/binary responses unchanged.
+
+        This delegates to :func:`pulpcore.cache.accept_prefers_json`, which is also used by
+        :meth:`pulpcore.cache.AsyncContentCache.make_key` (via ``CacheKeys.format``) to key
+        cache entries by negotiated representation. Keeping a single implementation guarantees
+        the caching layer and this negotiation decision can never disagree.
+
+        Args:
+            request (aiohttp.web.Request): The incoming request.
+
+        Returns:
+            bool: True if the highest-quality (per RFC 9110 q-values) media type the client
+                accepts is "application/json" or a "+json" subtype, False otherwise.
+        """
+        return accept_prefers_json(request.headers.get("Accept"))
+
+    @staticmethod
+    def _pagination_params(request):
+        """
+        Parse and bound the ?limit=&offset= query params used by the generic JSON listing.
+
+        Invalid or missing values fall back to defaults rather than raising, since pagination
+        parameters should never be the reason a request fails.
+
+        Args:
+            request (aiohttp.web.Request): The incoming request.
+
+        Returns:
+            (int, int): A (limit, offset) tuple.
+        """
+        return json_listing_pagination(getattr(request, "query", None))
+
+    @staticmethod
+    def _json_response(data):
+        """
+        Wrap a JSON-serializable object as a JSON HTTP response, passing responses through as-is.
+
+        Args:
+            data: Either an aiohttp.web.StreamResponse (returned as-is, giving a Distribution
+                full control over headers/status) or a JSON-serializable object (dict/list).
+
+        Returns:
+            aiohttp.web.StreamResponse: The response to return to the client.
+        """
+        if isinstance(data, StreamResponse):
+            return data
+        return HTTPOk(
+            headers={"Content-Type": "application/json", "Vary": "Accept"},
+            text=json.dumps(data, default=str),
+        )
+
+    @staticmethod
+    def _json_listing_response(request_path, entries, total, limit, offset):
+        """
+        Build the JSON response envelope for the generic, recursive directory listing.
+
+        Args:
+            request_path (str): The full request path, echoed back for client convenience.
+            entries (list): List of {"path", "size", "date"} dicts, already paginated.
+            total (int): Total number of matching entries before pagination was applied.
+            limit (int): The limit that was applied.
+            offset (int): The offset that was applied.
+
+        Returns:
+            aiohttp.web.HTTPOk: The JSON response.
+        """
+        body = {
+            "path": request_path,
+            "packages": entries,
+            "count": total,
+            "limit": limit,
+            "offset": offset,
+        }
+        if offset + len(entries) < total:
+            body["next_offset"] = offset + len(entries)
+        return HTTPOk(
+            headers={"Content-Type": "application/json", "Vary": "Accept"},
+            text=json.dumps(body, default=str),
+        )
+
+    async def list_directory_flat(self, repo_version, publication, path, limit, offset):
+        """
+        Generate a flat, recursive listing of every actual file under ``path``.
+
+        Unlike :meth:`list_directory`, this does not collapse nested files down to their
+        first path segment (i.e. it never synthesizes directory placeholders): every leaf
+        file anywhere below ``path`` is returned with its full path relative to ``path``.
+        This backs the generic, plugin-agnostic JSON representation of a distribution's
+        contents (see :meth:`negotiate_json`), so that a client can request the root (or
+        any subdirectory) of a distribution and get every package beneath it without having
+        to walk the directory tree itself.
+
+        Pagination (``limit``/``offset``) is applied in the database: only the requested page
+        of paths is loaded, and repository-membership dates are fetched for that page's
+        content IDs.
+
+        Args:
+            repo_version (pulpcore.app.models.RepositoryVersion) The repository version
+            publication (pulpcore.app.models.Publication) Publication
+            path (str): relative path inside the repo version or publication.
+            limit (int): Maximum number of entries to return.
+            offset (int): Number of matching entries (sorted by path) to skip.
+
+        Returns:
+            (list, int): A tuple of (entries, total) where entries is a list of
+                {"path": str, "size": int|None, "date": str|None} dicts already sliced to
+                [offset:offset + limit] and sorted by path, and total is the number of
+                matching entries before slicing (for pagination).
+        """
+
+        def list_directory_flat_blocking():
+            if not publication and not repo_version:
+                raise Exception("Either a repo_version or publication is required.")
+            if publication and repo_version:
+                raise Exception("Either a repo_version or publication can be specified.")
+            content_repo_ver = repo_version or publication.repository_version
+            use_content_artifacts = bool(repo_version or publication.pass_through)
+
+            pub_paths = None
+            if publication:
+                pub_paths = publication.published_artifact.filter(
+                    relative_path__startswith=path
+                ).values_list("relative_path", flat=True)
+            ca_paths = None
+            if use_content_artifacts:
+                ca_paths = ContentArtifact.objects.filter(
+                    content__in=content_repo_ver.content,
+                    relative_path__startswith=path,
+                ).values_list("relative_path", flat=True)
+
+            if pub_paths is not None and ca_paths is not None:
+                path_qs = pub_paths.union(ca_paths).order_by("relative_path")
+            elif ca_paths is not None:
+                path_qs = ca_paths.distinct().order_by("relative_path")
+            else:
+                path_qs = pub_paths.order_by("relative_path")
+
+            total = path_qs.count()
+            if total == 0:
+                return [], 0
+
+            page_full_paths = list(path_qs[offset : offset + limit])
+            if not page_full_paths:
+                return [], total
+
+            # relative_path -> {date, size, content_id, ca_pk}. ContentArtifact overwrites
+            # PublishedArtifact for the same path (same last-write-wins as the unpaginated loop).
+            details = {
+                rel: {"date": None, "size": None, "content_id": None, "ca_pk": None}
+                for rel in page_full_paths
+            }
+
+            if publication:
+                for pa in publication.published_artifact.select_related(
+                    "content_artifact__artifact"
+                ).filter(relative_path__in=page_full_paths):
+                    details[pa.relative_path]["date"] = pa.pulp_created
+                    details[pa.relative_path]["content_id"] = pa.content_artifact.content_id
+                    if pa.content_artifact.artifact:
+                        details[pa.relative_path]["size"] = pa.content_artifact.artifact.size
+                    else:
+                        details[pa.relative_path]["ca_pk"] = pa.content_artifact.pk
+
+            if use_content_artifacts:
+                for ca in ContentArtifact.objects.select_related("artifact").filter(
+                    content__in=content_repo_ver.content,
+                    relative_path__in=page_full_paths,
+                ):
+                    details[ca.relative_path]["date"] = ca.pulp_created
+                    details[ca.relative_path]["content_id"] = ca.content_id
+                    if ca.artifact:
+                        details[ca.relative_path]["size"] = ca.artifact.size
+                        details[ca.relative_path]["ca_pk"] = None
+                    else:
+                        details[ca.relative_path]["ca_pk"] = ca.pk
+
+            content_ids = [
+                item["content_id"] for item in details.values() if item["content_id"] is not None
+            ]
+            if content_ids:
+                content_id_to_path = {
+                    item["content_id"]: rel
+                    for rel, item in details.items()
+                    if item["content_id"] is not None
+                }
+                for rc in content_repo_ver._content_relationships().filter(
+                    content_id__in=content_ids
+                ):
+                    rel = content_id_to_path.get(rc.content_id)
+                    if rel is not None:
+                        details[rel]["date"] = rc.pulp_created
+
+            artifacts_to_find = {
+                item["ca_pk"]: rel for rel, item in details.items() if item["ca_pk"] is not None
+            }
+            if artifacts_to_find:
+                r_artifacts = RemoteArtifact.objects.filter(
+                    content_artifact__in=artifacts_to_find.keys(), size__isnull=False
+                ).values_list("content_artifact_id", "size")
+                for ca_pk, size in r_artifacts:
+                    details[artifacts_to_find[ca_pk]]["size"] = size
+
+            result = [
+                {
+                    "path": rel[len(path) :],
+                    "size": details[rel]["size"],
+                    "date": (details[rel]["date"].isoformat() if details[rel]["date"] else None),
+                }
+                for rel in page_full_paths
+            ]
+            return result, total
+
+        return await sync_to_async(list_directory_flat_blocking)()
+
     async def list_directory(self, repo_version, publication, path):
         """
         Generate a set with directory listing of the path.
@@ -683,6 +923,13 @@ async def _match_and_stream(self, path, request):
         Finally, when nothing is served to client yet, we check if there is a remote for the
         Distribution. If so, the Artifact is pulled from the remote and streamed to the client.
 
+        If the client's ``Accept`` header prefers JSON (see :meth:`negotiate_json`), this method
+        instead calls :meth:`Distribution.content_handler_json` for a plugin-specific JSON
+        representation of ``path``; if that returns None, it falls back to a generic, recursive
+        JSON listing of every file at or below ``path`` (see :meth:`list_directory_flat`) when
+        ``path`` resolves to a directory. Concrete artifact paths are unaffected by ``Accept``
+        unless a plugin's ``content_handler_json`` explicitly handles them.
+
         Args:
             path (str): The path component of the URL.
             request(aiohttp.web.Request) The request to prepare a response for.
@@ -717,6 +964,11 @@ async def _match_and_stream(self, path, request):
 
         headers = self.response_headers(original_rel_path, distro)
 
+        # Determine, once, whether the client is asking for a JSON representation of this path
+        # rather than the default HTML/binary behavior. See negotiate_json() for details.
+        wants_json = self.negotiate_json(request)
+        json_limit, json_offset = self._pagination_params(request) if wants_json else (None, None)
+
         content_handler_result = await sync_to_async(distro.content_handler)(original_rel_path)
         if content_handler_result is not None:
             if isinstance(content_handler_result, ContentArtifact):
@@ -732,6 +984,13 @@ async def _match_and_stream(self, path, request):
                 # the result is a response so just return it
                 return content_handler_result
 
+        if wants_json:
+            content_handler_json_result = await sync_to_async(distro.content_handler_json)(
+                original_rel_path
+            )
+            if content_handler_json_result is not None:
+                return self._json_response(content_handler_json_result)
+
         if distro.checkpoint:
             repository = repo_version = None
             publication = await sync_to_async(self._select_checkpoint_publication)(
@@ -746,30 +1005,42 @@ async def _match_and_stream(self, path, request):
             )()
 
         if publication:
-            try:
-                index_path = "{}index.html".format(rel_path)
-
-                await publication.published_artifact.aget(relative_path=index_path)
-                if not ends_in_slash:
-                    # index.html found, but user didn't specify a trailing slash
-                    raise HTTPMovedPermanently(f"{request.path}/")
-                original_rel_path = index_path
-                headers = self.response_headers(original_rel_path, distro)
-            except ObjectDoesNotExist:
-                dir_list, dates, sizes = await self.list_directory(None, publication, rel_path)
-                dir_list.update(
-                    await sync_to_async(distro.content_handler_list_directory)(rel_path)
+            if wants_json:
+                entries, total = await self.list_directory_flat(
+                    None, publication, rel_path, json_limit, json_offset
                 )
-                if dir_list and not ends_in_slash:
-                    # Directory can be listed, but user did not specify trailing slash
-                    raise HTTPMovedPermanently(f"{request.path}/")
-                elif dir_list:
-                    return HTTPOk(
-                        headers={"Content-Type": "text/html"},
-                        text=self.render_html(
-                            dir_list, path=request.path, dates=dates, sizes=sizes
-                        ),
+                if total:
+                    if not ends_in_slash:
+                        # Directory can be listed, but user did not specify trailing slash
+                        raise HTTPMovedPermanently(f"{request.path}/")
+                    return self._json_listing_response(
+                        request.path, entries, total, json_limit, json_offset
                     )
+            else:
+                try:
+                    index_path = "{}index.html".format(rel_path)
+
+                    await publication.published_artifact.aget(relative_path=index_path)
+                    if not ends_in_slash:
+                        # index.html found, but user didn't specify a trailing slash
+                        raise HTTPMovedPermanently(f"{request.path}/")
+                    original_rel_path = index_path
+                    headers = self.response_headers(original_rel_path, distro)
+                except ObjectDoesNotExist:
+                    dir_list, dates, sizes = await self.list_directory(None, publication, rel_path)
+                    dir_list.update(
+                        await sync_to_async(distro.content_handler_list_directory)(rel_path)
+                    )
+                    if dir_list and not ends_in_slash:
+                        # Directory can be listed, but user did not specify trailing slash
+                        raise HTTPMovedPermanently(f"{request.path}/")
+                    elif dir_list:
+                        return HTTPOk(
+                            headers={"Content-Type": "text/html", "Vary": "Accept"},
+                            text=self.render_html(
+                                dir_list, path=request.path, dates=dates, sizes=sizes
+                            ),
+                        )
 
             # published artifact
             try:
@@ -832,30 +1103,42 @@ async def _match_and_stream(self, path, request):
                     )
 
         if repo_version and not publication and not distro.SERVE_FROM_PUBLICATION:
-            # Look for index.html or list the directory
-            index_path = "{}index.html".format(rel_path)
-
-            contentartifact_exists = await ContentArtifact.objects.filter(
-                content__in=repo_version.content, relative_path=index_path
-            ).aexists()
-            if contentartifact_exists:
-                original_rel_path = index_path
-                headers = self.response_headers(original_rel_path, distro)
-            else:
-                dir_list, dates, sizes = await self.list_directory(repo_version, None, rel_path)
-                dir_list.update(
-                    await sync_to_async(distro.content_handler_list_directory)(rel_path)
+            if wants_json:
+                entries, total = await self.list_directory_flat(
+                    repo_version, None, rel_path, json_limit, json_offset
                 )
-                if dir_list and not ends_in_slash:
-                    # Directory can be listed, but user did not specify trailing slash
-                    raise HTTPMovedPermanently(f"{request.path}/")
-                elif dir_list:
-                    return HTTPOk(
-                        headers={"Content-Type": "text/html"},
-                        text=self.render_html(
-                            dir_list, path=request.path, dates=dates, sizes=sizes
-                        ),
+                if total:
+                    if not ends_in_slash:
+                        # Directory can be listed, but user did not specify trailing slash
+                        raise HTTPMovedPermanently(f"{request.path}/")
+                    return self._json_listing_response(
+                        request.path, entries, total, json_limit, json_offset
                     )
+            else:
+                # Look for index.html or list the directory
+                index_path = "{}index.html".format(rel_path)
+
+                contentartifact_exists = await ContentArtifact.objects.filter(
+                    content__in=repo_version.content, relative_path=index_path
+                ).aexists()
+                if contentartifact_exists:
+                    original_rel_path = index_path
+                    headers = self.response_headers(original_rel_path, distro)
+                else:
+                    dir_list, dates, sizes = await self.list_directory(repo_version, None, rel_path)
+                    dir_list.update(
+                        await sync_to_async(distro.content_handler_list_directory)(rel_path)
+                    )
+                    if dir_list and not ends_in_slash:
+                        # Directory can be listed, but user did not specify trailing slash
+                        raise HTTPMovedPermanently(f"{request.path}/")
+                    elif dir_list:
+                        return HTTPOk(
+                            headers={"Content-Type": "text/html", "Vary": "Accept"},
+                            text=self.render_html(
+                                dir_list, path=request.path, dates=dates, sizes=sizes
+                            ),
+                        )
 
             try:
                 ca = await ContentArtifact.objects.select_related(
diff --git a/pulpcore/plugin/cache/__init__.py b/pulpcore/plugin/cache/__init__.py
index 6095a18baf4..fae29643ab4 100644
--- a/pulpcore/plugin/cache/__init__.py
+++ b/pulpcore/plugin/cache/__init__.py
@@ -1,3 +1,3 @@
 # ruff: noqa: F401
 # isort: skip_file
-from pulpcore.cache import CacheKeys, AsyncContentCache, SyncContentCache
+from pulpcore.cache import CacheKeys, AsyncContentCache, SyncContentCache, accept_prefers_json
diff --git a/pulpcore/tests/functional/api/using_plugin/test_content_json_listing.py b/pulpcore/tests/functional/api/using_plugin/test_content_json_listing.py
new file mode 100644
index 00000000000..564ed18cf1c
--- /dev/null
+++ b/pulpcore/tests/functional/api/using_plugin/test_content_json_listing.py
@@ -0,0 +1,167 @@
+"""Tests for content-app Accept negotiation and JSON directory listing."""
+
+import json
+from urllib.parse import urljoin
+
+import pytest
+import requests
+
+from pulpcore.tests.functional.utils import download_file
+
+JSON_ACCEPT = {"Accept": "application/json"}
+HTML_ACCEPT = {"Accept": "text/html"}
+
+
+def _add_files_to_repo(file_bindings, repo, contents, monitor_task):
+    monitor_task(
+        file_bindings.RepositoriesFileApi.modify(
+            repo.pulp_href,
+            {"add_content_units": [content.pulp_href for content in contents]},
+        ).task
+    )
+    return file_bindings.RepositoriesFileApi.read(repo.pulp_href)
+
+
+@pytest.mark.parallel
+def test_json_vs_html_listing_and_artifact(
+    file_bindings,
+    file_repo_with_auto_publish,
+    file_content_unit_with_name_factory,
+    file_distribution_factory,
+    distribution_base_url,
+    monitor_task,
+):
+    """JSON listing is recursive; default Accept stays HTML; artifacts stay binary."""
+    root_file = file_content_unit_with_name_factory("a.iso")
+    nested_file = file_content_unit_with_name_factory("subdir/b.iso")
+    repo = _add_files_to_repo(
+        file_bindings,
+        file_repo_with_auto_publish,
+        [root_file, nested_file],
+        monitor_task,
+    )
+    distro = file_distribution_factory(repository=repo.pulp_href)
+    distro_url = distribution_base_url(distro.base_url)
+
+    json_listing = download_file(distro_url, headers=JSON_ACCEPT)
+    assert "application/json" in json_listing.response_obj.headers["Content-Type"]
+    assert json_listing.response_obj.headers.get("Vary") == "Accept"
+    body = json.loads(json_listing.body)
+    assert body["path"].rstrip("/").endswith(distro.base_path)
+    listed_paths = [pkg["path"] for pkg in body["packages"]]
+    assert "a.iso" in listed_paths
+    assert "subdir/b.iso" in listed_paths
+    assert "subdir/" not in listed_paths
+    assert body["count"] == len(body["packages"])
+    assert body["limit"] == 1000
+    assert body["offset"] == 0
+    assert "next_offset" not in body
+
+    html_listing = download_file(distro_url, headers=HTML_ACCEPT)
+    html = html_listing.body.decode("utf-8")
+    assert "text/html" in html_listing.response_obj.headers["Content-Type"]
+    assert html_listing.response_obj.headers.get("Vary") == "Accept"
+    assert '<a href="./a.iso">' in html
+    assert '<a href="./subdir/">' in html
+    assert "./subdir/b.iso" not in html
+
+    default_listing = download_file(distro_url)
+    assert "text/html" in default_listing.response_obj.headers["Content-Type"]
+
+    artifact = download_file(urljoin(distro_url, "a.iso"), headers=JSON_ACCEPT)
+    assert "application/json" not in artifact.response_obj.headers.get("Content-Type", "")
+    assert artifact.body != json_listing.body
+
+
+@pytest.mark.parallel
+def test_json_listing_pagination_and_invalid_params(
+    file_bindings,
+    file_repo_with_auto_publish,
+    file_content_unit_with_name_factory,
+    file_distribution_factory,
+    distribution_base_url,
+    monitor_task,
+):
+    contents = [file_content_unit_with_name_factory(f"{i}.iso") for i in range(3)]
+    repo = _add_files_to_repo(file_bindings, file_repo_with_auto_publish, contents, monitor_task)
+    distro = file_distribution_factory(repository=repo.pulp_href)
+    distro_url = distribution_base_url(distro.base_url)
+
+    page = json.loads(download_file(f"{distro_url}?limit=1&offset=0", headers=JSON_ACCEPT).body)
+    assert page["limit"] == 1
+    assert page["offset"] == 0
+    assert len(page["packages"]) == 1
+    assert page["count"] >= 3
+    assert page["next_offset"] == 1
+
+    next_page = json.loads(
+        download_file(
+            f"{distro_url}?limit=1&offset={page['next_offset']}", headers=JSON_ACCEPT
+        ).body
+    )
+    assert next_page["offset"] == 1
+    assert next_page["packages"][0]["path"] != page["packages"][0]["path"]
+
+    invalid = json.loads(
+        download_file(f"{distro_url}?limit=nope&offset=nope", headers=JSON_ACCEPT).body
+    )
+    assert invalid["limit"] == 1000
+    assert invalid["offset"] == 0
+
+
+@pytest.mark.parallel
+def test_json_listing_trailing_slash_redirect(
+    file_bindings,
+    file_repo_with_auto_publish,
+    file_content_unit_with_name_factory,
+    file_distribution_factory,
+    distribution_base_url,
+    monitor_task,
+):
+    nested = file_content_unit_with_name_factory("subdir/b.iso")
+    repo = _add_files_to_repo(file_bindings, file_repo_with_auto_publish, [nested], monitor_task)
+    distro = file_distribution_factory(repository=repo.pulp_href)
+    distro_url = distribution_base_url(distro.base_url)
+    no_slash_url = urljoin(distro_url, "subdir")
+
+    redirect = requests.get(no_slash_url, headers=JSON_ACCEPT, allow_redirects=False, verify=False)
+    assert redirect.status_code == 301
+    assert redirect.headers["Location"].endswith("subdir/")
+
+    listed = download_file(urljoin(distro_url, "subdir/"), headers=JSON_ACCEPT)
+    body = json.loads(listed.body)
+    assert body["packages"]
+    assert body["packages"][0]["path"] == "b.iso"
+
+
+@pytest.mark.parallel
+def test_json_and_html_listings_are_cached_separately(
+    file_bindings,
+    file_repo_with_auto_publish,
+    file_content_unit_with_name_factory,
+    file_distribution_factory,
+    distribution_base_url,
+    monitor_task,
+    redis_status,
+):
+    if not redis_status:
+        pytest.xfail("Could not connect to the Redis server")
+
+    content = file_content_unit_with_name_factory("a.iso")
+    repo = _add_files_to_repo(file_bindings, file_repo_with_auto_publish, [content], monitor_task)
+    distro = file_distribution_factory(repository=repo.pulp_href)
+    distro_url = distribution_base_url(distro.base_url)
+
+    json_miss = download_file(distro_url, headers=JSON_ACCEPT)
+    json_hit = download_file(distro_url, headers=JSON_ACCEPT)
+    html_miss = download_file(distro_url, headers=HTML_ACCEPT)
+    html_hit = download_file(distro_url, headers=HTML_ACCEPT)
+
+    assert json_miss.response_obj.headers.get("X-PULP-CACHE") == "MISS"
+    assert json_hit.response_obj.headers.get("X-PULP-CACHE") == "HIT"
+    assert html_miss.response_obj.headers.get("X-PULP-CACHE") == "MISS"
+    assert html_hit.response_obj.headers.get("X-PULP-CACHE") == "HIT"
+    assert "application/json" in json_hit.response_obj.headers["Content-Type"]
+    assert "text/html" in html_hit.response_obj.headers["Content-Type"]
+    assert json.loads(json_hit.body)["packages"]
+    assert b"<html" in html_hit.body.lower() or b"<a href=" in html_hit.body.lower()
diff --git a/pulpcore/tests/unit/content/test_handler.py b/pulpcore/tests/unit/content/test_handler.py
index 045efdad6cd..08a8f652a51 100644
--- a/pulpcore/tests/unit/content/test_handler.py
+++ b/pulpcore/tests/unit/content/test_handler.py
@@ -1,10 +1,14 @@
+import hashlib
+import json
 import uuid
 from datetime import timedelta
 from unittest.mock import AsyncMock, Mock
 
 import pytest
 import pytest_asyncio
+from aiohttp.web import HTTPOk, StreamResponse
 from aiohttp.web_exceptions import HTTPMovedPermanently
+from asgiref.sync import async_to_sync
 from django.db import IntegrityError
 from django_guid import clear_guid, set_guid
 
@@ -17,6 +21,7 @@
     ContentArtifact,
     Distribution,
     Publication,
+    PublishedArtifact,
     Remote,
     RemoteArtifact,
     Repository,
@@ -607,3 +612,140 @@ async def test_async_pull_through_add(ca1, monkeypatch, app_status):
         await repo.adelete()
         if task:
             await task.adelete()
+
+
+def test_content_handler_json_default_returns_none():
+    assert Distribution().content_handler_json("any/path") is None
+
+
+@pytest.mark.parametrize(
+    "accept,expected",
+    [
+        (None, False),
+        ("text/html", False),
+        ("*/*", False),
+        ("application/json", True),
+        ("application/vnd.pypi.simple.v1+json", True),
+    ],
+)
+def test_negotiate_json(accept, expected):
+    request = Mock(headers={} if accept is None else {"Accept": accept})
+    assert Handler.negotiate_json(request) is expected
+
+
+def test_pagination_params_defaults_and_bounds():
+    request = Mock(query={})
+    assert Handler._pagination_params(request) == (Handler.DEFAULT_JSON_LIST_LIMIT, 0)
+
+    request = Mock(query={"limit": "nope", "offset": "also-nope"})
+    assert Handler._pagination_params(request) == (Handler.DEFAULT_JSON_LIST_LIMIT, 0)
+
+    request = Mock(query={"limit": "0", "offset": "-5"})
+    assert Handler._pagination_params(request) == (1, 0)
+
+    request = Mock(query={"limit": "99999", "offset": "10"})
+    assert Handler._pagination_params(request) == (Handler.MAX_JSON_LIST_LIMIT, 10)
+
+
+def test_json_listing_response_envelope_and_next_offset():
+    entries = [{"path": "a.iso", "size": 1, "date": "2026-01-01T00:00:00"}]
+    response = Handler._json_listing_response("/pulp/content/foo/", entries, 3, 1, 0)
+    assert isinstance(response, HTTPOk)
+    assert "application/json" in response.headers["Content-Type"]
+    assert response.headers["Vary"] == "Accept"
+    body = json.loads(response.text)
+    assert body["path"] == "/pulp/content/foo/"
+    assert body["packages"] == entries
+    assert body["count"] == 3
+    assert body["limit"] == 1
+    assert body["offset"] == 0
+    assert body["next_offset"] == 1
+
+    last_page = Handler._json_listing_response("/pulp/content/foo/", entries, 1, 1, 0)
+    assert "next_offset" not in json.loads(last_page.text)
+
+
+def test_json_response_passes_through_stream_response():
+    original = StreamResponse()
+    assert Handler._json_response(original) is original
+
+    response = Handler._json_response({"hello": "world"})
+    assert "application/json" in response.headers["Content-Type"]
+    assert json.loads(response.text) == {"hello": "world"}
+
+
+@pytest.mark.asyncio
+async def test_content_handler_json_hook_is_used_when_accept_prefers_json():
+    """A Distribution.content_handler_json override is returned as the JSON response."""
+    handler = Handler()
+    distro = Mock()
+    distro.base_path = "foo"
+    distro.checkpoint = False
+    distro.content_handler.return_value = None
+    distro.content_handler_json.return_value = {"packages": ["from-plugin"]}
+    distro.content_headers_for.return_value = {}
+    handler._match_distribution = Mock(return_value=distro)
+    handler._permit = Mock(return_value=False)
+
+    request = Mock()
+    request.headers = {"Accept": "application/json"}
+    request.path = "/pulp/content/foo/"
+    request.query = {}
+
+    response = await handler._match_and_stream("foo/", request)
+
+    distro.content_handler_json.assert_called_once_with("")
+    assert "application/json" in response.headers["Content-Type"]
+    assert json.loads(response.text) == {"packages": ["from-plugin"]}
+
+
+@pytest.mark.django_db
+def test_list_directory_flat_recursive_leaves_and_pagination():
+    from pulp_file.app.models import FileContent, FilePublication, FileRepository
+
+    repo = FileRepository.objects.create(name=str(uuid.uuid4()))
+    paths = ["a.iso", "subdir/b.iso"]
+    with repo.new_version() as repo_version:
+        for path in paths:
+            digest = hashlib.sha256(path.encode()).hexdigest()
+            content = FileContent.objects.create(relative_path=path, digest=digest)
+            repo_version.add_content(Content.objects.filter(pk=content.pk))
+            ContentArtifact.objects.create(content=content, relative_path=path)
+
+    publication = FilePublication.objects.create(repository_version=repo_version, complete=True)
+    for ca in ContentArtifact.objects.filter(content__in=repo_version.content.all()):
+        PublishedArtifact.objects.create(
+            publication=publication, content_artifact=ca, relative_path=ca.relative_path
+        )
+
+    handler = Handler()
+    list_flat = async_to_sync(handler.list_directory_flat)
+    entries, total = list_flat(None, publication, "", 1000, 0)
+    listed_paths = [entry["path"] for entry in entries]
+    assert total == 2
+    assert listed_paths == ["a.iso", "subdir/b.iso"]
+    assert "subdir/" not in listed_paths
+    assert all("size" in entry and "date" in entry for entry in entries)
+
+    nested, nested_total = list_flat(None, publication, "subdir/", 1000, 0)
+    assert nested_total == 1
+    assert nested[0]["path"] == "b.iso"
+
+    page, page_total = list_flat(None, publication, "", 1, 0)
+    assert page_total == 2
+    assert [entry["path"] for entry in page] == ["a.iso"]
+
+    empty, empty_total = list_flat(None, publication, "missing/", 1000, 0)
+    assert empty_total == 0
+    assert empty == []
+
+    rv_entries, rv_total = list_flat(repo_version, None, "", 1000, 0)
+    assert rv_total == 2
+    assert [entry["path"] for entry in rv_entries] == ["a.iso", "subdir/b.iso"]
+
+    from django.db import connection
+    from django.test.utils import CaptureQueriesContext
+
+    with CaptureQueriesContext(connection) as ctx:
+        list_flat(None, publication, "", 1, 0)
+    assert any("LIMIT" in q["sql"].upper() for q in ctx.captured_queries)
diff --git a/pulpcore/tests/unit/test_cache.py b/pulpcore/tests/unit/test_cache.py
index 6da69732e07..6921e650f66 100644
--- a/pulpcore/tests/unit/test_cache.py
+++ b/pulpcore/tests/unit/test_cache.py
@@ -1,9 +1,10 @@
 from time import sleep
+from unittest.mock import Mock
 
 import pytest
 
 import pulpcore.app.redis_connection
-from pulpcore.cache import Cache
+from pulpcore.cache import AsyncContentCache, Cache, CacheKeys, accept_prefers_json
 
 
 @pytest.fixture
@@ -107,3 +108,91 @@ def test_clear(pulp_redisdb):
     cache.redis.flushdb()
     for key, _, base_key in tuples:
         assert not cache.exists(key, base_key=base_key)
+
+
+@pytest.mark.parametrize(
+    "accept_header,expected",
+    [
+        (None, False),
+        ("", False),
+        ("*/*", False),
+        ("text/html", False),
+        ("text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", False),
+        ("application/json, text/html", True),
+        ("text/html, application/json", False),
+        ("application/json", True),
+        ("Application/JSON", True),
+        ("application/vnd.pypi.simple.v1+json", True),
+        ("application/json;q=0.5, text/html", False),
+        ("application/json;q=0.9, text/html;q=0.8", True),
+        ("application/json;q=0", False),
+        ("application/json;charset=utf-8", True),
+        ("text/html;q=0.9, application/json;q=0.9", False),
+        ("  application/json  ", True),
+        (Mock(), False),
+        (1, False),
+    ],
+)
+def test_accept_prefers_json(accept_header, expected):
+    assert accept_prefers_json(accept_header) is expected
+
+
+def test_content_cache_key_varies_by_accept(monkeypatch):
+    """JSON and HTML requests for the same path must not share a cache key."""
+    monkeypatch.setattr("pulpcore.cache.cache.get_async_redis_connection", lambda: None)
+    cache = AsyncContentCache(keys=(CacheKeys.path, CacheKeys.method, CacheKeys.format))
+
+    def make_request(accept):
+        request = Mock()
+        request.path = "/pulp/content/foo/"
+        request.method = "GET"
+        request.url.host = "example.com"
+        request.headers = {} if accept is None else {"Accept": accept}
+        return request
+
+    json_key = cache.make_key(make_request("application/json"))
+    html_key = cache.make_key(make_request("text/html"))
+    default_key = cache.make_key(make_request(None))
+    star_key = cache.make_key(make_request("*/*"))
+
+    assert json_key == "/pulp/content/foo/:GET:json"
+    assert html_key == "/pulp/content/foo/:GET:other"
+    assert default_key == html_key
+    assert star_key == html_key
+    assert json_key != html_key
+
+
+def test_content_cache_key_varies_by_query(monkeypatch):
+    """JSON pagination params must be in the cache key; junk query params and HTML must not."""
+    monkeypatch.setattr("pulpcore.cache.cache.get_async_redis_connection", lambda: None)
+    cache = AsyncContentCache(
+        keys=(CacheKeys.path, CacheKeys.method, CacheKeys.format, CacheKeys.query)
+    )
+
+    def make_request(accept, query=None):
+        request = Mock()
+        request.path = "/pulp/content/foo/"
+        request.method = "GET"
+        request.url.host = "example.com"
+        request.headers = {"Accept": accept}
+        request.query = {} if query is None else query
+        return request
+
+    page0 = cache.make_key(make_request("application/json", {"limit": "1", "offset": "0"}))
+    page0_reordered = cache.make_key(
+        make_request("application/json", {"offset": "0", "limit": "1"})
+    )
+    page1 = cache.make_key(make_request("application/json", {"limit": "1", "offset": "1"}))
+    no_query = cache.make_key(make_request("application/json", {}))
+    junk = cache.make_key(make_request("application/json", {"t": "12345", "foo": "bar"}))
+    html_junk = cache.make_key(make_request("text/html", {"t": "12345", "limit": "1"}))
+    html_plain = cache.make_key(make_request("text/html", {}))
+
+    assert page0 == "/pulp/content/foo/:GET:json:1:0"
+    assert page0_reordered == page0
+    assert page1 == "/pulp/content/foo/:GET:json:1:1"
+    assert no_query == "/pulp/content/foo/:GET:json:1000:0"
+    assert junk == no_query
+    assert html_junk == html_plain
+    assert html_plain == "/pulp/content/foo/:GET:other"
+    assert page0 != page1