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

# 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.utcnow() + 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:

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

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    user_permissions = get_user_permissions(current_user)
    required_permissions = TOOL_PERMISSIONS.get(name, [])

    if not any(p in user_permissions for p in required_permissions):
        raise McpError(
            code=-32603,
            message=f"Permission denied for tool: {name}"
        )

    # Execute tool
    return await execute_tool(name, arguments)

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