Linting and formating fixes

Author: jerome3o-anthropic

src/mcp/server/auth/__init__.py

@@ -1,3 +1,3 @@
 """
 MCP OAuth server authorization components.
-"""
+"""

src/mcp/server/auth/errors.py

@@ -4,138 +4,149 @@
 Corresponds to TypeScript file: src/server/auth/errors.ts
 """
 
-from typing import Any, TypedDict
+from typing import TypedDict
 
 
 class OAuthErrorResponse(TypedDict):
     """OAuth error response format."""
+
     error: str
     error_description: str
 
 
 class OAuthError(Exception):
     """
     Base class for all OAuth errors.
-    
+
     Corresponds to OAuthError in src/server/auth/errors.ts
     """
+
     error_code: str = "server_error"
-    
+
     def __init__(self, message: str):
         super().__init__(message)
         self.message = message
-        
+
     def to_response_object(self) -> OAuthErrorResponse:
         """Convert error to JSON response object."""
-        return {
-            "error": self.error_code,
-            "error_description": self.message
-        }
+        return {"error": self.error_code, "error_description": self.message}
 
 
 class ServerError(OAuthError):
     """
     Server error.
-    
+
     Corresponds to ServerError in src/server/auth/errors.ts
     """
+
     error_code = "server_error"
 
 
 class InvalidRequestError(OAuthError):
     """
     Invalid request error.
-    
+
     Corresponds to InvalidRequestError in src/server/auth/errors.ts
     """
+
     error_code = "invalid_request"
 
 
 class InvalidClientError(OAuthError):
     """
     Invalid client error.
-    
+
     Corresponds to InvalidClientError in src/server/auth/errors.ts
     """
+
     error_code = "invalid_client"
 
 
 class InvalidGrantError(OAuthError):
     """
     Invalid grant error.
-    
+
     Corresponds to InvalidGrantError in src/server/auth/errors.ts
     """
+
     error_code = "invalid_grant"
 
 
 class UnauthorizedClientError(OAuthError):
     """
     Unauthorized client error.
-    
+
     Corresponds to UnauthorizedClientError in src/server/auth/errors.ts
     """
+
     error_code = "unauthorized_client"
 
 
 class UnsupportedGrantTypeError(OAuthError):
     """
     Unsupported grant type error.
-    
+
     Corresponds to UnsupportedGrantTypeError in src/server/auth/errors.ts
     """
+
     error_code = "unsupported_grant_type"
 
 
 class UnsupportedResponseTypeError(OAuthError):
     """
     Unsupported response type error.
-    
+
     Corresponds to UnsupportedResponseTypeError in src/server/auth/errors.ts
     """
+
     error_code = "unsupported_response_type"
 
 
 class InvalidScopeError(OAuthError):
     """
     Invalid scope error.
-    
+
     Corresponds to InvalidScopeError in src/server/auth/errors.ts
     """
+
     error_code = "invalid_scope"
 
 
 class AccessDeniedError(OAuthError):
     """
     Access denied error.
-    
+
     Corresponds to AccessDeniedError in src/server/auth/errors.ts
     """
+
     error_code = "access_denied"
 
 
 class TemporarilyUnavailableError(OAuthError):
     """
     Temporarily unavailable error.
-    
+
     Corresponds to TemporarilyUnavailableError in src/server/auth/errors.ts
     """
+
     error_code = "temporarily_unavailable"
 
 
 class InvalidTokenError(OAuthError):
     """
     Invalid token error.
-    
+
     Corresponds to InvalidTokenError in src/server/auth/errors.ts
     """
+
     error_code = "invalid_token"
 
 
 class InsufficientScopeError(OAuthError):
     """
     Insufficient scope error.
-    
+
     Corresponds to InsufficientScopeError in src/server/auth/errors.ts
     """
-    error_code = "insufficient_scope"
+
+    error_code = "insufficient_scope"

src/mcp/server/auth/handlers/__init__.py

@@ -1,3 +1,3 @@
 """
 Request handlers for MCP authorization endpoints.
-"""
+"""

src/mcp/server/auth/handlers/authorize.py

@@ -4,45 +4,49 @@
 Corresponds to TypeScript file: src/server/auth/handlers/authorize.ts
 """
 
-import re
-from urllib.parse import urlparse, urlunparse, urlencode
-from typing import Any, Callable, Dict, List, Literal, Optional
-from urllib.parse import urlencode, parse_qs
+from typing import Literal
+from urllib.parse import urlencode, urlparse, urlunparse
 
-from starlette.requests import Request
-from starlette.responses import JSONResponse, RedirectResponse, Response
 from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, ValidationError
-from pydantic_core import Url
+from starlette.requests import Request
+from starlette.responses import RedirectResponse, Response
 
 from mcp.server.auth.errors import (
-    InvalidClientError, 
+    InvalidClientError,
     InvalidRequestError,
-    UnsupportedResponseTypeError,
-    ServerError,
     OAuthError,
 )
+from mcp.server.auth.handlers.types import HandlerFn
 from mcp.server.auth.provider import AuthorizationParams, OAuthServerProvider
-from mcp.shared.auth import OAuthClientInformationFull
 
 
 class AuthorizationRequest(BaseModel):
     """
     Model for the authorization request parameters.
-    
-    Corresponds to request schema in authorizationHandler in src/server/auth/handlers/authorize.ts
+
+    Corresponds to request schema in authorizationHandler in
+    src/server/auth/handlers/authorize.ts
     """
+
     client_id: str = Field(..., description="The client ID")
-    redirect_uri: AnyHttpUrl | None = Field(..., description="URL to redirect to after authorization")
+    redirect_uri: AnyHttpUrl | None = Field(
+        ..., description="URL to redirect to after authorization"
+    )
 
-    response_type: Literal["code"] = Field(..., description="Must be 'code' for authorization code flow")
+    response_type: Literal["code"] = Field(
+        ..., description="Must be 'code' for authorization code flow"
+    )
     code_challenge: str = Field(..., description="PKCE code challenge")
-    code_challenge_method: Literal["S256"] = Field("S256", description="PKCE code challenge method")
-    state: Optional[str] = Field(None, description="Optional state parameter")
-    scope: Optional[str] = Field(None, description="Optional scope parameter")
-    
+    code_challenge_method: Literal["S256"] = Field(
+        "S256", description="PKCE code challenge method"
+    )
+    state: str | None = Field(None, description="Optional state parameter")
+    scope: str | None = Field(None, description="Optional scope parameter")
+
     class Config:
         extra = "ignore"
 
+
 def validate_scope(requested_scope: str | None, scope: str | None) -> list[str] | None:
     if requested_scope is None:
         return None
@@ -53,7 +57,10 @@ def validate_scope(requested_scope: str | None, scope: str | None) -> list[str]
             raise InvalidRequestError(f"Client was not registered with scope {scope}")
     return requested_scopes
 
-def validate_redirect_uri(redirect_uri: AnyHttpUrl | None, redirect_uris: list[AnyHttpUrl]) -> AnyHttpUrl:
+
+def validate_redirect_uri(
+    redirect_uri: AnyHttpUrl | None, redirect_uris: list[AnyHttpUrl]
+) -> AnyHttpUrl:
     if not redirect_uris:
         raise InvalidClientError("Client has no registered redirect URIs")
 
@@ -67,16 +74,19 @@ def validate_redirect_uri(redirect_uri: AnyHttpUrl | None, redirect_uris: list[A
     elif len(redirect_uris) == 1:
         return redirect_uris[0]
     else:
-        raise InvalidRequestError("redirect_uri must be specified when client has multiple registered URIs")
+        raise InvalidRequestError(
+            "redirect_uri must be specified when client has multiple registered URIs"
+        )
+
 
-def create_authorization_handler(provider: OAuthServerProvider) -> Callable:
+def create_authorization_handler(provider: OAuthServerProvider) -> HandlerFn:
     """
     Create a handler for the OAuth 2.0 Authorization endpoint.
-    
+
     Corresponds to authorizationHandler in src/server/auth/handlers/authorize.ts
 
     """
-    
+
     async def authorization_handler(request: Request) -> Response:
         """
         Handler for the OAuth 2.0 Authorization endpoint.
@@ -94,65 +104,64 @@ async def authorization_handler(request: Request) -> Response:
                 auth_request = AuthorizationRequest.model_validate(params)
         except ValidationError as e:
             raise InvalidRequestError(str(e))
-        
+
         # Get client information
-        try:
-            client = await provider.clients_store.get_client(auth_request.client_id)
-        except OAuthError as e:
-            # TODO: proper error rendering
-            raise InvalidClientError(str(e))
-        
+        client = await provider.clients_store.get_client(auth_request.client_id)
+
         if not client:
             raise InvalidClientError(f"Client ID '{auth_request.client_id}' not found")
-        
- 
+
         # do validation which is dependent on the client configuration
-        redirect_uri = validate_redirect_uri(auth_request.redirect_uri, client.redirect_uris)
+        redirect_uri = validate_redirect_uri(
+            auth_request.redirect_uri, client.redirect_uris
+        )
         scopes = validate_scope(auth_request.scope, client.scope)
-        
+
         auth_params = AuthorizationParams(
             state=auth_request.state,
             scopes=scopes,
             code_challenge=auth_request.code_challenge,
             redirect_uri=redirect_uri,
         )
-        
-        response = RedirectResponse(url="", status_code=302, headers={"Cache-Control": "no-store"})
-            
+
+        response = RedirectResponse(
+            url="", status_code=302, headers={"Cache-Control": "no-store"}
+        )
+
         try:
             # Let the provider handle the authorization flow
             await provider.authorize(client, auth_params, response)
-            
+
             return response
         except Exception as e:
             return RedirectResponse(
                 url=create_error_redirect(redirect_uri, e, auth_request.state),
                 status_code=302,
                 headers={"Cache-Control": "no-store"},
-                )
-    
+            )
+
     return authorization_handler
 
-def create_error_redirect(redirect_uri: AnyUrl, error: Exception, state: Optional[str]) -> str:
+
+def create_error_redirect(
+    redirect_uri: AnyUrl, error: Exception, state: str | None
+) -> str:
     parsed_uri = urlparse(str(redirect_uri))
     if isinstance(error, OAuthError):
-        query_params = {
-            "error": error.error_code,
-            "error_description": str(error)
-        }
+        query_params = {"error": error.error_code, "error_description": str(error)}
     else:
         query_params = {
             "error": "internal_error",
-            "error_description": "An unknown error occurred"
+            "error_description": "An unknown error occurred",
         }
     # TODO: should we add error_uri?
     # if error.error_uri:
     #     query_params["error_uri"] = str(error.error_uri)
     if state:
         query_params["state"] = state
-    
+
     new_query = urlencode(query_params)
     if parsed_uri.query:
         new_query = f"{parsed_uri.query}&{new_query}"
-    
-    return urlunparse(parsed_uri._replace(query=new_query))
+
+    return urlunparse(parsed_uri._replace(query=new_query))