feat: add read-only query model tool

Signed-off-by: Rai Siqueira <rai93siqueira@gmail.com>

Author: raisiqueira

README.md

@@ -259,7 +259,28 @@ Reverse a named URL pattern to get its actual URL path. Supports both positional
 - `args`: Optional list of positional arguments
 - `kwargs`: Optional dict of keyword arguments
 
-### 10. `read_recent_logs`
+### 10. `query_model`
+Query a Django model with read-only operations using the Django ORM manager. This tool allows safe querying of any Django model with filtering, ordering, and pagination.
+
+**Arguments:**
+- `app_label`: The Django app label (e.g., "blog")
+- `model_name`: The model name (e.g., "Post")
+- `filters`: Optional dict of field lookups (e.g., `{"status": "published", "featured": true}`)
+- `order_by`: Optional list of fields to order by (e.g., `["-created_at", "title"]`)
+- `limit`: Maximum number of results to return (default: 100, max: 1000)
+
+**Returns:**
+- Total count of matching objects
+- Number of results returned
+- List of model instances as dictionaries with all field values
+- For foreign keys, includes both the ID and string representation
+
+**Example Queries:**
+- Get all published posts: `filters={"status": "published"}`
+- Get featured posts ordered by date: `filters={"featured": true}`, `order_by=["-created_at"]`
+- Get recent posts with limit: `order_by=["-created_at"]`, `limit=10`
+
+### 11. `read_recent_logs`
 Read recent log entries with optional filtering by log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
 
 ### Prompts
@@ -301,6 +322,9 @@ Once configured, you can ask your AI assistant questions like:
 - "What's the value of the DEBUG setting?"
 - "What's the URL for blog post with ID 5?"
 - "Reverse the 'post_detail' URL pattern with pk=10"
+- "Show me all published blog posts"
+- "Get the 10 most recent posts ordered by creation date"
+- "Find all featured posts in the blog"
 - "Show me recent error logs"
 
 **Using Prompts:**
@@ -316,6 +340,9 @@ The project includes a comprehensive test suite and a fixture Django project for
 # Test the MCP server with the fixture project
 uv run python test_server.py
 
+# Test the query_model tool
+uv run python test_query_model.py
+
 # Run the MCP server with the test project
 export PYTHONPATH="${PYTHONPATH}:./fixtures/testproject"
 uv run django-ai-boost --settings testproject.settings

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "django-ai-boost"
-version = "0.2.1"
+version = "0.3.0"
 description = "A Model Context Protocol (MCP) server for Django applications, inspired by Laravel Boost"
 readme = "README.md"
 authors = [

server.json

@@ -6,12 +6,12 @@
     "url": "https://github.com/vintasoftware/django-ai-boost",
     "source": "github"
   },
-  "version": "0.2.1",
+  "version": "0.3.0",
   "packages": [
     {
       "registryType": "pypi",
       "identifier": "django-ai-boost",
-      "version": "0.2.1",
+      "version": "0.3.0",
       "transport": {
         "type": "stdio"
       },

src/django_ai_boost/server_fastmcp.py

@@ -429,6 +429,114 @@ async def reverse_url(
         return {"error": f"Error reversing URL: {str(e)}"}
 
 
+@mcp.tool()
+async def query_model(
+    app_label: str,
+    model_name: str,
+    filters: dict[str, Any] | None = None,
+    order_by: list[str] | None = None,
+    limit: int = 100,
+) -> dict[str, Any]:
+    """
+    Query a Django model with read-only operations using the Django ORM manager.
+
+    Args:
+        app_label: The app label (e.g., "blog")
+        model_name: The model name (e.g., "Post")
+        filters: Optional dictionary of field lookups (e.g., {"status": "published", "featured": true})
+        order_by: Optional list of fields to order by (e.g., ["-created_at", "title"])
+        limit: Maximum number of results to return (default: 100, max: 1000)
+
+    Returns:
+        Dictionary containing query results or error message.
+    """
+
+    @sync_to_async
+    def execute_query():
+        try:
+            # Get the model
+            try:
+                model = apps.get_model(app_label, model_name)
+            except LookupError:
+                return {"error": f"Model '{app_label}.{model_name}' not found"}
+
+            # Enforce maximum limit for safety
+            max_limit = 1000
+            actual_limit = min(limit, max_limit) if limit else 100
+
+            # Start with all objects
+            queryset = model.objects.all()
+
+            # Apply filters if provided
+            if filters:
+                try:
+                    queryset = queryset.filter(**filters)
+                except Exception as e:
+                    return {"error": f"Invalid filter parameters: {str(e)}"}
+
+            # Apply ordering if provided
+            if order_by:
+                try:
+                    queryset = queryset.order_by(*order_by)
+                except Exception as e:
+                    return {"error": f"Invalid order_by parameters: {str(e)}"}
+
+            # Get total count before limiting
+            total_count = queryset.count()
+
+            # Limit results
+            queryset = queryset[:actual_limit]
+
+            # Convert queryset to list of dictionaries
+            results = []
+            for obj in queryset:
+                obj_dict = {}
+                for field in model._meta.get_fields():
+                    # Skip reverse relations
+                    if field.many_to_many or field.one_to_many:
+                        continue
+
+                    field_name = field.name
+                    try:
+                        value = getattr(obj, field_name)
+
+                        # Handle different field types
+                        if value is None:
+                            obj_dict[field_name] = None
+                        elif hasattr(field, "related_model") and field.related_model:
+                            # Foreign key - store the pk
+                            obj_dict[field_name] = value.pk if value else None
+                            obj_dict[f"{field_name}_str"] = (
+                                str(value) if value else None
+                            )
+                        elif isinstance(value, (str, int, float, bool)):
+                            obj_dict[field_name] = value
+                        else:
+                            # For dates, times, and other complex types
+                            obj_dict[field_name] = str(value)
+                    except Exception:
+                        # Skip fields that can't be accessed
+                        continue
+
+                results.append(obj_dict)
+
+            return {
+                "app": app_label,
+                "model": model_name,
+                "total_count": total_count,
+                "returned_count": len(results),
+                "limit": actual_limit,
+                "filters": filters or {},
+                "order_by": order_by or [],
+                "results": results,
+            }
+
+        except Exception as e:
+            return {"error": f"Error executing query: {str(e)}"}
+
+    return await execute_query()
+
+
 @mcp.prompt()
 async def search_django_docs(topic: str) -> str:
     """