modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Get baseline 100% clean coverage (#1553)

Browse source
This commit is contained in:
Max Isbey 2025-11-11 13:09:32 +00:00 • committed by GitHub
commit 89e9c43acf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 1032 additions and 1084 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/mcp/server/auth/handlers/authorize.py
View file

@ -50,7 +50,7 @@ class AuthorizationErrorResponse(BaseModel):
def best_effort_extract_string(key: str, params: None | FormData | QueryParams) -> str | None:
if params is None:
if params is None: # pragma: no cover
return None
value = params.get(key)
if isinstance(value, str):
@ -116,7 +116,7 @@ class AuthorizationHandler:
pass
# the error response MUST contain the state specified by the client, if any
if state is None:
if state is None: # pragma: no cover
# make last-ditch effort to load state
state = best_effort_extract_string("state", params)
@ -218,7 +218,7 @@ class AuthorizationHandler:
# Handle authorization errors as defined in RFC 6749 Section 4.1.2.1
return await error_response(error=e.error, error_description=e.error_description)
except Exception as validation_error:
except Exception as validation_error: # pragma: no cover
# 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")

4
src/mcp/server/auth/handlers/register.py
View file

@ -50,7 +50,7 @@ class RegistrationHandler:
client_id = str(uuid4())
client_secret = None
if client_metadata.token_endpoint_auth_method != "none":
if client_metadata.token_endpoint_auth_method != "none": # pragma: no branch
# cryptographically secure random 32-byte hex string
client_secret = secrets.token_hex(32)
@ -59,7 +59,7 @@ class RegistrationHandler:
elif client_metadata.scope is not None and self.options.valid_scopes is not None:
requested_scopes = set(client_metadata.scope.split())
valid_scopes = set(self.options.valid_scopes)
if not requested_scopes.issubset(valid_scopes):
if not requested_scopes.issubset(valid_scopes): # pragma: no branch
return PydanticJSONResponse(
content=RegistrationErrorResponse(
error="invalid_client_metadata",

4
src/mcp/server/auth/handlers/revoke.py
View file

@ -56,7 +56,7 @@ class RevocationHandler:
client = await self.client_authenticator.authenticate(
revocation_request.client_id, revocation_request.client_secret
)
except AuthenticationError as e:
except AuthenticationError as e: # pragma: no cover
return PydanticJSONResponse(
status_code=401,
content=RevocationErrorResponse(
@ -69,7 +69,7 @@ class RevocationHandler:
self.provider.load_access_token,
partial(self.provider.load_refresh_token, client),
]
if revocation_request.token_type_hint == "refresh_token":
if revocation_request.token_type_hint == "refresh_token": # pragma: no cover
loaders = reversed(loaders)
token: None | AccessToken | RefreshToken = None

8
src/mcp/server/auth/handlers/token.py
View file

@ -107,7 +107,7 @@ class TokenHandler:
client_id=token_request.client_id,
client_secret=token_request.client_secret,
)
except AuthenticationError as e:
except AuthenticationError as e: # pragma: no cover
return self.response(
TokenErrorResponse(
error="unauthorized_client",
@ -115,7 +115,7 @@ class TokenHandler:
)
)
if token_request.grant_type not in client_info.grant_types:
if token_request.grant_type not in client_info.grant_types: # pragma: no cover
return self.response(
TokenErrorResponse(
error="unsupported_grant_type",
@ -151,7 +151,7 @@ class TokenHandler:
# see https://datatracker.ietf.org/doc/html/rfc6749#section-10.6
if auth_code.redirect_uri_provided_explicitly:
authorize_request_redirect_uri = auth_code.redirect_uri
else:
else: # pragma: no cover
authorize_request_redirect_uri = None
# Convert both sides to strings for comparison to handle AnyUrl vs string issues
@ -192,7 +192,7 @@ class TokenHandler:
)
)
case RefreshTokenRequest():
case RefreshTokenRequest(): # pragma: no cover
refresh_token = await self.provider.load_refresh_token(client_info, token_request.refresh_token)
if refresh_token is None or refresh_token.client_id != token_request.client_id:
# if token belongs to different client, pretend it doesn't exist

2
src/mcp/server/auth/middleware/bearer_auth.py
View file

@ -99,7 +99,7 @@ class RequireAuthMiddleware:
"""Send an authentication error response with WWW-Authenticate header."""
# Build WWW-Authenticate header value
www_auth_parts = [f'error="{error}"', f'error_description="{description}"']
if self.resource_metadata_url:
if self.resource_metadata_url: # pragma: no cover
www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"')
www_authenticate = f"Bearer {', '.join(www_auth_parts)}"

12
src/mcp/server/auth/middleware/client_auth.py
View file

@ -7,7 +7,7 @@ from mcp.shared.auth import OAuthClientInformationFull
class AuthenticationError(Exception):
def __init__(self, message: str):
self.message = message
self.message = message # pragma: no cover
class ClientAuthenticator:
@ -34,18 +34,18 @@ class ClientAuthenticator:
# Look up client information
client = await self.provider.get_client(client_id)
if not client:
raise AuthenticationError("Invalid client_id")
raise AuthenticationError("Invalid client_id") # pragma: no cover
# If client from the store expects a secret, validate that the request provides
# that secret
if client.client_secret:
if client.client_secret: # pragma: no branch
if not client_secret:
raise AuthenticationError("Client secret is required")
raise AuthenticationError("Client secret is required") # pragma: no cover
if client.client_secret != client_secret:
raise AuthenticationError("Invalid client_secret")
raise AuthenticationError("Invalid client_secret") # pragma: no cover
if client.client_secret_expires_at and client.client_secret_expires_at < int(time.time()):
raise AuthenticationError("Client secret has expired")
raise AuthenticationError("Client secret has expired") # pragma: no cover
return client

7
src/mcp/server/auth/provider.py
View file

@ -117,7 +117,6 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
Returns:
The client information, or None if the client does not exist.
"""
...
async def register_client(self, client_info: OAuthClientInformationFull) -> None:
"""
@ -132,7 +131,6 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
Raises:
RegistrationError: If the client metadata is invalid.
"""
...
async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
"""
@ -221,8 +219,7 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
Returns:
The RefreshToken object if found, or None if not found.
"""
...
...
async def exchange_refresh_token(
self,
@ -258,7 +255,6 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
Returns:
The AuthInfo, or None if the token is invalid.
"""
...
async def revoke_token(
self,
@ -276,7 +272,6 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
Args:
token: the token to revoke
"""
...
def construct_redirect_uri(redirect_uri_base: str, **params: str | None) -> str:

14
src/mcp/server/auth/routes.py
View file

@ -38,13 +38,13 @@ def validate_issuer_url(url: AnyHttpUrl):
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")
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")
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")
raise ValueError("Issuer URL must not have a query string") # pragma: no cover
AUTHORIZATION_PATH = "/authorize"
@ -115,7 +115,7 @@ def create_auth_routes(
),
]
if client_registration_options.enabled:
if client_registration_options.enabled: # pragma: no branch
registration_handler = RegistrationHandler(
provider,
options=client_registration_options,
@ -131,7 +131,7 @@ def create_auth_routes(
)
)
if revocation_options.enabled:
if revocation_options.enabled: # pragma: no branch
revocation_handler = RevocationHandler(provider, client_authenticator)
routes.append(
Route(
@ -176,11 +176,11 @@ def build_metadata(
)
# Add registration endpoint if supported
if client_registration_options.enabled:
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:
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"]