14. Authentication

Chapter 14 of 22 · 20 min

KEY INSIGHT

Authentication verifies identity before granting accessΓÇöwithout it, every tool becomes a public interface vulnerable to unauthorized use. MCP supports multiple authentication mechanisms. The simplest, API keys, work for straightforward scenarios: ```python from fastapi import Request, HTTPException from functools import wraps API_KEYS = { "agent-key-001": {"name": "Development Agent", "scope": "full"}, "agent-key-002": {"name": "Production Agent", "scope": "limited"}, } async def authenticate_request(request: Request) -> dict: auth_header = request.headers.get("Authorization", "") if not auth_header.startswith("Bearer "): raise HTTPException(status_code=401, detail="Missing or invalid authorization") key = auth_header[7:] # Strip "Bearer " if key not in API_KEYS: raise HTTPException(status_code=401, detail="Invalid API key") return API_KEYS[key] # Apply to endpoints app.add_middleware(AuthMiddleware) ``` For production deployments, OAuth 2.0 provides delegated access with fine-grained permissions. The flow involves redirecting users to an authorization server, handling the callback with a code, and exchanging that code for access tokens: ```python from authlib.integrations.starlette_client import OAuth oauth = OAuth() oauth.register( name="mcp", client_id=OAUTH_CLIENT_ID, client_secret=OAUTH_CLIENT_SECRET, server_metadata_url=CONFIG["well_known_url"], client_kwargs={"scope": "mcp:read mcp:write mcp:admin"}, ) @app.get("/auth/login") async def login(request: Request): return await oauth.mcp.authorize_redirect( request, request.url_for("auth_callback") ) @app.get("/auth/callback") async def callback(request: Request): token = await oauth.mcp.authorize_access_token(request) # Store token for authenticated requests return {"access_token": token["access_token"]} ``` JWT tokens provide stateless authentication. After validating credentials, issue a signed token containing claims: ```python import jwt from datetime import datetime, timedelta def create_token(subject: str, scopes: list[str]) -> str: payload = { "sub": subject, "scopes": scopes, "iat": datetime.utcnow(), "exp": datetime.utcnow() + timedelta(hours=1), } 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 HTTPException(401, "Token expired") except jwt.InvalidTokenError: raise HTTPException(401, "Invalid token") ``` Token refresh handles long-lived sessions. When a client presents an expired token, issue a new one if the refresh token remains valid.

Local verification checkpoint

Run the smallest example from this chapter in a local workspace and record the package version, runtime, data path, and observed output. If the result depends on model size, vector count, CPU/GPU backend, or available memory, note that constraint beside the exercise so the lesson remains reproducible.

EXERCISE

Implement token-based authentication for an MCP server. Create endpoints for registration, login, and token refresh. Add a middleware that validates tokens on all tool calls. Test that unauthenticated requests receive 401 responses.