Build an MCP Server in Django: give AI agents safe access to your app.

Every AI Integration You've Written Twice

If you've shipped anything AI-flavoured on top of Django in the last two years, you've written the same adapter more than once. A function that queries the ORM, a JSON schema describing that function, a dispatch table mapping tool names back to Python callables, a loop that feeds results to a model. Then a different client turns up — a desktop assistant, an IDE, an agent framework — and you write the whole thing again in that client's shape.

The Model Context Protocol exists to end that. It's an open protocol that standardises the boundary between an AI client and the systems it needs to reach: the client discovers what a server can do, calls it, and gets structured results back. You implement it once against your Django app, and every MCP-capable client — Claude, Claude Code, an editor, your own agent — talks to it without a bespoke adapter.

This post is the version I'd want to read before starting: how MCP actually models capabilities, how to run a server inside an existing Django project rather than beside it, how to touch the ORM from async tool code without tripping over Django's threading rules, and — the part most tutorials skip entirely — how to make sure the agent on the other end can only see data the person driving it is allowed to see.

What MCP Actually Is

Strip away the branding and MCP is JSON-RPC 2.0 over a transport, with an agreed vocabulary for three kinds of capability. The vocabulary is the interesting part, because which one you choose changes who decides when your code runs.

That distinction matters more than it looks. A tool is something the model calls on its own initiative because your description convinced it the tool was relevant — so a tool that deletes records is a tool the model can decide to call. A resource is addressed by URI and pulled in by the client, typically because a user attached it; the model doesn't reach for it unprompted. A prompt is a template the user picks from a menu. When you're deciding where a piece of functionality belongs, ask who you want holding the trigger.

Transports

Two matter. stdio runs your server as a subprocess of the client and speaks over pipes — perfect for local developer tooling, useless for a Django app serving many users. Streamable HTTP exposes a single endpoint that accepts POSTed JSON-RPC messages and can upgrade to a server-sent-event stream when the server needs to push. That's the one you deploy. (An older HTTP+SSE transport with a separate endpoint pair is deprecated; don't build on it.)

Because streamable HTTP is a plain ASGI application, the whole thing slots into a Django deployment without a second service, a second image, or a second set of secrets.

Mounting an MCP Server Inside Django

Start with the server object and one trivial tool, in a normal Django app module:

# crm/mcp_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "acme-crm",
    stateless_http=True,        # no per-session state; safe behind N workers
    streamable_http_path="/",   # we'll mount this app at /mcp/ ourselves
)

@mcp.tool()
async def ping() -> str:
    """Health check. Returns 'pong' if the CRM server is reachable."""
    return "pong"

Now wire it into ASGI. The instinct is to reach for Django's URLconf, but the MCP app isn't a Django view — it's a sibling ASGI application. Put a tiny Starlette router in front of both:

# config/asgi.py
import os

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")

from django.core.asgi import get_asgi_application

django_app = get_asgi_application()   # must be built before importing app code

from starlette.applications import Starlette
from starlette.routing import Mount

from crm.mcp_server import mcp

mcp_app = mcp.streamable_http_app()

application = Starlette(
    routes=[
        Mount("/mcp", app=mcp_app),
        Mount("/", app=django_app),      # everything else is still Django
    ],
    # CRITICAL: the MCP app starts its session manager in its own lifespan.
    # A mounted sub-app never receives lifespan events, so hand it up here.
    lifespan=lambda app: mcp_app.router.lifespan_context(app),
)

Run it with any ASGI server and you have a working MCP endpoint at /mcp alongside your entire existing Django site:

uvicorn config.asgi:application --host 0.0.0.0 --port 8000

# Point a client at it — for Claude Code:
claude mcp add --transport http acme-crm http://localhost:8000/mcp

# Or inspect it interactively with the official dev tool:
npx @modelcontextprotocol/inspector

The inspector is worth the detour. It lists your tools, shows the exact JSON schema generated from your type hints, and lets you invoke each one by hand — which turns "the model didn't call my tool" from a mystery into a five-second check of whether the schema is what you thought it was.

Tools Over the Django ORM

Here's where Django's specifics start to bite. MCP tools run in an async context, and the Django ORM refuses to run synchronous queries there — you'll get SynchronousOnlyOperation the first time you touch a queryset. You have two ways through, and you'll use both.

For straightforward reads and writes, use Django's async query API: aget(), acreate(), aexists(), acount(), aupdate(), and async for over a queryset.

# crm/mcp_server.py
from typing import Annotated

from django.db.models import Q
from pydantic import Field

from crm.models import Customer


@mcp.tool()
async def search_customers(
    query: Annotated[str, Field(description="Name or email fragment to match")],
    limit: Annotated[int, Field(description="Max rows to return", ge=1, le=50)] = 10,
) -> list[dict]:
    """Search customers by name or email address.

    Returns the matching customers with their id, name, email and plan.
    Use the returned id with get_customer for the full record.
    """
    qs = (
        Customer.objects
        .filter(Q(name__icontains=query) | Q(email__icontains=query))
        .order_by("-created_at")[:limit]
    )
    return [
        {"id": c.id, "name": c.name, "email": c.email, "plan": c.plan}
        async for c in qs          # async iteration — no sync_to_async needed
    ]

For everything the async API doesn't cover — model methods that lazily follow relations, DRF serializers, anything calling into third-party sync code — wrap it in sync_to_async with thread_sensitive=True so it runs in the same thread as the rest of your sync work and shares its database connection state:

from asgiref.sync import sync_to_async

from crm.models import Invoice
from crm.serializers import InvoiceSerializer


@sync_to_async(thread_sensitive=True)
def _serialize_invoices(customer_id: int) -> list[dict]:
    # Ordinary synchronous Django. select_related here, not in the tool —
    # every lazy relation you forget becomes a query the agent waits on.
    qs = (
        Invoice.objects
        .filter(customer_id=customer_id)
        .select_related("customer")
        .order_by("-issued_at")[:20]
    )
    return InvoiceSerializer(qs, many=True).data


@mcp.tool()
async def list_invoices(customer_id: int) -> list[dict]:
    """List the 20 most recent invoices for a customer, newest first."""
    return await _serialize_invoices(customer_id)

Two things to keep front of mind. First, N+1 queries hurt more here than in a view: a page renders once, but an agent may call the same tool a dozen times in one conversation while the user watches a spinner. Put your select_related and prefetch_related in from the start. Second, transactions don't cross the async boundary the way you expect — if a tool needs atomicity, do the whole unit of work inside one sync_to_async function wrapped in transaction.atomic(), not by sprinkling awaits between writes.

Designing Tools a Model Can Actually Use

A tool is not an endpoint. Your REST API is consumed by a developer who read the docs; your MCP tools are consumed by a model that gets one paragraph of description and has to decide, unassisted, whether this is the right thing to call. The interface design rules are genuinely different.

• The docstring is the prompt. It's the entire basis on which the model chooses your tool. Say what it does, what it returns, and when not to use it. "Search customers by name or email. Use get_customer for the full record" beats "Searches customers." by a wide margin.
• Fewer, broader tools beat many narrow ones. Thirty tools is a menu the model has to reason over on every turn, and near-duplicates get confused with each other. One search_customers with a filter argument is better than search_by_name, search_by_email and search_by_plan.
• Return small, flat, self-describing data. Every field costs context. Return the eight fields that matter and an id for drilling down, not your full serializer output with forty nullable columns.
• Name arguments the way you'd name them to a colleague. customer_id, not pk. include_cancelled, not flag2. The parameter names go straight into the schema the model reads.
• Errors are messages, not exceptions. A raised exception becomes an opaque failure. Returning "No customer with id 4102. Use search_customers to find the right id." lets the model recover on its own turn.
• Separate reads from writes, loudly. Name destructive tools so they read as destructive, and mark them with the appropriate tool annotations so clients can gate them behind confirmation.

from mcp.server.fastmcp import FastMCP
from mcp.types import ToolAnnotations

@mcp.tool(
    annotations=ToolAnnotations(
        title="Cancel subscription",
        readOnlyHint=False,
        destructiveHint=True,      # clients can require confirmation
        idempotentHint=True,       # cancelling twice is harmless
    )
)
async def cancel_subscription(customer_id: int, reason: str) -> str:
    """Cancel a customer's active subscription at the end of the billing period.

    Does NOT issue a refund — use refund_invoice for that.
    Returns a confirmation string, or an explanation if there was
    no active subscription to cancel.
    """
    ...

Annotations are hints, not enforcement — a client is free to ignore them, so they never substitute for a real permission check. But they're how a well-behaved client knows to put a confirmation dialog in front of the call, and they cost you one line.

Resources and Prompts

Tools get the attention, but the other two primitives are what make a server feel native inside a client. Resources are read-only, URI-addressed content the client can attach to a conversation — think of them as the things a user would paste in if you hadn't provided them.

@mcp.resource("crm://customer/{customer_id}/summary")
async def customer_summary(customer_id: str) -> str:
    """A plain-text summary of one customer's account."""
    c = await Customer.objects.filter(pk=customer_id).afirst()
    if c is None:
        return f"No customer with id {customer_id}."
    open_invoices = await c.invoices.filter(status="open").acount()
    return (
        f"{c.name} <{c.email}>\n"
        f"Plan: {c.plan}  ·  Since: {c.created_at:%Y-%m-%d}\n"
        f"Open invoices: {open_invoices}"
    )

Prompts are user-triggered templates. They show up in the client's UI as something a person deliberately picks, which makes them the right home for the multi-step workflows your team runs by hand every week.

@mcp.prompt()
def churn_review(customer_id: str) -> str:
    """Draft a churn-risk review for a customer."""
    return (
        f"Review customer {customer_id} for churn risk.\n"
        "1. Fetch the account with get_customer.\n"
        "2. Pull the last 20 invoices with list_invoices.\n"
        "3. Flag: late payments, downgrades, support volume, usage decline.\n"
        "4. Finish with a risk rating (low/medium/high) and one recommended action.\n"
        "Do not contact the customer or change the account."
    )

That prompt is the honest answer to "how do I make the agent follow our process?" — you don't hope the model infers it, you ship the process as a prompt.

Authentication, and Why This Is the Whole Ballgame

Almost every MCP tutorial you'll find stops before this section, and the result is a lot of servers that expose an entire production database to anyone who can reach the port. An MCP endpoint is an unauthenticated remote-procedure-call surface until you make it otherwise, and the caller is a language model that can be talked into things.

The MCP specification defines an OAuth 2.1 authorization flow for HTTP transports, and if you're serving third parties that's the road to take. For an internal server — the common case — a bearer token resolved to a Django user in ASGI middleware is correct, simple, and auditable. The pattern: authenticate at the edge, stash the user in a ContextVar, and make every tool read it.

# crm/mcp_auth.py
from contextvars import ContextVar

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

from crm.models import AgentToken

_current_user: ContextVar = ContextVar("mcp_current_user", default=None)


def current_user():
    """The Django user behind this MCP call. Raises if there isn't one."""
    user = _current_user.get()
    if user is None:
        raise RuntimeError("No authenticated user on this MCP request")
    return user


async def _resolve(raw_token: str):
    token = await (
        AgentToken.objects
        .select_related("user")
        .filter(key=raw_token, revoked_at__isnull=True)
        .afirst()
    )
    if token is None or token.expires_at < timezone.now():
        return None
    return token.user


class MCPTokenAuth:
    """ASGI middleware: bearer token -> Django user, per request."""

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)

        headers = dict(scope["headers"])
        raw = headers.get(b"authorization", b"").decode()
        token = raw.removeprefix("Bearer ").strip()
        user = await _resolve(token) if token else None

        if user is None:
            await send({
                "type": "http.response.start",
                "status": 401,
                "headers": [
                    (b"content-type", b"application/json"),
                    (b"www-authenticate", b'Bearer realm="mcp"'),
                ],
            })
            await send({"type": "http.response.body",
                        "body": b'{"error": "unauthorized"}'})
            return

        reset = _current_user.set(user)
        try:
            await self.app(scope, receive, send)
        finally:
            _current_user.reset(reset)   # never leak a user across requests

Wrap the MCP app with it in asgi.py — Mount("/mcp", app=MCPTokenAuth(mcp_app)) — and now every tool has a real user to authorize against. Which is the point:

from crm.mcp_auth import current_user

@mcp.tool()
async def search_customers(query: str, limit: int = 10) -> list[dict]:
    """Search customers you have access to, by name or email."""
    user = current_user()
    qs = (
        Customer.objects
        .filter(organisation_id=user.organisation_id)   # <- the important line
        .filter(Q(name__icontains=query) | Q(email__icontains=query))
        .order_by("-created_at")[:limit]
    )
    return [{"id": c.id, "name": c.name, "email": c.email} async for c in qs]

Everything in the Django security checklist applies here unchanged, and one item applies double: authorize at the data layer. A tool is a view with no template, no CSRF token, and a caller you can't reason about.

Running It in Production

The gap between a working MCP server and a deployable one is smaller than you'd fear, but it isn't zero.

Statelessness and scaling

stateless_http=True is the setting that makes MCP behave like the rest of your Django app: each request is self-contained, so any worker can serve it and you don't need sticky sessions or shared session storage. Stateful mode buys you server-initiated notifications; unless you need those, don't pay for the affinity.

Timeouts and slow tools

An agent waiting on a tool call is a user watching nothing happen. Give every tool a hard budget and fail loudly inside it rather than letting the client time out with no explanation. For genuinely long work, don't block — hand back a job id and let the agent poll, exactly as you would for an async Celery task.

import asyncio

@mcp.tool()
async def generate_report(customer_id: int) -> str:
    """Generate the monthly usage report for a customer. Takes a few seconds."""
    try:
        async with asyncio.timeout(20):
            return await _build_report(customer_id)
    except TimeoutError:
        return (
            "Report generation exceeded 20s and was cancelled. "
            "Try a narrower date range, or use queue_report for a background run."
        )

Audit logging

Log every tool invocation with the resolved user, the tool name, the arguments and the outcome. You will need it — the first time someone asks "why did the agent change that record", a structured log line is the difference between an answer and a shrug. This is also the data that tells you which tools the model actually reaches for, which is the only honest signal about whether your descriptions are working.

The rest of the list

• Rate-limit per token. An agent in a retry loop can hammer a tool far harder than a human ever would.
• Cap every result set. Enforce a maximum limit server-side; a model that asks for 10,000 rows will blow up the context window and your database at the same time.
• Version through tool names. There's no URL to version. Renaming a tool or changing its arguments is a breaking change for every connected client — add search_customers_v2 rather than mutating the original.
• Keep the endpoint off the public internet if it can be. Internal MCP servers belong behind the VPN, with the token as defence in depth rather than the only defence.
• Test tools as ordinary async functions. They're plain Python — pytest.mark.asyncio, pytest.mark.django_db, call the function, assert on the dict. Save protocol-level testing for a couple of smoke tests through the inspector. The patterns in testing async Django code carry over directly.

Production Checklist

• Mount the MCP app in your ASGI stack, not as a second service. One Starlette router in front of get_asgi_application() and mcp.streamable_http_app(), and pass the MCP lifespan up to the outer app or nothing works.
• Use streamable HTTP with stateless_http=True. Any worker serves any request; no sticky sessions.
• Never touch the ORM synchronously in a tool. Async query API for simple cases, sync_to_async(thread_sensitive=True) for everything else, one atomic block per unit of work.
• Write docstrings for the model, not for the docs site. What it does, what it returns, when not to use it — that text is the routing logic.
• Prefer few broad tools to many narrow ones, and return compact flat data with an id for drilling down.
• Authenticate at the ASGI edge and scope every queryset to the resolved user. No unauthenticated MCP endpoint, ever. The ContextVar gets reset in a finally.
• Treat tool output as untrusted input. Anything a user can write into your database can reach the model as instructions.
• Annotate destructive tools and keep reads and writes clearly separated by name and by permission.
• Budget, cap and rate-limit: a timeout on every tool, a server-side maximum on every result set, a rate limit on every token.
• Log every invocation with user, tool, arguments and outcome. It's your audit trail and your only real usage analytics.

What surprised me most, building these, is how much of the work is interface design rather than integration. The plumbing is an afternoon. The rest is deciding what a language model should be trusted to do with your data, writing that down precisely enough that it picks the right tool, and enforcing it at the data layer for the times it doesn't. An MCP server is an API whose consumer can be talked into things — build it accordingly, and it becomes the most useful surface your Django app has.