Merge remote-tracking branch 'origin/main' into 100-coverage

Author: Kludex

.gitattribute

@@ -0,0 +1,2 @@
+# Generated
+uv.lock  linguist-generated=true

.gitignore

@@ -52,6 +52,7 @@ coverage.xml
 *.py,cover
 .hypothesis/
 .pytest_cache/
+.ruff_cache/
 cover/
 
 # Translations
@@ -168,3 +169,6 @@ cython_debug/
 .vscode/
 .windsurfrules
 **/CLAUDE.local.md
+
+# claude code
+.claude/

.pre-commit-config.yaml

@@ -25,22 +25,22 @@ repos:
     hooks:
       - id: ruff-format
         name: Ruff Format
-        entry: uv run ruff
+        entry: uv run --frozen ruff
         args: [format]
         language: system
         types: [python]
         pass_filenames: false
       - id: ruff
         name: Ruff
-        entry: uv run ruff
+        entry: uv run --frozen ruff
         args: ["check", "--fix", "--exit-non-zero-on-fix"]
         types: [python]
         language: system
         pass_filenames: false
         exclude: ^README\.md$
       - id: pyright
         name: pyright
-        entry: uv run pyright
+        entry: uv run --frozen pyright
         language: system
         types: [python]
         pass_filenames: false
@@ -52,7 +52,7 @@ repos:
         pass_filenames: false
       - id: readme-snippets
         name: Check README snippets are up to date
-        entry: uv run scripts/update_readme_snippets.py --check
+        entry: uv run --frozen python scripts/update_readme_snippets.py --check
         language: system
         files: ^(README\.md|examples/.*\.py|scripts/update_readme_snippets\.py)$
         pass_filenames: false

pyproject.toml

@@ -103,18 +103,36 @@ executionEnvironments = [
   { root = "examples/servers", reportUnusedFunction = false },
 ]
 
-[tool.ruff.lint]
-select = ["C4", "E", "F", "I", "PERF", "UP"]
-ignore = ["PERF203"]
-
 [tool.ruff]
 line-length = 120
 target-version = "py310"
 extend-exclude = ["README.md"]
 
+[tool.ruff.lint]
+select = [
+    "C4",    # flake8-comprehensions
+    "C90",   # mccabe
+    "E",     # pycodestyle
+    "F",     # pyflakes
+    "I",     # isort
+    "PERF",  # Perflint
+    "PL",    # Pylint
+    "UP",    # pyupgrade
+]
+ignore = ["PERF203", "PLC0415", "PLR0402"]
+mccabe.max-complexity = 24  # Default is 10
+
 [tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["F401"]
 "tests/server/fastmcp/test_func_metadata.py" = ["E501"]
+"tests/shared/test_progress_notifications.py" = ["PLW0603"]
+
+[tool.ruff.lint.pylint]
+allow-magic-value-types = ["bytes", "float", "int", "str"]
+max-args = 23         # Default is 5
+max-branches = 23     # Default is 12
+max-returns = 13      # Default is 6
+max-statements = 102  # Default is 50
 
 [tool.uv.workspace]
 members = ["examples/servers/*", "examples/snippets"]

src/mcp/server/auth/handlers/authorize.py

@@ -99,7 +99,7 @@ async def error_response(
             if client is None and attempt_load_client:
                 # make last-ditch attempt to load the client
                 client_id = best_effort_extract_string("client_id", params)
-                client = client_id and await self.provider.get_client(client_id)
+                client = await self.provider.get_client(client_id) if client_id else None
             if redirect_uri is None and client:
                 # make last-ditch effort to load the redirect uri
                 try:

src/mcp/server/elicitation.py

@@ -98,7 +98,7 @@ async def elicit_with_validation(
         related_request_id=related_request_id,
     )
 
-    if result.action == "accept" and result.content:
+    if result.action == "accept" and result.content is not None:
         # Validate and parse the content using the schema
         validated_data = schema.model_validate(result.content)
         return AcceptedElicitation(data=validated_data)

src/mcp/server/fastmcp/prompts/base.py

@@ -1,14 +1,23 @@
 """Base classes for FastMCP prompts."""
 
+from __future__ import annotations
+
 import inspect
 from collections.abc import Awaitable, Callable, Sequence
-from typing import Any, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 import pydantic_core
 from pydantic import BaseModel, Field, TypeAdapter, validate_call
 
+from mcp.server.fastmcp.utilities.context_injection import find_context_parameter, inject_context
+from mcp.server.fastmcp.utilities.func_metadata import func_metadata
 from mcp.types import ContentBlock, TextContent
 
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT, RequestT
+
 
 class Message(BaseModel):
     """Base class for all prompt messages."""
@@ -62,6 +71,7 @@ class Prompt(BaseModel):
     description: str | None = Field(None, description="Description of what the prompt does")
     arguments: list[PromptArgument] | None = Field(None, description="Arguments that can be passed to the prompt")
     fn: Callable[..., PromptResult | Awaitable[PromptResult]] = Field(exclude=True)
+    context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context", exclude=True)
 
     @classmethod
     def from_function(
@@ -70,7 +80,8 @@ def from_function(
         name: str | None = None,
         title: str | None = None,
         description: str | None = None,
-    ) -> "Prompt":
+        context_kwarg: str | None = None,
+    ) -> Prompt:
         """Create a Prompt from a function.
 
         The function can return:
@@ -84,8 +95,16 @@ def from_function(
         if func_name == "<lambda>":
             raise ValueError("You must provide a name for lambda functions")
 
-        # Get schema from TypeAdapter - will fail if function isn't properly typed
-        parameters = TypeAdapter(fn).json_schema()
+        # Find context parameter if it exists
+        if context_kwarg is None:
+            context_kwarg = find_context_parameter(fn)
+
+        # Get schema from func_metadata, excluding context parameter
+        func_arg_metadata = func_metadata(
+            fn,
+            skip_names=[context_kwarg] if context_kwarg is not None else [],
+        )
+        parameters = func_arg_metadata.arg_model.model_json_schema()
 
         # Convert parameters to PromptArguments
         arguments: list[PromptArgument] = []
@@ -109,9 +128,14 @@ def from_function(
             description=description or fn.__doc__ or "",
             arguments=arguments,
             fn=fn,
+            context_kwarg=context_kwarg,
         )
 
-    async def render(self, arguments: dict[str, Any] | None = None) -> list[Message]:
+    async def render(
+        self,
+        arguments: dict[str, Any] | None = None,
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+    ) -> list[Message]:
         """Render the prompt with arguments."""
         # Validate required arguments
         if self.arguments:
@@ -122,8 +146,11 @@ async def render(self, arguments: dict[str, Any] | None = None) -> list[Message]
                 raise ValueError(f"Missing required arguments: {missing}")
 
         try:
+            # Add context to arguments if needed
+            call_args = inject_context(self.fn, arguments or {}, context, self.context_kwarg)
+
             # Call function and check if result is a coroutine
-            result = self.fn(**(arguments or {}))
+            result = self.fn(**call_args)
             if inspect.iscoroutine(result):
                 result = await result
 

src/mcp/server/fastmcp/prompts/manager.py

@@ -1,10 +1,17 @@
 """Prompt management functionality."""
 
-from typing import Any
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
 
 from mcp.server.fastmcp.prompts.base import Message, Prompt
 from mcp.server.fastmcp.utilities.logging import get_logger
 
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT, RequestT
+
 logger = get_logger(__name__)
 
 
@@ -39,10 +46,15 @@ def add_prompt(
         self._prompts[prompt.name] = prompt
         return prompt
 
-    async def render_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> list[Message]:
+    async def render_prompt(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+    ) -> list[Message]:
         """Render a prompt by name with arguments."""
         prompt = self.get_prompt(name)
         if not prompt:
             raise ValueError(f"Unknown prompt: {name}")
 
-        return await prompt.render(arguments)
+        return await prompt.render(arguments, context=context)

src/mcp/server/fastmcp/resources/resource_manager.py

@@ -1,14 +1,21 @@
 """Resource manager functionality."""
 
+from __future__ import annotations
+
 from collections.abc import Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from pydantic import AnyUrl
 
 from mcp.server.fastmcp.resources.base import Resource
 from mcp.server.fastmcp.resources.templates import ResourceTemplate
 from mcp.server.fastmcp.utilities.logging import get_logger
 
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT, RequestT
+
 logger = get_logger(__name__)
 
 
@@ -67,7 +74,11 @@ def add_template(
         self._templates[template.uri_template] = template
         return template
 
-    async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
+    async def get_resource(
+        self,
+        uri: AnyUrl | str,
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+    ) -> Resource | None:
         """Get resource by URI, checking concrete resources first, then templates."""
         uri_str = str(uri)
         logger.debug("Getting resource", extra={"uri": uri_str})
@@ -80,7 +91,7 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
         for template in self._templates.values():
             if params := template.matches(uri_str):
                 try:
-                    return await template.create_resource(uri_str, params)
+                    return await template.create_resource(uri_str, params, context=context)
                 except Exception as e:
                     raise ValueError(f"Error creating resource from template: {e}")
 

src/mcp/server/fastmcp/resources/templates.py

@@ -5,11 +5,18 @@
 import inspect
 import re
 from collections.abc import Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
-from pydantic import BaseModel, Field, TypeAdapter, validate_call
+from pydantic import BaseModel, Field, validate_call
 
 from mcp.server.fastmcp.resources.types import FunctionResource, Resource
+from mcp.server.fastmcp.utilities.context_injection import find_context_parameter, inject_context
+from mcp.server.fastmcp.utilities.func_metadata import func_metadata
+
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT, RequestT
 
 
 class ResourceTemplate(BaseModel):
@@ -22,6 +29,7 @@ class ResourceTemplate(BaseModel):
     mime_type: str = Field(default="text/plain", description="MIME type of the resource content")
     fn: Callable[..., Any] = Field(exclude=True)
     parameters: dict[str, Any] = Field(description="JSON schema for function parameters")
+    context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context")
 
     @classmethod
     def from_function(
@@ -32,14 +40,23 @@ def from_function(
         title: str | None = None,
         description: str | None = None,
         mime_type: str | None = None,
+        context_kwarg: str | None = None,
     ) -> ResourceTemplate:
         """Create a template from a function."""
         func_name = name or fn.__name__
         if func_name == "<lambda>":
             raise ValueError("You must provide a name for lambda functions")
 
-        # Get schema from TypeAdapter - will fail if function isn't properly typed
-        parameters = TypeAdapter(fn).json_schema()
+        # Find context parameter if it exists
+        if context_kwarg is None:
+            context_kwarg = find_context_parameter(fn)
+
+        # Get schema from func_metadata, excluding context parameter
+        func_arg_metadata = func_metadata(
+            fn,
+            skip_names=[context_kwarg] if context_kwarg is not None else [],
+        )
+        parameters = func_arg_metadata.arg_model.model_json_schema()
 
         # ensure the arguments are properly cast
         fn = validate_call(fn)
@@ -52,6 +69,7 @@ def from_function(
             mime_type=mime_type or "text/plain",
             fn=fn,
             parameters=parameters,
+            context_kwarg=context_kwarg,
         )
 
     def matches(self, uri: str) -> dict[str, Any] | None:
@@ -63,9 +81,17 @@ def matches(self, uri: str) -> dict[str, Any] | None:
             return match.groupdict()
         return None
 
-    async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
+    async def create_resource(
+        self,
+        uri: str,
+        params: dict[str, Any],
+        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
+    ) -> Resource:
         """Create a resource from the template with the given parameters."""
         try:
+            # Add context to params if needed
+            params = inject_context(self.fn, params, context, self.context_kwarg)
+
             # Call function and check if result is a coroutine
             result = self.fn(**params)
             if inspect.iscoroutine(result):