Fix type errors for ty 0.0.1-alpha.31 upgrade

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:

src/fastmcp/server/auth/oauth_proxy.py

@@ -1006,7 +1006,8 @@ async def authorize(
         # Store transaction data for IdP callback processing
         if client.client_id is None:
             raise AuthorizeError(
-                error="invalid_client", error_description="Client ID is required"
+                error="invalid_client",  # type: ignore[arg-type]
+                error_description="Client ID is required",
             )
         transaction = OAuthTransaction(
             txn_id=txn_id,
@@ -1088,7 +1089,8 @@ async def load_authorization_code(
         # Create authorization code object with PKCE challenge
         if client.client_id is None:
             raise AuthorizeError(
-                error="invalid_client", error_description="Client ID is required"
+                error="invalid_client",  # type: ignore[arg-type]
+                error_description="Client ID is required",
             )
         return AuthorizationCode(
             code=authorization_code,
@@ -1512,7 +1514,7 @@ async def exchange_refresh_token(
     # Token Validation
     # -------------------------------------------------------------------------
 
-    async def load_access_token(self, token: str) -> AccessToken | None:
+    async def load_access_token(self, token: str) -> AccessToken | None:  # type: ignore[override]
         """Validate FastMCP JWT by swapping for upstream token.
 
         This implements the token swap pattern:

src/fastmcp/server/auth/providers/in_memory.py

@@ -284,7 +284,7 @@ async def exchange_refresh_token(
             scope=" ".join(scopes),
         )
 
-    async def load_access_token(self, token: str) -> AccessToken | None:
+    async def load_access_token(self, token: str) -> AccessToken | None:  # type: ignore[override]
         token_obj = self.access_tokens.get(token)
         if token_obj:
             if token_obj.expires_at is not None and token_obj.expires_at < time.time():
@@ -295,7 +295,7 @@ async def load_access_token(self, token: str) -> AccessToken | None:
             return token_obj
         return None
 
-    async def verify_token(self, token: str) -> AccessToken | None:
+    async def verify_token(self, token: str) -> AccessToken | None:  # type: ignore[override]
         """
         Verify a bearer token and return access info if valid.
 

src/fastmcp/server/auth/providers/oci.py

@@ -171,7 +171,7 @@ def __init__(
             allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
         """
 
-        overrides = {
+        overrides: dict[str, object] = {
             k: v
             for k, v in {
                 "config_url": config_url,
@@ -187,7 +187,7 @@ def __init__(
             }.items()
             if v is not NotSet
         }
-        settings = OCIProviderSettings(**overrides)
+        settings = OCIProviderSettings(**overrides)  # type: ignore[arg-type]
 
         if not settings.config_url:
             raise ValueError(

src/fastmcp/server/context.py

@@ -388,7 +388,7 @@ def store_data(data: dict, ctx: Context) -> str:
             session_id = str(uuid4())
 
         # Save the session id to the session attributes
-        session._fastmcp_id = session_id
+        session._fastmcp_id = session_id  # type: ignore[attr-defined]
         return session_id
 
     @property
@@ -540,7 +540,7 @@ async def sample(
                     maxTokens=max_tokens,
                     modelPreferences=_parse_model_preferences(model_preferences),
                 ),
-                self.request_context,
+                self.request_context,  # type: ignore[arg-type]
             )
 
             if inspect.isawaitable(create_message_result):

src/fastmcp/server/elicitation.py

@@ -37,13 +37,13 @@ class ElicitationJsonSchema(GenerateJsonSchema):
     Optionally adds enumNames for better UI display when available.
     """
 
-    def generate_inner(self, schema: core_schema.CoreSchema) -> JsonSchemaValue:
+    def generate_inner(self, schema: core_schema.CoreSchema) -> JsonSchemaValue:  # type: ignore[override]
         """Override to prevent ref generation for enums."""
         # For enum schemas, bypass the ref mechanism entirely
         if schema["type"] == "enum":
             # Directly call our custom enum_schema without going through handler
             # This prevents the ref/defs mechanism from being invoked
-            return self.enum_schema(schema)
+            return self.enum_schema(schema)  # type: ignore[arg-type]
         # For all other types, use the default implementation
         return super().generate_inner(schema)
 

src/fastmcp/server/http.py

@@ -117,7 +117,8 @@ def create_base_app(
         A Starlette application
     """
     # Always add RequestContextMiddleware as the outermost middleware
-    middleware.insert(0, Middleware(RequestContextMiddleware))
+    # TODO(ty): remove type ignore when ty supports Starlette Middleware typing
+    middleware.insert(0, Middleware(RequestContextMiddleware))  # type: ignore[arg-type]
 
     return StarletteWithLifespan(
         routes=routes,

src/fastmcp/server/proxy.py

@@ -14,6 +14,7 @@
 from mcp.types import (
     METHOD_NOT_FOUND,
     BlobResourceContents,
+    ElicitRequestFormParams,
     GetPromptResult,
     TextResourceContents,
 )
@@ -363,7 +364,7 @@ def __init__(self, client: Client, **kwargs: Any):
         self._client = client
 
     @classmethod
-    def from_mcp_template(
+    def from_mcp_template(  # type: ignore[override]
         cls, client: Client, mcp_template: mcp.types.ResourceTemplate
     ) -> ProxyTemplate:
         """Factory method to create a ProxyTemplate from a raw MCP template schema."""
@@ -454,7 +455,7 @@ def from_mcp_prompt(
             _mirrored=True,
         )
 
-    async def render(self, arguments: dict[str, Any]) -> list[PromptMessage]:
+    async def render(self, arguments: dict[str, Any]) -> list[PromptMessage]:  # type: ignore[override]
         """Render the prompt by making a call through the client."""
         async with self._client:
             result = await self._client.get_prompt(self.name, arguments)
@@ -567,7 +568,8 @@ async def default_sampling_handler(
         return mcp.types.CreateMessageResult(
             role="assistant",
             model="fastmcp-client",
-            content=content,
+            # TODO(ty): remove when ty supports isinstance exclusion narrowing
+            content=content,  # type: ignore[arg-type]
         )
 
     @classmethod
@@ -582,9 +584,15 @@ async def default_elicitation_handler(
         A handler that forwards the elicitation request from the remote server to the proxy's connected clients and relays the response back to the remote server.
         """
         ctx = get_context()
+        # requestedSchema only exists on ElicitRequestFormParams, not ElicitRequestURLParams
+        requested_schema = (
+            params.requestedSchema
+            if isinstance(params, ElicitRequestFormParams)
+            else {"type": "object", "properties": {}}
+        )
         result = await ctx.session.elicit(
             message=message,
-            requestedSchema=params.requestedSchema,
+            requestedSchema=requested_schema,
             related_request_id=ctx.request_id,
         )
         return ElicitResult(action=result.action, content=result.content)
@@ -628,7 +636,7 @@ def __init__(self, *args: Any, **kwargs: Any):
         super().__init__(*args, **kwargs)
         self._caches: dict[ServerSession, Client[ClientTransportT]] = {}
 
-    async def __aexit__(self, exc_type, exc_value, traceback) -> None:
+    async def __aexit__(self, exc_type, exc_value, traceback) -> None:  # type: ignore[override]
         """
         The stateful proxy client will be forced disconnected when the session is exited.
         So we do nothing here.

src/fastmcp/server/server.py

@@ -407,7 +407,8 @@ async def _docket_lifespan(self) -> AsyncIterator[None]:
 
                 for prompt in self._prompt_manager._prompts.values():
                     if isinstance(prompt, FunctionPrompt) and prompt.task:
-                        docket.register(prompt.fn)
+                        # task=True requires async fn (validated at creation time)
+                        docket.register(cast(Callable[..., Awaitable[Any]], prompt.fn))
 
                 for resource in self._resource_manager._resources.values():
                     if isinstance(resource, FunctionResource) and resource.task:

src/fastmcp/utilities/components.py

@@ -96,7 +96,7 @@ def get_meta(
 
         return meta or None
 
-    def model_copy(
+    def model_copy(  # type: ignore[override]
         self,
         *,
         update: dict[str, Any] | None = None,
@@ -137,7 +137,7 @@ def disable(self) -> None:
         """Disable the component."""
         self.enabled = False
 
-    def copy(self) -> Self:
+    def copy(self) -> Self:  # type: ignore[override]
         """Create a copy of the component."""
         return self.model_copy()
 
@@ -173,7 +173,7 @@ def disable(self) -> None:
             )
         super().disable()
 
-    def copy(self) -> Self:
+    def copy(self) -> Self:  # type: ignore[override]
         """Create a copy of the component that can be modified."""
         # Create a copy and mark it as not mirrored
         copied = self.model_copy()

src/fastmcp/utilities/json_schema_type.py

@@ -248,7 +248,7 @@ def _create_numeric_type(
         if v is not None
     }
 
-    return Annotated[base, Field(**constraints)] if constraints else base
+    return Annotated[base, Field(**constraints)] if constraints else base  # type: ignore[return-value]
 
 
 def _create_enum(name: str, values: list[Any]) -> type:
@@ -265,8 +265,8 @@ def _create_array_type(
     if isinstance(items, list):
         # Handle positional item schemas
         item_types = [_schema_to_type(s, schemas) for s in items]
-        combined = Union[tuple(item_types)]  # type: ignore # noqa: UP007
-        base = list[combined]
+        combined = Union[tuple(item_types)]  # type: ignore[arg-type] # noqa: UP007
+        base = list[combined]  # type: ignore[valid-type]
     else:
         # Handle single item schema
         item_type = _schema_to_type(items, schemas)
@@ -282,7 +282,7 @@ def _create_array_type(
         if v is not None
     }
 
-    return Annotated[base, Field(**constraints)] if constraints else base
+    return Annotated[base, Field(**constraints)] if constraints else base  # type: ignore[return-value]
 
 
 def _return_Any() -> Any: