Linting

Author: jerome3o-anthropic

src/mcp/server/auth/errors.py

@@ -155,4 +155,7 @@ class InsufficientScopeError(OAuthError):
 
 
 def stringify_pydantic_error(validation_error: ValidationError) -> str:
-    return "\n".join(f"{'.'.join(str(loc) for loc in e['loc'])}: {e['msg']}" for e in validation_error.errors())
+    return "\n".join(
+        f"{'.'.join(str(loc) for loc in e['loc'])}: {e['msg']}"
+        for e in validation_error.errors()
+    )

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

@@ -4,25 +4,27 @@
 Corresponds to TypeScript file: src/server/auth/handlers/authorize.ts
 """
 
+import logging
 from typing import Callable, Literal, Optional, Union
-from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
+from urllib.parse import urlencode, urlparse, urlunparse
 
 from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, RootModel, ValidationError
 from starlette.datastructures import FormData, QueryParams
 from starlette.requests import Request
 from starlette.responses import RedirectResponse, Response
 
 from mcp.server.auth.errors import (
-    InvalidClientError,
     InvalidRequestError,
     OAuthError,
     stringify_pydantic_error,
 )
-from mcp.server.auth.provider import AuthorizationParams, OAuthServerProvider, construct_redirect_uri
-from mcp.shared.auth import OAuthClientInformationFull
 from mcp.server.auth.json_response import PydanticJSONResponse
-
-import logging
+from mcp.server.auth.provider import (
+    AuthorizationParams,
+    OAuthServerProvider,
+    construct_redirect_uri,
+)
+from mcp.shared.auth import OAuthClientInformationFull
 
 logger = logging.getLogger(__name__)
 
@@ -80,15 +82,17 @@ def validate_redirect_uri(
             "redirect_uri must be specified when client has multiple registered URIs"
         )
 
+
 ErrorCode = Literal[
-        "invalid_request",
-        "unauthorized_client",
-        "access_denied",
-        "unsupported_response_type",
-        "invalid_scope",
-        "server_error",
-        "temporarily_unavailable"
-    ]
+    "invalid_request",
+    "unauthorized_client",
+    "access_denied",
+    "unsupported_response_type",
+    "invalid_scope",
+    "server_error",
+    "temporarily_unavailable",
+]
+
 
 class ErrorResponse(BaseModel):
     error: ErrorCode
@@ -97,14 +101,18 @@ class ErrorResponse(BaseModel):
     # must be set if provided in the request
     state: Optional[str]
 
-def best_effort_extract_string(key: str, params: None | FormData | QueryParams) -> Optional[str]:
+
+def best_effort_extract_string(
+    key: str, params: None | FormData | QueryParams
+) -> Optional[str]:
     if params is None:
         return None
     value = params.get(key)
     if isinstance(value, str):
         return value
     return None
 
+
 class AnyHttpUrlModel(RootModel):
     root: AnyHttpUrl
 
@@ -119,18 +127,24 @@ async def authorization_handler(request: Request) -> Response:
         client = None
         params = None
 
-        async def error_response(error: ErrorCode, error_description: str, attempt_load_client: bool = True):
+        async def error_response(
+            error: ErrorCode, error_description: str, attempt_load_client: bool = True
+        ):
             nonlocal client, redirect_uri, state
             if client is None and attempt_load_client:
                 # make last-ditch attempt to load the client
                 client_id = best_effort_extract_string("client_id", params)
-                client = client_id and await provider.clients_store.get_client(client_id)
+                client = client_id and await provider.clients_store.get_client(
+                    client_id
+                )
             if redirect_uri is None and client:
                 # make last-ditch effort to load the redirect uri
                 if params is not None and "redirect_uri" not in params:
                     raw_redirect_uri = None
                 else:
-                    raw_redirect_uri = AnyHttpUrlModel.model_validate(best_effort_extract_string("redirect_uri", params)).root
+                    raw_redirect_uri = AnyHttpUrlModel.model_validate(
+                        best_effort_extract_string("redirect_uri", params)
+                    ).root
                 try:
                     redirect_uri = validate_redirect_uri(raw_redirect_uri, client)
                 except (ValidationError, InvalidRequestError):
@@ -147,7 +161,9 @@ async def error_response(error: ErrorCode, error_description: str, attempt_load_
 
             if redirect_uri and client:
                 return RedirectResponse(
-                    url=construct_redirect_uri(str(redirect_uri), **error_resp.model_dump(exclude_none=True)),
+                    url=construct_redirect_uri(
+                        str(redirect_uri), **error_resp.model_dump(exclude_none=True)
+                    ),
                     status_code=302,
                     headers={"Cache-Control": "no-store"},
                 )
@@ -157,7 +173,7 @@ async def error_response(error: ErrorCode, error_description: str, attempt_load_
                     content=error_resp,
                     headers={"Cache-Control": "no-store"},
                 )
-        
+
         try:
             # Parse request parameters
             if request.method == "GET":
@@ -166,20 +182,22 @@ async def error_response(error: ErrorCode, error_description: str, attempt_load_
             else:
                 # Parse form data for POST requests
                 params = await request.form()
-                
+
             # Save state if it exists, even before validation
             state = best_effort_extract_string("state", params)
-                
+
             try:
                 auth_request = AuthorizationRequest.model_validate(params)
                 state = auth_request.state  # Update with validated state
             except ValidationError as validation_error:
                 error: ErrorCode = "invalid_request"
                 for e in validation_error.errors():
-                    if e['loc'] == ('response_type',) and e['type'] == 'literal_error':
+                    if e["loc"] == ("response_type",) and e["type"] == "literal_error":
                         error = "unsupported_response_type"
                         break
-                return await error_response(error, stringify_pydantic_error(validation_error))
+                return await error_response(
+                    error, stringify_pydantic_error(validation_error)
+                )
 
             # Get client information
             client = await provider.clients_store.get_client(auth_request.client_id)
@@ -191,7 +209,6 @@ async def error_response(error: ErrorCode, error_description: str, attempt_load_
                     attempt_load_client=False,
                 )
 
-                
             # Validate redirect_uri against client's registered URIs
             try:
                 redirect_uri = validate_redirect_uri(auth_request.redirect_uri, client)
@@ -201,7 +218,7 @@ async def error_response(error: ErrorCode, error_description: str, attempt_load_
                     error="invalid_request",
                     error_description=validation_error.message,
                 )
-                
+
             # Validate scope - for scope errors, we can redirect
             try:
                 scopes = validate_scope(auth_request.scope, client)
@@ -211,28 +228,30 @@ async def error_response(error: ErrorCode, error_description: str, attempt_load_
                     error="invalid_scope",
                     error_description=validation_error.message,
                 )
-                
+
             # Setup authorization parameters
             auth_params = AuthorizationParams(
                 state=state,
                 scopes=scopes,
                 code_challenge=auth_request.code_challenge,
                 redirect_uri=redirect_uri,
             )
-            
+
             # Let the provider pick the next URI to redirect to
             response = RedirectResponse(
                 url="", status_code=302, headers={"Cache-Control": "no-store"}
             )
-            response.headers["location"] = await provider.authorize(
-                client, auth_params
-            )
+            response.headers["location"] = await provider.authorize(client, auth_params)
             return response
-                
+
         except Exception as validation_error:
             # Catch-all for unexpected errors
-            logger.exception("Unexpected error in authorization_handler", exc_info=validation_error)
-            return await error_response(error="server_error", error_description="An unexpected error occurred")
+            logger.exception(
+                "Unexpected error in authorization_handler", exc_info=validation_error
+            )
+            return await error_response(
+                error="server_error", error_description="An unexpected error occurred"
+            )
 
     return authorization_handler
 
@@ -241,7 +260,7 @@ def create_error_redirect(
     redirect_uri: AnyUrl, error: Union[Exception, ErrorResponse]
 ) -> str:
     parsed_uri = urlparse(str(redirect_uri))
-    
+
     if isinstance(error, ErrorResponse):
         # Convert ErrorResponse to dict
         error_dict = error.model_dump(exclude_none=True)
@@ -252,7 +271,7 @@ def create_error_redirect(
                     query_params[key] = str(value)
                 else:
                     query_params[key] = value
-                    
+
     elif isinstance(error, OAuthError):
         query_params = {"error": error.error_code, "error_description": str(error)}
     else:
@@ -265,4 +284,4 @@ def create_error_redirect(
     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))

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

@@ -4,7 +4,7 @@
 Corresponds to TypeScript file: src/server/auth/handlers/metadata.ts
 """
 
-from typing import Any, Callable, Dict
+from typing import Callable
 
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
@@ -35,12 +35,13 @@ async def metadata_handler(request: Request) -> Response:
         Returns:
             JSON response with the authorization server metadata
         """
-        # Remove any None values from metadata
-        clean_metadata = {k: v for k, v in metadata.items() if v is not None}
+        # Convert metadata to dict and remove any None values
+        metadata_dict = metadata.model_dump()
+        clean_metadata = {k: v for k, v in metadata_dict.items() if v is not None}
 
         return JSONResponse(
             content=clean_metadata,
             headers={"Cache-Control": "public, max-age=3600"},  # Cache for 1 hour
         )
 
-    return metadata_handler
+    return metadata_handler

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

@@ -14,7 +14,6 @@
 from starlette.responses import JSONResponse, Response
 
 from mcp.server.auth.errors import (
-    InvalidRequestError,
     OAuthError,
     ServerError,
     stringify_pydantic_error,
@@ -23,8 +22,14 @@
 from mcp.server.auth.provider import OAuthRegisteredClientsStore
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
 
+
 class ErrorResponse(BaseModel):
-    error: Literal["invalid_redirect_uri", "invalid_client_metadata", "invalid_software_statement", "unapproved_software_statement"]
+    error: Literal[
+        "invalid_redirect_uri",
+        "invalid_client_metadata",
+        "invalid_software_statement",
+        "unapproved_software_statement",
+    ]
     error_description: str
 
 
@@ -60,10 +65,13 @@ async def registration_handler(request: Request) -> Response:
                 body = await request.json()
                 client_metadata = OAuthClientMetadata.model_validate(body)
             except ValidationError as validation_error:
-                return PydanticJSONResponse(content=ErrorResponse(
-                    error="invalid_client_metadata",
-                    error_description=stringify_pydantic_error(validation_error)
-                ), status_code=400)
+                return PydanticJSONResponse(
+                    content=ErrorResponse(
+                        error="invalid_client_metadata",
+                        error_description=stringify_pydantic_error(validation_error),
+                    ),
+                    status_code=400,
+                )
 
             client_id = str(uuid4())
             client_secret = None
@@ -103,4 +111,4 @@ async def registration_handler(request: Request) -> Response:
             status_code = 500 if isinstance(e, ServerError) else 400
             return JSONResponse(status_code=status_code, content=e.to_response_object())
 
-    return registration_handler
+    return registration_handler

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

@@ -11,15 +11,14 @@
 from starlette.responses import Response
 
 from mcp.server.auth.errors import (
-    InvalidRequestError,
     stringify_pydantic_error,
 )
+from mcp.server.auth.json_response import PydanticJSONResponse
 from mcp.server.auth.middleware.client_auth import (
     ClientAuthenticator,
     ClientAuthRequest,
 )
 from mcp.server.auth.provider import OAuthServerProvider
-from mcp.server.auth.json_response import PydanticJSONResponse
 from mcp.shared.auth import OAuthTokenRevocationRequest, TokenErrorResponse
 
 
@@ -51,17 +50,25 @@ async def revocation_handler(request: Request) -> Response:
             form_data = await request.form()
             revocation_request = RevocationRequest.model_validate(dict(form_data))
         except ValidationError as e:
-            return PydanticJSONResponse(status_code=400, content=TokenErrorResponse(
-                error="invalid_request",
-                error_description=stringify_pydantic_error(e)
-            ))
+            return PydanticJSONResponse(
+                status_code=400,
+                content=TokenErrorResponse(
+                    error="invalid_request",
+                    error_description=stringify_pydantic_error(e),
+                ),
+            )
 
         # Authenticate client
         client_auth_result = await client_authenticator(revocation_request)
 
         # Revoke token
         if provider.revoke_token:
-            await provider.revoke_token(client_auth_result, revocation_request)
+            # Convert RevocationRequest to OAuthTokenRevocationRequest
+            oauth_revocation_request = OAuthTokenRevocationRequest(
+                token=revocation_request.token,
+                token_type_hint=revocation_request.token_type_hint,
+            )
+            await provider.revoke_token(client_auth_result, oauth_revocation_request)
 
         # Return successful empty response
         return Response(
@@ -72,4 +79,4 @@ async def revocation_handler(request: Request) -> Response:
             },
         )
 
-    return revocation_handler
+    return revocation_handler