Fix critical token consumption issue in list endpoints (#488)

python/src/mcp_server/features/documents/document_tools.py

@@ -144,7 +144,11 @@ async def list_documents(ctx: Context, project_id: str) -> str:
             timeout = get_default_timeout()
 
             async with httpx.AsyncClient(timeout=timeout) as client:
-                response = await client.get(urljoin(api_url, f"/api/projects/{project_id}/docs"))
+                # Pass include_content=False for lightweight response
+                response = await client.get(
+                    urljoin(api_url, f"/api/projects/{project_id}/docs"),
+                    params={"include_content": False}
+                )
 
                 if response.status_code == 200:
                     result = response.json()

python/src/mcp_server/features/projects/project_tools.py

@@ -175,7 +175,11 @@ async def list_projects(ctx: Context) -> str:
             timeout = get_default_timeout()
 
             async with httpx.AsyncClient(timeout=timeout) as client:
-                response = await client.get(urljoin(api_url, "/api/projects"))
+                # CRITICAL: Pass include_content=False for lightweight response
+                response = await client.get(
+                    urljoin(api_url, "/api/projects"),
+                    params={"include_content": False}
+                )
 
                 if response.status_code == 200:
                     projects = response.json()

python/src/server/api_routes/projects_api.py

@@ -9,7 +9,9 @@
 """
 
 import asyncio
+import json
 import secrets
+import sys
 from typing import Any
 
 from fastapi import APIRouter, HTTPException
@@ -74,23 +76,49 @@ class CreateTaskRequest(BaseModel):
 
 
 @router.get("/projects")
-async def list_projects():
-    """List all projects."""
+async def list_projects(include_content: bool = True):
+    """
+    List all projects.
+
+    Args:
+        include_content: If True (default), returns full project content.
+                        If False, returns lightweight metadata with statistics.
+    """
     try:
-        logfire.info("Listing all projects")
+        logfire.info(f"Listing all projects | include_content={include_content}")
 
-        # Use ProjectService to get projects
+        # Use ProjectService to get projects with include_content parameter
         project_service = ProjectService()
-        success, result = project_service.list_projects()
+        success, result = project_service.list_projects(include_content=include_content)
 
         if not success:
             raise HTTPException(status_code=500, detail=result)
 
-        # Use SourceLinkingService to format projects with sources
-        source_service = SourceLinkingService()
-        formatted_projects = source_service.format_projects_with_sources(result["projects"])
-
-        logfire.info(f"Projects listed successfully | count={len(formatted_projects)}")
+        # Only format with sources if we have full content
+        if include_content:
+            # Use SourceLinkingService to format projects with sources
+            source_service = SourceLinkingService()
+            formatted_projects = source_service.format_projects_with_sources(result["projects"])
+        else:
+            # Lightweight response doesn't need source formatting
+            formatted_projects = result["projects"]
+
+        # Monitor response size for optimization validation
+        response_json = json.dumps(formatted_projects)
+        response_size = len(response_json)
+
+        # Log response metrics
+        logfire.info(
+            f"Projects listed successfully | count={len(formatted_projects)} | "
+            f"size_bytes={response_size} | include_content={include_content}"
+        )
+
+        # Warning for large responses (>10KB)
+        if response_size > 10000:
+            logfire.warning(
+                f"Large response size detected | size_bytes={response_size} | "
+                f"include_content={include_content} | project_count={len(formatted_projects)}"
+            )
 
         return formatted_projects
 
@@ -473,39 +501,32 @@ async def get_project_features(project_id: str):
 
 
 @router.get("/projects/{project_id}/tasks")
-async def list_project_tasks(project_id: str, include_archived: bool = False):
+async def list_project_tasks(project_id: str, include_archived: bool = False, exclude_large_fields: bool = False):
     """List all tasks for a specific project. By default, filters out archived tasks."""
     try:
         logfire.info(
-            f"Listing project tasks | project_id={project_id} | include_archived={include_archived}"
+            f"Listing project tasks | project_id={project_id} | include_archived={include_archived} | exclude_large_fields={exclude_large_fields}"
         )
 
         # Use TaskService to list tasks
         task_service = TaskService()
         success, result = task_service.list_tasks(
             project_id=project_id,
-            include_closed=True,  # Get all tasks, we'll filter archived separately
+            include_closed=True,  # Get all tasks, including done
+            exclude_large_fields=exclude_large_fields,
+            include_archived=include_archived,  # Pass the flag down to service
         )
 
         if not success:
             raise HTTPException(status_code=500, detail=result)
 
         tasks = result.get("tasks", [])
 
-        # Apply filters
-        filtered_tasks = []
-        for task in tasks:
-            # Skip archived tasks if not including them (handle None as False)
-            if not include_archived and task.get("archived", False):
-                continue
-
-            filtered_tasks.append(task)
-
         logfire.info(
-            f"Project tasks retrieved | project_id={project_id} | task_count={len(filtered_tasks)}"
+            f"Project tasks retrieved | project_id={project_id} | task_count={len(tasks)}"
         )
 
-        return filtered_tasks
+        return tasks
 
     except HTTPException:
         raise
@@ -571,6 +592,7 @@ async def list_tasks(
             project_id=project_id,
             status=status,
             include_closed=include_closed,
+            exclude_large_fields=exclude_large_fields,
         )
 
         if not success:
@@ -591,8 +613,8 @@ async def list_tasks(
         end_idx = start_idx + per_page
         paginated_tasks = tasks[start_idx:end_idx]
 
-        # Return paginated response
-        return {
+        # Prepare response
+        response = {
             "tasks": paginated_tasks,
             "pagination": {
                 "total": len(tasks),
@@ -601,6 +623,25 @@ async def list_tasks(
                 "pages": (len(tasks) + per_page - 1) // per_page,
             },
         }
+
+        # Monitor response size for optimization validation
+        response_json = json.dumps(response)
+        response_size = len(response_json)
+
+        # Log response metrics
+        logfire.info(
+            f"Tasks listed successfully | count={len(paginated_tasks)} | "
+            f"size_bytes={response_size} | exclude_large_fields={exclude_large_fields}"
+        )
+
+        # Warning for large responses (>10KB)
+        if response_size > 10000:
+            logfire.warning(
+                f"Large task response size | size_bytes={response_size} | "
+                f"exclude_large_fields={exclude_large_fields} | task_count={len(paginated_tasks)}"
+            )
+
+        return response
 
     except HTTPException:
         raise
@@ -795,14 +836,23 @@ async def mcp_update_task_status_with_socketio(task_id: str, status: str):
 
 
 @router.get("/projects/{project_id}/docs")
-async def list_project_documents(project_id: str):
-    """List all documents for a specific project."""
+async def list_project_documents(project_id: str, include_content: bool = False):
+    """
+    List all documents for a specific project.
+
+    Args:
+        project_id: Project UUID
+        include_content: If True, includes full document content.
+                        If False (default), returns metadata only.
+    """
     try:
-        logfire.info(f"Listing documents for project | project_id={project_id}")
+        logfire.info(
+            f"Listing documents for project | project_id={project_id} | include_content={include_content}"
+        )
 
         # Use DocumentService to list documents
         document_service = DocumentService()
-        success, result = document_service.list_documents(project_id)
+        success, result = document_service.list_documents(project_id, include_content=include_content)
 
         if not success:
             if "not found" in result.get("error", "").lower():
@@ -811,7 +861,7 @@ async def list_project_documents(project_id: str):
                 raise HTTPException(status_code=500, detail=result)
 
         logfire.info(
-            f"Documents listed successfully | project_id={project_id} | count={result.get('total_count', 0)}"
+            f"Documents listed successfully | project_id={project_id} | count={result.get('total_count', 0)} | lightweight={not include_content}"
         )
 
         return result

python/src/server/services/projects/document_service.py

@@ -96,10 +96,15 @@ def add_document(
             logger.error(f"Error adding document: {e}")
             return False, {"error": f"Error adding document: {str(e)}"}
 
-    def list_documents(self, project_id: str) -> tuple[bool, dict[str, Any]]:
+    def list_documents(self, project_id: str, include_content: bool = False) -> tuple[bool, dict[str, Any]]:
         """
         List all documents in a project's docs JSONB field.
 
+        Args:
+            project_id: The project ID
+            include_content: If True, includes full document content.
+                           If False (default), returns metadata only.
+
         Returns:
             Tuple of (success, result_dict)
         """
@@ -116,20 +121,28 @@ def list_documents(self, project_id: str) -> tuple[bool, dict[str, Any]]:
 
             docs = response.data[0].get("docs", [])
 
-            # Format documents for response (exclude full content for listing)
+            # Format documents for response
             documents = []
             for doc in docs:
-                documents.append({
-                    "id": doc.get("id"),
-                    "document_type": doc.get("document_type"),
-                    "title": doc.get("title"),
-                    "status": doc.get("status"),
-                    "version": doc.get("version"),
-                    "tags": doc.get("tags", []),
-                    "author": doc.get("author"),
-                    "created_at": doc.get("created_at"),
-                    "updated_at": doc.get("updated_at"),
-                })
+                if include_content:
+                    # Return full document
+                    documents.append(doc)
+                else:
+                    # Return metadata only
+                    documents.append({
+                        "id": doc.get("id"),
+                        "document_type": doc.get("document_type"),
+                        "title": doc.get("title"),
+                        "status": doc.get("status"),
+                        "version": doc.get("version"),
+                        "tags": doc.get("tags", []),
+                        "author": doc.get("author"),
+                        "created_at": doc.get("created_at"),
+                        "updated_at": doc.get("updated_at"),
+                        "stats": {
+                            "content_size": len(str(doc.get("content", {})))
+                        }
+                    })
 
             return True, {
                 "project_id": project_id,

python/src/server/services/projects/project_service.py

@@ -73,35 +73,73 @@ def create_project(self, title: str, github_repo: str = None) -> tuple[bool, dic
             logger.error(f"Error creating project: {e}")
             return False, {"error": f"Database error: {str(e)}"}
 
-    def list_projects(self) -> tuple[bool, dict[str, Any]]:
+    def list_projects(self, include_content: bool = True) -> tuple[bool, dict[str, Any]]:
         """
         List all projects.
 
+        Args:
+            include_content: If True (default), includes docs, features, data fields.
+                           If False, returns lightweight metadata only with counts.
+
         Returns:
             Tuple of (success, result_dict)
         """
         try:
-            response = (
-                self.supabase_client.table("archon_projects")
-                .select("*")
-                .order("created_at", desc=True)
-                .execute()
-            )
-
-            projects = []
-            for project in response.data:
-                projects.append({
-                    "id": project["id"],
-                    "title": project["title"],
-                    "github_repo": project.get("github_repo"),
-                    "created_at": project["created_at"],
-                    "updated_at": project["updated_at"],
-                    "pinned": project.get("pinned", False),
-                    "description": project.get("description", ""),
-                    "docs": project.get("docs", []),
-                    "features": project.get("features", []),
-                    "data": project.get("data", []),
-                })
+            if include_content:
+                # Current behavior - maintain backward compatibility
+                response = (
+                    self.supabase_client.table("archon_projects")
+                    .select("*")
+                    .order("created_at", desc=True)
+                    .execute()
+                )
+
+                projects = []
+                for project in response.data:
+                    projects.append({
+                        "id": project["id"],
+                        "title": project["title"],
+                        "github_repo": project.get("github_repo"),
+                        "created_at": project["created_at"],
+                        "updated_at": project["updated_at"],
+                        "pinned": project.get("pinned", False),
+                        "description": project.get("description", ""),
+                        "docs": project.get("docs", []),
+                        "features": project.get("features", []),
+                        "data": project.get("data", []),
+                    })
+            else:
+                # Lightweight response for MCP - fetch all data but only return metadata + stats
+                # FIXED: N+1 query problem - now using single query
+                response = (
+                    self.supabase_client.table("archon_projects")
+                    .select("*")  # Fetch all fields in single query
+                    .order("created_at", desc=True)
+                    .execute()
+                )
+
+                projects = []
+                for project in response.data:
+                    # Calculate counts from fetched data (no additional queries)
+                    docs_count = len(project.get("docs", []))
+                    features_count = len(project.get("features", []))
+                    has_data = bool(project.get("data", []))
+
+                    # Return only metadata + stats, excluding large JSONB fields
+                    projects.append({
+                        "id": project["id"],
+                        "title": project["title"],
+                        "github_repo": project.get("github_repo"),
+                        "created_at": project["created_at"],
+                        "updated_at": project["updated_at"],
+                        "pinned": project.get("pinned", False),
+                        "description": project.get("description", ""),
+                        "stats": {
+                            "docs_count": docs_count,
+                            "features_count": features_count,
+                            "has_data": has_data
+                        }
+                    })
 
             return True, {"projects": projects, "total_count": len(projects)}