refactor: improve auth middleware for MCP operational endpoints

Author: Stephen Peterkins

examples/server_with_custom_routes.py

@@ -4,7 +4,8 @@
 from north_mcp_python_sdk import NorthMCPServer
 from north_mcp_python_sdk.auth import get_authenticated_user, get_authenticated_user_optional
 
-mcp = NorthMCPServer("Demo with Custom Routes", port=5222)
+# MCP server with operational endpoints for container orchestration
+mcp = NorthMCPServer("MCP Server with K8s Endpoints", port=5222)
 
 
 @mcp.tool()
@@ -21,45 +22,71 @@ def add(a: int, b: int) -> int:
 
 @mcp.custom_route("/health", methods=["GET"])
 async def health_check(request: Request) -> PlainTextResponse:
-    """Health check endpoint - no authentication required"""
+    """Kubernetes liveness probe endpoint - automatically bypasses authentication"""
     return PlainTextResponse("OK")
 
 
-@mcp.custom_route("/status", methods=["GET"])
-async def status_check(request: Request) -> JSONResponse:
-    """Status endpoint - no authentication required"""
+@mcp.custom_route("/ready", methods=["GET"])
+async def readiness_check(request: Request) -> JSONResponse:
+    """Kubernetes readiness probe - checks if server is ready to accept traffic"""
+    # In a real implementation, you might check database connections,
+    # external service availability, etc.
     return JSONResponse({
-        "status": "running",
-        "server": "NorthMCP Demo",
-        "authenticated": False
+        "status": "ready",
+        "server": "MCP Server with K8s Endpoints",
+        "checks": {
+            "mcp_protocol": "ok",
+            "tools_loaded": "ok"
+        }
     })
 
 
-@mcp.custom_route("/user-info", methods=["GET"])
-async def user_info(request: Request) -> JSONResponse:
-    """User info endpoint that shows auth is optional for custom routes"""
+@mcp.custom_route("/metrics", methods=["GET"])
+async def metrics_endpoint(request: Request) -> PlainTextResponse:
+    """Prometheus metrics endpoint for monitoring"""
+    # In a real implementation, you would return actual Prometheus metrics
+    metrics = """# HELP mcp_requests_total Total number of MCP requests
+# TYPE mcp_requests_total counter
+mcp_requests_total 42
+
+# HELP mcp_tools_total Number of available MCP tools
+# TYPE mcp_tools_total gauge
+mcp_tools_total 1
+"""
+    return PlainTextResponse(metrics, media_type="text/plain")
+
+
+@mcp.custom_route("/status", methods=["GET"])
+async def status_check(request: Request) -> JSONResponse:
+    """General status endpoint for monitoring dashboards"""
     user = get_authenticated_user_optional()
 
+    # This endpoint works without auth but can show auth info if provided
+    status_data = {
+        "status": "running",
+        "server": "MCP Server with K8s Endpoints",
+        "version": "1.0.0",
+        "uptime_seconds": 3600,  # In real implementation, track actual uptime
+        "authenticated_request": user is not None
+    }
+    
     if user:
-        return JSONResponse({
-            "authenticated": True,
-            "email": user.email,
-            "connectors": list(user.connector_access_tokens.keys())
-        })
-    else:
-        return JSONResponse({
-            "authenticated": False,
-            "message": "This custom route works without authentication!"
-        })
+        status_data["user_email"] = user.email
+        status_data["available_connectors"] = list(user.connector_access_tokens.keys())
+    
+    return JSONResponse(status_data)
 
 
 if __name__ == "__main__":
-    print("Starting server with custom routes...")
-    print("Try these endpoints:")
-    print("  GET /health - Simple health check")
-    print("  GET /status - Status information")
-    print("  GET /user-info - Shows authentication is optional")
-    print("  POST /mcp - MCP protocol endpoint (requires auth)")
-    print("  GET /sse - Server-sent events endpoint (requires auth)")
+    print("Starting MCP server with Kubernetes operational endpoints...")
+    print("\nOperational endpoints (no authentication required):")
+    print("  GET /health  - Liveness probe for Kubernetes")
+    print("  GET /ready   - Readiness probe for Kubernetes")
+    print("  GET /metrics - Prometheus metrics for monitoring")
+    print("  GET /status  - General status for dashboards")
+    print("\nMCP protocol endpoints (authentication required):")
+    print("  POST /mcp    - JSON-RPC MCP communication")
+    print("  GET /sse     - Server-sent events for streaming")
+    print("\nPerfect for deployment in Kubernetes with proper health checks!")
 
     mcp.run(transport="streamable-http")

examples/simple_custom_route.py

@@ -2,14 +2,24 @@
 from starlette.requests import Request
 from starlette.responses import PlainTextResponse
 
-# Create your North MCP server
-mcp = NorthMCPServer("MyServer")
+# Create MCP server ready for Kubernetes deployment
+mcp = NorthMCPServer("MyMCPServer")
 
-# Add a custom route - no auth middleware applied automatically!
+# Add health check for Kubernetes liveness probe
+# Custom routes automatically bypass authentication - perfect for operational endpoints!
 @mcp.custom_route("/health", methods=["GET"])
 async def health_check(request: Request) -> PlainTextResponse:
+    """Kubernetes liveness probe endpoint"""
     return PlainTextResponse("OK")
 
-# That's it! Custom routes bypass auth, MCP routes (/mcp, /sse) require auth
+@mcp.tool()
+def my_tool(message: str) -> str:
+    """Example MCP tool - accessible via authenticated /mcp endpoint"""
+    return f"Processed: {message}"
+
+# Deploy with confidence: /health works without auth, /mcp requires auth
 if __name__ == "__main__":
+    print("MCP server ready for Kubernetes deployment!")
+    print("Health check: GET /health (no auth)")
+    print("MCP protocol: POST /mcp (requires auth)")
     mcp.run()

src/north_mcp_python_sdk/auth.py

@@ -33,13 +33,20 @@ def __init__(
 
 class NorthAuthenticationMiddleware:
     """
-    North's default authentication middleware that only applies authentication 
-    to MCP protocol paths (/mcp, /sse). Custom routes bypass authentication entirely.
+    North's authentication middleware for MCP servers that applies authentication 
+    only to MCP protocol endpoints (/mcp, /sse). Custom routes bypass authentication
+    and are intended for operational purposes like Kubernetes health checks.
     
-    This is the standard authentication behavior for North MCP servers:
-    - MCP protocol routes (/mcp, /sse) require authentication
-    - All custom routes added via @mcp.custom_route() work without authentication
-    - No configuration needed - this behavior is automatic
+    MCP servers typically only need two authenticated endpoints:
+    - /mcp: JSON-RPC protocol endpoint for MCP communication
+    - /sse: Server-sent events endpoint for streaming transport
+    
+    Custom routes are automatically public and designed for:
+    - Kubernetes liveness/readiness probes (/health, /ready)
+    - Monitoring and metrics endpoints (/metrics, /status)
+    - Other operational/orchestration needs
+    
+    No configuration needed - this behavior follows MCP best practices.
     """
 
     def __init__(
@@ -65,7 +72,7 @@ def _should_authenticate(self, path: str) -> bool:
         Check if the given path requires authentication.
         Only MCP protocol paths (/mcp, /sse) require auth by default.
         """
-        return any(path.startswith(protected) for protected in self.protected_paths)
+        return path in self.protected_paths
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send):
         if scope["type"] == "lifespan":
@@ -74,13 +81,17 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send):
         path = scope.get("path", "")
 
         if not self._should_authenticate(path):
-            self.logger.debug("Path %s does not require authentication, bypassing", path)
+            self.logger.debug(
+                "Path %s is a custom route (likely operational endpoint like health check), "
+                "bypassing authentication as intended for k8s/orchestration use", 
+                path
+            )
             # For non-protected paths, create a minimal unauthenticated user
             scope["user"] = None
             scope["auth"] = AuthCredentials()
             return await self.app(scope, receive, send)
 
-        self.logger.debug("Path %s requires authentication", path)
+        self.logger.debug("Path %s is an MCP protocol endpoint, applying authentication", path)
 
         # Apply authentication for protected paths
         conn = HTTPConnection(scope)
@@ -147,7 +158,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send):
 
         # For custom routes that don't require auth, user will be None
         if user is None:
-            self.logger.debug("No authentication required for this route")
+            self.logger.debug("Custom route accessed without authentication (operational endpoint)")
             token = auth_context_var.set(None)
             try:
                 await self.app(scope, receive, send)