Skip to content

resource_manager

Resource manager functionality.

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())