# Django Ninja Aio Crud

> Async dynamic CRUD framework for Django Ninja — built-in auth, filtering, pagination, automatic serialization, and MCP tools for AI agents.

- **Type:** MCP server
- **Install:** `agentstack add mcp-caspel26-django-ninja-aio-crud`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [caspel26](https://agentstack.voostack.com/s/caspel26)
- **Installs:** 0
- **Category:** [Integrations](https://agentstack.voostack.com/c/integrations)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [caspel26](https://github.com/caspel26)
- **Source:** https://github.com/caspel26/django-ninja-aio-crud
- **Website:** https://django-ninja-aio.com

## Install

```sh
agentstack add mcp-caspel26-django-ninja-aio-crud
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

Async CRUD framework for Django Ninja
  Automatic schema generation · Filtering · Pagination · Auth · M2M management

  
  
  
  
  
  
  

  Documentation ·
  PyPI ·
  Framework Comparison ·
  Performance Benchmarks ·
  Example Project ·
  Issues

---

## Features

| | Feature | Description |
|---|---|---|
| **🔒 Type Safety** | Generic classes | Full IDE autocomplete and type checking with generic `ModelUtil`, `Serializer`, and `APIViewSet` |
| **Meta-driven Serializer** | Dynamic schemas | Generate CRUD schemas for existing Django models without changing base classes |
| **Async CRUD ViewSets** | Full operations | Create, list, retrieve, update, delete — all async |
| **Auto Schemas** | Pydantic generation | Automatic read/create/update schemas from `ModelSerializer` |
| **Dynamic Query Params** | Runtime schemas | Built with `pydantic.create_model` for flexible filtering |
| **Per-method Auth** | Granular control | `auth`, `get_auth`, `post_auth`, etc. |
| **Async Pagination** | Customizable | `PageNumberPagination`, `CursorPagination`, or custom — DB-level slicing |
| **M2M Relations** | Add/remove/list | Endpoints via `M2MRelationSchema` with filtering support |
| **Reverse Relations** | Nested serialization | Automatic handling of reverse FK and M2M |
| **Bulk Operations** | Create/update/delete | Opt-in bulk endpoints with partial success semantics and configurable response fields |
| **Custom Actions** | `@action` decorator | Detail and list actions with auth inheritance, custom decorators, and auto URL generation |
| **Lifecycle Hooks** | Extensible | `before_save`, `after_save`, `custom_actions`, `on_delete`, and more |
| **Schema Validators** | Pydantic validators | `@field_validator` and `@model_validator` on serializer classes |
| **ORJSON Renderer** | Performance | Built-in fast JSON rendering via `NinjaAIO` |
| **AI Agent Integration** | MCP tools | Expose ViewSets as [MCP](https://modelcontextprotocol.io) tools for any MCP client |

---

## See It In Action

  

  A ModelSerializer-based model, wired to a viewset, serving full CRUD in a few lines — no manual schemas or endpoint wiring. See the docs for the full walkthrough.

---

## Quick Start

### Option A: Meta-driven Serializer (existing models)

Use this if you already have Django models and don't want to change their base class.

```python
from ninja_aio.models import serializers
from ninja_aio.views import APIViewSet
from ninja_aio import NinjaAIO
from . import models

class BookSerializer(serializers.Serializer):
    class Meta:
        model = models.Book
        schema_in = serializers.SchemaModelConfig(fields=["title", "published"])
        schema_out = serializers.SchemaModelConfig(fields=["id", "title", "published"])
        schema_update = serializers.SchemaModelConfig(
            optionals=[("title", str), ("published", bool)]
        )

api = NinjaAIO()

@api.viewset(models.Book)
class BookViewSet(APIViewSet):
    serializer_class = BookSerializer
```

### Option B: ModelSerializer (new projects)

Define models with built-in serialization for minimal boilerplate.

**models.py**

```python
from django.db import models
from ninja_aio.models import ModelSerializer

class Book(ModelSerializer):
    title = models.CharField(max_length=120)
    published = models.BooleanField(default=True)

    class ReadSerializer:
        fields = ["id", "title", "published"]

    class CreateSerializer:
        fields = ["title", "published"]

    class UpdateSerializer:
        optionals = [("title", str), ("published", bool)]
```

**views.py**

```python
from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from .models import Book

api = NinjaAIO()

@api.viewset(Book)
class BookViewSet(APIViewSet):
    pass
```

> Visit `/docs` — CRUD endpoints ready.

---

## Query Filtering

```python
@api.viewset(Book)
class BookViewSet(APIViewSet):
    query_params = {"published": (bool, None), "title": (str, None)}

    async def query_params_handler(self, queryset, filters):
        if filters.get("published") is not None:
            queryset = queryset.filter(published=filters["published"])
        if filters.get("title"):
            queryset = queryset.filter(title__icontains=filters["title"])
        return queryset
```

```
GET /book/?published=true&title=python
```

---

## Many-to-Many Relations

```python
from ninja_aio.schemas import M2MRelationSchema

class Tag(ModelSerializer):
    name = models.CharField(max_length=50)
    class ReadSerializer:
        fields = ["id", "name"]

class Article(ModelSerializer):
    title = models.CharField(max_length=120)
    tags = models.ManyToManyField(Tag, related_name="articles")
    class ReadSerializer:
        fields = ["id", "title", "tags"]

@api.viewset(Article)
class ArticleViewSet(APIViewSet):
    m2m_relations = [
        M2MRelationSchema(
            model=Tag,
            related_name="tags",
            filters={"name": (str, "")}
        )
    ]

    async def tags_query_params_handler(self, queryset, filters):
        n = filters.get("name")
        if n:
            queryset = queryset.filter(name__icontains=n)
        return queryset
```

**Endpoints:**

```
GET  /article/{pk}/tag?name=dev
POST /article/{pk}/tag/    body: {"add": [1, 2], "remove": [3]}
```

---

## Authentication (JWT)

```python
from ninja_aio.auth import AsyncJwtBearer
from joserfc import jwk

class JWTAuth(AsyncJwtBearer):
    jwt_public = jwk.RSAKey.import_key("-----BEGIN PUBLIC KEY----- ...")
    jwt_alg = "RS256"
    claims = {"sub": {"essential": True}}

    async def auth_handler(self, request):
        book_id = self.dcd.claims.get("sub")
        return await Book.objects.aget(id=book_id)

@api.viewset(Book)
class SecureBookViewSet(APIViewSet):
    auth = [JWTAuth()]
    get_auth = None  # list/retrieve remain public
```

---

## Lifecycle Hooks

Available on every save/delete cycle:

| Hook | When |
|---|---|
| `on_create_before_save` | Before first save |
| `on_create_after_save` | After first save |
| `before_save` | Before any save |
| `after_save` | After any save |
| `on_delete` | After deletion |
| `custom_actions(payload)` | Create/update custom field logic |
| `post_create()` | After create commit |

---

## Custom Endpoints

### Option A: `@action` Decorator (recommended)

```python
from ninja import Schema, Status
from ninja_aio.decorators import action

class StatsSchema(Schema):
    total: int

@api.viewset(Book)
class BookViewSet(APIViewSet):
    @action(detail=False, methods=["get"], url_path="stats", response=StatsSchema)
    async def stats(self, request):
        total = await Book.objects.acount()
        return {"total": total}

    @action(detail=True, methods=["post"], url_path="publish")
    async def publish(self, request, pk):
        book = await self.model_util.get_object(request, pk)
        book.published = True
        await book.asave()
        return Status(200, {"message": "published"})
```

```
GET  /book/stats/       → {"total": 42}
POST /book/{pk}/publish/ → {"message": "published"}
```

### Option B: operations Decorators

```python
from ninja_aio.decorators import api_get

@api.viewset(Book)
class BookViewSet(APIViewSet):
    @api_get("/stats/")
    async def stats(self, request):
        total = await Book.objects.acount()
        return {"total": total}
```

---

## AI Agent Integration (MCP)

Expose every registered `APIViewSet` — CRUD, bulk operations, and custom `@action`/`@on` endpoints — as well as any custom `APIView`, as [MCP](https://modelcontextprotocol.io) tools any MCP client can call directly.

```sh
pip install "django-ninja-aio-crud[mcp]"
```

### Option A: `manage.py mcp_server` (recommended)

Add `"ninja_aio"` to `INSTALLED_APPS` to pick up the bundled management command:

```python
INSTALLED_APPS = [
    ...,
    "ninja_aio",
]
```

```sh
python manage.py mcp_server myproject.api.api
```

Or set a default so you can drop the argument:

```python
# settings.py
NINJA_AIO_MCP_API = "myproject.api.api"
```

```json
{
  "mcpServers": {
    "myproject": {
      "type": "stdio",
      "command": "python",
      "args": ["manage.py", "mcp_server"]
    }
  }
}
```

### Option B: standalone script

```python
# mcp_server.py
import asyncio
import django
django.setup()

from myproject.api import api  # your NinjaAIO() instance with @api.viewset(...) registered
from ninja_aio.mcp import run_mcp_server

if __name__ == "__main__":
    asyncio.run(run_mcp_server(api))
```

```json
{
  "mcpServers": {
    "myproject": {
      "type": "stdio",
      "command": "python",
      "args": ["mcp_server.py"]
    }
  }
}
```

Every `@api.viewset(...)`-registered ViewSet and `@api.view(...)`-registered View is picked up automatically (or pass `viewsets=[...]`/`views=[...]` explicitly). Tools are named `_` for ViewSets — e.g. `book_create`, `book_list`, `book_retrieve`, `book_update`, `book_delete`, `book_bulk_create`, `book_publish` — and `__` for plain Views — e.g. `bookview_stats_get`.

> **⚠️ Auth caveat:** tool calls invoke the same registered view logic as HTTP requests (filters, pagination, and `on_before_operation`/`on_before_object_operation`/`query_params_handler` hooks all run identically) but bypass django-ninja's `auth=` wiring, since that applies at the router layer, not inside the handler. Pass `request_factory` to attach your own `request.user`/auth context, and use viewset hooks to enforce authorization for MCP-driven calls:

```python
from ninja_aio.mcp import NinjaAIOMCPServer

def mcp_request_factory():
    from django.test.client import AsyncRequestFactory
    request = AsyncRequestFactory().get("/mcp/")
    request.user = get_service_account_user()  # your own resolution logic
    return request

server = NinjaAIOMCPServer(api, request_factory=mcp_request_factory)
```

---

## Bulk Operations

```python
@api.viewset(Book)
class BookViewSet(APIViewSet):
    bulk_operations = ["create", "update", "delete"]
    bulk_response_fields = "title"  # Optional: return titles instead of PKs
```

```
POST   /book/bulk/  body: [{...}, {...}]         → {"success": {"count": 2, "details": ["Book 1", "Book 2"]}}
PATCH  /book/bulk/  body: [{id, ...}, {id, ...}] → {"success": {"count": 2, "details": ["Updated 1", "Updated 2"]}}
DELETE /book/bulk/  body: {"ids": [1, 2]}         → {"success": {"count": 2, "details": ["Book 1", "Book 2"]}}
```

---

## Pagination

Default: `PageNumberPagination`. Override per ViewSet:

```python
from ninja.pagination import PageNumberPagination, CursorPagination

class LargePagination(PageNumberPagination):
    page_size = 50
    max_page_size = 200

@api.viewset(Book)
class BookViewSet(APIViewSet):
    pagination_class = LargePagination
    # Or use cursor-based pagination for large datasets:
    # pagination_class = CursorPagination
```

---

## Schema Validators

Add Pydantic `@field_validator` and `@model_validator` directly on serializer classes for input validation.

### ModelSerializer

Declare validators on inner serializer classes:

```python
from django.db import models
from pydantic import field_validator, model_validator
from ninja_aio.models import ModelSerializer

class Book(ModelSerializer):
    title = models.CharField(max_length=120)
    description = models.TextField(blank=True)

    class CreateSerializer:
        fields = ["title", "description"]

        @field_validator("title")
        @classmethod
        def validate_title_min_length(cls, v):
            if len(v) 

---

## License

MIT License. See [LICENSE](LICENSE).

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [caspel26](https://github.com/caspel26)
- **Source:** [caspel26/django-ninja-aio-crud](https://github.com/caspel26/django-ninja-aio-crud)
- **License:** MIT
- **Homepage:** https://django-ninja-aio.com

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/mcp-caspel26-django-ninja-aio-crud
- Seller: https://agentstack.voostack.com/s/caspel26
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.