Fix type errors for ty 0.0.1-alpha.31 upgrade (#2561)

* Fix type errors for ty 0.0.1-alpha.31 upgrade

Add type ignores and fixes for ty's stricter checking:
- Path(None) guards in cli.py
- isinstance checks for ElicitRequestFormParams (URL elicitation support)
- TODO(ty) comments for match/isinstance narrowing bugs
- Method override type ignores for generic covariance
- Starlette Middleware typing workarounds
- Dynamic type construction ignores in json_schema_type.py

* Fix remaining type errors for ty 0.0.1-alpha.31

- Add asserts for optional attribute access in tests
- Add type ignores for dynamic httpx transport internals
- Add TODO(ty) comments for `in` operator on str|bytes
- Add TODO(ty) comments for Starlette Middleware typing
- Use cast for prompt.fn async validation in server.py

* Upgrade ty to 0.0.1-alpha.31

Fixes additional test file type errors discovered after upgrade.

Author: jlowin

pyproject.toml

@@ -72,7 +72,7 @@ dev = [
     "pytest-timeout>=2.4.0",
     "pytest-xdist>=3.6.1",
     "ruff>=0.12.8",
-    "ty==0.0.1a25",
+    "ty==0.0.1a31",
     "prek>=0.2.12",
 ]
 

src/fastmcp/cli/cli.py

@@ -103,7 +103,7 @@ def version(
         "MCP version": importlib.metadata.version("mcp"),
         "Python version": platform.python_version(),
         "Platform": platform.platform(),
-        "FastMCP root path": Path(fastmcp.__file__).resolve().parents[1],
+        "FastMCP root path": Path(fastmcp.__file__ or ".").resolve().parents[1],
     }
 
     g = Table.grid(padding=(0, 1))
@@ -819,11 +819,13 @@ async def prepare(
             )
             sys.exit(1)
 
+    assert config_path is not None
     config_file = Path(config_path)
     if not config_file.exists():
         logger.error(f"Configuration file not found: {config_path}")
         sys.exit(1)
 
+    assert output_dir is not None
     output_path = Path(output_dir)
 
     try:

src/fastmcp/client/elicitation.py

@@ -7,7 +7,7 @@
 from mcp import ClientSession
 from mcp.client.session import ElicitationFnT
 from mcp.shared.context import LifespanContextT, RequestContext
-from mcp.types import ElicitRequestParams
+from mcp.types import ElicitRequestFormParams, ElicitRequestParams
 from mcp.types import ElicitResult as MCPElicitResult
 from pydantic_core import to_jsonable_python
 from typing_extensions import TypeVar
@@ -26,7 +26,8 @@ class ElicitResult(MCPElicitResult, Generic[T]):
 ElicitationHandler: TypeAlias = Callable[
     [
         str,  # message
-        type[T],  # a class for creating a structured response
+        type[T]
+        | None,  # a class for creating a structured response (None for URL elicitation)
         ElicitRequestParams,
         RequestContext[ClientSession, LifespanContextT],
     ],
@@ -42,10 +43,15 @@ async def _elicitation_handler(
         params: ElicitRequestParams,
     ) -> MCPElicitResult | mcp.types.ErrorData:
         try:
-            if params.requestedSchema == {"type": "object", "properties": {}}:
-                response_type = None
+            # requestedSchema only exists on ElicitRequestFormParams, not ElicitRequestURLParams
+            if isinstance(params, ElicitRequestFormParams):
+                if params.requestedSchema == {"type": "object", "properties": {}}:
+                    response_type = None
+                else:
+                    response_type = json_schema_to_type(params.requestedSchema)
             else:
-                response_type = json_schema_to_type(params.requestedSchema)
+                # URL-based elicitation doesn't have a schema
+                response_type = None
 
             result = await elicitation_handler(
                 params.message, response_type, params, context

src/fastmcp/client/messages.py

@@ -35,16 +35,18 @@ async def dispatch(self, message: Message) -> None:
             # requests
             case RequestResponder():
                 # handle all requests
-                await self.on_request(message)
+                # TODO(ty): remove when ty supports match statement narrowing
+                await self.on_request(message)  # type: ignore[arg-type]
 
                 # handle specific requests
-                match message.request.root:
+                # TODO(ty): remove type ignores when ty supports match statement narrowing
+                match message.request.root:  # type: ignore[union-attr]
                     case mcp.types.PingRequest():
-                        await self.on_ping(message.request.root)
+                        await self.on_ping(message.request.root)  # type: ignore[union-attr]
                     case mcp.types.ListRootsRequest():
-                        await self.on_list_roots(message.request.root)
+                        await self.on_list_roots(message.request.root)  # type: ignore[union-attr]
                     case mcp.types.CreateMessageRequest():
-                        await self.on_create_message(message.request.root)
+                        await self.on_create_message(message.request.root)  # type: ignore[union-attr]
 
             # notifications
             case mcp.types.ServerNotification():

src/fastmcp/client/roots.py

@@ -34,7 +34,8 @@ def create_roots_callback(
     handler: RootsList | RootsHandler,
 ) -> ListRootsFnT:
     if isinstance(handler, list):
-        return _create_roots_callback_from_roots(handler)
+        # TODO(ty): remove when ty supports isinstance union narrowing
+        return _create_roots_callback_from_roots(handler)  # type: ignore[arg-type]
     elif inspect.isfunction(handler):
         return _create_roots_callback_from_fn(handler)
     else:

src/fastmcp/client/tasks.py

@@ -201,7 +201,7 @@ def _handle_status_notification(self, status: GetTaskResult) -> None:
                 result = callback(status)
                 if inspect.isawaitable(result):
                     # Fire and forget async callbacks
-                    asyncio.create_task(result)  # noqa: RUF006
+                    asyncio.create_task(result)  # type: ignore[arg-type] # noqa: RUF006
             except Exception as e:
                 logger.warning(f"Task callback error: {e}", exc_info=True)
 

src/fastmcp/client/transports.py

@@ -375,7 +375,8 @@ async def connect(
                 env=self.env,
                 cwd=self.cwd,
                 log_file=self.log_file,
-                session_kwargs=session_kwargs,
+                # TODO(ty): remove when ty supports Unpack[TypedDict] inference
+                session_kwargs=session_kwargs,  # type: ignore[arg-type]
                 ready_event=self._ready_event,
                 stop_event=self._stop_event,
                 session_future=session_future,

src/fastmcp/experimental/sampling/handlers/openai.py

@@ -164,7 +164,7 @@ def _select_model_from_preferences(
     ) -> ChatModel:
         for model_option in self._iter_models_from_preferences(model_preferences):
             if model_option in get_args(ChatModel):
-                chosen_model: ChatModel = model_option  # pyright: ignore[reportAssignmentType]
+                chosen_model: ChatModel = model_option  # type: ignore[assignment]
                 return chosen_model
 
         return self.default_model

src/fastmcp/prompts/prompt.py

@@ -206,7 +206,7 @@ def from_function(
             fn = fn.__call__
         # if the fn is a staticmethod, we need to work with the underlying function
         if isinstance(fn, staticmethod):
-            fn = fn.__func__
+            fn = fn.__func__  # type: ignore[assignment]
 
         # Validate that task=True requires async functions (after unwrapping)
         if task and not inspect.iscoroutinefunction(fn):

src/fastmcp/server/auth/auth.py

@@ -189,12 +189,13 @@ def get_middleware(self) -> list:
         Returns:
             List of Starlette Middleware instances to apply to the HTTP app
         """
+        # TODO(ty): remove type ignores when ty supports Starlette Middleware typing
         return [
             Middleware(
-                AuthenticationMiddleware,
+                AuthenticationMiddleware,  # type: ignore[arg-type]
                 backend=BearerAuthBackend(self),
             ),
-            Middleware(AuthContextMiddleware),
+            Middleware(AuthContextMiddleware),  # type: ignore[arg-type]
         ]
 
     def _get_resource_url(self, path: str | None = None) -> AnyHttpUrl | None: