Skip to content

resource_manager

Resource manager functionality.

ResourceManager

Manages MCPServer resources.

Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
 22
 23
 24
 25
 26
 27
 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
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,
    ) -> 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,
        )
        self._templates[template.uri_template] = template
        return template

    async def get_resource(self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]) -> Resource:
        """Get resource by URI, checking concrete resources first, then templates."""
        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():
            if params := template.matches(uri_str):
                try:
                    return await template.create_resource(uri_str, params, context=context)
                except Exception as e:  # pragma: no cover
                    raise ValueError(f"Error creating resource from template: {e}")

        raise ValueError(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 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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,
) -> ResourceTemplate

Add a template from a function.

Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
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
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,
) -> 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,
    )
    self._templates[template.uri_template] = template
    return template

get_resource async 

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

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

Source code in src/mcp/server/mcpserver/resources/resource_manager.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
async def get_resource(self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]) -> Resource:
    """Get resource by URI, checking concrete resources first, then templates."""
    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():
        if params := template.matches(uri_str):
            try:
                return await template.create_resource(uri_str, params, context=context)
            except Exception as e:  # pragma: no cover
                raise ValueError(f"Error creating resource from template: {e}")

    raise ValueError(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 
100
101
102
103
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 
105
106
107
108
def list_templates(self) -> list[ResourceTemplate]:
    """List all registered templates."""
    logger.debug("Listing templates", extra={"count": len(self._templates)})
    return list(self._templates.values())