refactor: use sync function for fetch server config

Author: gao-sun

mcpauth/__init__.py

@@ -3,16 +3,23 @@
 
 from .middleware.create_bearer_auth import BaseBearerAuthConfig, BearerAuthConfig
 from .types import VerifyAccessTokenFunction
-from .utils.fetch_server_config import ServerMetadataPaths
 from .config import MCPAuthConfig
 from .exceptions import MCPAuthAuthServerException, AuthServerExceptionCode
-from .utils.validate_server_config import validate_server_config
-from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
-from starlette.requests import Request
-from starlette.responses import Response, JSONResponse
+from .utils import validate_server_config
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.responses import JSONResponse
 
 
 class MCPAuth:
+    """
+    The main class for the mcp-auth library, which provides methods for creating middleware
+    functions for handling OAuth 2.0-related tasks and bearer token auth.
+
+    See Also: https://mcp-auth.dev for more information about the library and its usage.
+
+    :param config: An instance of `MCPAuthConfig` containing the server configuration.
+    """
+
     def __init__(self, config: MCPAuthConfig):
         result = validate_server_config(config.server)
 
@@ -32,41 +39,24 @@ def __init__(self, config: MCPAuthConfig):
 
         self.config = config
 
-    def delegated_middleware(self) -> type[BaseHTTPMiddleware]:
+    def metadata_response(self) -> JSONResponse:
         """
-        Returns a middleware that handles OAuth 2.0 Authorization Metadata endpoint
-        (`/.well-known/oauth-authorization-server`) with CORS support (delegated mode).
-
-        :return: A middleware class that can be used in a Starlette or FastAPI application.
+        Returns a response containing the server metadata in JSON format with CORS support.
         """
         server_config = self.config.server
 
-        class DelegatedMiddleware(BaseHTTPMiddleware):
-            async def dispatch(
-                self, request: Request, call_next: RequestResponseEndpoint
-            ) -> Response:
-                path = request.url.path
-                if path == ServerMetadataPaths.OAUTH:
-                    response = JSONResponse(
-                        {
-                            k: v
-                            for k, v in server_config.metadata.model_dump().items()
-                            if v is not None
-                        },
-                        status_code=200,
-                    )
-                    response.headers["Access-Control-Allow-Origin"] = "*"
-                    response.headers["Access-Control-Allow-Methods"] = "GET, OPTIONS"
-                    return response
-                else:
-                    return await call_next(request)
-
-        return DelegatedMiddleware
+        response = JSONResponse(
+            server_config.metadata.model_dump(exclude_none=True),
+            status_code=200,
+        )
+        response.headers["Access-Control-Allow-Origin"] = "*"
+        response.headers["Access-Control-Allow-Methods"] = "GET, OPTIONS"
+        return response
 
     def bearer_auth_middleware(
         self,
         mode_or_verify: Union[Literal["jwt"], VerifyAccessTokenFunction],
-        config: BaseBearerAuthConfig,
+        config: BaseBearerAuthConfig = BaseBearerAuthConfig(),
         jwt_options: dict[str, Any] = {},
     ) -> type[BaseHTTPMiddleware]:
         """
@@ -83,7 +73,7 @@ def bearer_auth_middleware(
 
         metadata = self.config.server.metadata
         if isinstance(mode_or_verify, str) and mode_or_verify == "jwt":
-            from .utils.create_verify_jwt import create_verify_jwt
+            from .utils import create_verify_jwt
 
             if not metadata.jwks_uri:
                 raise MCPAuthAuthServerException(

mcpauth/exceptions.py

@@ -36,7 +36,7 @@ def to_json(self, show_cause: bool = False) -> Record:
             "error_description": self.message,
             "cause": (
                 (
-                    {k: v for k, v in self.cause.model_dump().items() if v is not None}
+                    self.cause.model_dump(exclude_none=True)
                     if isinstance(self.cause, BaseModel)
                     else str(self.cause)
                 )

mcpauth/models/__init__.py

@@ -0,0 +1,7 @@
+from .auth_server import (
+    AuthServerConfig as AuthServerConfig,
+    AuthServerType as AuthServerType,
+)
+from .oauth import (
+    AuthorizationServerMetadata as AuthorizationServerMetadata,
+)

mcpauth/utils/__init__.py

@@ -0,0 +1,14 @@
+from ._create_verify_jwt import create_verify_jwt as create_verify_jwt
+from ._fetch_server_config import (
+    fetch_server_config as fetch_server_config,
+    fetch_server_config_by_well_known_url as fetch_server_config_by_well_known_url,
+    ServerMetadataPaths as ServerMetadataPaths,
+)
+from ._validate_server_config import (
+    validate_server_config as validate_server_config,
+    AuthServerConfigErrorCode as AuthServerConfigErrorCode,
+    AuthServerConfigError as AuthServerConfigError,
+    AuthServerConfigWarningCode as AuthServerConfigWarningCode,
+    AuthServerConfigWarning as AuthServerConfigWarning,
+    AuthServerConfigValidationResult as AuthServerConfigValidationResult,
+)

mcpauth/utils/create_verify_jwt.py renamed to mcpauth/utils/_create_verify_jwt.py

@@ -1,7 +1,7 @@
 from enum import Enum
 from typing import Callable, Optional
 from urllib.parse import urlparse, urlunparse
-import aiohttp
+import requests
 import pydantic
 from pathlib import Path
 
@@ -46,7 +46,7 @@ def get_oidc_well_known_url(issuer: str) -> str:
     return urlunparse(parsed._replace(path=new_path))
 
 
-async def fetch_server_config_by_well_known_url(
+def fetch_server_config_by_well_known_url(
     well_known_url: str,
     type: AuthServerType,
     transpile_data: Optional[Callable[[Record], Record]] = None,
@@ -69,14 +69,13 @@ async def fetch_server_config_by_well_known_url(
     """
 
     try:
-        async with aiohttp.ClientSession() as session:
-            async with session.get(well_known_url) as response:
-                response.raise_for_status()
-                json = await response.json()
-                transpiled_data = transpile_data(json) if transpile_data else json
-                return AuthServerConfig(
-                    metadata=AuthorizationServerMetadata(**transpiled_data), type=type
-                )
+        response = requests.get(well_known_url, timeout=10)
+        response.raise_for_status()
+        json = response.json()
+        transpiled_data = transpile_data(json) if transpile_data else json
+        return AuthServerConfig(
+            metadata=AuthorizationServerMetadata(**transpiled_data), type=type
+        )
     except pydantic.ValidationError as e:
         raise MCPAuthAuthServerException(
             AuthServerExceptionCode.INVALID_SERVER_METADATA,
@@ -90,7 +89,7 @@ async def fetch_server_config_by_well_known_url(
         ) from e
 
 
-async def fetch_server_config(
+def fetch_server_config(
     issuer: str,
     type: AuthServerType,
     transpile_data: Optional[Callable[[Record], Record]] = None,
@@ -141,6 +140,4 @@ async def fetch_server_config(
         if type == AuthServerType.OAUTH
         else get_oidc_well_known_url(issuer)
     )
-    return await fetch_server_config_by_well_known_url(
-        well_known_url, type, transpile_data
-    )
+    return fetch_server_config_by_well_known_url(well_known_url, type, transpile_data)

mcpauth/utils/fetch_server_config.py renamed to mcpauth/utils/_fetch_server_config.py

@@ -14,9 +14,9 @@ keywords = [
   "openid-connect",
 ]
 dependencies = [
-  "aiohttp>=3.11.18",
   "pydantic>=2.11.3",
   "pyjwt[crypto]>=2.9.0",
+  "requests>=2.32.3",
   "starlette>=0.46.2",
 ]
 
@@ -27,9 +27,14 @@ documentation = "https://mcp-auth.dev/docs"
 
 [dependency-groups]
 dev = [
-  "aresponses>=3.0.0",
   "black>=24.8.0",
   "pytest>=8.3.5",
   "pytest-asyncio>=0.26.0",
   "pytest-cov>=6.1.1",
+  "responses>=0.25.7",
+  "uvicorn>=0.34.2",
 ]
+
+[tool.coverage.run]
+branch = true
+source = ["mcpauth"]

mcpauth/utils/validate_server_config.py renamed to mcpauth/utils/_validate_server_config.py

@@ -0,0 +1,30 @@
+from mcpauth import MCPAuth
+from mcpauth.config import MCPAuthConfig
+from mcpauth.models import AuthServerType
+from mcpauth.utils import fetch_server_config, ServerMetadataPaths
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.responses import JSONResponse
+from starlette.requests import Request
+
+mcpAuth = MCPAuth(
+    MCPAuthConfig(
+        server=fetch_server_config("https://auth.logto.io/oidc", AuthServerType.OIDC)
+    )
+)
+
+protected_app = Starlette(
+    middleware=[Middleware(mcpAuth.bearer_auth_middleware("jwt"))]
+)
+
+
+@protected_app.route("/")  # type: ignore
+async def secret_endpoint(_: Request):
+    return JSONResponse({"secret": True})
+
+
+app = Starlette(
+    debug=True,
+)
+app.mount(ServerMetadataPaths.OAUTH.value, mcpAuth.metadata_response())
+app.mount("/mcp", protected_app)