fix(session): satisfy strict CI for method classification

Author: Ranga-Prasath-22

src/mcp/shared/session.py

@@ -3,7 +3,7 @@
 from contextlib import AsyncExitStack
 from datetime import timedelta
 from types import TracebackType
-from typing import Any, Generic, Protocol, TypeVar, get_args
+from typing import Any, Generic, Protocol, TypeVar, cast, get_args
 
 import anyio
 import httpx
@@ -167,15 +167,15 @@ def _extract_known_request_methods(request_type: type[Any]) -> frozenset[str]:
         if hasattr(union_type, "model_fields") and "root" in union_type.model_fields:
             union_type = union_type.model_fields["root"].annotation
 
-        methods = set()
+        methods: set[str] = set()
 
         def _unpack_union(t: Any) -> None:
             args = get_args(t)
             if args:
                 for arg in args:
                     _unpack_union(arg)
             elif hasattr(t, "model_fields") and "method" in t.model_fields:
-                m = t.model_fields["method"].default
+                m = cast(Any, t.model_fields["method"].default)
                 if isinstance(m, str):
                     methods.add(m)
 
@@ -376,9 +376,7 @@ async def _send_response(self, request_id: RequestId, response: SendResultT | Er
             await self._write_stream.send(session_message)
 
     def _get_request_validation_error(self, request: JSONRPCRequest) -> ErrorData:
-        if self._known_request_methods and (
-            not isinstance(request.method, str) or request.method not in self._known_request_methods
-        ):
+        if self._known_request_methods and request.method not in self._known_request_methods:
             return ErrorData(code=METHOD_NOT_FOUND, message="Method not found")
         return ErrorData(code=INVALID_PARAMS, message="Invalid request parameters")
 

tests/shared/test_session.py

@@ -1,4 +1,5 @@
 from collections.abc import AsyncGenerator
+from types import SimpleNamespace
 from typing import Any
 
 import anyio
@@ -10,6 +11,7 @@
 from mcp.shared.exceptions import McpError
 from mcp.shared.memory import create_client_server_memory_streams, create_connected_server_and_client_session
 from mcp.shared.message import SessionMessage
+from mcp.shared.session import _extract_known_request_methods
 from mcp.types import (
     CancelledNotification,
     CancelledNotificationParams,
@@ -363,6 +365,7 @@ async def mock_server():
                 )
             )
             resp = await server_read.receive()
+            assert isinstance(resp, SessionMessage)
             assert isinstance(resp.message.root, JSONRPCError)
             assert resp.message.root.id == 1
             assert resp.message.root.error.code == types.METHOD_NOT_FOUND
@@ -373,3 +376,22 @@ async def mock_server():
             anyio.create_task_group() as tg,
         ):
             tg.start_soon(mock_server)
+
+
+def test_extract_known_request_methods_ignores_non_string_defaults():
+    class RequestWithNonStringMethod:
+        model_fields = {"method": SimpleNamespace(default=None)}
+
+    assert _extract_known_request_methods(RequestWithNonStringMethod) == frozenset()
+
+
+def test_extract_known_request_methods_fails_closed_on_schema_introspection_error():
+    class ExplodingMeta(type):
+        @property
+        def __value__(cls) -> type[Any]:
+            raise RuntimeError("broken schema")
+
+    class BrokenRequest(metaclass=ExplodingMeta):
+        pass
+
+    assert _extract_known_request_methods(BrokenRequest) == frozenset()