Skip to content

Index

Resource

Bases: BaseModel, ABC

Base class for all resources.

Source code in src/mcp/server/mcpserver/resources/base.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class Resource(BaseModel, abc.ABC):
    """Base class for all resources."""

    model_config = ConfigDict(validate_default=True)

    uri: str = Field(default=..., description="URI of the resource")
    name: str | None = Field(description="Name of the resource", default=None)
    title: str | None = Field(description="Human-readable title of the resource", default=None)
    description: str | None = Field(description="Description of the resource", default=None)
    mime_type: str = Field(default="text/plain", description="MIME type of the resource content")
    icons: list[Icon] | None = Field(default=None, description="Optional list of icons for this resource")
    annotations: Annotations | None = Field(default=None, description="Optional annotations for the resource")
    meta: dict[str, Any] | None = Field(default=None, description="Optional metadata for this resource")

    @field_validator("name", mode="before")
    @classmethod
    def set_default_name(cls, name: str | None, info: ValidationInfo) -> str:
        """Set default name from URI if not provided."""
        if name:
            return name
        if uri := info.data.get("uri"):
            return str(uri)
        raise ValueError("Either name or uri must be provided")

    @abc.abstractmethod
    async def read(self) -> str | bytes:
        """Read the resource content."""
        pass  # pragma: no cover

set_default_name classmethod 

set_default_name(
    name: str | None, info: ValidationInfo
) -> str

Set default name from URI if not provided.

Source code in src/mcp/server/mcpserver/resources/base.py 
30
31
32
33
34
35
36
37
38
@field_validator("name", mode="before")
@classmethod
def set_default_name(cls, name: str | None, info: ValidationInfo) -> str:
    """Set default name from URI if not provided."""
    if name:
        return name
    if uri := info.data.get("uri"):
        return str(uri)
    raise ValueError("Either name or uri must be provided")

read abstractmethod async 

read() -> str | bytes

Read the resource content.

Source code in src/mcp/server/mcpserver/resources/base.py 
40
41
42
43
@abc.abstractmethod
async def read(self) -> str | bytes:
    """Read the resource content."""
    pass  # pragma: no cover

ResourceManager

Manages MCPServer resources.

Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class ResourceManager:
    """Manages MCPServer resources."""

    def __init__(self, warn_on_duplicate_resources: bool = True, *, resources: list[Resource] | None = None):
        self._resources: dict[str, Resource] = {}
        self._templates: dict[str, ResourceTemplate] = {}
        self.warn_on_duplicate_resources = warn_on_duplicate_resources

        for resource in resources or ():
            self.add_resource(resource)

    def add_resource(self, resource: Resource) -> Resource:
        """Add a resource to the manager.

        Args:
            resource: A Resource instance to add.

        Returns:
            The added resource. If a resource with the same URI already exists, returns the existing resource.
        """
        logger.debug(
            "Adding resource",
            extra={"uri": resource.uri, "type": type(resource).__name__, "resource_name": resource.name},
        )
        existing = self._resources.get(str(resource.uri))
        if existing:
            if self.warn_on_duplicate_resources:
                logger.warning(f"Resource already exists: {resource.uri}")
            return existing
        self._resources[str(resource.uri)] = resource
        return resource

    def add_template(
        self,
        fn: Callable[..., Any],
        uri_template: str,
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        mime_type: str | None = None,
        icons: list[Icon] | None = None,
        annotations: Annotations | None = None,
        meta: dict[str, Any] | None = None,
        security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
    ) -> ResourceTemplate:
        """Add a template from a function."""
        template = ResourceTemplate.from_function(
            fn,
            uri_template=uri_template,
            name=name,
            title=title,
            description=description,
            mime_type=mime_type,
            icons=icons,
            annotations=annotations,
            meta=meta,
            security=security,
        )
        self._templates[template.uri_template] = template
        return template

    async def get_resource(
        self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]
    ) -> Resource | InputRequiredResult:
        """Get resource by URI, checking concrete resources first, then templates.

        A template function may return an `InputRequiredResult` instead of
        resource content (the 2026-07-28 multi-round-trip flow); it is passed
        through unchanged.

        Raises:
            ResourceNotFoundError: If no resource or template matches the URI.
            ResourceError: If a matching template fails to create the resource.

        Note:
            Pydantic's ``AnyUrl`` normalises percent-encoding and
            resolves ``..`` segments during validation, so a value
            constructed as ``AnyUrl("file:///a/%2E%2E/b")`` arrives
            here as ``file:///b``. The JSON-RPC protocol layer passes
            raw ``str`` values and is unaffected, but internal callers
            wrapping URIs in ``AnyUrl`` should be aware that security
            checks see the already-normalised form.
        """
        uri_str = str(uri)
        logger.debug("Getting resource", extra={"uri": uri_str})

        # First check concrete resources
        if resource := self._resources.get(uri_str):
            return resource

        # Then check templates
        for template in self._templates.values():
            try:
                params = template.matches(uri_str)
            except ResourceSecurityError as e:
                raise ResourceNotFoundError(f"Unknown resource: {uri}") from e
            if params is not None:
                return await template.create_resource(uri_str, params, context=context)

        raise ResourceNotFoundError(f"Unknown resource: {uri}")

    def list_resources(self) -> list[Resource]:
        """List all registered resources."""
        logger.debug("Listing resources", extra={"count": len(self._resources)})
        return list(self._resources.values())

    def list_templates(self) -> list[ResourceTemplate]:
        """List all registered templates."""
        logger.debug("Listing templates", extra={"count": len(self._templates)})
        return list(self._templates.values())

add_resource 

add_resource(resource: Resource) -> Resource

Add a resource to the manager.

Parameters:

Name Type Description Default
resource Resource

A Resource instance to add.

 required

Returns:

Type Description
Resource

The added resource. If a resource with the same URI already exists, returns the existing resource.
Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def add_resource(self, resource: Resource) -> Resource:
    """Add a resource to the manager.

    Args:
        resource: A Resource instance to add.

    Returns:
        The added resource. If a resource with the same URI already exists, returns the existing resource.
    """
    logger.debug(
        "Adding resource",
        extra={"uri": resource.uri, "type": type(resource).__name__, "resource_name": resource.name},
    )
    existing = self._resources.get(str(resource.uri))
    if existing:
        if self.warn_on_duplicate_resources:
            logger.warning(f"Resource already exists: {resource.uri}")
        return existing
    self._resources[str(resource.uri)] = resource
    return resource

add_template 

add_template(
    fn: Callable[..., Any],
    uri_template: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
    security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
) -> ResourceTemplate

Add a template from a function.

Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def add_template(
    self,
    fn: Callable[..., Any],
    uri_template: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
    security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
) -> ResourceTemplate:
    """Add a template from a function."""
    template = ResourceTemplate.from_function(
        fn,
        uri_template=uri_template,
        name=name,
        title=title,
        description=description,
        mime_type=mime_type,
        icons=icons,
        annotations=annotations,
        meta=meta,
        security=security,
    )
    self._templates[template.uri_template] = template
    return template

get_resource async 

get_resource(
    uri: AnyUrl | str,
    context: Context[LifespanContextT, RequestT],
) -> Resource | InputRequiredResult

Get resource by URI, checking concrete resources first, then templates.

A template function may return an InputRequiredResult instead of resource content (the 2026-07-28 multi-round-trip flow); it is passed through unchanged.

Raises:

Type Description
ResourceNotFoundError

If no resource or template matches the URI.
ResourceError

If a matching template fails to create the resource.
Note

Pydantic's AnyUrl normalises percent-encoding and resolves .. segments during validation, so a value constructed as AnyUrl("file:///a/%2E%2E/b") arrives here as file:///b. The JSON-RPC protocol layer passes raw str values and is unaffected, but internal callers wrapping URIs in AnyUrl should be aware that security checks see the already-normalised form.

Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def get_resource(
    self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]
) -> Resource | InputRequiredResult:
    """Get resource by URI, checking concrete resources first, then templates.

    A template function may return an `InputRequiredResult` instead of
    resource content (the 2026-07-28 multi-round-trip flow); it is passed
    through unchanged.

    Raises:
        ResourceNotFoundError: If no resource or template matches the URI.
        ResourceError: If a matching template fails to create the resource.

    Note:
        Pydantic's ``AnyUrl`` normalises percent-encoding and
        resolves ``..`` segments during validation, so a value
        constructed as ``AnyUrl("file:///a/%2E%2E/b")`` arrives
        here as ``file:///b``. The JSON-RPC protocol layer passes
        raw ``str`` values and is unaffected, but internal callers
        wrapping URIs in ``AnyUrl`` should be aware that security
        checks see the already-normalised form.
    """
    uri_str = str(uri)
    logger.debug("Getting resource", extra={"uri": uri_str})

    # First check concrete resources
    if resource := self._resources.get(uri_str):
        return resource

    # Then check templates
    for template in self._templates.values():
        try:
            params = template.matches(uri_str)
        except ResourceSecurityError as e:
            raise ResourceNotFoundError(f"Unknown resource: {uri}") from e
        if params is not None:
            return await template.create_resource(uri_str, params, context=context)

    raise ResourceNotFoundError(f"Unknown resource: {uri}")

list_resources 

list_resources() -> list[Resource]

List all registered resources.

Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
129
130
131
132
def list_resources(self) -> list[Resource]:
    """List all registered resources."""
    logger.debug("Listing resources", extra={"count": len(self._resources)})
    return list(self._resources.values())

list_templates 

list_templates() -> list[ResourceTemplate]

List all registered templates.

Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
134
135
136
137
def list_templates(self) -> list[ResourceTemplate]:
    """List all registered templates."""
    logger.debug("Listing templates", extra={"count": len(self._templates)})
    return list(self._templates.values())

DEFAULT_RESOURCE_SECURITY module-attribute 

DEFAULT_RESOURCE_SECURITY = ResourceSecurity()

Secure-by-default policy: traversal, absolute paths, and null bytes rejected.

ResourceSecurity dataclass

Security policy applied to extracted resource template parameters.

These checks run after :meth:~mcp.shared.uri_template.UriTemplate.match has extracted and decoded parameter values. They catch path-traversal and absolute-path injection regardless of how the value was encoded in the URI (literal, %2F, %5C, %2E%2E).

Example::

# Opt out for a parameter that legitimately contains ..
@mcp.resource(
    "git://diff/{+range}",
    security=ResourceSecurity(exempt_params={"range"}),
)
def git_diff(range: str) -> str: ...
Source code in src/mcp/server/mcpserver/resources/templates.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@dataclass(frozen=True)
class ResourceSecurity:
    """Security policy applied to extracted resource template parameters.

    These checks run after :meth:`~mcp.shared.uri_template.UriTemplate.match`
    has extracted and decoded parameter values. They catch path-traversal
    and absolute-path injection regardless of how the value was encoded in
    the URI (literal, ``%2F``, ``%5C``, ``%2E%2E``).

    Example::

        # Opt out for a parameter that legitimately contains ..
        @mcp.resource(
            "git://diff/{+range}",
            security=ResourceSecurity(exempt_params={"range"}),
        )
        def git_diff(range: str) -> str: ...
    """

    reject_path_traversal: bool = True
    """Reject values containing ``..`` as a path component."""

    reject_absolute_paths: bool = True
    """Reject values that look like absolute filesystem paths."""

    reject_null_bytes: bool = True
    """Reject values containing NUL (``\\x00``). Null bytes defeat string
    comparisons (``"..\\x00" != ".."``) and can cause truncation in C
    extensions or subprocess calls."""

    exempt_params: Set[str] = field(default_factory=frozenset[str])
    """Parameter names to skip all checks for."""

    def validate(self, params: Mapping[str, str | list[str]]) -> str | None:
        """Check all parameter values against the configured policy.

        Args:
            params: Extracted template parameters. List values (from
                explode variables) are checked element-wise.

        Returns:
            The name of the first parameter that fails, or ``None`` if
            all values pass.
        """
        for name, value in params.items():
            if name in self.exempt_params:
                continue
            values = value if isinstance(value, list) else [value]
            for v in values:
                if self.reject_null_bytes and "\0" in v:
                    return name
                if self.reject_path_traversal and contains_path_traversal(v):
                    return name
                if self.reject_absolute_paths and is_absolute_path(v):
                    return name
        return None

reject_path_traversal class-attribute instance-attribute 

reject_path_traversal: bool = True

Reject values containing .. as a path component.

reject_absolute_paths class-attribute instance-attribute 

reject_absolute_paths: bool = True

Reject values that look like absolute filesystem paths.

reject_null_bytes class-attribute instance-attribute 

reject_null_bytes: bool = True

Reject values containing NUL (\x00). Null bytes defeat string comparisons ("..\x00" != "..") and can cause truncation in C extensions or subprocess calls.

exempt_params class-attribute instance-attribute 

exempt_params: Set[str] = field(
    default_factory=frozenset[str]
)

Parameter names to skip all checks for.

validate 

validate(
    params: Mapping[str, str | list[str]],
) -> str | None

Check all parameter values against the configured policy.

Parameters:

Name Type Description Default
params Mapping[str, str | list[str]]

Extracted template parameters. List values (from explode variables) are checked element-wise.

 required

Returns:

Type Description
str | None

The name of the first parameter that fails, or None if
str | None

all values pass.
Source code in src/mcp/server/mcpserver/resources/templates.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def validate(self, params: Mapping[str, str | list[str]]) -> str | None:
    """Check all parameter values against the configured policy.

    Args:
        params: Extracted template parameters. List values (from
            explode variables) are checked element-wise.

    Returns:
        The name of the first parameter that fails, or ``None`` if
        all values pass.
    """
    for name, value in params.items():
        if name in self.exempt_params:
            continue
        values = value if isinstance(value, list) else [value]
        for v in values:
            if self.reject_null_bytes and "\0" in v:
                return name
            if self.reject_path_traversal and contains_path_traversal(v):
                return name
            if self.reject_absolute_paths and is_absolute_path(v):
                return name
    return None

ResourceSecurityError

Bases: ValueError

Raised when an extracted parameter fails :class:ResourceSecurity checks.

Distinct from a simple None non-match so that template iteration can stop at the first security rejection rather than falling through to a later, possibly more permissive, template.

Source code in src/mcp/server/mcpserver/resources/templates.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class ResourceSecurityError(ValueError):
    """Raised when an extracted parameter fails :class:`ResourceSecurity` checks.

    Distinct from a simple ``None`` non-match so that template
    iteration can stop at the first security rejection rather than
    falling through to a later, possibly more permissive, template.
    """

    def __init__(self, template: str, param: str) -> None:
        super().__init__(f"Parameter {param!r} of template {template!r} failed security validation")
        self.template = template
        self.param = param

ResourceTemplate

Bases: BaseModel

A template for dynamically creating resources.

Source code in src/mcp/server/mcpserver/resources/templates.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class ResourceTemplate(BaseModel):
    """A template for dynamically creating resources."""

    uri_template: str = Field(description="URI template with parameters (e.g. weather://{city}/current)")
    name: str = Field(description="Name of the resource")
    title: str | None = Field(description="Human-readable title of the resource", default=None)
    description: str | None = Field(description="Description of what the resource does")
    mime_type: str = Field(default="text/plain", description="MIME type of the resource content")
    icons: list[Icon] | None = Field(default=None, description="Optional list of icons for the resource template")
    annotations: Annotations | None = Field(default=None, description="Optional annotations for the resource template")
    meta: dict[str, Any] | None = Field(default=None, description="Optional metadata for this resource template")
    fn: Callable[..., Any] = Field(exclude=True)
    parameters: dict[str, Any] = Field(description="JSON schema for function parameters")
    context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context")
    parsed_template: UriTemplate = Field(exclude=True, description="Parsed RFC 6570 template")
    security: ResourceSecurity = Field(exclude=True, description="Path-safety policy for extracted parameters")

    @classmethod
    def from_function(
        cls,
        fn: Callable[..., Any],
        uri_template: str,
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        mime_type: str | None = None,
        icons: list[Icon] | None = None,
        annotations: Annotations | None = None,
        meta: dict[str, Any] | None = None,
        context_kwarg: str | None = None,
        security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
    ) -> ResourceTemplate:
        """Create a template from a function.

        Raises:
            InvalidUriTemplate: If ``uri_template`` is malformed or uses
                unsupported RFC 6570 features.
        """
        func_name = name or fn.__name__
        if func_name == "<lambda>":
            raise ValueError("You must provide a name for lambda functions")  # pragma: no cover

        parsed = UriTemplate.parse(uri_template)

        # Find context parameter if it exists
        if context_kwarg is None:  # pragma: no branch
            context_kwarg = find_context_parameter(fn)

        # Get schema from func_metadata, excluding context parameter
        func_arg_metadata = func_metadata(
            fn,
            skip_names=[context_kwarg] if context_kwarg is not None else [],
        )
        parameters = func_arg_metadata.arg_model.model_json_schema()

        # ensure the arguments are properly cast
        fn = validate_call(fn)

        return cls(
            uri_template=uri_template,
            name=func_name,
            title=title,
            description=description or fn.__doc__ or "",
            mime_type=mime_type or "text/plain",
            icons=icons,
            annotations=annotations,
            meta=meta,
            fn=fn,
            parameters=parameters,
            context_kwarg=context_kwarg,
            parsed_template=parsed,
            security=security,
        )

    def matches(self, uri: str) -> dict[str, str | list[str]] | None:
        """Check if a URI matches this template and extract parameters.

        Delegates to :meth:`UriTemplate.match` for RFC 6570 extraction,
        then applies this template's :class:`ResourceSecurity` policy
        (path traversal, absolute paths).

        Returns:
            Extracted parameters on success, or ``None`` if the URI
            doesn't match the template.

        Raises:
            ResourceSecurityError: If the URI matches but an extracted
                parameter fails security validation. Raising (rather
                than returning ``None``) prevents the resource manager
                from silently falling through to a later, possibly more
                permissive, template.
        """
        params = self.parsed_template.match(uri)
        if params is None:
            return None
        failed = self.security.validate(params)
        if failed is not None:
            raise ResourceSecurityError(self.uri_template, failed)
        return params

    async def create_resource(
        self,
        uri: str,
        params: dict[str, Any],
        context: Context[LifespanContextT, RequestT],
    ) -> Resource | InputRequiredResult:
        """Create a resource from the template with the given parameters.

        An `InputRequiredResult` returned by the template function is passed
        through unchanged (the 2026-07-28 multi-round-trip flow); the retry's
        answers arrive on `ctx.input_responses`, with `ctx.request_state`
        carrying the echoed opaque state.

        Raises:
            ResourceError: If creating the resource fails.
        """
        try:
            # Add context to params if needed
            params = inject_context(self.fn, params, context, self.context_kwarg)

            fn = self.fn
            if is_async_callable(fn):
                result = await fn(**params)
            else:
                result = await anyio.to_thread.run_sync(functools.partial(self.fn, **params))

            if isinstance(result, InputRequiredResult):
                return result

            return FunctionResource(
                uri=uri,  # type: ignore
                name=self.name,
                title=self.title,
                description=self.description,
                mime_type=self.mime_type,
                icons=self.icons,
                annotations=self.annotations,
                meta=self.meta,
                fn=lambda: result,  # Capture result in closure
            )
        except (ResourceError, MCPError):
            raise
        except Exception as exc:
            logger.exception(f"Error creating resource from template {uri}")
            raise ResourceError(f"Error creating resource from template {uri}") from exc

from_function classmethod 

from_function(
    fn: Callable[..., Any],
    uri_template: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
    context_kwarg: str | None = None,
    security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
) -> ResourceTemplate

Create a template from a function.

Raises:

Type Description
InvalidUriTemplate

If uri_template is malformed or uses unsupported RFC 6570 features.
Source code in src/mcp/server/mcpserver/resources/templates.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
@classmethod
def from_function(
    cls,
    fn: Callable[..., Any],
    uri_template: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
    context_kwarg: str | None = None,
    security: ResourceSecurity = DEFAULT_RESOURCE_SECURITY,
) -> ResourceTemplate:
    """Create a template from a function.

    Raises:
        InvalidUriTemplate: If ``uri_template`` is malformed or uses
            unsupported RFC 6570 features.
    """
    func_name = name or fn.__name__
    if func_name == "<lambda>":
        raise ValueError("You must provide a name for lambda functions")  # pragma: no cover

    parsed = UriTemplate.parse(uri_template)

    # Find context parameter if it exists
    if context_kwarg is None:  # pragma: no branch
        context_kwarg = find_context_parameter(fn)

    # Get schema from func_metadata, excluding context parameter
    func_arg_metadata = func_metadata(
        fn,
        skip_names=[context_kwarg] if context_kwarg is not None else [],
    )
    parameters = func_arg_metadata.arg_model.model_json_schema()

    # ensure the arguments are properly cast
    fn = validate_call(fn)

    return cls(
        uri_template=uri_template,
        name=func_name,
        title=title,
        description=description or fn.__doc__ or "",
        mime_type=mime_type or "text/plain",
        icons=icons,
        annotations=annotations,
        meta=meta,
        fn=fn,
        parameters=parameters,
        context_kwarg=context_kwarg,
        parsed_template=parsed,
        security=security,
    )

matches 

matches(uri: str) -> dict[str, str | list[str]] | None

Check if a URI matches this template and extract parameters.

Delegates to :meth:UriTemplate.match for RFC 6570 extraction, then applies this template's :class:ResourceSecurity policy (path traversal, absolute paths).

Returns:

Type Description
dict[str, str | list[str]] | None

Extracted parameters on success, or None if the URI
dict[str, str | list[str]] | None

doesn't match the template.

Raises:

Type Description
ResourceSecurityError

If the URI matches but an extracted parameter fails security validation. Raising (rather than returning None) prevents the resource manager from silently falling through to a later, possibly more permissive, template.
Source code in src/mcp/server/mcpserver/resources/templates.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def matches(self, uri: str) -> dict[str, str | list[str]] | None:
    """Check if a URI matches this template and extract parameters.

    Delegates to :meth:`UriTemplate.match` for RFC 6570 extraction,
    then applies this template's :class:`ResourceSecurity` policy
    (path traversal, absolute paths).

    Returns:
        Extracted parameters on success, or ``None`` if the URI
        doesn't match the template.

    Raises:
        ResourceSecurityError: If the URI matches but an extracted
            parameter fails security validation. Raising (rather
            than returning ``None``) prevents the resource manager
            from silently falling through to a later, possibly more
            permissive, template.
    """
    params = self.parsed_template.match(uri)
    if params is None:
        return None
    failed = self.security.validate(params)
    if failed is not None:
        raise ResourceSecurityError(self.uri_template, failed)
    return params

create_resource async 

create_resource(
    uri: str,
    params: dict[str, Any],
    context: Context[LifespanContextT, RequestT],
) -> Resource | InputRequiredResult

Create a resource from the template with the given parameters.

An InputRequiredResult returned by the template function is passed through unchanged (the 2026-07-28 multi-round-trip flow); the retry's answers arrive on ctx.input_responses, with ctx.request_state carrying the echoed opaque state.

Raises:

Type Description
ResourceError

If creating the resource fails.
Source code in src/mcp/server/mcpserver/resources/templates.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
async def create_resource(
    self,
    uri: str,
    params: dict[str, Any],
    context: Context[LifespanContextT, RequestT],
) -> Resource | InputRequiredResult:
    """Create a resource from the template with the given parameters.

    An `InputRequiredResult` returned by the template function is passed
    through unchanged (the 2026-07-28 multi-round-trip flow); the retry's
    answers arrive on `ctx.input_responses`, with `ctx.request_state`
    carrying the echoed opaque state.

    Raises:
        ResourceError: If creating the resource fails.
    """
    try:
        # Add context to params if needed
        params = inject_context(self.fn, params, context, self.context_kwarg)

        fn = self.fn
        if is_async_callable(fn):
            result = await fn(**params)
        else:
            result = await anyio.to_thread.run_sync(functools.partial(self.fn, **params))

        if isinstance(result, InputRequiredResult):
            return result

        return FunctionResource(
            uri=uri,  # type: ignore
            name=self.name,
            title=self.title,
            description=self.description,
            mime_type=self.mime_type,
            icons=self.icons,
            annotations=self.annotations,
            meta=self.meta,
            fn=lambda: result,  # Capture result in closure
        )
    except (ResourceError, MCPError):
        raise
    except Exception as exc:
        logger.exception(f"Error creating resource from template {uri}")
        raise ResourceError(f"Error creating resource from template {uri}") from exc

BinaryResource

Bases: Resource

A resource that reads from bytes.

Source code in src/mcp/server/mcpserver/resources/types.py 
33
34
35
36
37
38
39
40
class BinaryResource(Resource):
    """A resource that reads from bytes."""

    data: bytes = Field(description="Binary content of the resource")

    async def read(self) -> bytes:
        """Read the binary content."""
        return self.data  # pragma: no cover

read async 

read() -> bytes

Read the binary content.

Source code in src/mcp/server/mcpserver/resources/types.py 
38
39
40
async def read(self) -> bytes:
    """Read the binary content."""
    return self.data  # pragma: no cover

DirectoryResource

Bases: Resource

A resource that lists files in a directory.

Source code in src/mcp/server/mcpserver/resources/types.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
class DirectoryResource(Resource):
    """A resource that lists files in a directory."""

    path: Path = Field(description="Path to the directory")
    recursive: bool = Field(default=False, description="Whether to list files recursively")
    pattern: str | None = Field(default=None, description="Optional glob pattern to filter files")
    mime_type: str = Field(default="application/json", description="MIME type of the resource content")

    @pydantic.field_validator("path")
    @classmethod
    def validate_absolute_path(cls, path: Path) -> Path:  # pragma: no cover
        """Ensure path is absolute."""
        if not path.is_absolute():
            raise ValueError("Path must be absolute")
        return path

    def list_files(self) -> list[Path]:  # pragma: no cover
        """List files in the directory."""
        if not self.path.exists():
            raise FileNotFoundError(f"Directory not found: {self.path}")
        if not self.path.is_dir():
            raise NotADirectoryError(f"Not a directory: {self.path}")

        try:
            if self.pattern:
                return list(self.path.glob(self.pattern)) if not self.recursive else list(self.path.rglob(self.pattern))
            return list(self.path.glob("*")) if not self.recursive else list(self.path.rglob("*"))
        except Exception as e:
            raise ValueError(f"Error listing directory {self.path}: {e}")

    async def read(self) -> str:  # Always returns JSON string  # pragma: no cover
        """Read the directory listing."""
        try:
            files = await anyio.to_thread.run_sync(self.list_files)
            file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
            return json.dumps({"files": file_list}, indent=2)
        except Exception as e:
            raise ValueError(f"Error reading directory {self.path}: {e}")

validate_absolute_path classmethod 

validate_absolute_path(path: Path) -> Path

Ensure path is absolute.

Source code in src/mcp/server/mcpserver/resources/types.py 
187
188
189
190
191
192
193
@pydantic.field_validator("path")
@classmethod
def validate_absolute_path(cls, path: Path) -> Path:  # pragma: no cover
    """Ensure path is absolute."""
    if not path.is_absolute():
        raise ValueError("Path must be absolute")
    return path

list_files 

list_files() -> list[Path]

List files in the directory.

Source code in src/mcp/server/mcpserver/resources/types.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
def list_files(self) -> list[Path]:  # pragma: no cover
    """List files in the directory."""
    if not self.path.exists():
        raise FileNotFoundError(f"Directory not found: {self.path}")
    if not self.path.is_dir():
        raise NotADirectoryError(f"Not a directory: {self.path}")

    try:
        if self.pattern:
            return list(self.path.glob(self.pattern)) if not self.recursive else list(self.path.rglob(self.pattern))
        return list(self.path.glob("*")) if not self.recursive else list(self.path.rglob("*"))
    except Exception as e:
        raise ValueError(f"Error listing directory {self.path}: {e}")

read async 

read() -> str

Read the directory listing.

Source code in src/mcp/server/mcpserver/resources/types.py 
209
210
211
212
213
214
215
216
async def read(self) -> str:  # Always returns JSON string  # pragma: no cover
    """Read the directory listing."""
    try:
        files = await anyio.to_thread.run_sync(self.list_files)
        file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
        return json.dumps({"files": file_list}, indent=2)
    except Exception as e:
        raise ValueError(f"Error reading directory {self.path}: {e}")

FileResource

Bases: Resource

A resource that reads from a file.

Set is_binary=True to read the file as binary data instead of text.

Source code in src/mcp/server/mcpserver/resources/types.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class FileResource(Resource):
    """A resource that reads from a file.

    Set is_binary=True to read the file as binary data instead of text.
    """

    path: Path = Field(description="Path to the file")
    is_binary: bool = Field(
        default=False,
        description="Whether to read the file as binary data",
    )
    mime_type: str = Field(
        default="text/plain",
        description="MIME type of the resource content",
    )

    @pydantic.field_validator("path")
    @classmethod
    def validate_absolute_path(cls, path: Path) -> Path:
        """Ensure path is absolute."""
        if not path.is_absolute():
            raise ValueError("Path must be absolute")
        return path

    @pydantic.field_validator("is_binary")
    @classmethod
    def set_binary_from_mime_type(cls, is_binary: bool, info: ValidationInfo) -> bool:
        """Set is_binary based on mime_type if not explicitly set."""
        if is_binary:
            return True
        mime_type = info.data.get("mime_type", "text/plain")
        return not mime_type.startswith("text/")

    async def read(self) -> str | bytes:
        """Read the file content."""
        try:
            if self.is_binary:
                return await anyio.to_thread.run_sync(self.path.read_bytes)
            return await anyio.to_thread.run_sync(self.path.read_text)
        except Exception as e:
            raise ValueError(f"Error reading file {self.path}: {e}")

validate_absolute_path classmethod 

validate_absolute_path(path: Path) -> Path

Ensure path is absolute.

Source code in src/mcp/server/mcpserver/resources/types.py 
138
139
140
141
142
143
144
@pydantic.field_validator("path")
@classmethod
def validate_absolute_path(cls, path: Path) -> Path:
    """Ensure path is absolute."""
    if not path.is_absolute():
        raise ValueError("Path must be absolute")
    return path

set_binary_from_mime_type classmethod 

set_binary_from_mime_type(
    is_binary: bool, info: ValidationInfo
) -> bool

Set is_binary based on mime_type if not explicitly set.

Source code in src/mcp/server/mcpserver/resources/types.py 
146
147
148
149
150
151
152
153
@pydantic.field_validator("is_binary")
@classmethod
def set_binary_from_mime_type(cls, is_binary: bool, info: ValidationInfo) -> bool:
    """Set is_binary based on mime_type if not explicitly set."""
    if is_binary:
        return True
    mime_type = info.data.get("mime_type", "text/plain")
    return not mime_type.startswith("text/")

read async 

read() -> str | bytes

Read the file content.

Source code in src/mcp/server/mcpserver/resources/types.py 
155
156
157
158
159
160
161
162
async def read(self) -> str | bytes:
    """Read the file content."""
    try:
        if self.is_binary:
            return await anyio.to_thread.run_sync(self.path.read_bytes)
        return await anyio.to_thread.run_sync(self.path.read_text)
    except Exception as e:
        raise ValueError(f"Error reading file {self.path}: {e}")

FunctionResource

Bases: Resource

A resource that defers data loading by wrapping a function.

The function is only called when the resource is read, allowing for lazy loading of potentially expensive data. This is particularly useful when listing resources, as the function won't be called until the resource is actually accessed.

The function can return: - str for text content (default) - bytes for binary content - other types will be converted to JSON

Source code in src/mcp/server/mcpserver/resources/types.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class FunctionResource(Resource):
    """A resource that defers data loading by wrapping a function.

    The function is only called when the resource is read, allowing for lazy loading
    of potentially expensive data. This is particularly useful when listing resources,
    as the function won't be called until the resource is actually accessed.

    The function can return:
    - str for text content (default)
    - bytes for binary content
    - other types will be converted to JSON
    """

    fn: Callable[[], Any] = Field(exclude=True)

    async def read(self) -> str | bytes:
        """Read the resource by calling the wrapped function."""
        try:
            fn = self.fn
            if is_async_callable(fn):
                result = await fn()
            else:
                result = await anyio.to_thread.run_sync(self.fn)

            if isinstance(result, InputRequiredResult):
                # A static resource function can never read the retry's
                # input_responses (it takes no Context), so this can only be a
                # mistake — reject it instead of JSON-dumping it as content.
                raise ValueError(
                    "static resources cannot return InputRequiredResult; only resource "
                    "template functions participate in the multi-round-trip flow"
                )
            if isinstance(result, Resource):  # pragma: no cover
                return await result.read()
            elif isinstance(result, bytes):
                return result
            elif isinstance(result, str):
                return result
            else:
                return pydantic_core.to_json(result, fallback=str, indent=2).decode()
        except MCPError:
            raise
        except Exception as e:
            raise ValueError(f"Error reading resource {self.uri}: {e}")

    @classmethod
    def from_function(
        cls,
        fn: Callable[..., Any],
        uri: str,
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        mime_type: str | None = None,
        icons: list[Icon] | None = None,
        annotations: Annotations | None = None,
        meta: dict[str, Any] | None = None,
    ) -> FunctionResource:
        """Create a FunctionResource from a function."""
        func_name = name or fn.__name__
        if func_name == "<lambda>":  # pragma: no cover
            raise ValueError("You must provide a name for lambda functions")

        # ensure the arguments are properly cast
        fn = validate_call(fn)

        return cls(
            uri=uri,
            name=func_name,
            title=title,
            description=description or fn.__doc__ or "",
            mime_type=mime_type or "text/plain",
            fn=fn,
            icons=icons,
            annotations=annotations,
            meta=meta,
        )

read async 

read() -> str | bytes

Read the resource by calling the wrapped function.

Source code in src/mcp/server/mcpserver/resources/types.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def read(self) -> str | bytes:
    """Read the resource by calling the wrapped function."""
    try:
        fn = self.fn
        if is_async_callable(fn):
            result = await fn()
        else:
            result = await anyio.to_thread.run_sync(self.fn)

        if isinstance(result, InputRequiredResult):
            # A static resource function can never read the retry's
            # input_responses (it takes no Context), so this can only be a
            # mistake — reject it instead of JSON-dumping it as content.
            raise ValueError(
                "static resources cannot return InputRequiredResult; only resource "
                "template functions participate in the multi-round-trip flow"
            )
        if isinstance(result, Resource):  # pragma: no cover
            return await result.read()
        elif isinstance(result, bytes):
            return result
        elif isinstance(result, str):
            return result
        else:
            return pydantic_core.to_json(result, fallback=str, indent=2).decode()
    except MCPError:
        raise
    except Exception as e:
        raise ValueError(f"Error reading resource {self.uri}: {e}")

from_function classmethod 

from_function(
    fn: Callable[..., Any],
    uri: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
) -> FunctionResource

Create a FunctionResource from a function.

Source code in src/mcp/server/mcpserver/resources/types.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@classmethod
def from_function(
    cls,
    fn: Callable[..., Any],
    uri: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
) -> FunctionResource:
    """Create a FunctionResource from a function."""
    func_name = name or fn.__name__
    if func_name == "<lambda>":  # pragma: no cover
        raise ValueError("You must provide a name for lambda functions")

    # ensure the arguments are properly cast
    fn = validate_call(fn)

    return cls(
        uri=uri,
        name=func_name,
        title=title,
        description=description or fn.__doc__ or "",
        mime_type=mime_type or "text/plain",
        fn=fn,
        icons=icons,
        annotations=annotations,
        meta=meta,
    )

HttpResource

Bases: Resource

A resource that reads from an HTTP endpoint.

Source code in src/mcp/server/mcpserver/resources/types.py 
165
166
167
168
169
170
171
172
173
174
175
176
class HttpResource(Resource):
    """A resource that reads from an HTTP endpoint."""

    url: str = Field(description="URL to fetch content from")
    mime_type: str = Field(default="application/json", description="MIME type of the resource content")

    async def read(self) -> str | bytes:
        """Read the HTTP content."""
        async with httpx.AsyncClient() as client:  # pragma: no cover
            response = await client.get(self.url)
            response.raise_for_status()
            return response.text

read async 

read() -> str | bytes

Read the HTTP content.

Source code in src/mcp/server/mcpserver/resources/types.py 
171
172
173
174
175
176
async def read(self) -> str | bytes:
    """Read the HTTP content."""
    async with httpx.AsyncClient() as client:  # pragma: no cover
        response = await client.get(self.url)
        response.raise_for_status()
        return response.text

TextResource

Bases: Resource

A resource that reads from a string.

Source code in src/mcp/server/mcpserver/resources/types.py 
23
24
25
26
27
28
29
30
class TextResource(Resource):
    """A resource that reads from a string."""

    text: str = Field(description="Text content of the resource")

    async def read(self) -> str:
        """Read the text content."""
        return self.text

read async 

read() -> str

Read the text content.

Source code in src/mcp/server/mcpserver/resources/types.py 
28
29
30
async def read(self) -> str:
    """Read the text content."""
    return self.text