modelcontextprotocol__python-sdk-1864

{
  "status": "ok",
  "stdout": "src/mcp/server/fastmcp/resources/templates.py:94:    async def create_resource(\nsrc/mcp/server/fastmcp/resources/resource_manager.py:31:    def add_resource(self, resource: Resource) -> Resource:\nsrc/mcp/server/fastmcp/resources/resource_manager.py:84:    async def get_resource(\nsrc/mcp/server/fastmcp/resources/resource_manager.py:101:                    return await template.create_resource(uri_str, params, context=context)\nsrc/mcp/server/fastmcp/server.py:295:        self._mcp_server.read_resource()(self.read_resource)\nsrc/mcp/server/fastmcp/server.py:367:    async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:\nsrc/mcp/server/fastmcp/server.py:371:        resource = await self._resource_manager.get_resource(uri, context=context)\nsrc/mcp/server/fastmcp/server.py:511:    def add_resource(self, resource: Resource) -> None:\nsrc/mcp/server/fastmcp/server.py:517:        self._resource_manager.add_resource(resource)\nsrc/mcp/server/fastmcp/server.py:519:    def resource(\nsrc/mcp/server/fastmcp/server.py:551:            @server.resource(\"resource://my-resource\")\nsrc/mcp/server/fastmcp/server.py:555:            @server.resource(\"resource://my-resource\")\nsrc/mcp/server/fastmcp/server.py:560:            @server.resource(\"resource://{city}/weather\")\nsrc/mcp/server/fastmcp/server.py:564:            @server.resource(\"resource://{city}/weather\")\nsrc/mcp/server/fastmcp/server.py:573:                \"Did you forget to call it? Use @resource('uri') instead of @resource\"\nsrc/mcp/server/fastmcp/server.py:622:                self.add_resource(resource)\nsrc/mcp/server/fastmcp/server.py:1056:        data = ctx.read_resource(\"resource://data\")\nsrc/mcp/server/fastmcp/server.py:1119:    async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContents]:\nsrc/mcp/server/fastmcp/server.py:1129:        return await self._fastmcp.read_resource(uri)\nsrc/mcp/server/lowlevel/server.py:337:    def read_resource(self):\nsrc/mcp/server/lowlevel/server.py:413:    def subscribe_resource(self):  # pragma: no cover\nsrc/mcp/server/lowlevel/server.py:426:    def unsubscribe_resource(self):  # pragma: no cover\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/resources/resource_manager.py",
  "content": "\"\"\"Resource manager functionality.\"\"\"\n\nfrom __future__ import annotations\n\nfrom collections.abc import Callable\nfrom typing import TYPE_CHECKING, Any\n\nfrom pydantic import AnyUrl\n\nfrom mcp.server.fastmcp.resources.base import Resource\nfrom mcp.server.fastmcp.resources.templates import ResourceTemplate\nfrom mcp.server.fastmcp.utilities.logging import get_logger\nfrom mcp.types import Annotations, Icon\n\nif TYPE_CHECKING:\n    from mcp.server.fastmcp.server import Context\n    from mcp.server.session import ServerSessionT\n    from mcp.shared.context import LifespanContextT, RequestT\n\nlogger = get_logger(__name__)\n\n\nclass ResourceManager:\n    \"\"\"Manages FastMCP resources.\"\"\"\n\n    def __init__(self, warn_on_duplicate_resources: bool = True):\n        self._resources: dict[str, Resource] = {}\n        self._templates: dict[str, ResourceTemplate] = {}\n        self.warn_on_duplicate_resources = warn_on_duplicate_resources\n\n    def add_resource(self, resource: Resource) -> Resource:\n        \"\"\"Add a resource to the manager.\n\n        Args:\n            resource: A Resource instance to add\n\n        Returns:\n            The added resource. If a resource with the same URI already exists,\n            returns the existing resource.\n        \"\"\"\n        logger.debug(\n            \"Adding resource\",\n            extra={\n                \"uri\": resource.uri,\n                \"type\": type(resource).__name__,\n                \"resource_name\": resource.name,\n            },\n        )\n        existing = self._resources.get(str(resource.uri))\n        if existing:\n            if self.warn_on_duplicate_resources:\n                logger.warning(f\"Resource already exists: {resource.uri}\")\n            return existing\n        self._resources[str(resource.uri)] = resource\n        return resource\n\n    def add_template(\n        self,\n        fn: Callable[..., Any],\n        uri_template: str,\n        name: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        mime_type: str | None = None,\n        icons: list[Icon] | None = None,\n        annotations: Annotations | None = None,\n        meta: dict[str, Any] | None = None,\n    ) -> ResourceTemplate:\n        \"\"\"Add a template from a function.\"\"\"\n        template = ResourceTemplate.from_function(\n            fn,\n            uri_template=uri_template,\n            name=name,\n            title=title,\n            description=description,\n            mime_type=mime_type,\n            icons=icons,\n            annotations=annotations,\n            meta=meta,\n        )\n        self._templates[template.uri_template] = template\n        return template\n\n    async def get_resource(\n        self,\n        uri: AnyUrl | str,\n        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,\n    ) -> Resource | None:\n        \"\"\"Get resource by URI, checking concrete resources first, then templates.\"\"\"\n        uri_str = str(uri)\n        logger.debug(\"Getting resource\", extra={\"uri\": uri_str})\n\n        # First check concrete resources\n        if resource := self._resources.get(uri_str):\n            return resource\n\n        # Then check templates\n        for template in self._templates.values():\n            if params := template.matches(uri_str):\n                try:\n                    return await template.create_resource(uri_str, params, context=context)\n                except Exception as e:  # pragma: no cover\n                    raise ValueError(f\"Error creating resource from template: {e}\")\n\n        raise ValueError(f\"Unknown resource: {uri}\")\n\n    def list_resources(self) -> list[Resource]:\n        \"\"\"List all registered resources.\"\"\"\n        logger.debug(\"Listing resources\", extra={\"count\": len(self._resources)})\n        return list(self._resources.values())\n\n    def list_templates(self) -> list[ResourceTemplate]:\n        \"\"\"List all registered templates.\"\"\"\n        logger.debug(\"Listing templates\", extra={\"count\": len(self._templates)})\n        return list(self._templates.values())",
  "start_line": 1,
  "end_line": 115,
  "total_lines": 115,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/resources/templates.py",
  "content": "\"\"\"Resource template functionality.\"\"\"\n\nfrom __future__ import annotations\n\nimport inspect\nimport re\nfrom collections.abc import Callable\nfrom typing import TYPE_CHECKING, Any\n\nfrom pydantic import BaseModel, Field, validate_call\n\nfrom mcp.server.fastmcp.resources.types import FunctionResource, Resource\nfrom mcp.server.fastmcp.utilities.context_injection import find_context_parameter, inject_context\nfrom mcp.server.fastmcp.utilities.func_metadata import func_metadata\nfrom mcp.types import Annotations, Icon\n\nif TYPE_CHECKING:\n    from mcp.server.fastmcp.server import Context\n    from mcp.server.session import ServerSessionT\n    from mcp.shared.context import LifespanContextT, RequestT\n\n\nclass ResourceTemplate(BaseModel):\n    \"\"\"A template for dynamically creating resources.\"\"\"\n\n    uri_template: str = Field(description=\"URI template with parameters (e.g. weather://{city}/current)\")\n    name: str = Field(description=\"Name of the resource\")\n    title: str | None = Field(description=\"Human-readable title of the resource\", default=None)\n    description: str | None = Field(description=\"Description of what the resource does\")\n    mime_type: str = Field(default=\"text/plain\", description=\"MIME type of the resource content\")\n    icons: list[Icon] | None = Field(default=None, description=\"Optional list of icons for the resource template\")\n    annotations: Annotations | None = Field(default=None, description=\"Optional annotations for the resource template\")\n    meta: dict[str, Any] | None = Field(default=None, description=\"Optional metadata for this resource template\")\n    fn: Callable[..., Any] = Field(exclude=True)\n    parameters: dict[str, Any] = Field(description=\"JSON schema for function parameters\")\n    context_kwarg: str | None = Field(None, description=\"Name of the kwarg that should receive context\")\n\n    @classmethod\n    def from_function(\n        cls,\n        fn: Callable[..., Any],\n        uri_template: str,\n        name: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        mime_type: str | None = None,\n        icons: list[Icon] | None = None,\n        annotations: Annotations | None = None,\n        meta: dict[str, Any] | None = None,\n        context_kwarg: str | None = None,\n    ) -> ResourceTemplate:\n        \"\"\"Create a template from a function.\"\"\"\n        func_name = name or fn.__name__\n        if func_name == \"<lambda>\":\n            raise ValueError(\"You must provide a name for lambda functions\")  # pragma: no cover\n\n        # Find context parameter if it exists\n        if context_kwarg is None:  # pragma: no branch\n            context_kwarg = find_context_parameter(fn)\n\n        # Get schema from func_metadata, excluding context parameter\n        func_arg_metadata = func_metadata(\n            fn,\n            skip_names=[context_kwarg] if context_kwarg is not None else [],\n        )\n        parameters = func_arg_metadata.arg_model.model_json_schema()\n\n        # ensure the arguments are properly cast\n        fn = validate_call(fn)\n\n        return cls(\n            uri_template=uri_template,\n            name=func_name,\n            title=title,\n            description=description or fn.__doc__ or \"\",\n            mime_type=mime_type or \"text/plain\",\n            icons=icons,\n            annotations=annotations,\n            meta=meta,\n            fn=fn,\n            parameters=parameters,\n            context_kwarg=context_kwarg,\n        )\n\n    def matches(self, uri: str) -> dict[str, Any] | None:\n        \"\"\"Check if URI matches template and extract parameters.\"\"\"\n        # Convert template to regex pattern\n        pattern = self.uri_template.replace(\"{\", \"(?P<\").replace(\"}\", \">[^/]+)\")\n        match = re.match(f\"^{pattern}$\", uri)\n        if match:\n            return match.groupdict()\n        return None\n\n    async def create_resource(\n        self,\n        uri: str,\n        params: dict[str, Any],\n        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,\n    ) -> Resource:\n        \"\"\"Create a resource from the template with the given parameters.\"\"\"\n        try:\n            # Add context to params if needed\n            params = inject_context(self.fn, params, context, self.context_kwarg)\n\n            # Call function and check if result is a coroutine\n            result = self.fn(**params)\n            if inspect.iscoroutine(result):\n                result = await result\n\n            return FunctionResource(\n                uri=uri,  # type: ignore\n                name=self.name,\n                title=self.title,\n                description=self.description,\n                mime_type=self.mime_type,\n                icons=self.icons,\n                annotations=self.annotations,\n                meta=self.meta,\n                fn=lambda: result,  # Capture result in closure\n            )\n        except Exception as e:\n            raise ValueError(f\"Error creating resource from template: {e}\")",
  "start_line": 1,
  "end_line": 122,
  "total_lines": 122,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Params: {'product_identifier': 'stick%20correcteur%20teint%C3%A9'}\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/resources/templates.py",
  "occurrences": 1,
  "strategy": "exact",
  "diff": "--- a/src/mcp/server/fastmcp/resources/templates.py\n+++ b/src/mcp/server/fastmcp/resources/templates.py\n@@ -6,6 +6,7 @@\n import re\n from collections.abc import Callable\n from typing import TYPE_CHECKING, Any\n+from urllib.parse import unquote\n \n from pydantic import BaseModel, Field, validate_call\n \n",
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/resources/templates.py",
  "occurrences": 1,
  "strategy": "unescaped",
  "diff": "--- a/src/mcp/server/fastmcp/resources/templates.py\n+++ b/src/mcp/server/fastmcp/resources/templates.py\n@@ -89,7 +89,7 @@\n         pattern = self.uri_template.replace(\"{\", \"(?P<\").replace(\"}\", \">[^/]+)\")\n         match = re.match(f\"^{pattern}$\", uri)\n         if match:\n-            return match.groupdict()\n+            return {k: unquote(v) for k, v in match.groupdict().items()}\n         return None\n \n     async def create_resource(\n",
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Params: {'product_identifier': 'stick correcteur teinté'}\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/lowlevel/server.py",
  "content": "    def read_resource(self):\n        def decorator(\n            func: Callable[[str], Awaitable[str | bytes | Iterable[ReadResourceContents]]],\n        ):\n            logger.debug(\"Registering handler for ReadResourceRequest\")\n\n            async def handler(req: types.ReadResourceRequest):\n                result = await func(req.params.uri)\n\n                def create_content(data: str | bytes, mime_type: str | None, meta: dict[str, Any] | None = None):\n                    # Note: ResourceContents uses Field(alias=\"_meta\"), so we must use the alias key\n                    meta_kwargs: dict[str, Any] = {\"_meta\": meta} if meta is not None else {}\n                    match data:\n                        case str() as data:",
  "start_line": 337,
  "end_line": 350,
  "total_lines": 808,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/server.py",
  "content": "\"\"\"FastMCP - A more ergonomic interface for MCP servers.\"\"\"\n\nfrom __future__ import annotations\n\nimport inspect\nimport re\nfrom collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence\nfrom contextlib import AbstractAsyncContextManager, asynccontextmanager\nfrom typing import Any, Generic, Literal\n\nimport anyio\nimport pydantic_core\nfrom pydantic import BaseModel\nfrom pydantic.networks import AnyUrl\nfrom pydantic_settings import BaseSettings, SettingsConfigDict\nfrom starlette.applications import Starlette\nfrom starlette.middleware import Middleware\nfrom starlette.middleware.authentication import AuthenticationMiddleware\nfrom starlette.requests import Request\nfrom starlette.responses import Response\nfrom starlette.routing import Mount, Route\nfrom starlette.types import Receive, Scope, Send\n\nfrom mcp.server.auth.middleware.auth_context import AuthContextMiddleware\nfrom mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware\nfrom mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier\nfrom mcp.server.auth.settings import AuthSettings\nfrom mcp.server.elicitation import ElicitationResult, ElicitSchemaModelT, UrlElicitationResult, elicit_with_validation\nfrom mcp.server.elicitation import elicit_url as _elicit_url\nfrom mcp.server.fastmcp.exceptions import ResourceError\nfrom mcp.server.fastmcp.prompts import Prompt, PromptManager\nfrom mcp.server.fastmcp.resources import FunctionResource, Resource, ResourceManager\nfrom mcp.server.fastmcp.tools import Tool, ToolManager\nfrom mcp.server.fastmcp.utilities.context_injection import find_context_parameter\nfrom mcp.server.fastmcp.utilities.logging import configure_logging, get_logger\nfrom mcp.server.lowlevel.helper_types import ReadResourceContents\nfrom mcp.server.lowlevel.server import LifespanResultT\nfrom mcp.server.lowlevel.server import Server as MCPServer\nfrom mcp.server.lowlevel.server import lifespan as default_lifespan\nfrom mcp.server.session import ServerSession, ServerSessionT\nfrom mcp.server.sse import SseServerTransport\nfrom mcp.server.stdio import stdio_server\nfrom mcp.server.streamable_http import EventStore\nfrom mcp.server.streamable_http_manager import StreamableHTTPSessionManager\nfrom mcp.server.transport_security import TransportSecuritySettings\nfrom mcp.shared.context import LifespanContextT, RequestContext, RequestT\nfrom mcp.types import Annotations, AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations\nfrom mcp.types import Prompt as MCPPrompt\nfrom mcp.types import PromptArgument as MCPPromptArgument\nfrom mcp.types import Resource as MCPResource\nfrom mcp.types import ResourceTemplate as MCPResourceTemplate\nfrom mcp.types import Tool as MCPTool\n\nlogger = get_logger(__name__)\n\n\nclass Settings(BaseSettings, Generic[LifespanResultT]):\n    \"\"\"FastMCP server settings.\n\n    All settings can be configured via environment variables with the prefix FASTMCP_.\n    For example, FASTMCP_DEBUG=true will set debug=True.\n    \"\"\"\n\n    model_config = SettingsConfigDict(\n        env_prefix=\"FASTMCP_\",\n        env_file=\".env\",\n        env_nested_delimiter=\"__\",\n        nested_model_default_partial_update=True,\n        extra=\"ignore\",\n    )\n\n    # Server settings\n    debug: bool\n    log_level: Literal[\"DEBUG\", \"INFO\", \"WARNING\", \"ERROR\", \"CRITICAL\"]\n\n    # HTTP settings\n    host: str\n    port: int\n    sse_path: str\n    message_path: str\n    streamable_http_path: str\n\n    # StreamableHTTP settings\n    json_response: bool\n    stateless_http: bool\n    \"\"\"Define if the server should create a new transport per request.\"\"\"\n\n    # resource settings\n    warn_on_duplicate_resources: bool\n\n    # tool settings\n    warn_on_duplicate_tools: bool\n\n    # prompt settings\n    warn_on_duplicate_prompts: bool\n\n    lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None\n    \"\"\"A async context manager that will be called when the server is started.\"\"\"\n\n    auth: AuthSettings | None\n\n    # Transport security settings (DNS rebinding protection)\n    transport_security: TransportSecuritySettings | None\n\n\ndef lifespan_wrapper(\n    app: FastMCP[LifespanResultT],\n    lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]],\n) -> Callable[[MCPServer[LifespanResultT, Request]], AbstractAsyncContextManager[LifespanResultT]]:\n    @asynccontextmanager\n    async def wrap(\n        _: MCPServer[LifespanResultT, Request],\n    ) -> AsyncIterator[LifespanResultT]:\n        async with lifespan(app) as context:\n            yield context\n\n    return wrap\n\n\nclass FastMCP(Generic[LifespanResultT]):\n    def __init__(  # noqa: PLR0913\n        self,\n        name: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        instructions: str | None = None,\n        website_url: str | None = None,\n        icons: list[Icon] | None = None,\n        version: str | None = None,\n        auth_server_provider: (OAuthAuthorizationServerProvider[Any, Any, Any] | None) = None,\n        token_verifier: TokenVerifier | None = None,\n        event_store: EventStore | None = None,\n        retry_interval: int | None = None,\n        *,\n        tools: list[Tool] | None = None,\n        debug: bool = False,\n        log_level: Literal[\"DEBUG\", \"INFO\", \"WARNING\", \"ERROR\", \"CRITICAL\"] = \"INFO\",\n        host: str = \"127.0.0.1\",\n        port: int = 8000,\n        sse_path: str = \"/sse\",\n        message_path: str = \"/messages/\",\n        streamable_http_path: str = \"/mcp\",\n        json_response: bool = False,\n        stateless_http: bool = False,\n        warn_on_duplicate_resources: bool = True,\n        warn_on_duplicate_tools: bool = True,\n        warn_on_duplicate_prompts: bool = True,\n        lifespan: (Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None) = None,\n        auth: AuthSettings | None = None,\n        transport_security: TransportSecuritySettings | None = None,",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 1298,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "src/mcp/server/fastmcp/resources/templates.py:9:from urllib.parse import unquote\nsrc/mcp/server/fastmcp/resources/templates.py:92:            return {k: unquote(v) for k, v in match.groupdict().items()}\nsrc/mcp/server/auth/middleware/client_auth.py:6:from urllib.parse import unquote\nsrc/mcp/server/auth/middleware/client_auth.py:79:                basic_client_id = unquote(basic_client_id)\nsrc/mcp/server/auth/middleware/client_auth.py:80:                request_client_secret = unquote(request_client_secret)\nsrc/mcp/client/auth/utils.py:33:    # Pattern matches: field_name=\"value\" or field_name=value (unquoted)\nsrc/mcp/client/auth/utils.py:38:        # Return quoted value if present, otherwise unquoted value\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "tests\ntests/test_examples.py\ntests/shared\ntests/shared/test_win32_utils.py\ntests/shared/test_auth.py\ntests/shared/test_exceptions.py\ntests/shared/test_sse.py\ntests/shared/test_httpx_utils.py\ntests/shared/test_streamable_http.py\ntests/shared/__init__.py\ntests/shared/test_tool_name_validation.py\ntests/shared/test_progress_notifications.py\ntests/shared/test_memory.py\ntests/shared/test_auth_utils.py\ntests/shared/test_session.py\ntests/shared/test_ws.py\ntests/cli\ntests/cli/__init__.py\ntests/cli/test_utils.py\ntests/experimental\ntests/experimental/tasks\ntests/experimental/__init__.py\ntests/server\ntests/server/fastmcp\ntests/server/test_cancel_handling.py\ntests/server/test_stateless_mode.py\ntests/server/test_streamable_http_security.py\ntests/server/test_lifespan.py\ntests/server/auth\ntests/server/test_lowlevel_input_validation.py\ntests/server/test_streamable_http_manager.py\ntests/server/lowlevel\ntests/server/test_read_resource.py\ntests/server/test_sse_security.py\ntests/server/test_lowlevel_output_validation.py\ntests/server/test_lowlevel_tool_annotations.py\ntests/server/__init__.py\ntests/server/test_stdio.py\ntests/server/test_lowlevel_exception_handling.py\ntests/server/test_session_race_condition.py\ntests/server/test_session.py\ntests/server/test_completion_with_context.py\ntests/server/test_validation.py\ntests/__init__.py\ntests/test_types.py\ntests/issues\ntests/issues/test_100_tool_listing.py\ntests/issues/test_1027_win_unreachable_cleanup.py\ntests/issues/test_88_random_error.py\ntests/issues/test_176_progress_token.py\ntests/issues/test_152_resource_mime_type.py\ntests/issues/test_1754_mime_type_parameters.py\ntests/issues/test_552_windows_hang.py\ntests/issues/test_1574_resource_uri_validation.py\ntests/issues/test_malformed_input.py\ntests/issues/test_129_resource_templates.py\ntests/issues/test_1338_icons_and_metadata.py\ntests/issues/test_188_concurrency.py\ntests/issues/test_192_request_id.py\ntests/issues/test_355_type_error.py\ntests/issues/test_141_resource_templates.py\ntests/issues/test_342_base64_encoding.py\ntests/issues/test_1363_race_condition_streamable_http.py\ntests/client\ntests/client/test_sampling_callback.py\ntests/client/auth\ntests/client/test_resource_cleanup.py\ntests/client/test_list_roots_callback.py\ntests/client/test_auth.py\ntests/client/test_session_group.py\ntests/client/test_scope_bug_1630.py\ntests/client/test_config.py\ntests/client/test_output_schema_validation.py\ntests/client/test_list_methods_cursor.py\ntests/client/__init__.py\ntests/client/test_stdio.py\ntests/client/test_http_unicode.py\ntests/client/conftest.py\ntests/client/test_logging_callback.py\ntests/client/test_notification_response.py\ntests/client/test_session.py\ntests/test_helpers.py\ntests/conftest.py\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "tests/issues/test_129_resource_templates.py",
  "content": "import pytest\n\nfrom mcp import types\nfrom mcp.server.fastmcp import FastMCP\n\n\n@pytest.mark.anyio\nasync def test_resource_templates():\n    # Create an MCP server\n    mcp = FastMCP(\"Demo\")\n\n    # Add a dynamic greeting resource\n    @mcp.resource(\"greeting://{name}\")\n    def get_greeting(name: str) -> str:  # pragma: no cover\n        \"\"\"Get a personalized greeting\"\"\"\n        return f\"Hello, {name}!\"\n\n    @mcp.resource(\"users://{user_id}/profile\")\n    def get_user_profile(user_id: str) -> str:  # pragma: no cover\n        \"\"\"Dynamic user data\"\"\"\n        return f\"Profile data for user {user_id}\"\n\n    # Get the list of resource templates using the underlying server\n    # Note: list_resource_templates() returns a decorator that wraps the handler\n    # The handler returns a ServerResult with a ListResourceTemplatesResult inside\n    result = await mcp._mcp_server.request_handlers[types.ListResourceTemplatesRequest](\n        types.ListResourceTemplatesRequest(params=None)\n    )\n    assert isinstance(result.root, types.ListResourceTemplatesResult)\n    templates = result.root.resourceTemplates\n\n    # Verify we get both templates back\n    assert len(templates) == 2\n\n    # Verify template details\n    greeting_template = next(t for t in templates if t.name == \"get_greeting\")  # pragma: no cover\n    assert greeting_template.uriTemplate == \"greeting://{name}\"\n    assert greeting_template.description == \"Get a personalized greeting\"\n\n    profile_template = next(t for t in templates if t.name == \"get_user_profile\")  # pragma: no cover\n    assert profile_template.uriTemplate == \"users://{user_id}/profile\"\n    assert profile_template.description == \"Dynamic user data\"",
  "start_line": 1,
  "end_line": 42,
  "total_lines": 42,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "tests/issues/test_141_resource_templates.py",
  "content": "import pytest\nfrom pydantic import AnyUrl\n\nfrom mcp.server.fastmcp import FastMCP\nfrom mcp.shared.memory import (\n    create_connected_server_and_client_session as client_session,\n)\nfrom mcp.types import (\n    ListResourceTemplatesResult,\n    TextResourceContents,\n)\n\n\n@pytest.mark.anyio\nasync def test_resource_template_edge_cases():\n    \"\"\"Test server-side resource template validation\"\"\"\n    mcp = FastMCP(\"Demo\")\n\n    # Test case 1: Template with multiple parameters\n    @mcp.resource(\"resource://users/{user_id}/posts/{post_id}\")\n    def get_user_post(user_id: str, post_id: str) -> str:\n        return f\"Post {post_id} by user {user_id}\"\n\n    # Test case 2: Template with optional parameter (should fail)\n    with pytest.raises(ValueError, match=\"Mismatch between URI parameters\"):\n\n        @mcp.resource(\"resource://users/{user_id}/profile\")\n        def get_user_profile(user_id: str, optional_param: str | None = None) -> str:  # pragma: no cover\n            return f\"Profile for user {user_id}\"\n\n    # Test case 3: Template with mismatched parameters\n    with pytest.raises(ValueError, match=\"Mismatch between URI parameters\"):\n\n        @mcp.resource(\"resource://users/{user_id}/profile\")\n        def get_user_profile_mismatch(different_param: str) -> str:  # pragma: no cover\n            return f\"Profile for user {different_param}\"\n\n    # Test case 4: Template with extra function parameters\n    with pytest.raises(ValueError, match=\"Mismatch between URI parameters\"):\n\n        @mcp.resource(\"resource://users/{user_id}/profile\")\n        def get_user_profile_extra(user_id: str, extra_param: str) -> str:  # pragma: no cover\n            return f\"Profile for user {user_id}\"\n\n    # Test case 5: Template with missing function parameters\n    with pytest.raises(ValueError, match=\"Mismatch between URI parameters\"):\n\n        @mcp.resource(\"resource://users/{user_id}/profile/{section}\")\n        def get_user_profile_missing(user_id: str) -> str:  # pragma: no cover\n            return f\"Profile for user {user_id}\"\n\n    # Verify valid template works\n    result = await mcp.read_resource(\"resource://users/123/posts/456\")\n    result_list = list(result)\n    assert len(result_list) == 1\n    assert result_list[0].content == \"Post 456 by user 123\"\n    assert result_list[0].mime_type == \"text/plain\"\n\n    # Verify invalid parameters raise error\n    with pytest.raises(ValueError, match=\"Unknown resource\"):\n        await mcp.read_resource(\"resource://users/123/posts\")  # Missing post_id\n\n    with pytest.raises(ValueError, match=\"Unknown resource\"):\n        await mcp.read_resource(\"resource://users/123/posts/456/extra\")  # Extra path component\n\n\n@pytest.mark.anyio\nasync def test_resource_template_client_interaction():\n    \"\"\"Test client-side resource template interaction\"\"\"\n    mcp = FastMCP(\"Demo\")\n\n    # Register some templated resources\n    @mcp.resource(\"resource://users/{user_id}/posts/{post_id}\")\n    def get_user_post(user_id: str, post_id: str) -> str:\n        return f\"Post {post_id} by user {user_id}\"\n\n    @mcp.resource(\"resource://users/{user_id}/profile\")\n    def get_user_profile(user_id: str) -> str:\n        return f\"Profile for user {user_id}\"\n\n    async with client_session(mcp._mcp_server) as session:\n        # Initialize the session\n        await session.initialize()\n\n        # List available resources\n        resources = await session.list_resource_templates()\n        assert isinstance(resources, ListResourceTemplatesResult)\n        assert len(resources.resourceTemplates) == 2\n\n        # Verify resource templates are listed correctly\n        templates = [r.uriTemplate for r in resources.resourceTemplates]\n        assert \"resource://users/{user_id}/posts/{post_id}\" in templates\n        assert \"resource://users/{user_id}/profile\" in templates\n\n        # Read a resource with valid parameters\n        result = await session.read_resource(AnyUrl(\"resource://users/123/posts/456\"))\n        contents = result.contents[0]\n        assert isinstance(contents, TextResourceContents)\n        assert contents.text == \"Post 456 by user 123\"\n        assert contents.mimeType == \"text/plain\"\n\n        # Read another resource with valid parameters\n        result = await session.read_resource(AnyUrl(\"resource://users/789/profile\"))\n        contents = result.contents[0]\n        assert isinstance(contents, TextResourceContents)\n        assert contents.text == \"Profile for user 789\"\n        assert contents.mimeType == \"text/plain\"\n\n        # Verify invalid resource URIs raise appropriate errors\n        with pytest.raises(Exception):  # Specific exception type may vary\n            await session.read_resource(AnyUrl(\"resource://users/123/posts\"))  # Missing post_id\n\n        with pytest.raises(Exception):  # Specific exception type may vary\n            await session.read_resource(AnyUrl(\"resource://users/123/invalid\"))  # Invalid template",
  "start_line": 1,
  "end_line": 114,
  "total_lines": 114,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "bringing up nodes...\n48 workers [3 items]\n\nscheduling tests via LoadScheduling\n\ntests/issues/test_129_resource_templates.py::test_resource_templates \ntests/issues/test_141_resource_templates.py::test_resource_template_client_interaction \ntests/issues/test_141_resource_templates.py::test_resource_template_edge_cases \n[gw2]\u001b[36m [ 33%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/issues/test_141_resource_templates.py::test_resource_template_client_interaction \n[gw0]\u001b[36m [ 66%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/issues/test_129_resource_templates.py::test_resource_templates \n[gw1]\u001b[36m [100%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/issues/test_141_resource_templates.py::test_resource_template_edge_cases \n\n\u001b[92m═══════════════════════════════ \u001b[0m\u001b[34minline-snapshot\u001b[0m\u001b[92m ════════════════════════════════\u001b[0m\nINFO: inline-snapshot was disabled because you used xdist. This means that tests\nwith snapshots will continue to run, but snapshot(x) will only return x and \ninline-snapshot will not be able to fix snapshots or generate reports.\n\n\n\n\u001b[32m============================== \u001b[32m\u001b[1m3 passed\u001b[0m\u001b[32m in 24.55s\u001b[0m\u001b[32m ==============================\u001b[0m\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/resources/templates.py",
  "content": "\"\"\"Resource template functionality.\"\"\"\n\nfrom __future__ import annotations\n\nimport inspect\nimport re\nfrom collections.abc import Callable\nfrom typing import TYPE_CHECKING, Any\nfrom urllib.parse import unquote\n\nfrom pydantic import BaseModel, Field, validate_call\n\nfrom mcp.server.fastmcp.resources.types import FunctionResource, Resource\nfrom mcp.server.fastmcp.utilities.context_injection import find_context_parameter, inject_context\nfrom mcp.server.fastmcp.utilities.func_metadata import func_metadata\nfrom mcp.types import Annotations, Icon\n\nif TYPE_CHECKING:\n    from mcp.server.fastmcp.server import Context\n    from mcp.server.session import ServerSessionT\n    from mcp.shared.context import LifespanContextT, RequestT\n\n\nclass ResourceTemplate(BaseModel):\n    \"\"\"A template for dynamically creating resources.\"\"\"\n\n    uri_template: str = Field(description=\"URI template with parameters (e.g. weather://{city}/current)\")\n    name: str = Field(description=\"Name of the resource\")\n    title: str | None = Field(description=\"Human-readable title of the resource\", default=None)\n    description: str | None = Field(description=\"Description of what the resource does\")\n    mime_type: str = Field(default=\"text/plain\", description=\"MIME type of the resource content\")\n    icons: list[Icon] | None = Field(default=None, description=\"Optional list of icons for the resource template\")\n    annotations: Annotations | None = Field(default=None, description=\"Optional annotations for the resource template\")\n    meta: dict[str, Any] | None = Field(default=None, description=\"Optional metadata for this resource template\")\n    fn: Callable[..., Any] = Field(exclude=True)\n    parameters: dict[str, Any] = Field(description=\"JSON schema for function parameters\")\n    context_kwarg: str | None = Field(None, description=\"Name of the kwarg that should receive context\")\n\n    @classmethod\n    def from_function(\n        cls,\n        fn: Callable[..., Any],\n        uri_template: str,\n        name: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        mime_type: str | None = None,\n        icons: list[Icon] | None = None,\n        annotations: Annotations | None = None,\n        meta: dict[str, Any] | None = None,\n        context_kwarg: str | None = None,\n    ) -> ResourceTemplate:\n        \"\"\"Create a template from a function.\"\"\"\n        func_name = name or fn.__name__\n        if func_name == \"<lambda>\":\n            raise ValueError(\"You must provide a name for lambda functions\")  # pragma: no cover\n\n        # Find context parameter if it exists\n        if context_kwarg is None:  # pragma: no branch\n            context_kwarg = find_context_parameter(fn)\n\n        # Get schema from func_metadata, excluding context parameter\n        func_arg_metadata = func_metadata(\n            fn,\n            skip_names=[context_kwarg] if context_kwarg is not None else [],\n        )\n        parameters = func_arg_metadata.arg_model.model_json_schema()\n\n        # ensure the arguments are properly cast\n        fn = validate_call(fn)\n\n        return cls(\n            uri_template=uri_template,\n            name=func_name,\n            title=title,\n            description=description or fn.__doc__ or \"\",\n            mime_type=mime_type or \"text/plain\",\n            icons=icons,\n            annotations=annotations,\n            meta=meta,\n            fn=fn,\n            parameters=parameters,\n            context_kwarg=context_kwarg,\n        )\n\n    def matches(self, uri: str) -> dict[str, Any] | None:\n        \"\"\"Check if URI matches template and extract parameters.\"\"\"\n        # Convert template to regex pattern\n        pattern = self.uri_template.replace(\"{\", \"(?P<\").replace(\"}\", \">[^/]+)\")\n        match = re.match(f\"^{pattern}$\", uri)\n        if match:\n            return {k: unquote(v) for k, v in match.groupdict().items()}\n        return None\n\n    async def create_resource(\n        self,\n        uri: str,\n        params: dict[str, Any],\n        context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,\n    ) -> Resource:\n        \"\"\"Create a resource from the template with the given parameters.\"\"\"\n        try:\n            # Add context to params if needed\n            params = inject_context(self.fn, params, context, self.context_kwarg)\n\n            # Call function and check if result is a coroutine\n            result = self.fn(**params)\n            if inspect.iscoroutine(result):\n                result = await result\n\n            return FunctionResource(\n                uri=uri,  # type: ignore\n                name=self.name,\n                title=self.title,\n                description=self.description,\n                mime_type=self.mime_type,\n                icons=self.icons,\n                annotations=self.annotations,\n                meta=self.meta,\n                fn=lambda: result,  # Capture result in closure\n            )\n        except Exception as e:\n            raise ValueError(f\"Error creating resource from template: {e}\")",
  "start_line": 1,
  "end_line": 123,
  "total_lines": 123,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Requested URI: products://search/my%20product%20name\nReceived parameter: my product name\nSUCCESS: Parameter was decoded\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": " M src/mcp/server/fastmcp/resources/templates.py\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "src/mcp/shared/metadata_utils.py:7:from mcp.types import Implementation, Prompt, Resource, ResourceTemplate, Tool\nsrc/mcp/shared/metadata_utils.py:10:def get_display_name(obj: Tool | Resource | Prompt | ResourceTemplate | Implementation) -> str:\nsrc/mcp/types.py:716:class ResourceTemplate(BaseMetadata):\nsrc/mcp/types.py:747:class ListResourceTemplatesRequest(PaginatedRequest[Literal[\"resources/templates/list\"]]):\nsrc/mcp/types.py:753:class ListResourceTemplatesResult(PaginatedResult):\nsrc/mcp/types.py:756:    resourceTemplates: list[ResourceTemplate]\nsrc/mcp/types.py:1480:class ResourceTemplateReference(MCPModel):\nsrc/mcp/types.py:1515:    ref: ResourceTemplateReference | PromptReference\nsrc/mcp/types.py:1671:    | ListResourceTemplatesRequest\nsrc/mcp/types.py:1850:    | ListResourceTemplatesResult\nsrc/mcp/server/fastmcp/resources/templates.py:24:class ResourceTemplate(BaseModel):\nsrc/mcp/server/fastmcp/resources/templates.py:52:    ) -> ResourceTemplate:\nsrc/mcp/server/fastmcp/resources/__init__.py:3:from .templates import ResourceTemplate\nsrc/mcp/server/fastmcp/resources/__init__.py:21:    \"ResourceTemplate\",\nsrc/mcp/server/fastmcp/resources/resource_manager.py:11:from mcp.server.fastmcp.resources.templates import ResourceTemplate\nsrc/mcp/server/fastmcp/resources/resource_manager.py:28:        self._templates: dict[str, ResourceTemplate] = {}\nsrc/mcp/server/fastmcp/resources/resource_manager.py:68:    ) -> ResourceTemplate:\nsrc/mcp/server/fastmcp/resources/resource_manager.py:70:        template = ResourceTemplate.from_function(\nsrc/mcp/server/fastmcp/resources/resource_manager.py:112:    def list_templates(self) -> list[ResourceTemplate]:\nsrc/mcp/server/fastmcp/server.py:51:from mcp.types import ResourceTemplate as MCPResourceTemplate\nsrc/mcp/server/fastmcp/server.py:351:    async def list_resource_templates(self) -> list[MCPResourceTemplate]:\nsrc/mcp/server/fastmcp/server.py:354:            MCPResourceTemplate(\nsrc/mcp/server/fastmcp/server.py:497:        - ref: PromptReference or ResourceTemplateReference\nsrc/mcp/server/fastmcp/server.py:504:                if isinstance(ref, ResourceTemplateReference):\nsrc/mcp/server/lowlevel/server.py:34:   async def handle_list_resource_templates() -> list[types.ResourceTemplate]:\nsrc/mcp/server/lowlevel/server.py:325:        def decorator(func: Callable[[], Awaitable[list[types.ResourceTemplate]]]):\nsrc/mcp/server/lowlevel/server.py:326:            logger.debug(\"Registering handler for ListResourceTemplatesRequest\")\nsrc/mcp/server/lowlevel/server.py:330:                return types.ServerResult(types.ListResourceTemplatesResult(resourceTemplates=templates))\nsrc/mcp/server/lowlevel/server.py:332:            self.request_handlers[types.ListResourceTemplatesRequest] = handler\nsrc/mcp/server/lowlevel/server.py:621:                    types.PromptReference | types.ResourceTemplateReference,\nsrc/mcp/client/session.py:271:    ) -> types.ListResourceTemplatesResult:\nsrc/mcp/client/session.py:278:            types.ClientRequest(types.ListResourceTemplatesRequest(params=params)),\nsrc/mcp/client/session.py:279:            types.ListResourceTemplatesResult,\nsrc/mcp/client/session.py:384:        ref: types.ResourceTemplateReference | types.PromptReference,\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "src/mcp/server/fastmcp/resources/templates.py:9:from urllib.parse import unquote\nsrc/mcp/server/fastmcp/resources/templates.py:92:            return {k: unquote(v) for k, v in match.groupdict().items()}\nsrc/mcp/server/auth/middleware/client_auth.py:6:from urllib.parse import unquote\nsrc/mcp/server/auth/middleware/client_auth.py:79:                basic_client_id = unquote(basic_client_id)\nsrc/mcp/server/auth/middleware/client_auth.py:80:                request_client_secret = unquote(request_client_secret)\nsrc/mcp/client/auth/utils.py:33:    # Pattern matches: field_name=\"value\" or field_name=value (unquoted)\nsrc/mcp/client/auth/utils.py:38:        # Return quoted value if present, otherwise unquoted value\n",
  "stderr": "",
  "exit_code": 0,
  "budget_warning": "Only 10 tool call(s) remaining (20/30 used). Finalize your edits and call submit_patch soon."
}

{
  "status": "ok",
  "stdout": "diff --git a/src/mcp/server/fastmcp/resources/templates.py b/src/mcp/server/fastmcp/resources/templates.py\nindex 89a8ceb..ba96694 100644\n--- a/src/mcp/server/fastmcp/resources/templates.py\n+++ b/src/mcp/server/fastmcp/resources/templates.py\n@@ -6,6 +6,7 @@ import inspect\n import re\n from collections.abc import Callable\n from typing import TYPE_CHECKING, Any\n+from urllib.parse import unquote\n \n from pydantic import BaseModel, Field, validate_call\n \n@@ -88,7 +89,7 @@ class ResourceTemplate(BaseModel):\n         pattern = self.uri_template.replace(\"{\", \"(?P<\").replace(\"}\", \">[^/]+)\")\n         match = re.match(f\"^{pattern}$\", uri)\n         if match:\n-            return match.groupdict()\n+            return {k: unquote(v) for k, v in match.groupdict().items()}\n         return None\n \n     async def create_resource(\n",
  "stderr": "",
  "exit_code": 0,
  "budget_warning": "Only 9 tool call(s) remaining (21/30 used). Finalize your edits and call submit_patch soon."
}

{
  "status": "ok",
  "patch_size": 820,
  "files_changed": 1
}