ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-23 14:04:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add loq file size limits and clean up type ignores (#2859)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-01-13 07:29:12 -05:00 • committed by GitHub
commit 2b6a0faf1c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
49 changed files with 815 additions and 973 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.pre-commit-config.yaml
View file

@ -35,9 +35,10 @@ repos:
- id: loq
name: loq (file size limits)
entry: uv run loq
entry: bash -c 'uv run loq || printf "\nloq violations not enforced... yet!\n"'
language: system
pass_filenames: false
verbose: true
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v6.0.0

292
loq.toml
View file

@ -1,158 +1,42 @@
# loq configuration - file size enforcement
# Run `loq baseline` to update when files exceed limits
default_max_lines = 500
default_max_lines = 1000
respect_gitignore = true
exclude = [
"**/uv.lock",
".git/",
"docs/",
]
exclude = ["**/uv.lock", ".git/**", "docs/**"]
[[rules]]
path = "src/fastmcp/server/auth/oauth_proxy.py"
max_lines = 2282
path = "tests/**"
max_lines = 1000
[[rules]]
path = "tests/server/providers/test_local_provider_tools.py"
max_lines = 1554
[[rules]]
path = "tests/client/test_client.py"
max_lines = 1438
[[rules]]
path = "src/fastmcp/client/tasks.py"
max_lines = 551
[[rules]]
path = "tests/server/test_auth_integration.py"
max_lines = 1242
[[rules]]
path = "tests/server/test_dependencies.py"
max_lines = 1046
[[rules]]
path = "tests/prompts/test_prompt.py"
max_lines = 552
[[rules]]
path = "tests/client/test_elicitation.py"
max_lines = 1132
[[rules]]
path = "src/fastmcp/tools/tool_transform.py"
max_lines = 954
[[rules]]
path = "tests/server/middleware/test_tool_injection.py"
max_lines = 509
path = "tests/server/auth/test_oauth_proxy.py"
max_lines = 1899
[[rules]]
path = "tests/server/middleware/test_middleware.py"
max_lines = 1066
[[rules]]
path = "tests/server/auth/providers/test_azure.py"
max_lines = 809
[[rules]]
path = "docs/servers/auth/oauth-proxy.mdx"
max_lines = 590
[[rules]]
path = "src/fastmcp/server/context.py"
max_lines = 1272
[[rules]]
path = "tests/test_mcp_config.py"
max_lines = 930
[[rules]]
path = "docs/deployment/http.mdx"
max_lines = 736
[[rules]]
path = "src/fastmcp/server/auth/providers/jwt.py"
max_lines = 527
[[rules]]
path = "src/fastmcp/server/dependencies.py"
max_lines = 908
[[rules]]
path = "docs/changelog.mdx"
max_lines = 2280
[[rules]]
path = "tests/server/tasks/test_task_mount.py"
max_lines = 965
[[rules]]
path = "src/fastmcp/utilities/json_schema_type.py"
max_lines = 648
[[rules]]
path = "tests/utilities/openapi/test_schemas.py"
max_lines = 641
[[rules]]
path = "src/fastmcp/utilities/openapi/schemas.py"
max_lines = 593
[[rules]]
path = "src/fastmcp/server/middleware/caching.py"
max_lines = 523
[[rules]]
path = "docs/servers/middleware.mdx"
max_lines = 846
[[rules]]
path = "tests/contrib/test_component_manager.py"
max_lines = 767
[[rules]]
path = "src/fastmcp/server/auth/auth.py"
max_lines = 558
[[rules]]
path = "tests/server/providers/test_local_provider_resources.py"
max_lines = 974
[[rules]]
path = "docs/deployment/server-configuration.mdx"
max_lines = 640
[[rules]]
path = "tests/cli/test_run.py"
max_lines = 628
[[rules]]
path = "tests/utilities/openapi/test_transitive_references.py"
max_lines = 842
[[rules]]
path = "src/fastmcp/resources/template.py"
max_lines = 576
[[rules]]
path = "tests/server/providers/openapi/test_comprehensive.py"
max_lines = 741
[[rules]]
path = "tests/client/test_sampling.py"
max_lines = 1002
[[rules]]
path = "tests/resources/test_resource_template.py"
max_lines = 1009
[[rules]]
path = "tests/server/providers/test_local_provider.py"
max_lines = 784
[[rules]]
path = "tests/server/middleware/test_caching.py"
max_lines = 617
path = "tests/tools/test_tool_transform.py"
max_lines = 1748
[[rules]]
path = "tests/server/test_mount.py"
@ -163,153 +47,57 @@ path = "tests/utilities/test_inspect.py"
max_lines = 1111
[[rules]]
path = "src/fastmcp/utilities/openapi/parser.py"
max_lines = 821
[[rules]]
path = "docs/python-sdk/fastmcp-server-context.mdx"
max_lines = 550
[[rules]]
path = "src/fastmcp/server/providers/local_provider.py"
max_lines = 962
[[rules]]
path = "tests/client/test_notifications.py"
max_lines = 556
[[rules]]
path = "tests/server/middleware/test_logging.py"
max_lines = 670
[[rules]]
path = "tests/utilities/test_json_schema.py"
max_lines = 543
[[rules]]
path = "tests/server/auth/test_jwt_provider.py"
max_lines = 1101
[[rules]]
path = "tests/server/auth/test_oauth_proxy.py"
max_lines = 1899
[[rules]]
path = "tests/utilities/test_types.py"
max_lines = 673
[[rules]]
path = "docs/servers/resources.mdx"
max_lines = 731
[[rules]]
path = "src/fastmcp/client/client.py"
max_lines = 1722
[[rules]]
path = "src/fastmcp/server/providers/proxy.py"
max_lines = 823
[[rules]]
path = "docs/python-sdk/fastmcp-server-server.mdx"
max_lines = 999
path = "tests/resources/test_resource_template.py"
max_lines = 1009
[[rules]]
path = "tests/server/auth/test_oauth_consent_flow.py"
max_lines = 1056
[[rules]]
path = "README.md"
max_lines = 515
[[rules]]
path = "tests/server/providers/proxy/test_proxy_server.py"
max_lines = 729
[[rules]]
path = "tests/tools/test_tool_transform.py"
max_lines = 1748
[[rules]]
path = "src/fastmcp/utilities/ui.py"
max_lines = 626
[[rules]]
path = "tests/server/auth/test_oidc_proxy.py"
max_lines = 878
[[rules]]
path = "docs/patterns/cli.mdx"
max_lines = 581
[[rules]]
path = "tests/cli/test_cli.py"
max_lines = 618
[[rules]]
path = "tests/server/tasks/test_task_return_types.py"
max_lines = 662
[[rules]]
path = "tests/server/providers/test_local_provider_tools.py"
max_lines = 1554
[[rules]]
path = "src/fastmcp/server/providers/fastmcp_provider.py"
max_lines = 586
[[rules]]
path = "tests/server/middleware/test_error_handling.py"
max_lines = 652
path = "src/fastmcp/server/server.py"
max_lines = 2942
[[rules]]
path = "tests/tools/test_tool.py"
max_lines = 1923
[[rules]]
path = "docs/python-sdk/fastmcp-client-client.mdx"
max_lines = 704
path = "tests/client/test_elicitation.py"
max_lines = 1132
[[rules]]
path = "docs/servers/tools.mdx"
max_lines = 1040
[[rules]]
path = "src/fastmcp/server/server.py"
max_lines = 2942
[[rules]]
path = "tests/deprecated/test_import_server.py"
max_lines = 714
[[rules]]
path = "src/fastmcp/client/transports.py"
max_lines = 1208
[[rules]]
path = "docs/docs.json"
max_lines = 589
[[rules]]
path = "src/fastmcp/cli/cli.py"
max_lines = 957
path = "src/fastmcp/client/client.py"
max_lines = 1722
[[rules]]
path = "tests/utilities/test_json_schema_type.py"
max_lines = 1584
[[rules]]
path = "docs/servers/context.mdx"
max_lines = 650
path = "src/fastmcp/server/auth/oauth_proxy.py"
max_lines = 2282
[[rules]]
path = "tests/server/auth/test_authorization.py"
max_lines = 604
path = "src/fastmcp/client/transports.py"
max_lines = 1208
[[rules]]
path = "docs/development/v3-notes/v3-features.mdx"
max_lines = 536
path = "tests/server/test_dependencies.py"
max_lines = 1046
[[rules]]
path = "src/fastmcp/tools/function_tool.py"
max_lines = 594
path = "tests/client/test_sampling.py"
max_lines = 1002
[[rules]]
path = "tests/server/auth/test_jwt_provider.py"
max_lines = 1101
[[rules]]
path = "docs/servers/tools.mdx"
max_lines = 1040
[[rules]]
path = "docs/changelog.mdx"
max_lines = 2280

14
src/fastmcp/client/client.py
View file

@ -296,7 +296,7 @@ class Client(Generic[ClientTransportT]):
"list_roots_callback": None,
"logging_callback": create_log_callback(log_handler),
"message_handler": message_handler or TaskNotificationHandler(self),
"read_timeout_seconds": timeout, # ty: ignore[invalid-argument-type]
"read_timeout_seconds": timeout,
"client_info": client_info,
}
@ -999,7 +999,7 @@ class Client(Generic[ClientTransportT]):
wrapped_result = await self._await_with_session_monitoring(
self.session.send_request(
request=request, # type: ignore[arg-type]
result_type=TaskResponseUnion, # type: ignore[arg-type]
result_type=TaskResponseUnion,
)
)
raw_result = wrapped_result.root
@ -1228,7 +1228,7 @@ class Client(Generic[ClientTransportT]):
wrapped_result = await self._await_with_session_monitoring(
self.session.send_request(
request=request, # type: ignore[arg-type]
result_type=TaskResponseUnion, # type: ignore[arg-type]
result_type=TaskResponseUnion,
)
)
raw_result = wrapped_result.root
@ -1383,7 +1383,7 @@ class Client(Generic[ClientTransportT]):
self.session.call_tool(
name=name,
arguments=arguments,
read_timeout_seconds=timeout, # ty: ignore[invalid-argument-type]
read_timeout_seconds=timeout,
progress_callback=progress_handler or self._progress_handler,
meta=meta,
)
@ -1564,7 +1564,7 @@ class Client(Generic[ClientTransportT]):
wrapped_result = await self._await_with_session_monitoring(
self.session.send_request(
request=request, # type: ignore[arg-type]
result_type=TaskResponseUnion, # type: ignore[arg-type]
result_type=TaskResponseUnion,
)
)
raw_result = wrapped_result.root
@ -1606,7 +1606,7 @@ class Client(Generic[ClientTransportT]):
return await self._await_with_session_monitoring(
self.session.send_request(
request=request, # type: ignore[arg-type]
result_type=GetTaskResult, # type: ignore[arg-type]
result_type=GetTaskResult,
)
)
@ -1633,7 +1633,7 @@ class Client(Generic[ClientTransportT]):
result = await self._await_with_session_monitoring(
self.session.send_request(
request=request, # type: ignore[arg-type]
result_type=GetTaskPayloadResult, # type: ignore[arg-type]
result_type=GetTaskPayloadResult,
)
)
# Return as dict for compatibility with Task class parsing

4
src/fastmcp/client/sampling/handlers/anthropic.py
View file

@ -198,7 +198,7 @@ class AnthropicSamplingHandler:
anthropic_messages.append(
MessageParam(
role=message.role,
content=content_blocks, # type: ignore[arg-type]
content=content_blocks,
)
)
continue
@ -310,7 +310,7 @@ class AnthropicSamplingHandler:
ToolParam(
name=tool.name,
description=tool.description or "",
input_schema=input_schema, # type: ignore[arg-type]
input_schema=input_schema,
)
)
return anthropic_tools

2
src/fastmcp/client/sampling/handlers/openai.py
View file

@ -368,7 +368,7 @@ class OpenAISamplingHandler:
# Skip non-function tool calls
if not hasattr(tool_call, "function"):
continue
func = tool_call.function # type: ignore[union-attr]
func = tool_call.function
# Parse the arguments JSON string
try:
arguments = json.loads(func.arguments) # type: ignore[union-attr]

4
src/fastmcp/client/tasks.py
View file

@ -378,7 +378,7 @@ class ToolTask(Task["CallToolResult"]):
):
mcp_result = mcp.types.CallToolResult(
content=raw_result.content,
structuredContent=raw_result.structured_content, # type: ignore[arg-type]
structuredContent=raw_result.structured_content,
_meta=raw_result.meta, # type: ignore[call-arg] # _meta is Pydantic alias for meta field
)
result = await self._client._parse_call_tool_result(
@ -386,7 +386,7 @@ class ToolTask(Task["CallToolResult"]):
)
else:
# Unknown type - just return it
result = raw_result # type: ignore[assignment]
result = raw_result
# Cache before returning
self._cached_result = result

2
src/fastmcp/client/transports.py
View file

@ -113,7 +113,7 @@ class ClientTransport(abc.ABC):
A mcp.ClientSession instance.
"""
raise NotImplementedError
yield # type: ignore
yield
def __repr__(self) -> str:
# Basic representation for subclasses

6
src/fastmcp/prompts/function_prompt.py
View file

@ -159,7 +159,7 @@ class FunctionPrompt(Prompt):
fn = fn.__call__
# if the fn is a staticmethod, we need to work with the underlying function
if isinstance(fn, staticmethod):
fn = fn.__func__ # type: ignore[assignment]
fn = fn.__func__
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
@ -310,7 +310,7 @@ class FunctionPrompt(Prompt):
"""
if not self.task_config.supports_tasks():
return
docket.register(self.fn, names=[self.key]) # type: ignore[arg-type]
docket.register(self.fn, names=[self.key])
async def add_to_docket( # type: ignore[override]
self,
@ -418,7 +418,7 @@ def prompt(
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
target.__fastmcp__ = metadata
return fn
def decorator(fn: F, prompt_name: str | None) -> F:

2
src/fastmcp/resources/function_resource.py
View file

@ -294,7 +294,7 @@ def resource(
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
target.__fastmcp__ = metadata
return fn
def decorator(fn: F) -> F:

2
src/fastmcp/resources/template.py
View file

@ -93,7 +93,7 @@ def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None:
for name in query_param_names:
if name in parsed_query:
# Take first value if multiple provided
params[name] = parsed_query[name][0] # type: ignore[index]
params[name] = parsed_query[name][0]
return params

4
src/fastmcp/server/auth/oauth_proxy.py
View file

@ -1464,7 +1464,7 @@ class OAuthProxy(OAuthProvider):
try:
logger.debug("Refreshing upstream token (jti=%s)", refresh_jti[:8])
token_response: dict[str, Any] = await oauth_client.refresh_token( # type: ignore[assignment]
token_response: dict[str, Any] = await oauth_client.refresh_token(
url=self._upstream_token_endpoint,
refresh_token=upstream_token_set.refresh_token,
scope=" ".join(upstream_scopes) if upstream_scopes else None,
@ -1890,7 +1890,7 @@ class OAuthProxy(OAuthProvider):
idp_tokens: dict[str, Any] = await oauth_client.fetch_token(
**token_params
) # type: ignore[assignment]
)
logger.debug(
f"Successfully exchanged IdP code for tokens (transaction: {txn_id}, PKCE: {bool(proxy_code_verifier)})"

2
src/fastmcp/server/auth/providers/jwt.py
View file

@ -275,7 +275,7 @@ class JWTVerifier(TokenVerifier):
for key_data in jwks_data.get("keys", []):
key_kid = key_data.get("kid")
jwk = JsonWebKey.import_key(key_data)
public_key = jwk.get_public_key() # type: ignore
public_key = jwk.get_public_key()
if key_kid:
self._jwks_cache[key_kid] = public_key

4
src/fastmcp/server/context.py
View file

@ -75,7 +75,7 @@ ResultT = TypeVar("ResultT", default=str)
# Simplified tool choice type - just the mode string instead of the full MCP object
ToolChoiceOption = Literal["auto", "required", "none"]
_current_context: ContextVar[Context | None] = ContextVar("context", default=None) # type: ignore[assignment]
_current_context: ContextVar[Context | None] = ContextVar("context", default=None)
TransportType = Literal["stdio", "sse", "streamable-http"]
_current_transport: ContextVar[TransportType | None] = ContextVar(
@ -961,7 +961,7 @@ class Context:
],
isError=True,
)
], # type: ignore[arg-type]
],
)
)

10
src/fastmcp/server/dependencies.py
View file

@ -230,7 +230,7 @@ def transform_context_annotations(fn: Callable[..., Any]) -> Callable[..., Any]:
# inspect.signature(bound_method) automatically removes the first param
if inspect.ismethod(fn):
# Get the original __func__ signature which includes 'self'
func_sig = inspect.signature(fn.__func__) # type: ignore[union-attr]
func_sig = inspect.signature(fn.__func__)
# Insert 'self' at the beginning of our new params
self_param = next(iter(func_sig.parameters.values())) # Should be 'self'
new_sig = func_sig.replace(parameters=[self_param, *new_params])
@ -261,8 +261,8 @@ def _clear_signature_caches(fn: Callable[..., Any]) -> None:
# Also clear for __func__ if it's a method
if inspect.ismethod(fn):
_signature_cache.pop(fn.__func__, None) # type: ignore[union-attr]
_parameter_cache.pop(fn.__func__, None) # type: ignore[union-attr]
_signature_cache.pop(fn.__func__, None)
_parameter_cache.pop(fn.__func__, None)
# Try to clear docket caches if docket is installed
if is_docket_available():
@ -273,8 +273,8 @@ def _clear_signature_caches(fn: Callable[..., Any]) -> None:
docket_sig_cache.pop(fn, None)
docket_param_cache.pop(fn, None)
if inspect.ismethod(fn):
docket_sig_cache.pop(fn.__func__, None) # type: ignore[union-attr]
docket_param_cache.pop(fn.__func__, None) # type: ignore[union-attr]
docket_sig_cache.pop(fn.__func__, None)
docket_param_cache.pop(fn.__func__, None)
except (ImportError, AttributeError):
pass # Cache access not available in this docket version

26
src/fastmcp/server/elicitation.py
View file

@ -2,7 +2,7 @@ from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import Any, Generic, Literal, get_origin
from typing import Any, Generic, Literal, cast, get_origin
from mcp.server.elicitation import (
CancelledElicitation,
@ -47,10 +47,10 @@ class ElicitationJsonSchema(GenerateJsonSchema):
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) # type: ignore[arg-type]
return self.enum_schema(schema)
# For list schemas, check if items are enums
if schema["type"] == "list":
return self.list_schema(schema) # type: ignore[arg-type]
return self.list_schema(schema)
# For all other types, use the default implementation
return super().generate_inner(schema)
@ -229,11 +229,13 @@ def _parse_list_syntax(lst: list[Any]) -> ElicitConfig:
# ["a", "b", "c"] -> single-select untitled
if lst and all(isinstance(item, str) for item in lst):
choice_literal = Literal[tuple(lst)] # type: ignore[valid-type]
# Construct Literal type from tuple - use cast since we can't construct Literal dynamically
# but we know the values are all strings
choice_literal: type[Any] = cast(type[Any], Literal[tuple(lst)]) # type: ignore[valid-type]
wrapped = ScalarElicitationType[choice_literal] # type: ignore[valid-type]
return ElicitConfig(
schema=get_elicitation_schema(wrapped), # type: ignore[arg-type]
response_type=wrapped, # type: ignore[assignment]
schema=get_elicitation_schema(wrapped),
response_type=wrapped,
is_raw=False,
)
@ -242,20 +244,20 @@ def _parse_list_syntax(lst: list[Any]) -> ElicitConfig:
def _parse_generic_list(response_type: Any) -> ElicitConfig:
"""Parse list[X] type annotation -> multi-select."""
wrapped = ScalarElicitationType[response_type] # type: ignore[valid-type]
wrapped = ScalarElicitationType[response_type]
return ElicitConfig(
schema=get_elicitation_schema(wrapped), # type: ignore[arg-type]
response_type=wrapped, # type: ignore[assignment]
schema=get_elicitation_schema(wrapped),
response_type=wrapped,
is_raw=False,
)
def _parse_scalar_type(response_type: Any) -> ElicitConfig:
"""Parse scalar types (bool, int, float, str, Literal, Enum)."""
wrapped = ScalarElicitationType[response_type] # type: ignore[valid-type]
wrapped = ScalarElicitationType[response_type]
return ElicitConfig(
schema=get_elicitation_schema(wrapped), # type: ignore[arg-type]
response_type=wrapped, # type: ignore[assignment]
schema=get_elicitation_schema(wrapped),
response_type=wrapped,
is_raw=False,
)

4
src/fastmcp/server/http.py
View file

@ -62,7 +62,7 @@ class StreamableHTTPASGIApp:
raise
_current_http_request: ContextVar[Request | None] = ContextVar( # type: ignore[assignment]
_current_http_request: ContextVar[Request | None] = ContextVar(
"http_request",
default=None,
)
@ -224,7 +224,7 @@ def create_sse_app(
else:
# No auth required
async def sse_endpoint(request: Request) -> Response:
return await handle_sse(request.scope, request.receive, request._send) # type: ignore[reportPrivateUsage]
return await handle_sse(request.scope, request.receive, request._send)
server_routes.append(
Route(

2
src/fastmcp/server/lifespan.py
View file

@ -117,7 +117,7 @@ class ContextManagerLifespan(Lifespan):
def __init__(self, fn: LifespanContextManagerFn) -> None:
"""Initialize with a context manager factory function."""
self._fn = fn # type: ignore[assignment]
self._fn = fn
@asynccontextmanager
async def __call__(self, server: FastMCP[Any]) -> AsyncIterator[dict[str, Any]]:

10
src/fastmcp/server/providers/proxy.py
View file

@ -86,7 +86,7 @@ class ProxyTool(Tool):
# First time name is being changed, preserve original for backend calls
update = {**update, "_backend_name": self.name}
kwargs["update"] = update
return super().model_copy(**kwargs) # type: ignore[return-value]
return super().model_copy(**kwargs)
@classmethod
def from_mcp_tool(
@ -180,7 +180,7 @@ class ProxyResource(Resource):
# First time uri is being changed, preserve original for backend calls
update = {**update, "_backend_uri": str(self.uri)}
kwargs["update"] = update
return super().model_copy(**kwargs) # type: ignore[return-value]
return super().model_copy(**kwargs)
@classmethod
def from_mcp_resource(
@ -266,7 +266,7 @@ class ProxyTemplate(ResourceTemplate):
# First time uri_template is being changed, preserve original for backend
update = {**update, "_backend_uri_template": self.uri_template}
kwargs["update"] = update
return super().model_copy(**kwargs) # type: ignore[return-value]
return super().model_copy(**kwargs)
@classmethod
def from_mcp_template( # type: ignore[override]
@ -375,7 +375,7 @@ class ProxyPrompt(Prompt):
# First time name is being changed, preserve original for backend calls
update = {**update, "_backend_name": self.name}
kwargs["update"] = update
return super().model_copy(**kwargs) # type: ignore[return-value]
return super().model_copy(**kwargs)
@classmethod
def from_mcp_prompt(
@ -693,7 +693,7 @@ async def default_proxy_sampling_handler(
role="assistant",
model="fastmcp-client",
# TODO(ty): remove when ty supports isinstance exclusion narrowing
content=content, # type: ignore[arg-type]
content=content,
)

2
src/fastmcp/server/sampling/sampling_tool.py
View file

@ -9,7 +9,7 @@ from typing import Any
from mcp.types import Tool as SDKTool
from pydantic import ConfigDict
from fastmcp.tools.function_tool import ParsedFunction
from fastmcp.tools.function_parsing import ParsedFunction
from fastmcp.utilities.types import FastMCPBaseModel

2
src/fastmcp/server/server.py
View file

@ -566,7 +566,7 @@ class FastMCP(Generic[LifespanResultT]):
worker_kwargs["name"] = settings.docket.worker_name
# Create and start Worker
async with Worker(docket, **worker_kwargs) as worker: # type: ignore[arg-type]
async with Worker(docket, **worker_kwargs) as worker:
# Store on server instance for cross-context access
self._worker = worker
# Set Worker in ContextVar so CurrentWorker can access it

2
src/fastmcp/server/tasks/handlers.py
View file

@ -132,7 +132,7 @@ async def submit_to_docket(
# Start subscription in session's task group (persists for connection lifetime)
if hasattr(ctx.session, "_subscription_task_group"):
tg = ctx.session._subscription_task_group # type: ignore[attr-defined]
tg = ctx.session._subscription_task_group
if tg:
tg.start_soon( # type: ignore[union-attr]
subscribe_to_task_updates,

2
src/fastmcp/server/tasks/subscriptions.py
View file

@ -62,7 +62,7 @@ async def subscribe_to_task_updates(
task_id=task_id,
task_key=task_key,
docket=docket,
state=event["state"], # type: ignore[typeddict-item]
state=event["state"],
poll_interval_ms=poll_interval_ms,
)
elif event["type"] == "progress":

201
src/fastmcp/tools/function_parsing.py Normal file
View file

@ -0,0 +1,201 @@
"""Function introspection and schema generation for FastMCP tools."""
from __future__ import annotations
import inspect
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, Generic, get_type_hints
import mcp.types
from pydantic import PydanticSchemaGenerationError
from typing_extensions import TypeVar as TypeVarExt
from fastmcp.server.dependencies import (
transform_context_annotations,
without_injected_parameters,
)
from fastmcp.tools.tool import ToolResult
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
Audio,
File,
Image,
create_function_without_params,
get_cached_typeadapter,
replace_type,
)
T = TypeVarExt("T", default=Any)
logger = get_logger(__name__)
@dataclass
class _WrappedResult(Generic[T]):
"""Generic wrapper for non-object return types."""
result: T
class _UnserializableType:
pass
def _is_object_schema(schema: dict[str, Any]) -> bool:
"""Check if a JSON schema represents an object type."""
# Direct object type
if schema.get("type") == "object":
return True
# Schema with properties but no explicit type is treated as object
if "properties" in schema:
return True
# Self-referencing types use $ref pointing to $defs
# The referenced type is always an object in our use case
return "$ref" in schema and "$defs" in schema
@dataclass
class ParsedFunction:
fn: Callable[..., Any]
name: str
description: str | None
input_schema: dict[str, Any]
output_schema: dict[str, Any] | None
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
exclude_args: list[str] | None = None,
validate: bool = True,
wrap_non_object_output_schema: bool = True,
) -> ParsedFunction:
if validate:
sig = inspect.signature(fn)
# Reject functions with *args or **kwargs
for param in sig.parameters.values():
if param.kind == inspect.Parameter.VAR_POSITIONAL:
raise ValueError("Functions with *args are not supported as tools")
if param.kind == inspect.Parameter.VAR_KEYWORD:
raise ValueError(
"Functions with **kwargs are not supported as tools"
)
# Reject exclude_args that don't exist in the function or don't have a default value
if exclude_args:
for arg_name in exclude_args:
if arg_name not in sig.parameters:
raise ValueError(
f"Parameter '{arg_name}' in exclude_args does not exist in function."
)
param = sig.parameters[arg_name]
if param.default == inspect.Parameter.empty:
raise ValueError(
f"Parameter '{arg_name}' in exclude_args must have a default value."
)
# collect name and doc before we potentially modify the function
fn_name = getattr(fn, "__name__", None) or fn.__class__.__name__
fn_doc = inspect.getdoc(fn)
# if the fn is a callable class, we need to get the __call__ method from here out
if not inspect.isroutine(fn):
fn = fn.__call__
# if the fn is a staticmethod, we need to work with the underlying function
if isinstance(fn, staticmethod):
fn = fn.__func__
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
# Handle injected parameters (Context, Docket dependencies)
wrapper_fn = without_injected_parameters(fn)
# Also handle exclude_args with non-serializable types (issue #2431)
# This must happen before Pydantic tries to serialize the parameters
if exclude_args:
wrapper_fn = create_function_without_params(wrapper_fn, list(exclude_args))
input_type_adapter = get_cached_typeadapter(wrapper_fn)
input_schema = input_type_adapter.json_schema()
# Compress and handle exclude_args
prune_params = list(exclude_args) if exclude_args else None
input_schema = compress_schema(
input_schema, prune_params=prune_params, prune_titles=True
)
output_schema = None
# Get the return annotation from the signature
sig = inspect.signature(fn)
output_type = sig.return_annotation
# If the annotation is a string (from __future__ annotations), resolve it
if isinstance(output_type, str):
try:
# Use get_type_hints to resolve the return type
# include_extras=True preserves Annotated metadata
type_hints = get_type_hints(fn, include_extras=True)
output_type = type_hints.get("return", output_type)
except Exception as e:
# If resolution fails, keep the string annotation
logger.debug("Failed to resolve type hint for return annotation: %s", e)
if output_type not in (inspect._empty, None, Any, ...):
# there are a variety of types that we don't want to attempt to
# serialize because they are either used by FastMCP internally,
# or are MCP content types that explicitly don't form structured
# content. By replacing them with an explicitly unserializable type,
# we ensure that no output schema is automatically generated.
clean_output_type = replace_type(
output_type,
dict.fromkeys(
(
Image,
Audio,
File,
ToolResult,
mcp.types.TextContent,
mcp.types.ImageContent,
mcp.types.AudioContent,
mcp.types.ResourceLink,
mcp.types.EmbeddedResource,
),
_UnserializableType,
),
)
try:
type_adapter = get_cached_typeadapter(clean_output_type)
base_schema = type_adapter.json_schema(mode="serialization")
# Generate schema for wrapped type if it's non-object
# because MCP requires that output schemas are objects
# Check if schema is an object type, resolving $ref references
# (self-referencing types use $ref at root level)
if wrap_non_object_output_schema and not _is_object_schema(base_schema):
# Use the wrapped result schema directly
wrapped_type = _WrappedResult[clean_output_type]
wrapped_adapter = get_cached_typeadapter(wrapped_type)
output_schema = wrapped_adapter.json_schema(mode="serialization")
output_schema["x-fastmcp-wrap-result"] = True
else:
output_schema = base_schema
output_schema = compress_schema(output_schema, prune_titles=True)
except PydanticSchemaGenerationError as e:
if "_UnserializableType" not in str(e):
logger.debug(f"Unable to generate schema for type {output_type!r}")
return cls(
fn=fn,
name=fn_name,
description=fn_doc,
input_schema=input_schema,
output_schema=output_schema or None,
)

187
src/fastmcp/tools/function_tool.py
View file

@ -9,44 +9,33 @@ from dataclasses import dataclass, field
from typing import (
TYPE_CHECKING,
Any,
Generic,
Literal,
Protocol,
TypeVar,
get_type_hints,
overload,
runtime_checkable,
)
import mcp.types
from mcp.types import Icon, ToolAnnotations, ToolExecution
from pydantic import PydanticSchemaGenerationError
from typing_extensions import TypeVar as TypeVarExt
import fastmcp
from fastmcp.decorators import resolve_task_config
from fastmcp.server.dependencies import (
transform_context_annotations,
without_injected_parameters,
)
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.function_parsing import ParsedFunction, _is_object_schema
from fastmcp.tools.tool import (
AuthCheckCallable,
Tool,
ToolResult,
ToolResultSerializerType,
)
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
Audio,
File,
Image,
NotSet,
NotSetT,
create_function_without_params,
get_cached_typeadapter,
replace_type,
)
if TYPE_CHECKING:
@ -54,9 +43,6 @@ if TYPE_CHECKING:
from docket.execution import Execution
F = TypeVar("F", bound=Callable[..., Any])
T = TypeVarExt("T", default=Any)
logger = get_logger(__name__)
@runtime_checkable
@ -87,175 +73,6 @@ class ToolMeta:
auth: AuthCheckCallable | list[AuthCheckCallable] | None = None
@dataclass
class _WrappedResult(Generic[T]):
"""Generic wrapper for non-object return types."""
result: T
class _UnserializableType:
pass
def _is_object_schema(schema: dict[str, Any]) -> bool:
"""Check if a JSON schema represents an object type."""
# Direct object type
if schema.get("type") == "object":
return True
# Schema with properties but no explicit type is treated as object
if "properties" in schema:
return True
# Self-referencing types use $ref pointing to $defs
# The referenced type is always an object in our use case
return "$ref" in schema and "$defs" in schema
@dataclass
class ParsedFunction:
fn: Callable[..., Any]
name: str
description: str | None
input_schema: dict[str, Any]
output_schema: dict[str, Any] | None
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
exclude_args: list[str] | None = None,
validate: bool = True,
wrap_non_object_output_schema: bool = True,
) -> ParsedFunction:
if validate:
sig = inspect.signature(fn)
# Reject functions with *args or **kwargs
for param in sig.parameters.values():
if param.kind == inspect.Parameter.VAR_POSITIONAL:
raise ValueError("Functions with *args are not supported as tools")
if param.kind == inspect.Parameter.VAR_KEYWORD:
raise ValueError(
"Functions with **kwargs are not supported as tools"
)
# Reject exclude_args that don't exist in the function or don't have a default value
if exclude_args:
for arg_name in exclude_args:
if arg_name not in sig.parameters:
raise ValueError(
f"Parameter '{arg_name}' in exclude_args does not exist in function."
)
param = sig.parameters[arg_name]
if param.default == inspect.Parameter.empty:
raise ValueError(
f"Parameter '{arg_name}' in exclude_args must have a default value."
)
# collect name and doc before we potentially modify the function
fn_name = getattr(fn, "__name__", None) or fn.__class__.__name__
fn_doc = inspect.getdoc(fn)
# if the fn is a callable class, we need to get the __call__ method from here out
if not inspect.isroutine(fn):
fn = fn.__call__
# if the fn is a staticmethod, we need to work with the underlying function
if isinstance(fn, staticmethod):
fn = fn.__func__
# Transform Context type annotations to Depends() for unified DI
fn = transform_context_annotations(fn)
# Handle injected parameters (Context, Docket dependencies)
wrapper_fn = without_injected_parameters(fn)
# Also handle exclude_args with non-serializable types (issue #2431)
# This must happen before Pydantic tries to serialize the parameters
if exclude_args:
wrapper_fn = create_function_without_params(wrapper_fn, list(exclude_args))
input_type_adapter = get_cached_typeadapter(wrapper_fn)
input_schema = input_type_adapter.json_schema()
# Compress and handle exclude_args
prune_params = list(exclude_args) if exclude_args else None
input_schema = compress_schema(
input_schema, prune_params=prune_params, prune_titles=True
)
output_schema = None
# Get the return annotation from the signature
sig = inspect.signature(fn)
output_type = sig.return_annotation
# If the annotation is a string (from __future__ annotations), resolve it
if isinstance(output_type, str):
try:
# Use get_type_hints to resolve the return type
# include_extras=True preserves Annotated metadata
type_hints = get_type_hints(fn, include_extras=True)
output_type = type_hints.get("return", output_type)
except Exception as e:
# If resolution fails, keep the string annotation
logger.debug("Failed to resolve type hint for return annotation: %s", e)
if output_type not in (inspect._empty, None, Any, ...):
# there are a variety of types that we don't want to attempt to
# serialize because they are either used by FastMCP internally,
# or are MCP content types that explicitly don't form structured
# content. By replacing them with an explicitly unserializable type,
# we ensure that no output schema is automatically generated.
clean_output_type = replace_type(
output_type,
dict.fromkeys( # type: ignore[arg-type]
(
Image,
Audio,
File,
ToolResult,
mcp.types.TextContent,
mcp.types.ImageContent,
mcp.types.AudioContent,
mcp.types.ResourceLink,
mcp.types.EmbeddedResource,
),
_UnserializableType,
),
)
try:
type_adapter = get_cached_typeadapter(clean_output_type)
base_schema = type_adapter.json_schema(mode="serialization")
# Generate schema for wrapped type if it's non-object
# because MCP requires that output schemas are objects
# Check if schema is an object type, resolving $ref references
# (self-referencing types use $ref at root level)
if wrap_non_object_output_schema and not _is_object_schema(base_schema):
# Use the wrapped result schema directly
wrapped_type = _WrappedResult[clean_output_type]
wrapped_adapter = get_cached_typeadapter(wrapped_type)
output_schema = wrapped_adapter.json_schema(mode="serialization")
output_schema["x-fastmcp-wrap-result"] = True
else:
output_schema = base_schema
output_schema = compress_schema(output_schema, prune_titles=True)
except PydanticSchemaGenerationError as e:
if "_UnserializableType" not in str(e):
logger.debug(f"Unable to generate schema for type {output_type!r}")
return cls(
fn=fn,
name=fn_name,
description=fn_doc,
input_schema=input_schema,
output_schema=output_schema or None,
)
class FunctionTool(Tool):
fn: Callable[..., Any]
@ -563,7 +380,7 @@ def tool(
auth=auth,
)
target = fn.__func__ if hasattr(fn, "__func__") else fn
target.__fastmcp__ = metadata # type: ignore[attr-defined]
target.__fastmcp__ = metadata
return fn
def decorator(fn: F, tool_name: str | None) -> F:

4
src/fastmcp/tools/tool_transform.py
View file

@ -15,7 +15,7 @@ from pydantic.fields import Field
from pydantic.functional_validators import BeforeValidator
import fastmcp
from fastmcp.tools.function_tool import ParsedFunction
from fastmcp.tools.function_parsing import ParsedFunction
from fastmcp.tools.tool import Tool, ToolResult, _convert_to_content
from fastmcp.utilities.components import _convert_set_default_none
from fastmcp.utilities.json_schema import compress_schema
@ -31,7 +31,7 @@ logger = get_logger(__name__)
# Context variable to store current transformed tool
_current_tool: ContextVar[TransformedTool | None] = ContextVar( # type: ignore[assignment]
_current_tool: ContextVar[TransformedTool | None] = ContextVar(
"_current_tool", default=None
)

32
src/fastmcp/utilities/json_schema_type.py
View file

@ -47,6 +47,7 @@ from typing import (
ForwardRef,
Literal,
Union,
cast,
)
from pydantic import (
@ -171,14 +172,15 @@ def json_schema_to_type(
if not schema.get("properties") and schema.get("additionalProperties"):
additional_props = schema["additionalProperties"]
if additional_props is True:
return dict[str, Any] # type: ignore - additionalProperties: true means dict[str, Any]
return dict[str, Any]
else:
# Handle typed dictionaries like dict[str, str]
value_type = _schema_to_type(additional_props, schemas=schema)
return dict[str, value_type] # type: ignore
# value_type might be ForwardRef or type - cast to Any for dynamic type construction
return cast(type[Any], dict[str, value_type]) # type: ignore[valid-type]
# If no properties and no additionalProperties, default to dict[str, Any] for safety
elif not schema.get("properties") and not schema.get("additionalProperties"):
return dict[str, Any] # type: ignore
return dict[str, Any]
# If has properties AND additionalProperties is True, use Pydantic BaseModel
elif schema.get("properties") and schema.get("additionalProperties") is True:
return _create_pydantic_model(schema, name, schemas=schema)
@ -265,13 +267,13 @@ 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[arg-type] # noqa: UP007
combined = Union[tuple(item_types)] # noqa: UP007
base = list[combined] # type: ignore[valid-type]
else:
# Handle single item schema
item_type = _schema_to_type(items, schemas)
base_class = set if schema.get("uniqueItems") else list
base = base_class[item_type] # type: ignore[misc]
base = base_class[item_type]
constraints = {
k: v
@ -295,17 +297,17 @@ def _get_from_type_handler(
"""Get the appropriate type handler for the schema."""
type_handlers: dict[str, Callable[..., Any]] = { # TODO
"string": lambda s: _create_string_type(s), # type: ignore
"integer": lambda s: _create_numeric_type(int, s), # type: ignore
"number": lambda s: _create_numeric_type(float, s), # type: ignore
"boolean": lambda _: bool, # type: ignore
"null": lambda _: type(None), # type: ignore
"array": lambda s: _create_array_type(s, schemas), # type: ignore
"string": lambda s: _create_string_type(s),
"integer": lambda s: _create_numeric_type(int, s),
"number": lambda s: _create_numeric_type(float, s),
"boolean": lambda _: bool,
"null": lambda _: type(None),
"array": lambda s: _create_array_type(s, schemas),
"object": lambda s: (
_create_pydantic_model(s, s.get("title"), schemas)
if s.get("properties") and s.get("additionalProperties") is True
else _create_dataclass(s, s.get("title"), schemas)
), # type: ignore
),
}
return type_handlers.get(schema.get("type", None), _return_Any)
@ -326,7 +328,7 @@ def _schema_to_type(
ref = schema["$ref"]
# Handle self-reference
if ref == "#":
return ForwardRef(schema.get("title", "Root")) # type: ignore[return-value]
return ForwardRef(schema.get("title", "Root"))
return _schema_to_type(_resolve_ref(ref, schemas), schemas)
if "const" in schema:
@ -348,7 +350,7 @@ def _schema_to_type(
# This is a dict type, handle it directly
additional_props = subschema["additionalProperties"]
if additional_props is True:
types.append(dict[str, Any]) # type: ignore
types.append(dict[str, Any])
else:
value_type = _schema_to_type(additional_props, schemas)
types.append(dict[str, value_type]) # type: ignore
@ -374,7 +376,7 @@ def _schema_to_type(
schema_type = schema.get("type")
if not schema_type:
return Any # type: ignore[return-value]
return Any
if isinstance(schema_type, list):
# Create a copy of the schema for each type, but keep all constraints

6
src/fastmcp/utilities/mcp_server_config/v1/mcp_server_config.py
View file

@ -216,7 +216,7 @@ class MCPServerConfig(BaseModel):
"""
if isinstance(v, dict):
return Deployment(**v) # type: ignore[arg-type]
return Deployment(**v)
return cast(Deployment, v) # type: ignore[return-value]
@classmethod
@ -301,9 +301,9 @@ class MCPServerConfig(BaseModel):
if any([transport, host, port, path, log_level, env, cwd, args]):
# Convert streamable-http to http for backward compatibility
if transport == "streamable-http":
transport = "http" # type: ignore[assignment]
transport = "http"
deployment = Deployment(
transport=transport, # type: ignore[arg-type]
transport=transport,
host=host,
port=port,
path=path,

6
src/fastmcp/utilities/openapi/parser.py
View file

@ -1,6 +1,6 @@
"""OpenAPI parsing logic for converting OpenAPI specs to HTTPRoute objects."""
from typing import Any, Generic, TypeVar
from typing import Any, Generic, TypeVar, cast
from openapi_pydantic import (
OpenAPI,
@ -146,9 +146,9 @@ class OpenAPIParser(
def _convert_to_parameter_location(self, param_in: str) -> ParameterLocation:
"""Convert string parameter location to our ParameterLocation type."""
if param_in in ["path", "query", "header", "cookie"]:
return param_in # type: ignore[return-value] # Safe cast since we checked values
return cast(ParameterLocation, param_in)
logger.warning(f"Unknown parameter location: {param_in}, defaulting to 'query'")
return "query" # type: ignore[return-value] # Safe cast to default value
return cast(ParameterLocation, "query")
def _resolve_ref(self, item: Any) -> Any:
"""Resolves a reference to its target definition."""

2
src/fastmcp/utilities/types.py
View file

@ -480,7 +480,7 @@ def replace_type(type_, type_map: dict[type, type]):
new_args = tuple(replace_type(arg, type_map) for arg in args)
if origin is UnionType:
return Union[new_args] # type: ignore # noqa: UP007
return Union[new_args] # noqa: UP007
else:
return origin[new_args]

20
tests/cli/test_mcp_server_config_integration.py
View file

@ -136,7 +136,7 @@ class TestEnvironmentExecution:
"""Test that environment with dependencies needs UV."""
config = MCPServerConfig(
source={"path": "server.py"},
environment={"dependencies": ["requests", "numpy"]}, # type: ignore[arg-type]
environment={"dependencies": ["requests", "numpy"]},
)
assert config.environment is not None
@ -146,7 +146,7 @@ class TestEnvironmentExecution:
"""Test that environment with Python version needs UV."""
config = MCPServerConfig(
source={"path": "server.py"},
environment={"python": "3.12"}, # type: ignore[arg-type]
environment={"python": "3.12"},
)
assert config.environment is not None
@ -164,7 +164,7 @@ class TestEnvironmentExecution:
"""Test that no UV is needed with empty environment config."""
config = MCPServerConfig(
source={"path": "server.py"},
environment={}, # type: ignore[arg-type]
environment={},
)
assert config.environment is not None
@ -202,7 +202,7 @@ class TestPathResolution:
config = MCPServerConfig(
source={"path": "server.py"},
deployment={"cwd": "work"}, # type: ignore[arg-type]
deployment={"cwd": "work"},
)
original_cwd = os.getcwd()
@ -226,7 +226,7 @@ class TestPathResolution:
config = MCPServerConfig(
source={"path": "server.py"},
environment={"requirements": "requirements.txt"}, # type: ignore[arg-type]
environment={"requirements": "requirements.txt"},
)
# Build UV command
@ -248,14 +248,14 @@ class TestConfigValidation:
with pytest.raises(ValueError):
MCPServerConfig(
source={"path": "server.py"},
deployment={"transport": "invalid_transport"}, # type: ignore[arg-type]
deployment={"transport": "invalid_transport"},
)
def test_streamable_http_transport_accepted(self):
"""Test that streamable-http transport is accepted as a valid value."""
config = MCPServerConfig(
source={"path": "server.py"},
deployment={"transport": "streamable-http"}, # type: ignore[arg-type]
deployment={"transport": "streamable-http"},
)
assert config.deployment.transport == "streamable-http"
@ -264,7 +264,7 @@ class TestConfigValidation:
with pytest.raises(ValueError):
MCPServerConfig(
source={"path": "server.py"},
deployment={"log_level": "INVALID"}, # type: ignore[arg-type]
deployment={"log_level": "INVALID"},
)
def test_missing_source_rejected(self):
@ -277,7 +277,7 @@ class TestConfigValidation:
for transport in ["stdio", "http", "sse"]:
config = MCPServerConfig(
source={"path": "server.py"},
deployment={"transport": transport}, # type: ignore[arg-type]
deployment={"transport": transport},
)
assert config.deployment is not None
assert config.deployment.transport == transport
@ -287,7 +287,7 @@ class TestConfigValidation:
for level in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]:
config = MCPServerConfig(
source={"path": "server.py"},
deployment={"log_level": level}, # type: ignore[arg-type]
deployment={"log_level": level},
)
assert config.deployment is not None
assert config.deployment.log_level == level

2
tests/client/test_notifications.py
View file

@ -515,7 +515,7 @@ class TestNotificationAPI:
async with Client(server, message_handler=recording_message_handler) as client:
recording_message_handler.reset()
result = await client.call_tool("trigger_and_wait", {})
tool_finished = datetime.fromisoformat(result.data) # type: ignore[arg-type]
tool_finished = datetime.fromisoformat(result.data)
# Notification should have been received at least 1.5s before tool finished
# (proves background flusher sent it, not final flush)

2
tests/prompts/test_standalone_decorator.py
View file

@ -106,7 +106,7 @@ class TestPromptDecorator:
with pytest.raises(TypeError, match="classmethod"):
class MyClass:
@prompt # type: ignore[arg-type]
@prompt
@classmethod
def my_prompt(cls, topic: str) -> str:
return f"Analyze: {topic}"

2
tests/resources/test_resources.py
View file

@ -93,7 +93,7 @@ class TestResourceValidation:
class ConcreteResource(Resource):
pass
resource = ConcreteResource(uri=AnyUrl("test://test"), name="test") # type: ignore
resource = ConcreteResource(uri=AnyUrl("test://test"), name="test")
with pytest.raises(NotImplementedError, match="Subclasses must implement read"):
await resource.read()

4
tests/server/auth/test_jwt_provider.py
View file

@ -395,8 +395,8 @@ class TestBearerTokenJWKS:
from authlib.jose import JsonWebKey
# Create JWK from the RSA public key
jwk = JsonWebKey.import_key(rsa_key_pair.public_key) # type: ignore
jwk_data: JWKData = jwk.as_dict() # type: ignore
jwk = JsonWebKey.import_key(rsa_key_pair.public_key)
jwk_data: JWKData = jwk.as_dict()
jwk_data["kid"] = "test-key-1"
jwk_data["alg"] = "RS256"

4
tests/server/http/test_bearer_auth_backend.py
View file

@ -140,7 +140,7 @@ class TestBearerAuthBackendWithMockVerifier:
expires_at=None,
)
mock_verifier = MockTokenVerifier(return_value=mock_access_token)
backend = BearerAuthBackend(mock_verifier) # type: ignore[arg-type]
backend = BearerAuthBackend(mock_verifier)
scope = {
"type": "http",
@ -162,7 +162,7 @@ class TestBearerAuthBackendWithMockVerifier:
async def test_backend_handles_verify_token_none_result(self):
"""Test that BearerAuthBackend handles None result from verify_token."""
mock_verifier = MockTokenVerifier(return_value=None)
backend = BearerAuthBackend(mock_verifier) # type: ignore[arg-type]
backend = BearerAuthBackend(mock_verifier)
scope = {
"type": "http",

12
tests/server/middleware/test_middleware.py
View file

@ -565,8 +565,10 @@ class TestApplyMiddlewareParameter:
result = await server.render_prompt("greet", {"name": "World"})
assert len(result.messages) == 1 # type: ignore[union-attr]
assert result.messages[0].content.text == "Hello, World!" # type: ignore[union-attr]
assert len(result.messages) == 1
# content is TextContent | EmbeddedResource, but we know it's TextContent from the test
assert isinstance(result.messages[0].content, mcp.types.TextContent)
assert result.messages[0].content.text == "Hello, World!"
assert recording.assert_called(hook="on_get_prompt", times=1)
async def test_render_prompt_with_run_middleware_false(self):
@ -584,8 +586,10 @@ class TestApplyMiddlewareParameter:
"greet", {"name": "World"}, run_middleware=False
)
assert len(result.messages) == 1 # type: ignore[union-attr]
assert result.messages[0].content.text == "Hello, World!" # type: ignore[union-attr]
assert len(result.messages) == 1
# content is TextContent | EmbeddedResource, but we know it's TextContent from the test
assert isinstance(result.messages[0].content, mcp.types.TextContent)
assert result.messages[0].content.text == "Hello, World!"
# Middleware should not have been called
assert len(recording.calls) == 0

4
tests/server/providers/proxy/test_proxy_client.py
View file

@ -88,7 +88,7 @@ async def proxy_server(fastmcp_server: FastMCP):
"""
A proxy server that forwards interactions with the proxy client to the given fastmcp server.
"""
return FastMCP.as_proxy(ProxyClient(fastmcp_server)) # type: ignore
return FastMCP.as_proxy(ProxyClient(fastmcp_server))
class TestProxyClient:
@ -377,7 +377,7 @@ class TestProxyClient:
else:
return f"Elicitation {result.action}"
proxy_server = FastMCP.as_proxy(ProxyClient(fastmcp_server)) # type: ignore
proxy_server = FastMCP.as_proxy(ProxyClient(fastmcp_server))
# Test that elicitation works correctly through the proxy
async def elicitation_handler(

2
tests/server/providers/proxy/test_proxy_server.py
View file

@ -214,7 +214,7 @@ async def test_proxy_with_async_client_factory():
proxy = FastMCPProxy(client_factory=async_factory)
assert isinstance(proxy, FastMCPProxy)
assert inspect.iscoroutinefunction(proxy.client_factory)
client = await proxy.client_factory() # type: ignore[misc]
client = await proxy.client_factory()
assert isinstance(client, Client)
assert isinstance(client.transport, StreamableHttpTransport)
assert client.transport.url == "http://example.com/mcp/"

2
tests/server/providers/test_local_provider_prompts.py
View file

@ -293,7 +293,7 @@ class TestPromptDecorator:
mcp = FastMCP()
class MyClass:
@mcp.prompt # type: ignore[misc]
@mcp.prompt
@staticmethod
def test_prompt() -> str:
return "Static Hello, world!"

2
tests/server/providers/test_local_provider_resources.py
View file

@ -438,7 +438,7 @@ class TestResourceDecorator:
mcp = FastMCP()
class MyClass:
@mcp.resource("resource://data") # type: ignore[misc]
@mcp.resource("resource://data")
@staticmethod
def get_data() -> str:
return "Static Hello, world!"

3
tests/server/tasks/test_task_dependencies.py
View file

@ -6,6 +6,7 @@ CurrentFastMCP(), and Depends() should be resolved in the worker context.
"""
from contextlib import asynccontextmanager
from typing import Any, cast
import pytest
@ -252,7 +253,7 @@ async def test_dependency_errors_propagate_to_task_failure():
@mcp.tool(task=True)
async def tool_with_failing_dep(
value: str, dep: str = Depends(failing_dependency)
value: str, dep: str = cast(Any, Depends(failing_dependency))
) -> str:
return f"Got: {dep}"

2
tests/server/tasks/test_task_meta_parameter.py
View file

@ -152,7 +152,7 @@ class TrackingMiddleware(Middleware):
) -> ToolResult:
if context.method:
self._calls.append(context.method)
return await call_next(context) # type: ignore[return-value]
return await call_next(context)
class TestTaskMetaMiddleware:

11
tests/server/test_context.py
View file

@ -1,3 +1,4 @@
from typing import Any, cast
from unittest.mock import MagicMock
import pytest
@ -48,7 +49,7 @@ class TestSessionId:
mock_headers = {"mcp-session-id": "test-session-123"}
token = request_ctx.set(
RequestContext( # type: ignore[arg-type]
RequestContext(
request_id=0,
meta=None,
session=MagicMock(wraps={}),
@ -69,7 +70,7 @@ class TestSessionId:
from mcp.shared.context import RequestContext
token = request_ctx.set(
RequestContext( # type: ignore[arg-type]
RequestContext(
request_id=0,
meta=None,
session=MagicMock(wraps={}),
@ -146,9 +147,9 @@ class TestContextMeta:
mock_meta = MockMeta()
token = request_ctx.set(
RequestContext( # type: ignore[arg-type]
RequestContext(
request_id=0,
meta=mock_meta, # type: ignore[arg-type]
meta=cast(Any, mock_meta), # Mock object for testing
session=MagicMock(wraps={}),
lifespan_context=MagicMock(),
)
@ -169,7 +170,7 @@ class TestContextMeta:
from mcp.shared.context import RequestContext
token = request_ctx.set(
RequestContext( # type: ignore[arg-type]
RequestContext(
request_id=0,
meta=None,
session=MagicMock(wraps={}),

2
tests/tools/test_standalone_decorator.py
View file

@ -106,7 +106,7 @@ class TestToolDecorator:
with pytest.raises(TypeError, match="classmethod"):
class MyClass:
@tool # type: ignore[arg-type]
@tool
@classmethod
def my_method(cls) -> str:
return "hello"

6
tests/tools/test_tool.py
View file

@ -450,7 +450,7 @@ class TestToolFromFunctionOutputSchema:
],
)
async def test_simple_return_annotation(self, annotation):
def func() -> annotation: # type: ignore
def func() -> annotation:
return 1
tool = Tool.from_function(func)
@ -484,7 +484,7 @@ class TestToolFromFunctionOutputSchema:
],
)
async def test_complex_return_annotation(self, annotation):
def func() -> annotation: # type: ignore
def func() -> annotation:
return 1
tool = Tool.from_function(func)
@ -524,7 +524,7 @@ class TestToolFromFunctionOutputSchema:
],
)
async def test_converted_return_annotation(self, annotation, expected):
def func() -> annotation: # type: ignore
def func() -> annotation:
return 1
tool = Tool.from_function(func)

4
tests/utilities/openapi/test_models.py
View file

@ -55,7 +55,7 @@ class TestParameterInfo:
"""Test that all valid parameter locations are accepted."""
param = ParameterInfo(
name="test",
location=location, # type: ignore
location=location,
required=False,
schema={"type": "string"},
)
@ -290,7 +290,7 @@ class TestHTTPRoute:
"""Test that all valid HTTP methods are accepted."""
route = HTTPRoute(
path="/test",
method=method, # type: ignore
method=method,
operation_id="test",
)
assert route.method == method

18
tests/utilities/openapi/test_nullable_fields.py
View file

@ -122,10 +122,20 @@ class TestHandleNullableFields:
def test_non_dict_input_unchanged(self):
"""Test that non-dict inputs are returned unchanged."""
assert convert_openapi_schema_to_json_schema("string", "3.0.0") == "string" # type: ignore[arg-type]
assert convert_openapi_schema_to_json_schema(123, "3.0.0") == 123 # type: ignore[arg-type]
assert convert_openapi_schema_to_json_schema(None, "3.0.0") is None # type: ignore[arg-type]
assert convert_openapi_schema_to_json_schema([1, 2, 3], "3.0.0") == [1, 2, 3] # type: ignore[arg-type]
# These tests intentionally pass invalid types to check edge case handling
from typing import Any, cast
assert (
convert_openapi_schema_to_json_schema(cast(Any, "string"), "3.0.0")
== "string"
)
assert convert_openapi_schema_to_json_schema(cast(Any, 123), "3.0.0") == 123
assert convert_openapi_schema_to_json_schema(cast(Any, None), "3.0.0") is None
assert convert_openapi_schema_to_json_schema(cast(Any, [1, 2, 3]), "3.0.0") == [
1,
2,
3,
]
def test_performance_optimization_no_copy_when_unchanged(self):
"""Test that schemas without nullable fields return the same object (no copy)."""

6
tests/utilities/test_types.py
View file

@ -1,6 +1,6 @@
import base64
import os
from typing import Annotated, Any
from typing import Annotated, Any, cast
import pytest
from mcp.types import BlobResourceContents, TextResourceContents
@ -98,7 +98,7 @@ class TestIsClassMemberOfType:
def test_none_is_not_member(self):
"""Test that None is not a member of any class."""
assert not is_class_member_of_type(None, BaseClass) # type: ignore
assert not is_class_member_of_type(None, BaseClass)
def test_generic_type_is_not_member(self):
"""Test that generic types are not members based on their parameter types."""
@ -121,7 +121,7 @@ class TestIsSubclassSafe:
def test_none_type_handled_safely(self):
"""Test that None type is handled safely without raising TypeError."""
assert not issubclass_safe(None, BaseClass) # type: ignore
assert not issubclass_safe(cast(Any, None), BaseClass)
class TestImage:

823
uv.lock generated
View file

File diff suppressed because it is too large Load diff