# Django MCP Pattern v1.0.0

Standardizes embedding a FastMCP server inside a Django ASGI process with token-based authentication, modular tool registration, and dual transport (Streamable HTTP + SSE). Used by Angelia 2 (Wagtail CMS); applicable to any Django project.

## 🐾 Red Panda Approval™

This pattern follows Red Panda Approval standards.

---

## Why a Pattern, Not a Shared App

Every Django project that exposes MCP tools has different domain models, different permission requirements, and different admin UX needs. A single reusable Django app cannot accommodate this variability:

- A **CMS platform** needs page-tree tools, media upload tools, and editorial permissions scoped to Wagtail collections
- An **e-commerce project** needs product-catalog tools, order-status tools, and Stripe-scoped API tokens
- An **internal dashboard** needs reporting tools, data-export tools, and LDAP-group-scoped permissions
- A **DevOps platform** needs deployment tools, log-query tools, and service-account tokens

The tools, models, and admin surfaces differ — but the **wiring** is always the same:

Instead, this pattern defines:

- **Required components** — token model, auth middleware, context helpers, server instance, ASGI mount
- **Recommended behaviors** — audit trail, metrics, masked display, dev-mode bypass
- **Extension guidelines** — tool registration, resource registration, admin UI, CLI management
- **Standard constants** — token length, state keys, settings names, metric names

---

## Required Components

The non-negotiable minimum every implementation must provide.

### MCPToken Model

```python
import secrets
from django.conf import settings
from django.db import models
from django.utils import timezone

class MCPToken(models.Model):
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        related_name='mcp_tokens',
    )
    token = models.CharField(max_length=64, unique=True, db_index=True)
    name = models.CharField(max_length=100)
    is_active = models.BooleanField(default=True)
    expires_at = models.DateTimeField(null=True, blank=True)
    last_used_at = models.DateTimeField(null=True, blank=True)
    allowed_tools = models.JSONField(default=list, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def save(self, **kwargs):
        if not self.token:
            self.token = secrets.token_urlsafe(48)  # 64-char URL-safe string
        super().save(**kwargs)

    @property
    def is_valid(self):
        if not self.is_active:
            return False
        if self.expires_at and self.expires_at < timezone.now():
            return False
        return True

    def can_use_tool(self, tool_name: str) -> bool:
        if not self.allowed_tools:
            return True  # Empty list = unrestricted
        return tool_name in self.allowed_tools
```

**Required fields:**

| Field | Type | Purpose |
|-------|------|---------|
| `user` | FK → User | Auth scope — all operations run as this user |
| `token` | CharField(64) | Auto-generated bearer token, unique + indexed |
| `name` | CharField(100) | Friendly label ("Claude Desktop", "CI Script") |
| `is_active` | BooleanField | Revocation flag |
| `expires_at` | DateTimeField | Optional expiry (null = never) |
| `last_used_at` | DateTimeField | Audit trail — updated on each request |
| `allowed_tools` | JSONField | Tool whitelist (empty list = all tools) |
| `created_at` | DateTimeField | Auto-set on creation |
| `updated_at` | DateTimeField | Auto-set on save |

**Required methods:**

| Method | Returns | Purpose |
|--------|---------|---------|
| `is_valid` | bool | Checks active + not expired |
| `can_use_tool(name)` | bool | Whitelist check (empty = permit all) |

### Auth Resolution

```python
from django.contrib.auth import get_user_model
from django.utils import timezone

User = get_user_model()

class MCPAuthError(Exception):
    pass

def resolve_mcp_user(token_string: str) -> tuple:
    try:
        token = (
            MCPToken.objects
            .select_related('user')
            .get(token=token_string)
        )
    except MCPToken.DoesNotExist:
        raise MCPAuthError("Invalid MCP token.")

    if not token.is_active:
        raise MCPAuthError("Token has been deactivated.")
    if token.expires_at and token.expires_at < timezone.now():
        raise MCPAuthError("Token has expired.")
    if not token.user.is_active:
        raise MCPAuthError("User account is disabled.")

    token.record_usage()
    return token.user, token

def check_tool_permission(token: MCPToken, tool_name: str) -> bool:
    return token.can_use_tool(tool_name)
```

**Validation chain (order matters):**

1. Token exists in database (`select_related('user')` for single query)
2. Token is active (`is_active=True`)
3. Token is not expired (`expires_at` is null or in the future)
4. Bound user is active (`user.is_active=True`)
5. Record usage for audit trail

### Auth Middleware

```python
from asgiref.sync import sync_to_async
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.server.dependencies import get_http_request

STATE_KEY_USER = "mcp_user"
STATE_KEY_TOKEN = "mcp_token"

class MCPAuthMiddleware(Middleware):
    async def __call__(self, context: MiddlewareContext, call_next):
        require_auth = getattr(settings, 'MCP_REQUIRE_AUTH', True)

        token_string = self._extract_token(context)

        user = None
        token = None

        if token_string:
            try:
                user, token = await sync_to_async(
                    resolve_mcp_user, thread_sensitive=True
                )(token_string)
            except MCPAuthError as e:
                if require_auth:
                    raise PermissionError(str(e))
        elif require_auth and context.method == "tools/call":
            raise PermissionError(
                "Authentication required. Provide a Bearer token."
            )

        # Tool-level permission check
        if token and context.method == "tools/call":
            tool_name = self._extract_tool_name(context)
            if tool_name and not check_tool_permission(token, tool_name):
                raise PermissionError(
                    f"Token does not have permission to call '{tool_name}'."
                )

        # Store on request-scoped state
        fastmcp_ctx = context.fastmcp_context
        if fastmcp_ctx and user:
            await fastmcp_ctx.set_state(
                STATE_KEY_USER, user, serializable=False
            )
            await fastmcp_ctx.set_state(
                STATE_KEY_TOKEN, token, serializable=False
            )

        return await call_next(context)

    def _extract_token(self, context: MiddlewareContext) -> str | None:
        try:
            request = get_http_request()
            auth_header = request.headers.get("Authorization", "")
            if auth_header.startswith("Bearer "):
                return auth_header[7:]
        except RuntimeError:
            pass  # No HTTP request (e.g., stdio transport)
        return None

    def _extract_tool_name(self, context: MiddlewareContext) -> str | None:
        msg = context.message
        if hasattr(msg, 'params') and hasattr(msg.params, 'name'):
            return msg.params.name
        return None
```

**Middleware responsibilities:**

1. Extract Bearer token from HTTP `Authorization` header
2. Resolve token to Django User via `sync_to_async` (ORM is synchronous)
3. Check tool-level permissions on `tools/call` requests
4. Store user and token on FastMCP's request-scoped state (`serializable=False`)
5. Handle dev-mode bypass when `MCP_REQUIRE_AUTH=False`
6. Gracefully skip auth when no HTTP request exists (stdio transport)

### Context Helpers

```python
from fastmcp.server.context import Context

async def get_mcp_user(ctx: Context):
    return await ctx.get_state(STATE_KEY_USER)

async def get_mcp_token(ctx: Context):
    return await ctx.get_state(STATE_KEY_TOKEN)
```

Tools use these to access the authenticated user:

```python
@mcp.tool
async def create_item(title: str, ctx: Context = None) -> dict:
    user = await get_mcp_user(ctx)
    # ORM operations run as this user
```

### FastMCP Server Instance

```python
from fastmcp import FastMCP

mcp = FastMCP(
    "my-project",
    instructions="System prompt describing your domain for LLMs.",
    middleware=[MCPAuthMiddleware()],
)

# Register tools by domain
register_product_tools(mcp)
register_order_tools(mcp)
```

**Requirements:**

- Single global `FastMCP` instance created at module import time
- Auth middleware injected at server level (not per-tool)
- `instructions` string guides LLMs on domain concepts (page types, field meanings, workflows)
- Tool registration via modular functions, one per domain

### ASGI Mount

```python
import os
import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
django.setup()

from contextlib import asynccontextmanager
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.responses import JSONResponse

from mcp_server.server import mcp

mcp_http_app = mcp.http_app(path="/", transport="streamable-http")
mcp_sse_app = mcp.http_app(path="/", transport="sse")

async def health(request):
    return JSONResponse({"status": "ok"})

@asynccontextmanager
async def lifespan(app):
    async with mcp_http_app.lifespan(app):
        async with mcp_sse_app.lifespan(app):
            yield

app = Starlette(
    routes=[
        Route("/mcp/health", health),
        Mount("/mcp/sse", app=mcp_sse_app),  # More specific path first
        Mount("/mcp", app=mcp_http_app),
    ],
    lifespan=lifespan,
)
```

**Requirements:**

- `django.setup()` before any Django imports (ORM, models)
- Health check endpoint at `/mcp/health`
- `/mcp/sse` route listed before `/mcp` (Starlette matches first hit)
- Lifespan combines both transport apps

---

## Standard Constants

Use these values for consistency across implementations.

### Token Generation

```python
# 64-character URL-safe token via stdlib
token = secrets.token_urlsafe(48)
```

### State Keys

```python
STATE_KEY_USER = "mcp_user"
STATE_KEY_TOKEN = "mcp_token"
```

### Settings

```python
# settings.py
MCP_REQUIRE_AUTH = env.bool('MCP_REQUIRE_AUTH', default=True)
```

### Metric Names

```python
mcp_tool_invocations_total   # Counter — labels: tool, status
mcp_tool_duration_seconds    # Histogram — labels: tool
mcp_auth_failures_total      # Counter — labels: reason
mcp_active_sessions          # Gauge
```

### Tool Description Limit

The MCP specification requires tool descriptions ≤ **1024 characters**. Use a stricter internal limit of **750 characters** to leave headroom for protocol overhead. Validate with:

```python
import inspect
assert len(inspect.cleandoc(tool_fn.__doc__)) <= 750
```

---

## Recommended Behaviors

Most implementations should include these, but they are not strictly required.

### Audit Trail

```python
def record_usage(self):
    self.last_used_at = timezone.now()
    self.save(update_fields=['last_used_at'])
```

Called by `resolve_mcp_user()` on every authenticated request. Provides admin visibility into token activity.

### Masked Token Display

```python
def get_masked_token(self):
    if len(self.token) > 8:
        return f"{'*' * (len(self.token) - 8)}{self.token[-8:]}"
    return "********"
```

Used in admin list views and logs. Never expose full tokens after initial creation.

### Prometheus Metrics

```python
from prometheus_client import Counter, Histogram, Gauge

mcp_tool_invocations_total = Counter(
    'mcp_tool_invocations_total',
    'Total MCP tool invocations',
    ['tool', 'status'],
)

mcp_tool_duration_seconds = Histogram(
    'mcp_tool_duration_seconds',
    'MCP tool execution duration in seconds',
    ['tool'],
    buckets=(0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0),
)

mcp_auth_failures_total = Counter(
    'mcp_auth_failures_total',
    'Total MCP authentication failures',
    ['reason'],
)

def record_tool_call(tool_name: str, status: str, duration: float):
    mcp_tool_invocations_total.labels(tool=tool_name, status=status).inc()
    mcp_tool_duration_seconds.labels(tool=tool_name).observe(duration)
```

### sync_to_async for ORM

All Django ORM calls inside async tool functions **must** be wrapped with `sync_to_async`:

```python
from asgiref.sync import sync_to_async

@mcp.tool
async def create_item(title: str, ctx: Context = None) -> dict:
    user = await get_mcp_user(ctx)

    def _do_create():
        item = MyModel(title=title, created_by=user)
        item.save()
        return {"id": item.id, "title": item.title}

    return await sync_to_async(_do_create, thread_sensitive=True)()
```

The `thread_sensitive=True` parameter ensures ORM operations run in Django's main thread, avoiding database connection issues.

### Dev Mode Auth Bypass

When `MCP_REQUIRE_AUTH=False`, the middleware skips token validation. Tools run without a user context. This is only for local development — never disable in production.

---

## Tool Registration Pattern

Tools are organized by domain in separate modules. Each module exports a `register_*_tools(mcp)` factory function that defines tools as closures, capturing the `mcp` instance.

### File Structure

```
mcp_server/
  tools/
    __init__.py
    pages.py          # register_page_tools(mcp)
    media.py          # register_media_tools(mcp)
    blog.py           # register_blog_tools(mcp)
    orders.py         # register_order_tools(mcp)
```

### Factory Function

```python
# mcp_server/tools/products.py

import time
from asgiref.sync import sync_to_async
from fastmcp.server.context import Context
from ..context import get_mcp_user
from ..metrics import record_tool_call

def register_product_tools(mcp):

    @mcp.tool
    def list_products(category: str | None = None, limit: int = 20) -> dict:
        """List products with optional category filter. Returns id, name,
        price, and category for each product."""
        start = time.time()
        try:
            qs = Product.objects.all()
            if category:
                qs = qs.filter(category__slug=category)
            result = {
                "products": [
                    {"id": p.id, "name": p.name, "price": str(p.price)}
                    for p in qs[:limit]
                ]
            }
            record_tool_call("list_products", "success", time.time() - start)
            return result
        except Exception:
            record_tool_call("list_products", "error", time.time() - start)
            raise

    @mcp.tool
    async def create_product(
        name: str, price: str, description: str,
        ctx: Context = None,
    ) -> dict:
        """Create a new product. Price as decimal string (e.g. '29.99')."""
        start = time.time()
        try:
            user = await get_mcp_user(ctx) if ctx else None

            def _do_create():
                product = Product(
                    name=name,
                    price=Decimal(price),
                    description=description,
                    created_by=user,
                )
                product.save()
                return {"id": product.id, "name": product.name}

            result = await sync_to_async(_do_create, thread_sensitive=True)()
            record_tool_call("create_product", "success", time.time() - start)
            return result
        except Exception:
            record_tool_call("create_product", "error", time.time() - start)
            raise
```

### Sync vs Async Decision

| Tool Type | Define As | ORM Access | Use When |
|-----------|-----------|------------|----------|
| Read-only queries | `def tool(...)` | Direct | Simple lookups, listing, search |
| Mutations | `async def tool(..., ctx)` | `sync_to_async` | Create, update, delete — needs user context |

Read-only tools can be synchronous because FastMCP handles the async bridge. Mutation tools must be async to access the request-scoped user context via `await get_mcp_user(ctx)`.

### Server Registration

```python
# mcp_server/server.py

mcp = FastMCP("my-project", instructions="...", middleware=[...])

register_product_tools(mcp)
register_order_tools(mcp)
register_inventory_tools(mcp)
```

All `register_*` calls happen at module import time. The tools are available immediately when the ASGI app starts.

---

## ASGI Dual Transport Mount

Two MCP transports share a single FastMCP instance, served by one Uvicorn process.

### Streamable HTTP (Standard)

POST-based JSON-RPC at `/mcp/`. Stateless — could support multiple workers (but single worker is simpler when SSE is also served).

### SSE (Legacy)

Server-Sent Events at `/mcp/sse/`. Stateful — session state lives in the worker's memory. **Requires single Uvicorn worker.** Supported for backward compatibility with older MCP clients.

### Health Check

`GET /mcp/health` returns `{"status": "ok"}`. Used by load balancers, Docker health checks, and monitoring.

### Deployment

```bash
# Separate from the Django WSGI server
uvicorn myproject.asgi:app --host 0.0.0.0 --port 8001 --workers 1
```

The MCP server runs on a separate port from Django's WSGI server (Gunicorn). Nginx routes `/mcp/` to Uvicorn:

```nginx
# Streamable HTTP + SSE
location /mcp/ {
    proxy_pass http://mcp:8001;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 300s;
}
```

> ⚠️ `proxy_buffering off` and `proxy_cache off` are required for SSE. Without them, Nginx buffers the event stream and clients see no output.

---

## Admin UI & CLI Token Management

Two approaches for creating and managing tokens.

### Django Admin

Register `MCPToken` with Django's admin. Show the full token **once** after creation — it cannot be retrieved later.

```python
# admin.py
from django.contrib import admin
from .models import MCPToken

@admin.register(MCPToken)
class MCPTokenAdmin(admin.ModelAdmin):
    list_display = ['name', 'user', 'is_active', 'masked_token_display',
                    'expires_at', 'last_used_at', 'created_at']
    list_filter = ['is_active']
    search_fields = ['name', 'user__email']
    readonly_fields = ['token', 'last_used_at', 'created_at', 'updated_at']
```

For frameworks with richer admin UIs (e.g., Wagtail snippets), register as a snippet with grouped tool selection widgets and token-shown-once creation views.

### Management Command

For scripted or CI token creation:

```python
# management/commands/create_mcp_token.py
from django.core.management.base import BaseCommand, CommandError

class Command(BaseCommand):
    help = "Create an MCP token for a user and print the full token."

    def add_arguments(self, parser):
        parser.add_argument("--user", required=True, help="User email.")
        parser.add_argument("--name", required=True, help="Token name.")
        parser.add_argument("--tools", default="",
                            help="Comma-separated allowed tools (empty = all).")
        parser.add_argument("--expires-days", type=int, default=None,
                            help="Days until expiry (omit for no expiry).")

    def handle(self, *args, **options):
        User = get_user_model()
        try:
            user = User.objects.get(email=options["user"])
        except User.DoesNotExist:
            raise CommandError(f'User "{options["user"]}" not found.')

        if not user.is_active:
            raise CommandError(f'User "{options["user"]}" is inactive.')

        allowed_tools = []
        if options["tools"]:
            allowed_tools = [t.strip() for t in options["tools"].split(",")
                             if t.strip()]

        expires_at = None
        if options["expires_days"] is not None:
            if options["expires_days"] < 1:
                raise CommandError("--expires-days must be at least 1.")
            expires_at = timezone.now() + timedelta(days=options["expires_days"])

        token = MCPToken.objects.create(
            user=user,
            name=options["name"],
            allowed_tools=allowed_tools,
            expires_at=expires_at,
        )

        self.stdout.write(self.style.SUCCESS("✔ MCP token created"))
        self.stdout.write(f"  Name:  {token.name}")
        self.stdout.write(f"  User:  {user.email}")
        self.stdout.write(self.style.WARNING("  Token (shown once):"))
        self.stdout.write(f"  {token.token}")
```

**Usage:**

```bash
# All tools, no expiry
python manage.py create_mcp_token --user admin@example.com --name "Claude Desktop"

# Restricted tools
python manage.py create_mcp_token --user admin@example.com --name "Read Only" \
    --tools list_products,get_order_status

# 30-day expiry
python manage.py create_mcp_token --user admin@example.com --name "Temp" \
    --expires-days 30
```

---

## MCP Resources

Read-only reference data registered with `@mcp.resource()`. Resources give LLMs context about your domain without requiring tool calls.

```python
# mcp_server/tools/reference.py
from pathlib import Path

def register_resources(mcp):

    @mcp.resource("myapp://api-schema")
    def api_schema() -> str:
        """OpenAPI schema for the public API."""
        return Path("static/openapi.yaml").read_text()

    @mcp.resource("myapp://style-guide")
    def style_guide() -> str:
        """Content style guide for consistent authoring."""
        return Path("static/style-guide.md").read_text()
```

**Good candidates for resources:**

- Design tokens / CSS custom properties
- Template structure descriptions
- API schemas or field references
- Content style guides
- Image/media specifications

---

## Domain Extension Examples

### Wagtail CMS (Angelia)

Angelia adds Wagtail-specific patterns on top of the core:

- **Page tree tools** — `get_page_tree()`, `get_page_content()` navigate Wagtail's hierarchical page model
- **Type-specific CRUD** — `create_flex_page()`, `create_blog_post()`, `create_event()` with type-aware field handling
- **Media tools** — `search_images()` returns pre-generated rendition URLs; `upload_image()` uses async httpx
- **Wagtail snippet admin** — MCPToken registered as a Wagtail snippet with `GroupedToolWidget` for tool selection
- **Design resources** — CSS custom properties, base template structure, rendition spec reference
- **Permissions** — Token inherits user's Wagtail page permissions, collection permissions, group memberships
- **Audit trail** — All page revisions record `user=token.user`

### Generic Django (Inventory API)

A hypothetical inventory system using standard Django:

```python
# mcp_server/tools/inventory.py

def register_inventory_tools(mcp):

    @mcp.tool
    def search_products(query: str, in_stock: bool = True, limit: int = 20) -> dict:
        """Search products by name or SKU. Filter by stock availability."""
        qs = Product.objects.filter(name__icontains=query)
        if in_stock:
            qs = qs.filter(stock_quantity__gt=0)
        return {"products": [
            {"id": p.id, "name": p.name, "sku": p.sku, "stock": p.stock_quantity}
            for p in qs[:limit]
        ]}

    @mcp.tool
    async def adjust_stock(
        product_id: int, quantity_change: int, reason: str,
        ctx: Context = None,
    ) -> dict:
        """Adjust stock for a product. Positive = restock, negative = deduct."""
        user = await get_mcp_user(ctx)

        def _do_adjust():
            product = Product.objects.get(id=product_id)
            product.stock_quantity += quantity_change
            product.save(update_fields=['stock_quantity'])
            StockAdjustment.objects.create(
                product=product, quantity=quantity_change,
                reason=reason, user=user,
            )
            return {"id": product.id, "new_stock": product.stock_quantity}

        return await sync_to_async(_do_adjust, thread_sensitive=True)()
```

- **Django admin** — Standard `MCPTokenAdmin` with `list_display` and `list_filter`
- **Permissions** — Django model permissions (`can_change_product`, `can_view_order`)
- **Resources** — Product category taxonomy, warehouse location reference

---

## Anti-Patterns

- ❌ Don't run multiple Uvicorn workers with SSE transport — sessions live in the worker's memory and POSTs hit random workers, causing `404 Not Found` on `/mcp/messages/`
- ❌ Don't store tokens hashed — the middleware needs plaintext lookup via `MCPToken.objects.get(token=token_string)`. Use `unique=True` + `db_index=True` instead
- ❌ Don't skip `sync_to_async` for ORM calls in async tools — Django raises `SynchronousOnlyOperation` when ORM methods are called from an async context
- ❌ Don't put auth logic inside individual tools — use middleware so auth is enforced uniformly before any tool executes
- ❌ Don't exceed 750-character tool descriptions — the MCP spec allows 1024, but leaving headroom avoids protocol overhead issues
- ❌ Don't inline metrics recording in every tool — extract a shared `record_tool_call(name, status, duration)` helper to keep tools focused on business logic
- ❌ Don't serialize user/token to state — use `serializable=False` to prevent Django model instances from leaking into logs or JSON responses
- ❌ Don't show full tokens after initial creation — display the masked version (`get_masked_token()`) in admin list views and logs

---

## Settings

```python
# settings.py

# Require Bearer token authentication for MCP requests.
# Set to False only for local development.
MCP_REQUIRE_AUTH = env.bool('MCP_REQUIRE_AUTH', default=True)
```

Additional deployment settings (not MCP-specific, but required for the pattern):

| Setting | Example | Purpose |
|---------|---------|---------|
| `ASGI_APPLICATION` | `'myproject.asgi.app'` | Uvicorn entrypoint |
| Uvicorn port | `8001` | Separate from WSGI server |
| Uvicorn workers | `1` | Required for SSE transport |
| Nginx proxy target | `http://mcp:8001` | Route `/mcp/` to Uvicorn |

---

## Testing

Standard test cases every implementation should cover.

### Token Model Tests

```python
class MCPTokenModelTest(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            email="test@example.com", password="pass123"
        )

    def test_token_auto_generated(self):
        """Token is auto-generated on creation."""
        token = MCPToken.objects.create(user=self.user, name="Test")
        self.assertIsNotNone(token.token)
        self.assertTrue(len(token.token) > 20)

    def test_active_token_is_valid(self):
        """Active non-expired token is valid."""
        token = MCPToken.objects.create(user=self.user, name="Valid")
        self.assertTrue(token.is_valid)

    def test_inactive_token_not_valid(self):
        """Deactivated token is not valid."""
        token = MCPToken.objects.create(
            user=self.user, name="Off", is_active=False
        )
        self.assertFalse(token.is_valid)

    def test_expired_token_not_valid(self):
        """Expired token is not valid."""
        token = MCPToken.objects.create(
            user=self.user, name="Old",
            expires_at=timezone.now() - timedelta(hours=1),
        )
        self.assertFalse(token.is_valid)

    def test_tool_restriction(self):
        """Restricted token only permits listed tools."""
        token = MCPToken.objects.create(
            user=self.user, name="Limited",
            allowed_tools=["list_products"],
        )
        self.assertTrue(token.can_use_tool("list_products"))
        self.assertFalse(token.can_use_tool("delete_product"))

    def test_unrestricted_permits_all(self):
        """Empty allowed_tools permits any tool."""
        token = MCPToken.objects.create(user=self.user, name="Open")
        self.assertTrue(token.can_use_tool("anything"))

    def test_record_usage(self):
        """record_usage updates last_used_at."""
        token = MCPToken.objects.create(user=self.user, name="Usage")
        self.assertIsNone(token.last_used_at)
        token.record_usage()
        token.refresh_from_db()
        self.assertIsNotNone(token.last_used_at)

    def test_masked_token(self):
        """Masked token hides most characters."""
        token = MCPToken.objects.create(user=self.user, name="Mask")
        masked = token.get_masked_token()
        self.assertTrue(masked.endswith(token.token[-8:]))
        self.assertIn("*", masked)
```

### Auth Resolution Tests

```python
class MCPAuthTest(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            email="auth@example.com", password="pass123"
        )
        self.token = MCPToken.objects.create(
            user=self.user, name="Auth Test"
        )

    def test_resolve_valid_token(self):
        """Valid token resolves to user and token."""
        user, token = resolve_mcp_user(self.token.token)
        self.assertEqual(user.email, "auth@example.com")

    def test_invalid_token_raises(self):
        with self.assertRaises(MCPAuthError):
            resolve_mcp_user("invalid-token-string")

    def test_inactive_token_raises(self):
        self.token.is_active = False
        self.token.save()
        with self.assertRaises(MCPAuthError):
            resolve_mcp_user(self.token.token)

    def test_expired_token_raises(self):
        self.token.expires_at = timezone.now() - timedelta(hours=1)
        self.token.save()
        with self.assertRaises(MCPAuthError):
            resolve_mcp_user(self.token.token)

    def test_disabled_user_raises(self):
        self.user.is_active = False
        self.user.save()
        with self.assertRaises(MCPAuthError):
            resolve_mcp_user(self.token.token)
```

### Server Registration Tests

```python
class MCPServerRegistrationTest(TestCase):
    def test_expected_tools_registered(self):
        """All expected tools are registered on the server."""
        from .server import mcp
        tools = asyncio.run(mcp.list_tools())
        tool_names = {t.name for t in tools}
        for expected in ["list_products", "create_product", "adjust_stock"]:
            self.assertIn(expected, tool_names)

    def test_resources_registered(self):
        """All expected resources are registered."""
        from .server import mcp
        resources = asyncio.run(mcp.list_resources())
        uris = {str(r.uri) for r in resources}
        self.assertIn("myapp://api-schema", uris)
```

### Management Command Tests

```python
class CreateMCPTokenCommandTest(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            email="cmd@example.com", password="pass123"
        )

    def test_create_basic_token(self):
        out = StringIO()
        call_command(
            "create_mcp_token",
            user="cmd@example.com",
            name="CLI Test",
            stdout=out,
        )
        self.assertEqual(MCPToken.objects.count(), 1)
        self.assertIn("CLI Test", out.getvalue())

    def test_invalid_user_raises(self):
        with self.assertRaises(CommandError):
            call_command(
                "create_mcp_token",
                user="nobody@example.com",
                name="Fail",
            )
```

---

## Deployment

### Dual-Worker Architecture

| Process | Server | Port | Protocol | Purpose |
|---------|--------|------|----------|---------|
| Web | Gunicorn | 8080 | WSGI | Django views, admin, static |
| MCP | Uvicorn | 8001 | ASGI | MCP tools (Streamable HTTP + SSE) |

Both processes share the same Django codebase and database. Nginx routes traffic:

- `/mcp/*` → Uvicorn (port 8001)
- Everything else → Gunicorn (port 8080)

### Docker Compose

```yaml
services:
  web:
    build: .
    command: gunicorn --bind :8080 --workers 3 myproject.wsgi
    ports:
      - "8080:8080"

  mcp:
    build: .
    command: uvicorn myproject.asgi:app --host 0.0.0.0 --port 8001 --workers 1
    ports:
      - "8001:8001"

  nginx:
    image: nginx:alpine
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    ports:
      - "443:443"
    depends_on:
      - web
      - mcp
```

### Entrypoint Pattern

Wait for dependencies before starting the server:

```bash
#!/bin/bash
set -e

# Wait for database
until PGPASSWORD=$DB_PASSWORD psql -h $DB_HOST -U $DB_USER -d $DB_NAME -c '\q'; do
  echo "Waiting for database..."
  sleep 2
done

# Run migrations
python manage.py migrate --noinput

# Collect static files
python manage.py collectstatic --noinput

# Start server (command passed via Docker CMD)
exec "$@"
```

### Client Configuration

```json
{
  "mcpServers": {
    "my-project": {
      "url": "https://my-site.com/mcp/",
      "headers": {
        "Authorization": "Bearer <token>"
      }
    }
  }
}
```

For SSE transport, change URL to `/mcp/sse/`.