modelcontextprotocol__python-sdk-1864

{
  "status": "ok",
  "stdout": "src/mcp/types.py:694:    \"\"\"The URI of this resource.\"\"\"\nsrc/mcp/types.py:696:    \"\"\"A description of what this resource represents.\"\"\"\nsrc/mcp/types.py:698:    \"\"\"The MIME type of this resource, if known.\"\"\"\nsrc/mcp/types.py:701:    The size of the raw resource content, in bytes (i.e., before base64 encoding\nsrc/mcp/types.py:707:    \"\"\"An optional list of icons for this resource.\"\"\"\nsrc/mcp/types.py:717:    \"\"\"A template description for resources available on the server.\"\"\"\nsrc/mcp/types.py:721:    A URI template (according to RFC 6570) that can be used to construct resource\nsrc/mcp/types.py:728:    The MIME type for all resources that match this template. This should only be\nsrc/mcp/types.py:729:    included if all resources matching this template have the same type.\nsrc/mcp/types.py:732:    \"\"\"An optional list of icons for this resource template.\"\"\"\nsrc/mcp/types.py:742:    \"\"\"The server's response to a resources/list request from the client.\"\"\"\nsrc/mcp/types.py:744:    resources: list[Resource]\nsrc/mcp/types.py:747:class ListResourceTemplatesRequest(PaginatedRequest[Literal[\"resources/templates/list\"]]):\nsrc/mcp/types.py:748:    \"\"\"Sent from the client to request a list of resource templates the server has.\"\"\"\nsrc/mcp/types.py:750:    method: Literal[\"resources/templates/list\"] = \"resources/templates/list\"\nsrc/mcp/types.py:754:    \"\"\"The server's response to a resources/templates/list request from the client.\"\"\"\nsrc/mcp/types.py:756:    resourceTemplates: list[ResourceTemplate]\nsrc/mcp/types.py:760:    \"\"\"Parameters for reading a resource.\"\"\"\nsrc/mcp/types.py:764:    The URI of the resource to read. The URI can use any protocol; it is up to the\nsrc/mcp/types.py:769:class ReadResourceRequest(Request[ReadResourceRequestParams, Literal[\"resources/read\"]]):\nsrc/mcp/types.py:770:    \"\"\"Sent from the client to the server, to read a specific resource URI.\"\"\"\n--\nsrc/mcp/types.py:1092:    type: Literal[\"resource\"] = \"resource\"\nsrc/mcp/types.py:1093:    resource: TextResourceContents | BlobResourceContents\nsrc/mcp/types.py:1104:    A resource that the server is capable of reading, included in a prompt or tool call result.\nsrc/mcp/types.py:1106:    Note: resource links returned by tools are not guaranteed to appear in the results of `resources/list` requests.\nsrc/mcp/types.py:1109:    type: Literal[\"resource_link\"] = \"resource_link\"\nsrc/mcp/types.py:1481:    \"\"\"A reference to a resource or resource template definition.\"\"\"\nsrc/mcp/types.py:1483:    type: Literal[\"ref/resource\"] = \"ref/resource\"\nsrc/mcp/types.py:1485:    \"\"\"The URI or URI template of the resource.\"\"\"\nsrc/mcp/server/fastmcp/resources/templates.py:12:from mcp.server.fastmcp.resources.types import FunctionResource, Resource\nsrc/mcp/server/fastmcp/resources/templates.py:24:    \"\"\"A template for dynamically creating resources.\"\"\"\nsrc/mcp/server/fastmcp/resources/templates.py:27:    name: str = Field(description=\"Name of the resource\")\nsrc/mcp/server/fastmcp/resources/templates.py:28:    title: str | None = Field(description=\"Human-readable title of the resource\", default=None)\nsrc/mcp/server/fastmcp/resources/templates.py:29:    description: str | None = Field(description=\"Description of what the resource does\")\nsrc/mcp/server/fastmcp/resources/templates.py:30:    mime_type: str = Field(default=\"text/plain\", description=\"MIME type of the resource content\")\nsrc/mcp/server/fastmcp/resources/templates.py:31:    icons: list[Icon] | None = Field(default=None, description=\"Optional list of icons for the resource template\")\nsrc/mcp/server/fastmcp/resources/templates.py:32:    annotations: Annotations | None = Field(default=None, description=\"Optional annotations for the resource template\")\nsrc/mcp/server/fastmcp/resources/templates.py:33:    meta: dict[str, Any] | None = Field(default=None, description=\"Optional metadata for this resource template\")\nsrc/mcp/server/fastmcp/resources/templates.py:94:    async def create_resource(\nsrc/mcp/server/fastmcp/resources/templates.py:100:        \"\"\"Create a resource from the template with the given parameters.\"\"\"\nsrc/mcp/server/fastmcp/resources/templates.py:122:            raise ValueError(f\"Error creating resource from template: {e}\")\nsrc/mcp/server/fastmcp/resources/base.py:1:\"\"\"Base classes and interfaces for FastMCP resources.\"\"\"\nsrc/mcp/server/fastmcp/resources/base.py:18:    \"\"\"Base class for all resources.\"\"\"\nsrc/mcp/server/fastmcp/resources/base.py:22:    uri: str = Field(default=..., description=\"URI of the resource\")\nsrc/mcp/server/fastmcp/resources/base.py:23:    name: str | None = Field(description=\"Name of the resource\", default=None)\nsrc/mcp/server/fastmcp/resources/base.py:24:    title: str | None = Field(description=\"Human-readable title of the resource\", default=None)\n--\nsrc/mcp/server/fastmcp/resources/types.py:156:    mime_type: str = Field(default=\"application/json\", description=\"MIME type of the resource content\")\nsrc/mcp/server/fastmcp/resources/types.py:167:    \"\"\"A resource that lists files in a d",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "src/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\")\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/server.py",
  "content": "            name: Optional name for the tool (defaults to function name)\n            title: Optional human-readable title for the tool\n            description: Optional description of what the tool does\n            annotations: Optional ToolAnnotations providing additional tool information\n            structured_output: Controls whether the tool's output is structured or unstructured\n                - If None, auto-detects based on the function's return type annotation\n                - If True, creates a structured tool (return type annotation permitting)\n                - If False, unconditionally creates an unstructured tool\n        \"\"\"\n        self._tool_manager.add_tool(\n            fn,\n            name=name,\n            title=title,\n            description=description,\n            annotations=annotations,\n            icons=icons,\n            meta=meta,\n            structured_output=structured_output,\n        )\n\n    def remove_tool(self, name: str) -> None:\n        \"\"\"Remove a tool from the server by name.\n\n        Args:\n            name: The name of the tool to remove\n\n        Raises:\n            ToolError: If the tool does not exist\n        \"\"\"\n        self._tool_manager.remove_tool(name)\n\n    def tool(\n        self,\n        name: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        annotations: ToolAnnotations | None = None,\n        icons: list[Icon] | None = None,\n        meta: dict[str, Any] | None = None,\n        structured_output: bool | None = None,\n    ) -> Callable[[AnyFunction], AnyFunction]:\n        \"\"\"Decorator to register a tool.\n\n        Tools can optionally request a Context object by adding a parameter with the\n        Context type annotation. The context provides access to MCP capabilities like\n        logging, progress reporting, and resource access.\n\n        Args:\n            name: Optional name for the tool (defaults to function name)\n            title: Optional human-readable title for the tool\n            description: Optional description of what the tool does\n            annotations: Optional ToolAnnotations providing additional tool information\n            structured_output: Controls whether the tool's output is structured or unstructured\n                - If None, auto-detects based on the function's return type annotation\n                - If True, creates a structured tool (return type annotation permitting)\n                - If False, unconditionally creates an unstructured tool\n\n        Example:\n            @server.tool()\n            def my_tool(x: int) -> str:\n                return str(x)\n\n            @server.tool()\n            def tool_with_context(x: int, ctx: Context) -> str:\n                ctx.info(f\"Processing {x}\")\n                return str(x)\n\n            @server.tool()\n            async def async_tool(x: int, context: Context) -> str:\n                await context.report_progress(50, 100)\n                return str(x)\n        \"\"\"\n        # Check if user passed function directly instead of calling decorator\n        if callable(name):\n            raise TypeError(\n                \"The @tool decorator was used incorrectly. Did you forget to call it? Use @tool() instead of @tool\"\n            )\n\n        def decorator(fn: AnyFunction) -> AnyFunction:\n            self.add_tool(\n                fn,\n                name=name,\n                title=title,\n                description=description,\n                annotations=annotations,\n                icons=icons,\n                meta=meta,\n                structured_output=structured_output,\n            )\n            return fn\n\n        return decorator\n\n    def completion(self):\n        \"\"\"Decorator to register a completion handler.\n\n        The completion handler receives:\n        - ref: PromptReference or ResourceTemplateReference\n        - argument: CompletionArgument with name and partial value\n        - context: Optional CompletionContext with previously resolved arguments\n\n        Example:\n            @mcp.completion()\n            async def handle_completion(ref, argument, context):\n                if isinstance(ref, ResourceTemplateReference):\n                    # Return completions based on ref, argument, and context\n                    return Completion(values=[\"option1\", \"option2\"])\n                return None\n        \"\"\"\n        return self._mcp_server.completion()\n\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",
  "start_line": 400,
  "end_line": 549,
  "total_lines": 1298,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/server.py",
  "content": "        Example:\n            @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,\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            else:\n                # Register as regular resource\n                resource = FunctionResource.from_function(\n                    fn=fn,\n                    uri=uri,\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.add_resource(resource)\n            return fn\n\n        return decorator\n\n    def add_prompt(self, prompt: Prompt) -> None:\n        \"\"\"Add a prompt to the server.\n\n        Args:\n            prompt: A Prompt instance to add\n        \"\"\"\n        self._prompt_manager.add_prompt(prompt)\n\n    def prompt(\n        self,\n        name: str | None = None,\n        title: str | None = None,\n        description: str | None = None,\n        icons: list[Icon] | None = None,\n    ) -> Callable[[AnyFunction], AnyFunction]:\n        \"\"\"Decorator to register a prompt.\n\n        Args:\n            name: Optional name for the prompt (defaults to function name)\n            title: Optional human-readable title for the prompt\n            description: Optional description of what the prompt does\n\n        Example:\n            @server.prompt()\n            def analyze_table(table_name: str) -> list[Message]:\n                schema = read_table_schema(table_name)\n                return [\n                    {\n                        \"role\": \"user\",\n                        \"content\": f\"Analyze this schema:\\n{schema}\"\n                    }\n                ]\n\n            @server.prompt()\n            async def analyze_file(path: str) -> list[Message]:\n                content = await read_file(path)\n                return [\n                    {\n                        \"role\": \"user\",\n                        \"content\": {\n                            \"type\": \"resource\",\n                            \"resource\": {\n                                \"uri\": f\"file://{path}\",\n                                \"text\": content\n                            }\n                        }\n                    }\n                ]\n        \"\"\"\n        # Check if user passed function directly instead of calling decorator\n        if callable(name):\n            raise TypeError(\n                \"The @prompt decorator was used incorrectly. \"\n                \"Did you forget to call it? Use @prompt() instead of @prompt\"\n            )\n\n        def decorator(func: AnyFunction) -> AnyFunction:\n            prompt = Prompt.from_function(func, name=name, title=title, description=description, icons=icons)\n            self.add_prompt(prompt)\n            return func\n\n        return decorator\n\n    def custom_route(\n        self,\n        path: str,\n        methods: list[str],\n        name: str | None = None,\n        include_in_schema: bool = True,\n    ):\n        \"\"\"\n        Decorator to register a custom HTTP route on the FastMCP server.\n",
  "start_line": 550,
  "end_line": 699,
  "total_lines": 1298,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "__init__.py\nbase.py\nresource_manager.py\ntemplates.py\ntypes.py\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": "error",
  "error_type": "FileWriteError",
  "error_message": "Path traversal detected: '/tmp/repro.py' escapes workspace root."
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Captured: ['stick%20correcteur', 'teint%C3%A9']\nFAILURE: Value stick%20correcteur is still encoded\n",
  "details": {
    "stdout": "Captured: ['stick%20correcteur', 'teint%C3%A9']\nFAILURE: Value stick%20correcteur is still encoded\n",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "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": 122,
  "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@@ -4,6 +4,7 @@\n \n import inspect\n import re\n+from urllib.parse import unquote\n from collections.abc import Callable\n from typing import TYPE_CHECKING, Any\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": "Captured: ['stick correcteur', 'teinté']\nSUCCESS: Values are decoded\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "src/mcp/server/fastmcp/resources/types.py",
  "content": "\"\"\"Concrete resource implementations.\"\"\"\n\nimport inspect\nimport json\nfrom collections.abc import Callable\nfrom pathlib import Path\nfrom typing import Any\n\nimport anyio\nimport anyio.to_thread\nimport httpx\nimport pydantic\nimport pydantic_core\nfrom pydantic import Field, ValidationInfo, validate_call\n\nfrom mcp.server.fastmcp.resources.base import Resource\nfrom mcp.types import Annotations, Icon\n\n\nclass TextResource(Resource):\n    \"\"\"A resource that reads from a string.\"\"\"\n\n    text: str = Field(description=\"Text content of the resource\")\n\n    async def read(self) -> str:\n        \"\"\"Read the text content.\"\"\"\n        return self.text  # pragma: no cover\n\n\nclass BinaryResource(Resource):\n    \"\"\"A resource that reads from bytes.\"\"\"\n\n    data: bytes = Field(description=\"Binary content of the resource\")\n\n    async def read(self) -> bytes:\n        \"\"\"Read the binary content.\"\"\"\n        return self.data  # pragma: no cover\n\n\nclass FunctionResource(Resource):\n    \"\"\"A resource that defers data loading by wrapping a function.\n\n    The function is only called when the resource is read, allowing for lazy loading\n    of potentially expensive data. This is particularly useful when listing resources,\n    as the function won't be called until the resource is actually accessed.\n\n    The function can return:\n    - str for text content (default)\n    - bytes for binary content\n    - other types will be converted to JSON\n    \"\"\"\n\n    fn: Callable[[], Any] = Field(exclude=True)\n\n    async def read(self) -> str | bytes:\n        \"\"\"Read the resource by calling the wrapped function.\"\"\"\n        try:\n            # Call the function first to see if it returns a coroutine\n            result = self.fn()\n            # If it's a coroutine, await it\n            if inspect.iscoroutine(result):\n                result = await result\n\n            if isinstance(result, Resource):  # pragma: no cover\n                return await result.read()\n            elif isinstance(result, bytes):\n                return result\n            elif isinstance(result, str):\n                return result\n            else:\n                return pydantic_core.to_json(result, fallback=str, indent=2).decode()\n        except Exception as e:\n            raise ValueError(f\"Error reading resource {self.uri}: {e}\")\n\n    @classmethod\n    def from_function(\n        cls,\n        fn: Callable[..., Any],\n        uri: 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    ) -> \"FunctionResource\":\n        \"\"\"Create a FunctionResource from a function.\"\"\"\n        func_name = name or fn.__name__\n        if func_name == \"<lambda>\":  # pragma: no cover\n            raise ValueError(\"You must provide a name for lambda functions\")\n\n        # ensure the arguments are properly cast\n        fn = validate_call(fn)\n\n        return cls(\n            uri=uri,\n            name=func_name,\n            title=title,\n            description=description or fn.__doc__ or \"\",\n            mime_type=mime_type or \"text/plain\",\n            fn=fn,\n            icons=icons,\n            annotations=annotations,\n            meta=meta,\n        )\n\n\nclass FileResource(Resource):\n    \"\"\"A resource that reads from a file.\n\n    Set is_binary=True to read file as binary data instead of text.\n    \"\"\"\n\n    path: Path = Field(description=\"Path to the file\")\n    is_binary: bool = Field(\n        default=False,\n        description=\"Whether to read the file as binary data\",\n    )\n    mime_type: str = Field(\n        default=\"text/plain\",\n        description=\"MIME type of the resource content\",\n    )\n\n    @pydantic.field_validator(\"path\")\n    @classmethod\n    def validate_absolute_path(cls, path: Path) -> Path:  # pragma: no cover\n        \"\"\"Ensure path is absolute.\"\"\"\n        if not path.is_absolute():\n            raise ValueError(\"Path must be absolute\")\n        return path\n\n    @pydantic.field_validator(\"is_binary\")\n    @classmethod\n    def set_binary_from_mime_type(cls, is_binary: bool, info: ValidationInfo) -> bool:\n        \"\"\"Set is_binary based on mime_type if not explicitly set.\"\"\"\n        if is_binary:\n            return True\n        mime_type = info.data.get(\"mime_type\", \"text/plain\")\n        return not mime_type.startswith(\"text/\")\n\n    async def read(self) -> str | bytes:\n        \"\"\"Read the file content.\"\"\"\n        try:\n            if self.is_binary:\n                return await anyio.to_thread.run_sync(self.path.read_bytes)\n            return await anyio.to_thread.run_sync(self.path.read_text)\n        except Exception as e:\n            raise ValueError(f\"Error reading file {self.path}: {e}\")\n",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 203,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "./src/mcp/server/fastmcp/resources/templates.py:92:            return {k: unquote(v) for k, v in match.groupdict().items()}\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "bringing up nodes...\n48 workers [0 items]\n\nscheduling tests via LoadScheduling\n\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\u001b[33m============================ \u001b[33mno tests ran\u001b[0m\u001b[33m in 14.23s\u001b[0m\u001b[33m ============================\u001b[0m\n",
  "details": {
    "stdout": "bringing up nodes...\n48 workers [0 items]\n\nscheduling tests via LoadScheduling\n\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\u001b[33m============================ \u001b[33mno tests ran\u001b[0m\u001b[33m in 14.23s\u001b[0m\u001b[33m ============================\u001b[0m\n",
    "stderr": "",
    "exit_code": 5
  }
}

{
  "status": "ok",
  "stdout": "tests\ntests/test_examples.py\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/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/test_utils.py\ntests/experimental/tasks/test_request_context.py\ntests/experimental/tasks/test_capabilities.py\ntests/experimental/tasks/test_spec_compliance.py\ntests/experimental/tasks/server/test_integration.py\ntests/experimental/tasks/server/test_run_task_flow.py\ntests/experimental/tasks/server/test_server.py\ntests/experimental/tasks/server/test_context.py\ntests/experimental/tasks/server/test_store.py\ntests/experimental/tasks/server/test_task_result_handler.py\ntests/experimental/tasks/server/test_server_task_context.py\ntests/experimental/tasks/test_message_queue.py\ntests/experimental/tasks/client/test_capabilities.py\ntests/experimental/tasks/client/test_handlers.py\ntests/experimental/tasks/client/test_tasks.py\ntests/experimental/tasks/client/test_poll_task.py\ntests/experimental/tasks/test_elicitation_scenarios.py\ntests/server/fastmcp/test_integration.py\ntests/server/fastmcp/test_url_elicitation.py\ntests/server/fastmcp/test_tool_manager.py\ntests/server/fastmcp/test_url_elicitation_error_throw.py\ntests/server/fastmcp/test_title.py\ntests/server/fastmcp/resources/test_file_resources.py\ntests/server/fastmcp/resources/test_function_resources.py\ntests/server/fastmcp/resources/test_resource_manager.py\ntests/server/fastmcp/resources/test_resources.py\ntests/server/fastmcp/resources/test_resource_template.py\ntests/server/fastmcp/auth/test_auth_integration.py\ntests/server/fastmcp/test_parameter_descriptions.py\ntests/server/fastmcp/test_func_metadata.py\ntests/server/fastmcp/servers/test_file_server.py\ntests/server/fastmcp/test_elicitation.py\ntests/server/fastmcp/test_server.py\ntests/server/fastmcp/prompts/test_base.py\ntests/server/fastmcp/prompts/test_manager.py\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/middleware/test_bearer_auth.py\ntests/server/auth/middleware/test_auth_context.py\ntests/server/auth/test_provider.py\ntests/server/auth/test_error_handling.py\ntests/server/auth/test_protected_resource.py\ntests/server/test_lowlevel_input_validation.py\ntests/server/test_streamable_http_manager.py\ntests/server/lowlevel/test_helper_types.py\ntests/server/lowlevel/test_server_pagination.py\ntests/server/lowlevel/test_func_inspection.py\ntests/server/lowlevel/test_server_listing.py\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/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/test_types.py\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/test_sampling_callback.py\ntests/client/auth/extensions/test_client_credentials.py\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/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",
  "stdout": "bringing up nodes...\n48 workers [14 items]\n\nscheduling tests via LoadScheduling\n\ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_template_matches \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_create_resource \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateMetadata::test_template_created_resources_inherit_metadata \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateMetadata::test_template_from_function_with_metadata \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateAnnotations::test_template_annotations_in_fastmcp \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_basemodel_conversion \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateAnnotations::test_template_with_annotations \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_async_binary_resource \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_template_error \n[gw3]\u001b[36m [  7%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_template_error \n[gw2]\u001b[36m [ 14%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_create_resource \n[gw7]\u001b[36m [ 21%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_basemodel_conversion \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_template_creation \n[gw0]\u001b[36m [ 28%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_template_creation \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateAnnotations::test_template_created_resources_inherit_annotations \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_async_text_resource \n[gw4]\u001b[36m [ 35%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_async_text_resource \n[gw1]\u001b[36m [ 42%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_template_matches \n[gw10]\u001b[36m [ 50%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateAnnotations::test_template_annotations_in_fastmcp \n[gw12]\u001b[36m [ 57%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateMetadata::test_template_from_function_with_metadata \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateAnnotations::test_template_without_annotations \n[gw9]\u001b[36m [ 64%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateAnnotations::test_template_without_annotations \n[gw6]\u001b[36m [ 71%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateAnnotations::test_template_with_annotations \n[gw11]\u001b[36m [ 78%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateAnnotations::test_template_created_resources_inherit_annotations \ntests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_custom_type_conversion \n[gw13]\u001b[36m [ 85%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplateMetadata::test_template_created_resources_inherit_metadata \n[gw5]\u001b[36m [ 92%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_async_binary_resource \n[gw8]\u001b[36m [100%] \u001b[0m\u001b[32mPASSED\u001b[0m tests/server/fastmcp/resources/test_resource_template.py::TestResourceTemplate::test_custom_type_conversion \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[1m14 passed\u001b[0m\u001b[32m in 24.23s\u001b[0m\u001b[32m ==============================\u001b[0m\n",
  "stderr": "",
  "exit_code": 0
}

{
  "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..e993cba 100644\n--- a/src/mcp/server/fastmcp/resources/templates.py\n+++ b/src/mcp/server/fastmcp/resources/templates.py\n@@ -4,6 +4,7 @@ from __future__ import annotations\n \n import inspect\n import re\n+from urllib.parse import unquote\n from collections.abc import Callable\n from typing import TYPE_CHECKING, Any\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
}

{
  "status": "error",
  "error_type": "FileEditError",
  "error_message": "Failed to replace: old_string not found. Ensure you're not escaping content incorrectly and check whitespace, indentation, and context.",
  "budget_warning": "Only 10 tool call(s) remaining (20/30 used). Finalize your edits and call submit_patch soon."
}

{
  "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 urllib.parse import unquote\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",
  "start_line": 1,
  "end_line": 20,
  "total_lines": 123,
  "is_truncated": false,
  "budget_warning": "Only 9 tool call(s) remaining (21/30 used). Finalize your edits and call submit_patch soon."
}