feat: Add pagination and reduce token usage in responses

README.md

@@ -247,6 +247,74 @@ asyncio.run(main())
 - `test_connection` - Verify API connectivity
 - `get_server_stats` - Server statistics
 
+## 📄 Pagination and Response Control
+
+All list operations support pagination to reduce token usage when working with LLM clients. Get operations support optional raw API response inclusion for debugging.
+
+### Pagination Parameters
+
+All `list_*` tools support these parameters:
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `limit` | integer | 100 | Maximum items to return (max: 500) |
+| `offset` | integer | 0 | Number of items to skip |
+
+**Response format:**
+```json
+{
+  "success": true,
+  "items": [...],
+  "metadata": { ... },
+  "pagination": {
+    "total": 250,
+    "limit": 100,
+    "offset": 0,
+    "returned": 100,
+    "has_more": true
+  }
+}
+```
+
+**Usage examples:**
+```
+"List the first 10 datasets"          → limit=10
+"Show users 50-100"                   → limit=50, offset=50
+"Get all SMB shares (up to 500)"      → limit=500
+```
+
+### Include Raw API Response
+
+Get operations for apps, instances, and VMs support the `include_raw` parameter:
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `include_raw` | boolean | false | Include full API response for debugging |
+
+**When to use `include_raw=true`:**
+- Debugging API response structure
+- Accessing fields not included in the formatted response
+- Troubleshooting integration issues
+
+**Tools supporting `include_raw`:**
+- `get_app` - App details
+- `get_instance` - Incus instance details
+- `get_legacy_vm` - Legacy VM details
+
+### Dataset Response Control
+
+The `list_datasets` and `get_dataset` tools support an additional parameter:
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `include_children` | boolean | true | Include child datasets (can reduce payload significantly) |
+
+**Usage:**
+```
+"List only top-level datasets"        → include_children=false
+"Get tank dataset without children"   → include_children=false
+```
+
 ## 🏗️ Architecture
 
 ```

truenas_mcp_server/tools/apps.py

@@ -27,10 +27,16 @@ class AppTools(BaseTool):
     def get_tool_definitions(self) -> list:
         """Get tool definitions for app management"""
         return [
-            ("list_apps", self.list_apps, "List all TrueNAS apps with status", {}),
+            ("list_apps", self.list_apps, "List all TrueNAS apps with status",
+             {"limit": {"type": "integer", "required": False,
+                       "description": "Max items to return (default: 100, max: 500)"},
+              "offset": {"type": "integer", "required": False,
+                        "description": "Items to skip for pagination"}}),
             ("get_app", self.get_app, "Get detailed information about a specific app",
              {"app_name": {"type": "string", "required": True,
-                         "description": "Name of the app"}}),
+                         "description": "Name of the app"},
+              "include_raw": {"type": "boolean", "required": False,
+                             "description": "Include full API response for debugging (default: false)"}}),
             ("get_app_config", self.get_app_config,
              "Get the full configuration of an app",
              {"app_name": {"type": "string", "required": True,
@@ -59,10 +65,18 @@ def get_tool_definitions(self) -> list:
         ]
 
     @tool_handler
-    async def list_apps(self) -> Dict[str, Any]:
+    async def list_apps(
+        self,
+        limit: int = BaseTool.DEFAULT_LIMIT,
+        offset: int = 0
+    ) -> Dict[str, Any]:
         """
         List all TrueNAS apps with their status
 
+        Args:
+            limit: Maximum number of items to return (default: 100, max: 500)
+            offset: Number of items to skip for pagination
+
         Returns:
             Dictionary containing list of apps with status and metadata
         """
@@ -95,28 +109,39 @@ async def list_apps(self) -> Dict[str, Any]:
             }
             app_list.append(app_info)
 
-        # Count by state
+        # Count by state (before pagination)
         state_counts = {}
         for app in app_list:
             state = app["state"]
             state_counts[state] = state_counts.get(state, 0) + 1
 
+        total_apps = len(app_list)
+
+        # Apply pagination
+        paginated_apps, pagination = self.apply_pagination(app_list, limit, offset)
+
         return {
             "success": True,
-            "apps": app_list,
+            "apps": paginated_apps,
             "metadata": {
-                "total_apps": len(app_list),
+                "total_apps": total_apps,
                 "state_counts": state_counts,
-            }
+            },
+            "pagination": pagination
         }
 
     @tool_handler
-    async def get_app(self, app_name: str) -> Dict[str, Any]:
+    async def get_app(
+        self,
+        app_name: str,
+        include_raw: bool = False
+    ) -> Dict[str, Any]:
         """
         Get detailed information about a specific app
 
         Args:
             app_name: Name of the app
+            include_raw: Include full API response for debugging (default: false)
 
         Returns:
             Dictionary containing app details
@@ -140,7 +165,7 @@ async def get_app(self, app_name: str) -> Dict[str, Any]:
                     "error": f"App '{app_name}' not found"
                 }
 
-        return {
+        result = {
             "success": True,
             "app": {
                 "name": app.get("id") or app.get("name"),
@@ -150,10 +175,14 @@ async def get_app(self, app_name: str) -> Dict[str, Any]:
                 "upgrade_available": app.get("upgrade_available", False),
                 "portal": app.get("portal"),
                 "metadata": app.get("metadata", {}),
-            },
-            "raw": app  # Include full response for debugging
+            }
         }
 
+        if include_raw:
+            result["raw"] = app
+
+        return result
+
     @tool_handler
     async def get_app_config(self, app_name: str) -> Dict[str, Any]:
         """

truenas_mcp_server/tools/base.py

@@ -4,7 +4,7 @@
 
 import logging
 from functools import wraps
-from typing import Any, Dict, Optional, Callable
+from typing import Any, Dict, List, Optional, Callable, Tuple
 from abc import ABC, abstractmethod
 
 from ..client import TrueNASClient
@@ -66,14 +66,19 @@ async def wrapper(self, *args, **kwargs) -> Dict[str, Any]:
 class BaseTool(ABC):
     """
     Base class for all MCP tools
-    
+
     Provides common functionality for tool implementations including:
     - Client management
     - Configuration access
     - Logging setup
     - Error handling utilities
+    - Pagination support
     """
-
+
+    # Pagination defaults
+    DEFAULT_LIMIT = 100
+    MAX_LIMIT = 500
+
     def __init__(self, client: Optional[TrueNASClient] = None, settings: Optional[Settings] = None):
         """
         Initialize the tool
@@ -179,18 +184,51 @@ def parse_size(self, size_str: str) -> int:
     def validate_required_fields(self, data: Dict[str, Any], required: list) -> bool:
         """
         Validate that required fields are present in data
-        
+
         Args:
             data: Data dictionary to validate
             required: List of required field names
-            
+
         Returns:
             True if all required fields are present
-            
+
         Raises:
             ValueError: If any required fields are missing
         """
         missing = [field for field in required if field not in data or data[field] is None]
         if missing:
             raise ValueError(f"Missing required fields: {', '.join(missing)}")
-        return True
+        return True
+
+    def apply_pagination(
+        self,
+        items: List[Any],
+        limit: int = DEFAULT_LIMIT,
+        offset: int = 0
+    ) -> Tuple[List[Any], Dict[str, Any]]:
+        """
+        Apply pagination to a list of items
+
+        Args:
+            items: Full list of items to paginate
+            limit: Maximum items to return (capped at MAX_LIMIT)
+            offset: Number of items to skip
+
+        Returns:
+            Tuple of (paginated_items, pagination_metadata)
+        """
+        # Cap limit at MAX_LIMIT
+        limit = min(limit, self.MAX_LIMIT)
+
+        total = len(items)
+        paginated = items[offset:offset + limit]
+
+        pagination = {
+            "total": total,
+            "limit": limit,
+            "offset": offset,
+            "returned": len(paginated),
+            "has_more": offset + limit < total
+        }
+
+        return paginated, pagination

truenas_mcp_server/tools/instances.py

@@ -31,11 +31,17 @@ def get_tool_definitions(self) -> list:
             ("list_instances", self.list_instances,
              "List all Incus instances (VMs and Containers)",
              {"instance_type": {"type": "string", "required": False,
-                               "description": "Filter by type: 'VM' or 'CONTAINER' (optional)"}}),
+                               "description": "Filter by type: 'VM' or 'CONTAINER' (optional)"},
+              "limit": {"type": "integer", "required": False,
+                       "description": "Max items to return (default: 100, max: 500)"},
+              "offset": {"type": "integer", "required": False,
+                        "description": "Items to skip for pagination"}}),
             ("get_instance", self.get_instance,
              "Get detailed information about a specific instance",
              {"instance_name": {"type": "string", "required": True,
-                               "description": "Name of the instance"}}),
+                               "description": "Name of the instance"},
+              "include_raw": {"type": "boolean", "required": False,
+                             "description": "Include full API response for debugging (default: false)"}}),
             ("start_instance", self.start_instance, "Start an Incus instance",
              {"instance_name": {"type": "string", "required": True,
                                "description": "Name of the instance to start"}}),
@@ -68,13 +74,17 @@ def get_tool_definitions(self) -> list:
     @tool_handler
     async def list_instances(
         self,
-        instance_type: Optional[str] = None
+        instance_type: Optional[str] = None,
+        limit: int = BaseTool.DEFAULT_LIMIT,
+        offset: int = 0
     ) -> Dict[str, Any]:
         """
         List all Incus instances (VMs and Containers)
 
         Args:
             instance_type: Optional filter by type ('VM' or 'CONTAINER')
+            limit: Maximum number of items to return (default: 100, max: 500)
+            offset: Number of items to skip for pagination
 
         Returns:
             Dictionary containing list of instances with status
@@ -107,7 +117,7 @@ async def list_instances(
             }
             instance_list.append(instance_info)
 
-        # Count by type and status
+        # Count by type and status (before pagination)
         type_counts = {}
         status_counts = {}
         for inst in instance_list:
@@ -117,23 +127,34 @@ async def list_instances(
             status = inst["status"]
             status_counts[status] = status_counts.get(status, 0) + 1
 
+        total_instances = len(instance_list)
+
+        # Apply pagination
+        paginated_instances, pagination = self.apply_pagination(instance_list, limit, offset)
+
         return {
             "success": True,
-            "instances": instance_list,
+            "instances": paginated_instances,
             "metadata": {
-                "total_instances": len(instance_list),
+                "total_instances": total_instances,
                 "type_counts": type_counts,
                 "status_counts": status_counts,
-            }
+            },
+            "pagination": pagination
         }
 
     @tool_handler
-    async def get_instance(self, instance_name: str) -> Dict[str, Any]:
+    async def get_instance(
+        self,
+        instance_name: str,
+        include_raw: bool = False
+    ) -> Dict[str, Any]:
         """
         Get detailed information about a specific instance
 
         Args:
             instance_name: Name of the instance
+            include_raw: Include full API response for debugging (default: false)
 
         Returns:
             Dictionary containing instance details
@@ -153,7 +174,7 @@ async def get_instance(self, instance_name: str) -> Dict[str, Any]:
         memory_bytes = inst.get("memory", 0)
         memory_gb = round(memory_bytes / (1024**3), 2)
 
-        return {
+        result = {
             "success": True,
             "instance": {
                 "id": inst.get("id"),
@@ -166,10 +187,14 @@ async def get_instance(self, instance_name: str) -> Dict[str, Any]:
                 "autostart": inst.get("autostart"),
                 "image": inst.get("image"),
                 "environment": inst.get("environment", {}),
-            },
-            "raw": inst  # Include full response for debugging
+            }
         }
 
+        if include_raw:
+            result["raw"] = inst
+
+        return result
+
     @tool_handler
     async def start_instance(self, instance_name: str) -> Dict[str, Any]:
         """