Lint

Author: praboud-ant

src/mcp/server/auth/__init__.py

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

src/mcp/server/auth/errors.py

@@ -4,132 +4,142 @@
 Corresponds to TypeScript file: src/server/auth/errors.ts
 """
 
-from typing import Dict, Optional, Any
+from typing import Dict
 
 
 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) -> Dict[str, str]:
         """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,21 +4,16 @@
 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 Callable, Literal, Optional
+from urllib.parse import parse_qs, 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.provider import AuthorizationParams, OAuthServerProvider
@@ -28,22 +23,35 @@
 class AuthorizationRequest(BaseModel):
     """
     Model for the authorization request parameters.
-    
+
     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, must be S256")
+    code_challenge_method: Literal["S256"] = Field(
+        "S256", description="PKCE code challenge method, must be S256"
+    )
     state: Optional[str] = Field(None, description="Optional state parameter")
-    scope: Optional[str] = Field(None, description="Optional scope; if specified, should be a space-separated list of scope strings")
-    
+    scope: Optional[str] = Field(
+        None,
+        description="Optional scope; if specified, should be a space-separated list of scope strings",
+    )
+
     class Config:
         extra = "ignore"
 
-def validate_scope(requested_scope: str | None, client: OAuthClientInformationFull) -> list[str] | None:
+
+def validate_scope(
+    requested_scope: str | None, client: OAuthClientInformationFull
+) -> list[str] | None:
     if requested_scope is None:
         return None
     requested_scopes = requested_scope.split(" ")
@@ -53,7 +61,10 @@ def validate_scope(requested_scope: str | None, client: OAuthClientInformationFu
             raise InvalidRequestError(f"Client was not registered with scope {scope}")
     return requested_scopes
 
-def validate_redirect_uri(auth_request: AuthorizationRequest, client: OAuthClientInformationFull) -> AnyHttpUrl:
+
+def validate_redirect_uri(
+    auth_request: AuthorizationRequest, client: OAuthClientInformationFull
+) -> AnyHttpUrl:
     if auth_request.redirect_uri is not None:
         # Validate redirect_uri against client's registered redirect URIs
         if auth_request.redirect_uri not in client.redirect_uris:
@@ -64,16 +75,19 @@ def validate_redirect_uri(auth_request: AuthorizationRequest, client: OAuthClien
     elif len(client.redirect_uris) == 1:
         return client.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:
     """
     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.
@@ -91,74 +105,79 @@ 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))
-        
+
         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, client)
         scopes = validate_scope(auth_request.scope, client)
-        
+
         auth_params = AuthorizationParams(
             state=auth_request.state,
             scopes=scopes,
             code_challenge=auth_request.code_challenge,
             redirect_uri=redirect_uri,
         )
-            
+
         try:
             # Let the provider handle the authorization flow
-            authorization_code = await provider.create_authorization_code(client, auth_params)
-            response = RedirectResponse(url="", status_code=302, headers={"Cache-Control": "no-store"})
-            
+            authorization_code = await provider.create_authorization_code(
+                client, auth_params
+            )
+            response = RedirectResponse(
+                url="", status_code=302, headers={"Cache-Control": "no-store"}
+            )
+
             # Redirect with code
             parsed_uri = urlparse(str(auth_params.redirect_uri))
             query_params = [(k, v) for k, vs in parse_qs(parsed_uri.query) for v in vs]
             query_params.append(("code", authorization_code))
             if auth_params.state:
                 query_params.append(("state", auth_params.state))
-            
-            redirect_url = urlunparse(parsed_uri._replace(query=urlencode(query_params)))
+
+            redirect_url = urlunparse(
+                parsed_uri._replace(query=urlencode(query_params))
+            )
             response.headers["location"] = redirect_url
-            
+
             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: Optional[str]
+) -> 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))