[codex] Add lightweight document metadata listing

core/database/postgres_database.py

@@ -29,6 +29,38 @@
     "summary_bucket",
     "summary_updated_at",
 }
+DOCUMENT_PROJECTION_COLUMN_MAP = {
+    "external_id": DocumentModel.external_id,
+    "content_type": DocumentModel.content_type,
+    "filename": DocumentModel.filename,
+    "metadata": DocumentModel.doc_metadata,
+    "metadata_types": DocumentModel.metadata_types,
+    "storage_info": DocumentModel.storage_info,
+    "system_metadata": DocumentModel.system_metadata,
+    "additional_metadata": DocumentModel.additional_metadata,
+    "chunk_ids": DocumentModel.chunk_ids,
+    "folder_name": DocumentModel.folder_name,
+    "folder_path": DocumentModel.folder_path,
+    "folder_id": DocumentModel.folder_id,
+    "app_id": DocumentModel.app_id,
+    "end_user_id": DocumentModel.end_user_id,
+}
+DOCUMENT_PROJECTION_ORDER = [
+    "external_id",
+    "content_type",
+    "filename",
+    "metadata",
+    "metadata_types",
+    "storage_info",
+    "system_metadata",
+    "additional_metadata",
+    "chunk_ids",
+    "folder_name",
+    "folder_path",
+    "folder_id",
+    "app_id",
+    "end_user_id",
+]
 
 
 class PostgresDatabase:
@@ -403,10 +435,11 @@ async def list_documents_flexible(
         include_status_counts: bool = False,
         include_folder_counts: bool = False,
         return_documents: bool = True,
+        fields: Optional[List[str]] = None,
         sort_by: Optional[str] = None,
         sort_direction: str = "desc",
     ) -> Dict[str, Any]:
-        """List documents with optional aggregate metadata. Field projection is handled at application layer."""
+        """List documents with optional aggregate metadata and projected document fields."""
         limit = max(limit, 0) if limit is not None else None
         skip = max(skip, 0)
 
@@ -440,16 +473,22 @@ async def list_documents_flexible(
 
                 final_where_clause = " AND ".join(where_clauses) if where_clauses else "TRUE"
 
-                documents: List[Document] = []
+                documents: List[Any] = []
                 returned_count = 0
                 has_more = False
 
                 fetch_documents = return_documents and (limit is None or limit > 0)
 
                 if fetch_documents:
-                    # Note: We always select all columns from the database
-                    # Field projection is handled at the application layer for simplicity
-                    base_query = select(DocumentModel).where(text(final_where_clause).bindparams(**filter_params))
+                    projection_fields = self._resolve_document_projection_fields(fields)
+                    if projection_fields:
+                        selected_columns = self._document_projection_columns(projection_fields)
+                        base_query = select(*selected_columns).where(
+                            text(final_where_clause).bindparams(**filter_params)
+                        )
+                    else:
+                        base_query = select(DocumentModel).where(text(final_where_clause).bindparams(**filter_params))
+
                     order_clause = self._resolve_document_sort_clause(sort_by, sort_direction)
                     if order_clause is not None:
                         base_query = base_query.order_by(order_clause, DocumentModel.external_id.asc())
@@ -462,13 +501,19 @@ async def list_documents_flexible(
                         base_query = base_query.limit(fetch_limit)
 
                     result = await session.execute(base_query)
-                    doc_models = result.scalars().all()
-
-                    if fetch_limit is not None and len(doc_models) > limit:
-                        has_more = True
-                        doc_models = doc_models[:limit]
-
-                    documents = [Document(**_document_model_to_dict(doc_model)) for doc_model in doc_models]
+                    if projection_fields:
+                        documents = [
+                            self._document_projection_row_to_dict(row, projection_fields) for row in result.mappings()
+                        ]
+                        if fetch_limit is not None and len(documents) > limit:
+                            has_more = True
+                            documents = documents[:limit]
+                    else:
+                        doc_models = result.scalars().all()
+                        if fetch_limit is not None and len(doc_models) > limit:
+                            has_more = True
+                            doc_models = doc_models[:limit]
+                        documents = [Document(**_document_model_to_dict(doc_model)) for doc_model in doc_models]
                     returned_count = len(documents)
 
                 total_count: Optional[int] = None
@@ -568,6 +613,53 @@ def _resolve_document_sort_clause(self, sort_by: Optional[str], sort_direction:
             f"{direction} NULLS LAST"
         )
 
+    @staticmethod
+    def _resolve_document_projection_fields(fields: Optional[List[str]]) -> Optional[set[str]]:
+        """Resolve requested API fields to the document table columns needed to serve them."""
+        if not fields:
+            return None
+
+        requested_roots = {field.strip().split(".", 1)[0] for field in fields if field and field.strip()}
+        if not requested_roots:
+            return None
+
+        resolved_fields = {"external_id"}
+        for root in requested_roots:
+            if root in DOCUMENT_PROJECTION_COLUMN_MAP:
+                resolved_fields.add(root)
+            elif root in SUMMARY_METADATA_KEYS:
+                resolved_fields.add("system_metadata")
+            elif root == "page_count":
+                resolved_fields.add("system_metadata")
+                resolved_fields.add("chunk_ids")
+
+        return resolved_fields
+
+    @staticmethod
+    def _document_projection_columns(fields: set[str]):
+        """Return a stable list of labeled SQLAlchemy columns for a document projection."""
+        return [
+            DOCUMENT_PROJECTION_COLUMN_MAP[field].label(field) for field in DOCUMENT_PROJECTION_ORDER if field in fields
+        ]
+
+    @staticmethod
+    def _document_projection_row_to_dict(row: Any, fields: set[str]) -> Dict[str, Any]:
+        """Convert a projected document row to the public document dictionary shape."""
+        document = dict(row)
+
+        for key in ("metadata", "metadata_types", "storage_info", "system_metadata", "additional_metadata"):
+            if key in document and document[key] is None:
+                document[key] = {}
+        if "chunk_ids" in document and document["chunk_ids"] is None:
+            document["chunk_ids"] = []
+
+        system_metadata = document.get("system_metadata") or {}
+        if "system_metadata" in fields and isinstance(system_metadata, dict):
+            for key in SUMMARY_METADATA_KEYS:
+                document[key] = system_metadata.get(key)
+
+        return document
+
     async def update_document(
         self,
         document_id: str,

core/routes/documents.py

@@ -102,6 +102,7 @@ async def list_docs(
             include_status_counts=request.include_status_counts,
             include_folder_counts=request.include_folder_counts,
             return_documents=request.return_documents,
+            fields=request.fields,
             sort_by=request.sort_by,
             sort_direction=request.sort_direction,
         )

core/routes/folders.py

@@ -198,6 +198,7 @@ async def folder_details(
                         include_status_counts=request.include_status_counts,
                         include_folder_counts=False,
                         return_documents=request.include_documents,
+                        fields=request.document_fields,
                         sort_by=request.sort_by,
                         sort_direction=request.sort_direction,
                     )

core/tests/unit/test_document_projection.py

@@ -0,0 +1,40 @@
+from core.database.postgres_database import PostgresDatabase
+
+
+def test_document_projection_fields_resolve_metadata_columns():
+    fields = PostgresDatabase._resolve_document_projection_fields(
+        ["metadata.source", "system_metadata.status", "summary_version", "unknown"]
+    )
+
+    assert fields == {"external_id", "metadata", "system_metadata"}
+
+
+def test_document_projection_fields_include_chunk_ids_for_page_count_fallback():
+    fields = PostgresDatabase._resolve_document_projection_fields(["page_count"])
+
+    assert fields == {"external_id", "system_metadata", "chunk_ids"}
+
+
+def test_document_projection_row_to_dict_normalizes_json_and_summary_fields():
+    row = {
+        "external_id": "doc-1",
+        "metadata": None,
+        "metadata_types": None,
+        "system_metadata": {
+            "status": "completed",
+            "summary_version": 3,
+            "summary_storage_key": "summaries/doc-1.md",
+        },
+        "chunk_ids": None,
+    }
+
+    document = PostgresDatabase._document_projection_row_to_dict(
+        row,
+        {"external_id", "metadata", "metadata_types", "system_metadata", "chunk_ids"},
+    )
+
+    assert document["metadata"] == {}
+    assert document["metadata_types"] == {}
+    assert document["chunk_ids"] == []
+    assert document["summary_version"] == 3
+    assert document["summary_storage_key"] == "summaries/doc-1.md"

sdks/python/CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `list_documents_metadata()` for sync, async, folder, and user-scoped clients. It returns a lightweight paginated metadata response and supports the same filters, folder scoping, aggregates, and sorting arguments as `list_documents()`.
+
 ## [1.2.2] - 2026-02-09
 
 ### Added

sdks/python/README.md

@@ -97,6 +97,18 @@ docs = db.list_documents(folder_name="/projects/alpha", folder_depth=-1)
 
 `Folder.full_path` is exposed on folder objects, and `Document.folder_path` mirrors server responses for tracing scope.
 
+### Listing Document Metadata
+
+Use `list_documents_metadata()` when you need a fast paginated inventory without large document fields such as chunk IDs or generated additional metadata:
+
+```python
+response = db.list_documents_metadata(limit=100, include_total_count=True)
+for doc in response.documents:
+    print(doc.external_id, doc.filename, doc.metadata)
+```
+
+The method accepts the same filters, folder scoping, pagination, aggregates, and sorting arguments as `list_documents()`. Pass `fields=["external_id", "metadata.source"]` to request a smaller custom projection.
+
 ### Asynchronous Usage
 
 ```python

sdks/python/morphik/__init__.py

@@ -3,13 +3,15 @@
 """
 
 from .async_ import AsyncMorphik
-from .models import Document, DocumentQueryResponse, Summary
+from .models import Document, DocumentMetadata, DocumentQueryResponse, ListDocumentMetadataResponse, Summary
 from .sync import Morphik
 
 __all__ = [
     "Morphik",
     "AsyncMorphik",
     "Document",
+    "DocumentMetadata",
+    "ListDocumentMetadataResponse",
     "Summary",
     "DocumentQueryResponse",
 ]

sdks/python/morphik/_internal.py

@@ -428,6 +428,8 @@ def _prepare_list_documents_request(
         completed_only: bool,
         sort_by: Optional[str],
         sort_direction: str,
+        return_documents: bool = True,
+        fields: Optional[List[str]] = None,
     ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
         """Prepare request for list_docs endpoint"""
         params = {}
@@ -442,14 +444,16 @@ def _prepare_list_documents_request(
             "skip": skip,
             "limit": limit,
             "document_filters": filters,
-            "return_documents": True,
+            "return_documents": return_documents,
             "include_total_count": include_total_count,
             "include_status_counts": include_status_counts,
             "include_folder_counts": include_folder_counts,
             "completed_only": completed_only,
             "sort_by": sort_by,
             "sort_direction": sort_direction,
         }
+        if fields is not None:
+            data["fields"] = fields
         return params, data
 
     def _prepare_batch_get_documents_request(

sdks/python/morphik/_scoped_ops.py

@@ -10,6 +10,23 @@
 T = TypeVar("T")
 
 logger = logging.getLogger(__name__)
+DEFAULT_DOCUMENT_METADATA_FIELDS = [
+    "external_id",
+    "content_type",
+    "filename",
+    "metadata",
+    "metadata_types",
+    "system_metadata",
+    "folder_name",
+    "folder_path",
+    "folder_id",
+    "end_user_id",
+    "app_id",
+    "summary_storage_key",
+    "summary_version",
+    "summary_bucket",
+    "summary_updated_at",
+]
 
 
 class _ScopedOperationsMixin:
@@ -301,6 +318,51 @@ def _scoped_list_documents(
             parser=self._parse_list_docs_response,
         )
 
+    def _scoped_list_documents_metadata(
+        self,
+        *,
+        skip: int,
+        limit: int,
+        filters: Optional[Dict[str, Any]],
+        folder_name: Optional[Union[str, List[str]]],
+        folder_depth: Optional[int],
+        end_user_id: Optional[str],
+        include_total_count: bool,
+        include_status_counts: bool,
+        include_folder_counts: bool,
+        completed_only: bool,
+        sort_by: Optional[str],
+        sort_direction: str,
+        fields: Optional[List[str]],
+    ):
+        selected_fields = fields if fields is not None else DEFAULT_DOCUMENT_METADATA_FIELDS
+        if not selected_fields:
+            selected_fields = ["external_id"]
+
+        params, data = self._logic._prepare_list_documents_request(
+            skip,
+            limit,
+            filters,
+            folder_name,
+            folder_depth,
+            end_user_id,
+            include_total_count,
+            include_status_counts,
+            include_folder_counts,
+            completed_only,
+            sort_by,
+            sort_direction,
+            fields=selected_fields,
+        )
+
+        return self._execute_scoped_operation(
+            "POST",
+            "documents/list_docs",
+            data=data,
+            params=params,
+            parser=self._parse_list_document_metadata_response,
+        )
+
     # ------------------------------------------------------------------
     # Parsers shared across clients
     # ------------------------------------------------------------------
@@ -322,3 +384,8 @@ def _parse_list_docs_response(self, response: Dict[str, Any]):
         for doc in result.documents:
             doc._client = self
         return result
+
+    def _parse_list_document_metadata_response(self, response: Dict[str, Any]):
+        from .models import ListDocumentMetadataResponse
+
+        return ListDocumentMetadataResponse(**response)