Add example proxy OAuth server implementation

Signed-off-by: Jesse Sanford <108698+jessesanford@users.noreply.github.com>

Author: jessesanford

examples/servers/proxy-auth/proxy_auth/auth_server.py

@@ -0,0 +1,230 @@
+# pyright: reportMissingImports=false
+import argparse
+import asyncio
+import logging
+import os
+import time
+
+from dotenv import load_dotenv  # type: ignore
+from mcp.server.auth.provider import AccessToken
+from mcp.server.auth.providers.transparent_proxy import (
+    ProxySettings,  # type: ignore
+    ProxyTokenHandler,
+    TransparentOAuthProxyProvider,
+)
+from mcp.server.auth.routes import cors_middleware, create_auth_routes
+from mcp.server.auth.settings import ClientRegistrationOptions
+from pydantic import AnyHttpUrl
+from starlette.applications import Starlette
+from starlette.requests import Request  # type: ignore
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+from uvicorn import Config, Server
+
+# Load environment variables from .env if present
+load_dotenv()
+
+# Configure logging after .env so LOG_LEVEL can come from environment
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
+
+logging.basicConfig(
+    level=LOG_LEVEL,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+
+# Dedicated logger for this server module
+logger = logging.getLogger("proxy_auth.auth_server")
+
+# Suppress noisy INFO messages from the FastMCP low-level server unless we are
+# explicitly running in DEBUG mode. These logs (e.g. "Processing request of type
+# ListToolsRequest") are helpful for debugging but clutter normal output.
+
+_mcp_lowlevel_logger = logging.getLogger("mcp.server.lowlevel.server")
+if LOG_LEVEL == "DEBUG":
+    # In full debug mode, allow the library to emit its detailed logs
+    _mcp_lowlevel_logger.setLevel(logging.DEBUG)
+else:
+    # Otherwise, only warnings and above
+    _mcp_lowlevel_logger.setLevel(logging.WARNING)
+
+# ----------------------------------------------------------------------------
+# Environment configuration
+# ----------------------------------------------------------------------------
+# Load and validate settings from the environment (uses .env automatically)
+settings = ProxySettings.load()
+
+# Upstream endpoints (fully-qualified URLs)
+UPSTREAM_AUTHORIZE: str = str(settings.upstream_authorize)
+UPSTREAM_TOKEN: str = str(settings.upstream_token)
+UPSTREAM_JWKS_URI = settings.jwks_uri
+# Derive base URL from the authorize endpoint for convenience / tests
+UPSTREAM_BASE: str = UPSTREAM_AUTHORIZE.rsplit("/", 1)[0]
+
+# Client credentials & defaults
+CLIENT_ID: str = settings.client_id or "demo-client-id"
+CLIENT_SECRET = settings.client_secret
+DEFAULT_SCOPE: str = settings.default_scope
+
+# Metadata URL (only used if we need to fetch from upstream)
+UPSTREAM_METADATA = f"{UPSTREAM_BASE}/.well-known/oauth-authorization-server"
+
+# ---------------------------------------------------------------------------
+# Logging helpers
+# ---------------------------------------------------------------------------
+
+
+def _mask_secret(secret: str | None) -> str | None:  # noqa: D401
+    """Return a masked version of the given secret.
+
+    The first and last four characters are preserved (if available) and the
+    middle section is replaced by asterisks. If the secret is shorter than
+    eight characters, the entire value is replaced by ``*``.
+    """
+
+    if not secret:
+        return None
+
+    if len(secret) <= 8:
+        return "*" * len(secret)
+
+    return f"{secret[:4]}{'*' * (len(secret) - 8)}{secret[-4:]}"
+
+
+# Consolidated configuration (with sensitive data redacted)
+_masked_settings = settings.model_dump(exclude_none=True).copy()
+
+if "client_secret" in _masked_settings:
+    _masked_settings["client_secret"] = _mask_secret(_masked_settings["client_secret"])
+
+# Log configuration at *debug* level only so it can be enabled when needed
+logger.debug("[Auth Proxy Config] %s", _masked_settings)
+
+# Server host/port
+AUTH_SERVER_PORT = int(os.getenv("AUTH_SERVER_PORT", "9000"))
+AUTH_SERVER_HOST = os.getenv("AUTH_SERVER_HOST", "localhost")
+AUTH_SERVER_URL = os.getenv(
+    "AUTH_SERVER_URL", f"http://{AUTH_SERVER_HOST}:{AUTH_SERVER_PORT}"
+)
+
+# ----------------------------------------------------------------------------
+# Auth Server
+# ----------------------------------------------------------------------------
+
+# Create auth provider
+oauth_provider = TransparentOAuthProxyProvider(settings=settings)
+
+# Enable client registration
+client_registration_options = ClientRegistrationOptions(
+    enabled=True,
+    valid_scopes=["openid"],
+    default_scopes=["openid"],
+)
+
+# Create auth routes
+routes = create_auth_routes(
+    provider=oauth_provider,
+    issuer_url=AnyHttpUrl(AUTH_SERVER_URL),
+    service_documentation_url=None,
+    client_registration_options=client_registration_options,
+    revocation_options=None,
+)
+
+# Add token endpoint handler
+# We need to replace any existing token endpoint route
+routes = [r for r in routes if not (hasattr(r, "path") and r.path == "/token")]
+
+# Create token handler and add it to routes
+proxy_token_handler = ProxyTokenHandler(oauth_provider)
+routes.append(Route("/token", endpoint=proxy_token_handler.handle, methods=["POST"]))
+
+
+# Add token introspection endpoint for Resource Servers
+async def introspect_handler(request: Request) -> Response:
+    """
+    Token introspection endpoint for Resource Servers.
+
+    Resource Servers call this endpoint to validate tokens without
+    needing direct access to token storage.
+    """
+    form = await request.form()
+    token = form.get("token")
+    if not token or not isinstance(token, str):
+        return JSONResponse({"active": False}, status_code=400)
+
+    # For the transparent proxy, we don't actually validate tokens
+    # Just create a dummy AccessToken like the provider does
+    access_token = AccessToken(
+        token=token, client_id=str(CLIENT_ID), scopes=[DEFAULT_SCOPE], expires_at=None
+    )
+
+    return JSONResponse(
+        {
+            "active": True,
+            "client_id": access_token.client_id,
+            "scope": " ".join(access_token.scopes),
+            "exp": access_token.expires_at,
+            "iat": int(time.time()),
+            "token_type": "Bearer",
+            "aud": access_token.resource,  # RFC 8707 audience claim
+        }
+    )
+
+
+routes.append(
+    Route(
+        "/introspect",
+        endpoint=cors_middleware(introspect_handler, ["POST", "OPTIONS"]),
+        methods=["POST", "OPTIONS"],
+    )
+)
+
+# Create Starlette app with routes
+auth_app = Starlette(routes=routes)
+
+
+async def run_server(host: str = AUTH_SERVER_HOST, port: int = AUTH_SERVER_PORT):
+    """Run the Authorization Server."""
+    config = Config(
+        auth_app,
+        host=host,
+        port=port,
+        log_level="info",
+    )
+    server = Server(config)
+
+    logger.info(f"🚀 MCP Authorization Server running on http://{host}:{port}")
+
+    await server.serve()
+
+
+def main():
+    """Command-line entry point for the Authorization Server."""
+    parser = argparse.ArgumentParser(description="MCP OAuth Proxy Authorization Server")
+    parser.add_argument(
+        "--host",
+        default=None,
+        help="Host to bind to (overrides AUTH_SERVER_HOST env var)",
+    )
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=None,
+        help="Port to bind to (overrides AUTH_SERVER_PORT env var)",
+    )
+
+    args = parser.parse_args()
+
+    # Use command-line arguments only if provided, otherwise use environment variables
+    host = args.host or AUTH_SERVER_HOST
+    port = args.port or AUTH_SERVER_PORT
+
+    # Log the configuration being used
+    logger.info(f"Starting Authorization Server with host={host}, port={port}")
+    logger.info("Using environment variables from .env file if present")
+
+    asyncio.run(run_server(host=host, port=port))
+
+
+if __name__ == "__main__":
+    main()