content: from collections.abc import Awaitable, Callable
  from typing import Any
  from urllib.parse import urlparse
  
  from pydantic import AnyHttpUrl
  from starlette.middleware.cors import CORSMiddleware
  from starlette.requests import Request
  from starlette.responses import Response
  from starlette.routing import Route, request_response  # type: ignore
  from starlette.types import ASGIApp
  
  from mcp.server.auth.handlers.authorize import AuthorizationHandler
  from mcp.server.auth.handlers.metadata import MetadataHandler
  from mcp.server.auth.handlers.register import RegistrationHandler
  from mcp.server.auth.handlers.revoke import RevocationHandler
  from mcp.server.auth.handlers.token import TokenHandler
  from mcp.server.auth.middleware.client_auth import ClientAuthenticator
  from mcp.server.auth.provider import OAuthAuthorizationServerProvider
  from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
  from mcp.server.streamable_http import MCP_PROTOCOL_VERSION_HEADER
  from mcp.shared.auth import OAuthMetadata
  
  
  def validate_issuer_url(url: AnyHttpUrl):
      """
      Validate that the issuer URL meets OAuth 2.0 requirements.
  
      Args:
          url: The issuer URL to validate
  
      Raises:
          ValueError: If the issuer URL is invalid
      """
  
      # RFC 8414 requires HTTPS, but we allow localhost HTTP for testing
      if (
          url.scheme != "https"
          and url.host != "localhost"
          and (url.host is not None and not url.host.startswith("127.0.0.1"))
      ):
          raise ValueError("Issuer URL must be HTTPS")  # pragma: no cover
  
      # No fragments or query parameters allowed
      if url.fragment:
          raise ValueError("Issuer URL must not have a fragment")  # pragma: no cover
      if url.query:
          raise ValueError("Issuer URL must not have a query string")  # pragma: no cover
  
  
  AUTHORIZATION_PATH = "/authorize"
  TOKEN_PATH = "/token"
  REGISTRATION_PATH = "/register"
  REVOCATION_PATH = "/revoke"
  
  
  def cors_middleware(
      handler: Callable[[Request], Response | Awaitable[Response]],
      allow_methods: list[str],
  ) -> ASGIApp:
      cors_app = CORSMiddleware(
          app=request_response(handler),
          allow_origins="*",
          allow_methods=allow_methods,
          allow_headers=[MCP_PROTOCOL_VERSION_HEADER],
      )
      return cors_app
  
  
  def create_auth_routes(
      provider: OAuthAuthorizationServerProvider[Any, Any, Any],
      issuer_url: AnyHttpUrl,
      service_documentation_url: AnyHttpUrl | None = None,
      client_registration_options: ClientRegistrationOptions | None = None,
      revocation_options: RevocationOptions | None = None,
  ) -> list[Route]:
      validate_issuer_url(issuer_url)
  
      client_registration_options = client_registration_options or ClientRegistrationOptions()
      revocation_options = revocation_options or RevocationOptions()
      metadata = build_metadata(
          issuer_url,
          service_documentation_url,
          client_registration_options,
          revocation_options,
      )
      client_authenticator = ClientAuthenticator(provider)
  
      # Create routes
      # Allow CORS requests for endpoints meant to be hit by the OAuth client
      # (with the client secret). This is intended to support things like MCP Inspector,
      # where the client runs in a web browser.
      routes = [
          Route(
              "/.well-known/oauth-authorization-server",
              endpoint=cors_middleware(
                  MetadataHandler(metadata).handle,
                  ["GET", "OPTIONS"],
              ),
              methods=["GET", "OPTIONS"],
          ),
          Route(
              AUTHORIZATION_PATH,
              # do not allow CORS for authorization endpoint;
              # clients should just redirect to this
              endpoint=AuthorizationHandler(provider).handle,
              methods=["GET", "POST"],
          ),
          Route(
              TOKEN_PATH,
              endpoint=cors_middleware(
                  TokenHandler(provider, client_authenticator).handle,
                  ["POST", "OPTIONS"],
              ),
              methods=["POST", "OPTIONS"],
          ),
      ]
  
      if client_registration_options.enabled:  # pragma: no branch
          registration_handler = RegistrationHandler(
              provider,
              options=client_registration_options,
          )
          routes.append(
              Route(
                  REGISTRATION_PATH,
                  endpoint=cors_middleware(
                      registration_handler.handle,
                      ["POST", "OPTIONS"],
                  ),
                  methods=["POST", "OPTIONS"],
              )
          )
  
      if revocation_options.enabled:  # pragma: no branch
          revocation_handler = RevocationHandler(provider, client_authenticator)
          routes.append(
              Route(
                  REVOCATION_PATH,
                  endpoint=cors_middleware(
                      revocation_handler.handle,
                      ["POST", "OPTIONS"],
                  ),
                  methods=["POST", "OPTIONS"],
              )
          )
  
      return routes
  
  
  def build_metadata(
      issuer_url: AnyHttpUrl,
      service_documentation_url: AnyHttpUrl | None,
      client_registration_options: ClientRegistrationOptions,
      revocation_options: RevocationOptions,
  ) -> OAuthMetadata:
      authorization_url = AnyHttpUrl(str(issuer_url).rstrip("/") + AUTHORIZATION_PATH)
      token_url = AnyHttpUrl(str(issuer_url).rstrip("/") + TOKEN_PATH)
  
      # Create metadata
      metadata = OAuthMetadata(
          issuer=issuer_url,
          authorization_endpoint=authorization_url,
          token_endpoint=token_url,
          scopes_supported=client_registration_options.valid_scopes,
          response_types_supported=["code"],
          response_modes_supported=None,
          grant_types_supported=["authorization_code", "refresh_token"],
          token_endpoint_auth_methods_supported=["client_secret_post", "client_secret_basic"],
          token_endpoint_auth_signing_alg_values_supported=None,
          service_documentation=service_documentation_url,
          ui_locales_supported=None,
          op_policy_uri=None,
          op_tos_uri=None,
          introspection_endpoint=None,
          code_challenge_methods_supported=["S256"],
      )
  
      # Add registration endpoint if supported
      if client_registration_options.enabled:  # pragma: no branch
          metadata.registration_endpoint = AnyHttpUrl(str(issuer_url).rstrip("/") + REGISTRATION_PATH)
  
      # Add revocation endpoint if supported
      if revocation_options.enabled:  # pragma: no branch
          metadata.revocation_endpoint = AnyHttpUrl(str(issuer_url).rstrip("/") + REVOCATION_PATH)
          metadata.revocation_endpoint_auth_methods_supported = ["client_secret_post", "client_secret_basic"]
  
      return metadata
  
  
  def build_resource_metadata_url(resource_server_url: AnyHttpUrl) -> AnyHttpUrl:
      """
      Build RFC 9728 compliant protected resource metadata URL.
  
      Inserts /.well-known/oauth-protected-resource between host and resource path
      as specified in RFC 9728 §3.1.
  
      Args:
          resource_server_url: The resource server URL (e.g., https://example.com/mcp)
  
      Returns:
          The metadata URL (e.g., https://example.com/.well-known/oauth-protected-resource/mcp)
      """
      parsed = urlparse(str(resource_server_url))
      # Handle trailing slash: if path is just "/", treat as empty
      resource_path = parsed.path if parsed.path != "/" else ""
      return AnyHttpUrl(f"{parsed.scheme}://{parsed.netloc}/.well-known/oauth-protected-resource{resource_path}")
  
  
  def create_protected_resource_routes(
      resource_url: AnyHttpUrl,
      authorization_servers: list[AnyHttpUrl],
      scopes_supported: list[str] | None = None,
      resource_name: str | None = None,
      resource_documentation: AnyHttpUrl | None = None,
  ) -> list[Route]:
      """
      Create routes for OAuth 2.0 Protected Resource Metadata (RFC 9728).
  
      Args:
          resource_url: The URL of this resource server
          authorization_servers: List of authorization servers that can issue tokens
          scopes_supported: Optional list of scopes supported by this resource
  
      Returns:
          List of Starlette routes for protected resource metadata
      """
      from mcp.server.auth.handlers.metadata import ProtectedResourceMetadataHandler
      from mcp.shared.auth import ProtectedResourceMetadata
  
      metadata = ProtectedResourceMetadata(
          resource=resource_url,
          authorization_servers=authorization_servers,
          scopes_supported=scopes_supported,
          resource_name=resource_name,
          resource_documentation=resource_documentation,
          # bearer_methods_supported defaults to ["header"] in the model
      )
  
      handler = ProtectedResourceMetadataHandler(metadata)
  
      # RFC 9728 §3.1: Register route at /.well-known/oauth-protected-resource + resource path
      metadata_url = build_resource_metadata_url(resource_url)
      # Extract just the path part for route registration
      parsed = urlparse(str(metadata_url))
      well_known_path = parsed.path
  
      return [
          Route(
              well_known_path,
              endpoint=cors_middleware(handler.handle, ["GET", "OPTIONS"]),
              methods=["GET", "OPTIONS"],
          )
      ]
  