test: cover invalid client response parsing

Author: daleselaji-dev

src/mcp/client/auth/oauth2.py

@@ -12,7 +12,7 @@
 import time
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from dataclasses import dataclass, field
-from typing import Any, Protocol, get_args
+from typing import Any, Protocol, cast, get_args
 from urllib.parse import quote, urlencode, urljoin, urlparse
 
 import anyio
@@ -122,7 +122,7 @@ def _is_invalid_client_response(body: bytes) -> bool:
         payload: Any = json.loads(body)
     except (json.JSONDecodeError, UnicodeDecodeError):
         return False
-    return isinstance(payload, dict) and payload.get("error") == "invalid_client"
+    return isinstance(payload, dict) and cast(dict[str, Any], payload).get("error") == "invalid_client"
 
 
 class PKCEParameters(BaseModel):

tests/client/test_auth.py

@@ -13,6 +13,7 @@
 
 from mcp.client.auth import OAuthClientProvider, PKCEParameters
 from mcp.client.auth.exceptions import OAuthFlowError, OAuthRegistrationError, OAuthTokenError
+from mcp.client.auth.oauth2 import _is_invalid_client_response
 from mcp.client.auth.utils import (
     build_oauth_authorization_server_metadata_discovery_urls,
     build_protected_resource_metadata_discovery_urls,
@@ -78,6 +79,20 @@ def client_metadata():
     )
 
 
+@pytest.mark.parametrize(
+    ("body", "expected"),
+    [
+        (b"not json", False),
+        (b"\xff", False),
+        (b"[]", False),
+        (b'{"error": "invalid_grant"}', False),
+        (b'{"error": "invalid_client"}', True),
+    ],
+)
+def test_invalid_client_response_detection(body: bytes, expected: bool) -> None:
+    assert _is_invalid_client_response(body) is expected
+
+
 @pytest.fixture
 def valid_tokens():
     return OAuthToken(