mnemosyne/docs/Pattern_Django-MCP_V1-00.md

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

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

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

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

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:

@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

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

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

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

State Keys

STATE_KEY_USER = "mcp_user"
STATE_KEY_TOKEN = "mcp_token"

Settings

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

Metric Names

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:

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

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

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

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:

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

# 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

# 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

# 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:

# 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.

# 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:

# 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:

# 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.

# 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:

# 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

# 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

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

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

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

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

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:

#!/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

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

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