Authentication and Authorization

A stdio server inherits its security from the operating system: it runs as you, on your machine, and nobody else can reach it. The moment you move to HTTP, all of that disappears and an endpoint that will run tools on request is sitting on the open internet. This lesson is about putting that back.

Choosing an authentication strategy

Pick by who the callers are, not by what sounds most rigorous. Over-engineered auth on a single-user server is a maintenance burden that buys nothing:

Whichever branch you land on, one rule holds throughout: the secret comes from the environment, never from the source file. Every example below is written that way.

API Key Authentication

Simple but effective for personal servers:

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

class APIKeyMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, api_keys: set):
        super().__init__(app)
        self.api_keys = api_keys

    async def dispatch(self, request, call_next):
        api_key = request.headers.get("X-API-Key")

        if api_key not in self.api_keys:
            return JSONResponse(
                {"error": "Invalid API key"},
                status_code=401
            )

        return await call_next(request)

# Apply middleware
app.add_middleware(APIKeyMiddleware, api_keys={"sk_live_abc123"})

JWT Authentication

For multi-user systems:

import os
import jwt
from datetime import datetime, timedelta, timezone

# Never a literal in source. Reading it this way fails loudly at startup
# if the variable is missing — better than a server that boots happily
# and rejects every token for reasons nobody can see.
SECRET_KEY = os.environ["MCP_JWT_SECRET"]

def create_token(user_id: str) -> str:
    payload = {
        "sub": user_id,
        "exp": datetime.now(timezone.utc) + timedelta(hours=24)
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def verify_token(token: str) -> dict:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        raise ValueError("Token expired")
    except jwt.InvalidTokenError:
        raise ValueError("Invalid token")

class JWTMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        auth_header = request.headers.get("Authorization", "")

        if not auth_header.startswith("Bearer "):
            return JSONResponse({"error": "Missing token"}, status_code=401)

        token = auth_header[7:]  # Remove "Bearer "

        try:
            payload = verify_token(token)
            request.state.user_id = payload["sub"]
        except ValueError as e:
            return JSONResponse({"error": str(e)}, status_code=401)

        return await call_next(request)

Tool-Level Authorization

Restrict tools based on user permissions:

from mcp.server import MCPServer
from mcp.shared.exceptions import MCPError

TOOL_PERMISSIONS = {
    "search_documents": ["read"],
    "add_document": ["read", "write"],
    "delete_document": ["admin"],
}


async def authz_middleware(ctx, call_next):
    """Check permissions before any tool handler runs."""
    if ctx.method != "tools/call":
        return await call_next(ctx)

    tool = (ctx.params or {}).get("name", "")
    required = TOOL_PERMISSIONS.get(tool)

    # A tool with no entry is denied, not allowed.
    if required is None:
        raise MCPError(code=-32602, message=f"Unknown tool: {tool}")

    granted = permissions_for(ctx)
    if not all(p in granted for p in required):
        raise MCPError(code=-32603, message=f"Permission denied for tool: {tool}")

    return await call_next(ctx)


mcp = MCPServer(name="secured-server", middleware=[authz_middleware])

Three decisions in that block are worth stating, because each one is a way authorisation usually goes wrong:

• The check runs in middleware, before any handler. Authorisation written inside each tool is authorisation you can forget to write, and the tool you forget is the one that matters.
• .get(tool) returning None denies. The earlier form of this pattern used .get(name, []), and an empty requirement list passes every check — so a tool nobody added to the table becomes a tool everybody can call. Default deny.
• all(), not any(). add_document needs read and write. With any(), read alone would let a reader write.

Rate Limiting

Protect against abuse:

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.route("/mcp")
@limiter.limit("100/minute")
async def handle_mcp(request):
    ...

Next, we'll explore real-time updates and notifications. :::