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/server.py:367:    async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:\nsrc/mcp/server/fastmcp/server.py:511:    def add_resource(self, resource: Resource) -> None:\nsrc/mcp/server/fastmcp/server.py:519:    def resource(\nsrc/mcp/server/fastmcp/server.py:1119:    async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContents]:\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\nsrc/mcp/client/session.py:282:    async def read_resource(self, uri: str | AnyUrl) -> types.ReadResourceResult:\nsrc/mcp/client/session.py:289:    async def subscribe_resource(self, uri: str | AnyUrl) -> types.EmptyResult:\nsrc/mcp/client/session.py:296:    async def unsubscribe_resource(self, uri: str | AnyUrl) -> types.EmptyResult:\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/server.py",
  "content": "\n    def add_resource(self, resource: Resource) -> None:\n        \"\"\"Add a resource to the server.\n\n        Args:\n            resource: A Resource instance to add\n        \"\"\"\n        self._resource_manager.add_resource(resource)\n\n    def resource(\n        self,\n        uri: str,\n        *,\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    ) -> Callable[[AnyFunction], AnyFunction]:\n        \"\"\"Decorator to register a function as a resource.\n\n        The function will be called when the resource is read to generate its content.\n        The function can return:\n        - str for text content\n        - bytes for binary content\n        - other types will be converted to JSON\n\n        If the URI contains parameters (e.g. \"resource://{param}\") or the function\n        has parameters, it will be registered as a template resource.\n\n        Args:\n            uri: URI for the resource (e.g. \"resource://my-resource\" or \"resource://{param}\")\n            name: Optional name for the resource\n            title: Optional human-readable title for the resource\n            description: Optional description of the resource\n            mime_type: Optional MIME type for the resource\n            meta: Optional metadata dictionary for the resource\n\n        Example:",
  "start_line": 510,
  "end_line": 550,
  "total_lines": 1298,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/server.py",
  "content": "            @server.resource(\"resource://my-resource\")\n            def get_data() -> str:\n                return \"Hello, world!\"\n\n            @server.resource(\"resource://my-resource\")\n            async get_data() -> str:\n                data = await fetch_data()\n                return f\"Hello, world! {data}\"\n\n            @server.resource(\"resource://{city}/weather\")\n            def get_weather(city: str) -> str:\n                return f\"Weather for {city}\"\n\n            @server.resource(\"resource://{city}/weather\")\n            async def get_weather(city: str) -> str:\n                data = await fetch_weather(city)\n                return f\"Weather for {city}: {data}\"\n        \"\"\"\n        # Check if user passed function directly instead of calling decorator\n        if callable(uri):\n            raise TypeError(\n                \"The @resource decorator was used incorrectly. \"\n                \"Did you forget to call it? Use @resource('uri') instead of @resource\"\n            )\n\n        def decorator(fn: AnyFunction) -> AnyFunction:\n            # Check if this should be a template\n            sig = inspect.signature(fn)\n            has_uri_params = \"{\" in uri and \"}\" in uri\n            has_func_params = bool(sig.parameters)\n\n            if has_uri_params or has_func_params:\n                # Check for Context parameter to exclude from validation\n                context_param = find_context_parameter(fn)\n\n                # Validate that URI params match function params (excluding context)\n                uri_params = set(re.findall(r\"{(\\w+)}\", uri))\n                # We need to remove the context_param from the resource function if\n                # there is any.\n                func_params = {p for p in sig.parameters.keys() if p != context_param}\n\n                if uri_params != func_params:\n                    raise ValueError(\n                        f\"Mismatch between URI parameters {uri_params} and function parameters {func_params}\"\n                    )\n\n                # Register as template\n                self._resource_manager.add_template(\n                    fn=fn,\n                    uri_template=uri,",
  "start_line": 551,
  "end_line": 600,
  "total_lines": 1298,
  "is_truncated": false
}

{
  "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": "URI: products://search/stick%20correcteur%20teint%C3%A9%20anti-imperfections%20rechargeable%20au%20zinc\nResult: ReadResourceContents(content='Searching for: stick%20correcteur%20teint%C3%A9%20anti-imperfections%20rechargeable%20au%20zinc', mime_type='text/plain', meta=None)\n",
  "stderr": "",
  "exit_code": 0
}

{
  "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@@ -88,7 +88,8 @@\n         pattern = self.uri_template.replace(\"{\", \"(?P<\").replace(\"}\", \">[^/]+)\")\n         match = re.match(f\"^{pattern}$\", uri)\n         if match:\n-            return match.groupdict()\n+            params = match.groupdict()\n+            return {k: unquote(v) for k, v in params.items()}\n         return None\n \n     async def create_resource(\n",
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "URI: products://search/stick%20correcteur%20teint%C3%A9%20anti-imperfections%20rechargeable%20au%20zinc\nError: name 'unquote' is not defined\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\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",
  "start_line": 1,
  "end_line": 20,
  "total_lines": 123,
  "is_truncated": false
}

{
  "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",
  "stdout": "URI: products://search/stick%20correcteur%20teint%C3%A9%20anti-imperfections%20rechargeable%20au%20zinc\nResult: ReadResourceContents(content='Searching for: stick correcteur teinté anti-imperfections rechargeable au zinc', mime_type='text/plain', meta=None)\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/lowlevel/server.py",
  "content": "\"\"\"\nMCP Server Module\n\nThis module provides a framework for creating an MCP (Model Context Protocol) server.\nIt allows you to easily define and handle various types of requests and notifications\nin an asynchronous manner.\n\nUsage:\n1. Create a Server instance:\n   server = Server(\"your_server_name\")\n\n2. Define request handlers using decorators:\n   @server.list_prompts()\n   async def handle_list_prompts(request: types.ListPromptsRequest) -> types.ListPromptsResult:\n       # Implementation\n\n   @server.get_prompt()\n   async def handle_get_prompt(\n       name: str, arguments: dict[str, str] | None\n   ) -> types.GetPromptResult:\n       # Implementation\n\n   @server.list_tools()\n   async def handle_list_tools(request: types.ListToolsRequest) -> types.ListToolsResult:\n       # Implementation\n\n   @server.call_tool()\n   async def handle_call_tool(\n       name: str, arguments: dict | None\n   ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:\n       # Implementation\n\n   @server.list_resource_templates()\n   async def handle_list_resource_templates() -> list[types.ResourceTemplate]:\n       # Implementation\n\n3. Define notification handlers if needed:\n   @server.progress_notification()\n   async def handle_progress(\n       progress_token: str | int, progress: float, total: float | None,\n       message: str | None\n   ) -> None:\n       # Implementation\n\n4. Run the server:\n   async def main():\n       async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):\n           await server.run(\n               read_stream,\n               write_stream,\n               InitializationOptions(\n                   server_name=\"your_server_name\",\n                   server_version=\"your_version\",\n                   capabilities=server.get_capabilities(\n                       notification_options=NotificationOptions(),\n                       experimental_capabilities={},\n                   ),\n               ),\n           )\n\n   asyncio.run(main())\n\nThe Server class provides methods to register handlers for various MCP requests and\nnotifications. It automatically manages the request context and handles incoming\nmessages from the client.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport base64\nimport contextvars\nimport json\nimport logging\nimport warnings\nfrom collections.abc import AsyncIterator, Awaitable, Callable, Iterable\nfrom contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager\nfrom typing import Any, Generic, TypeAlias, cast\n\nimport anyio\nimport jsonschema\nfrom anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream\nfrom typing_extensions import TypeVar\n\nimport mcp.types as types\nfrom mcp.server.experimental.request_context import Experimental\nfrom mcp.server.lowlevel.experimental import ExperimentalHandlers\nfrom mcp.server.lowlevel.func_inspection import create_call_wrapper\nfrom mcp.server.lowlevel.helper_types import ReadResourceContents\nfrom mcp.server.models import InitializationOptions\nfrom mcp.server.session import ServerSession\nfrom mcp.shared.context import RequestContext\nfrom mcp.shared.exceptions import McpError, UrlElicitationRequiredError\nfrom mcp.shared.message import ServerMessageMetadata, SessionMessage\nfrom mcp.shared.session import RequestResponder\nfrom mcp.shared.tool_name_validation import validate_and_warn_tool_name\n\nlogger = logging.getLogger(__name__)\n\nLifespanResultT = TypeVar(\"LifespanResultT\", default=Any)\nRequestT = TypeVar(\"RequestT\", default=Any)\n\n# type aliases for tool call results\nStructuredContent: TypeAlias = dict[str, Any]\nUnstructuredContent: TypeAlias = Iterable[types.ContentBlock]\nCombinationContent: TypeAlias = tuple[UnstructuredContent, StructuredContent]\n\n# This will be properly typed in each Server instance's context\nrequest_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any, Any]] = contextvars.ContextVar(\"request_ctx\")\n\n\nclass NotificationOptions:\n    def __init__(\n        self,\n        prompts_changed: bool = False,\n        resources_changed: bool = False,\n        tools_changed: bool = False,\n    ):\n        self.prompts_changed = prompts_changed\n        self.resources_changed = resources_changed\n        self.tools_changed = tools_changed\n\n\n@asynccontextmanager\nasync def lifespan(_: Server[LifespanResultT, RequestT]) -> AsyncIterator[dict[str, Any]]:\n    \"\"\"Default lifespan context manager that does nothing.\n\n    Args:\n        server: The server instance this lifespan is managing\n\n    Returns:\n        An empty context object\n    \"\"\"\n    yield {}\n\n\nclass Server(Generic[LifespanResultT, RequestT]):\n    def __init__(\n        self,\n        name: str,\n        version: 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[types.Icon] | None = None,\n        lifespan: Callable[\n            [Server[LifespanResultT, RequestT]],\n            AbstractAsyncContextManager[LifespanResultT],\n        ] = lifespan,\n    ):",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 808,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "337:    def read_resource(self):\n368:                            \"Returning str or bytes from read_resource is deprecated. \"\n387:                        raise ValueError(f\"Unexpected return type from read_resource: {type(result)}\")\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/lowlevel/server.py",
  "content": "                return types.ServerResult(types.ListResourceTemplatesResult(resourceTemplates=templates))\n\n            self.request_handlers[types.ListResourceTemplatesRequest] = handler\n            return func\n\n        return decorator\n\n    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:\n                            return types.TextResourceContents(\n                                uri=req.params.uri,\n                                text=data,\n                                mimeType=mime_type or \"text/plain\",\n                                **meta_kwargs,\n                            )\n                        case bytes() as data:  # pragma: no cover\n                            return types.BlobResourceContents(\n                                uri=req.params.uri,\n                                blob=base64.b64encode(data).decode(),",
  "start_line": 330,
  "end_line": 360,
  "total_lines": 808,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "src/mcp/server/fastmcp/resources/templates.py:87:        \"\"\"Check if URI matches template and extract parameters.\"\"\"\nsrc/mcp/server/fastmcp/resources/resource_manager.py:99:            if params := template.matches(uri_str):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "298:        self._mcp_server.list_resource_templates()(self.list_resource_templates)\n351:    async def list_resource_templates(self) -> list[MCPResourceTemplate]:\n352:        templates = self._resource_manager.list_templates()\n355:                uriTemplate=template.uri_template,\n356:                name=template.name,\n357:                title=template.title,\n358:                description=template.description,\n359:                mimeType=template.mime_type,\n360:                icons=template.icons,\n361:                annotations=template.annotations,\n362:                _meta=template.meta,\n364:            for template in templates\n540:        has parameters, it will be registered as a template resource.\n577:            # Check if this should be a template\n597:                # Register as template\n598:                self._resource_manager.add_template(\n600:                    uri_template=uri,\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/server.py",
  "content": "        \"\"\"Report progress for the current operation.\n\n        Args:\n            progress: Current progress value e.g. 24\n            total: Optional total value e.g. 100\n            message: Optional message e.g. Starting render...\n        \"\"\"\n        progress_token = self.request_context.meta.progressToken if self.request_context.meta else None\n\n        if progress_token is None:  # pragma: no cover\n            return\n\n        await self.request_context.session.send_progress_notification(\n            progress_token=progress_token,\n            progress=progress,\n            total=total,\n            message=message,\n        )\n\n    async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContents]:\n        \"\"\"Read a resource by URI.\n\n        Args:\n            uri: Resource URI to read\n\n        Returns:\n            The resource content as either text or bytes\n        \"\"\"\n        assert self._fastmcp is not None, \"Context is not available outside of a request\"\n        return await self._fastmcp.read_resource(uri)\n\n    async def elicit(\n        self,\n        message: str,\n        schema: type[ElicitSchemaModelT],\n    ) -> ElicitationResult[ElicitSchemaModelT]:\n        \"\"\"Elicit information from the client/user.\n\n        This method can be used to interactively ask for additional information from the\n        client within a tool's execution. The client might display the message to the\n        user and collect a response according to the provided schema. Or in case a\n        client is an agent, it might decide how to handle the elicitation -- either by asking\n        the user or automatically generating a response.\n\n        Args:\n            schema: A Pydantic model class defining the expected response structure, according to the specification,\n                    only primive types are allowed.\n            message: Optional message to present to the user. If not provided, will use\n                    a default message based on the schema\n\n        Returns:",
  "start_line": 1100,
  "end_line": 1150,
  "total_lines": 1298,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/server.py",
  "content": "                icons=template.icons,\n                annotations=template.annotations,\n                _meta=template.meta,\n            )\n            for template in templates\n        ]\n\n    async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:\n        \"\"\"Read a resource by URI.\"\"\"\n\n        context = self.get_context()\n        resource = await self._resource_manager.get_resource(uri, context=context)\n        if not resource:  # pragma: no cover\n            raise ResourceError(f\"Unknown resource: {uri}\")\n\n        try:\n            content = await resource.read()\n            return [ReadResourceContents(content=content, mime_type=resource.mime_type, meta=resource.meta)]\n        except Exception as e:  # pragma: no cover\n            logger.exception(f\"Error reading resource {uri}\")\n            raise ResourceError(str(e))",
  "start_line": 360,
  "end_line": 380,
  "total_lines": 1298,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": " M src/mcp/server/fastmcp/resources/templates.py\ndiff --git a/src/mcp/server/fastmcp/resources/templates.py b/src/mcp/server/fastmcp/resources/templates.py\nindex 89a8ceb..efaa596 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,8 @@ 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+            params = match.groupdict()\n+            return {k: unquote(v) for k, v in params.items()}\n         return None\n \n     async def create_resource(\n",
  "stderr": "",
  "exit_code": 0
}

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