Async deployment

MCPServer exposes two URL trees that mount the same registries, auth backend, and session store. Pick one based on infrastructure:

Mount	Use when
server.urls (default)	WSGI, mixed sync/async work, simplicity
server.async_urls	ASGI, high-concurrency workloads, async-native services

Both speak the same MCP wire format — clients cannot tell the deployments apart. The difference is internal: async_urls dispatches I/O-bound handlers through arun_service / arun_selector so a single Python process can serve many concurrent MCP calls without a thread pool.

Switch to ASGI

urls.py

from django.urls import include, path

from invoices.mcp import server

urlpatterns = [
    # Old: path("mcp/", include(server.urls)),
    path("mcp/", include(server.async_urls)),
]

Run with an ASGI server:

uvicorn myproject.asgi:application
# or
daphne -b 0.0.0.0 -p 8000 myproject.asgi:application
# or
hypercorn myproject.asgi:application --bind 0.0.0.0:8000

A standard myproject/asgi.py from django-admin startproject works without modification — Django's get_asgi_application() happily serves async views.

Sync collaborators are bridged automatically

AsyncStreamableHttpViewSet accepts the same auth backend and session store as the sync view. Sync methods on those collaborators are wrapped in asgiref.sync.sync_to_async at the call site, so the existing AllowAnyBackend, DjangoOAuthToolkitBackend, InMemorySessionStore, and DjangoCacheSessionStore work unchanged.

If you write a genuinely async backend or store — e.g. one that hits a remote IDP via httpx.AsyncClient — declare its methods async def and they are awaited directly without the thread hop:

class HttpxAuthBackend:
    async def authenticate(self, request):
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://idp.example/userinfo",
                headers={"Authorization": request.META.get("HTTP_AUTHORIZATION", "")},
            )
        if response.status_code != 200:
            return None
        return TokenInfo(user=response.json()["sub"], scopes=())

    def protected_resource_metadata(self):
        return {"resource": "https://example.com/mcp/"}

    def www_authenticate_challenge(self, *, scopes=None, error=None):
        return 'Bearer realm="mcp"'

server = MCPServer(name="my-app", auth_backend=HttpxAuthBackend())

The acall helper detects coroutine-functions at runtime via inspect.iscoroutinefunction and routes accordingly — no marker interface required.

Sync vs async services

Both work under async_urls:

• Async services (async def create_invoice(*, data)) run native via arun_service. The full request handling stays on the event loop.
• Sync services (def create_invoice(*, data)) are dispatched through sync_to_async. Django's connection pooling handles the thread hop correctly; ORM calls inside the service work without SynchronousOnlyOperation errors.

The same applies to selectors. Mix freely — the dispatch path picks the right strategy per call.

Server-initiated push (SSE on GET)

When a session opens GET /mcp/, the async view returns a text/event-stream response. The server pushes JSON-RPC payloads on that stream as events; the client interprets each data: line as one MCP message. Idle periods produce SSE keep-alive comments (: keepalive) every ~15 seconds so reverse proxies don't close the connection.

# from app code (a service, a Django signal, a background task — anything
# running in the same process as the MCP server):
await server.notify(session_id, {
    "jsonrpc": "2.0",
    "method": "notifications/progress",
    "params": {"progressToken": "task-7", "value": 0.42},
})

notify returns True if a subscriber was attached, False otherwise. A miss is normal — sessions without an open SSE stream just don't see the event. Most callers fire-and-forget.

Wire details

The endpoint enforces the same headers as POST: Mcp-Protocol-Version required, Mcp-Session-Id required and validated against the session store. Origin allowlist applies. With no broker configured (e.g. a MCPServer(sse_broker=None)), GET returns 405 — spec-compliant when the server has nothing to push.

Scaling across workers

The default InMemorySSEBroker is single-process. A multi-worker deployment can:

• Keep SSE on a single process by running one ASGI worker (or pinning SSE-enabled requests to one worker via session affinity). The simplest path and works for most apps.
• Use the first-party RedisSSEBroker (behind the [redis] extra) — pass it as MCPServer(sse_broker=RedisSSEBroker(client)). See Multi-worker SSE with Redis.
• Roll your own — implement subscribe / unsubscribe / publish against NATS, Kafka, etc., and pass it as MCPServer(sse_broker=...).

The single-subscriber rule applies per-session: re-subscribing replaces the previous queue. Message replay across reconnects is opt-in via an SSEReplayBuffer (InMemorySSEReplayBuffer or RedisSSEReplayBuffer) passed as MCPServer(sse_replay_buffer=...); with no buffer configured, Last-Event-ID is silently ignored. See SSE replay buffer.

When sync is the right answer

If you don't have async-native services and aren't running ASGI today, server.urls is the simpler path. Switching to async without genuine async work below the dispatch layer adds complexity (thread pool, connection management, more failure modes) without buying anything observable.