feat: add_resource_template to mcpserver and resource manager

src/mcp/server/mcpserver/resources/resource_manager.py

@@ -22,13 +22,21 @@
 class ResourceManager:
     """Manages MCPServer resources."""
 
-    def __init__(self, warn_on_duplicate_resources: bool = True, *, resources: list[Resource] | None = None):
+    def __init__(
+        self,
+        warn_on_duplicate_resources: bool = True,
+        *,
+        resources: list[Resource] | None = None,
+        resource_templates: list[ResourceTemplate] | 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)
+        for template in resource_templates or ():
+            self.add_resource_template(template)
 
     def add_resource(self, resource: Resource) -> Resource:
         """Add a resource to the manager.
@@ -51,6 +59,30 @@ def add_resource(self, resource: Resource) -> Resource:
         self._resources[str(resource.uri)] = resource
         return resource
 
+    def add_resource_template(self, template: ResourceTemplate) -> ResourceTemplate:
+        """Add a resource template to the manager.
+
+        Args:
+            template: A ResourceTemplate instance to add.
+
+        Returns:
+            The added template. If a template with the same URI template already exists, returns the existing template.
+        """
+        logger.debug(
+            "Adding resource template",
+            extra={
+                "uri_template": template.uri_template,
+                "type": type(template).__name__,
+                "resource_name": template.name,
+            },
+        )
+        if existing := self._templates.get(template.uri_template):
+            if self.warn_on_duplicate_resources:
+                logger.warning(f"Resource template already exists: {template.uri_template}")
+            return existing
+        self._templates[template.uri_template] = template
+        return template
+
     def add_template(
         self,
         fn: Callable[..., Any],
@@ -75,8 +107,7 @@ def add_template(
             annotations=annotations,
             meta=meta,
         )
-        self._templates[template.uri_template] = template
-        return template
+        return self.add_resource_template(template)
 
     async def get_resource(self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]) -> Resource:
         """Get resource by URI, checking concrete resources first, then templates."""

src/mcp/server/mcpserver/server.py

@@ -33,7 +33,7 @@
 from mcp.server.mcpserver.context import Context
 from mcp.server.mcpserver.exceptions import ResourceError
 from mcp.server.mcpserver.prompts import Prompt, PromptManager
-from mcp.server.mcpserver.resources import FunctionResource, Resource, ResourceManager
+from mcp.server.mcpserver.resources import FunctionResource, Resource, ResourceManager, ResourceTemplate
 from mcp.server.mcpserver.tools import Tool, ToolManager
 from mcp.server.mcpserver.utilities.context_injection import find_context_parameter
 from mcp.server.mcpserver.utilities.logging import configure_logging, get_logger
@@ -141,6 +141,7 @@ def __init__(
         *,
         tools: list[Tool] | None = None,
         resources: list[Resource] | None = None,
+        resource_templates: list[ResourceTemplate] | None = None,
         debug: bool = False,
         log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
         warn_on_duplicate_resources: bool = True,
@@ -164,7 +165,9 @@ def __init__(
 
         self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
         self._resource_manager = ResourceManager(
-            resources=resources, warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources
+            resources=resources,
+            resource_templates=resource_templates,
+            warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources,
         )
         self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
         self._lowlevel_server = Server(
@@ -624,6 +627,14 @@ def add_resource(self, resource: Resource) -> None:
         """
         self._resource_manager.add_resource(resource)
 
+    def add_resource_template(self, template: ResourceTemplate) -> None:
+        """Add a resource template to the server.
+
+        Args:
+            template: A ResourceTemplate instance to add
+        """
+        self._resource_manager.add_resource_template(template)
+
     def resource(
         self,
         uri: str,

tests/interaction/transports/test_stdio.py

@@ -92,7 +92,8 @@ async def collect(params: LoggingMessageNotificationParams) -> None:
     # seeing it proves the process exited on its own rather than via the transport's terminate
     # escalation, without a timing-based assertion. The capture itself proves stderr passthrough:
     # the transport routes the child's stderr to the caller's `errlog` without consuming it.
-    assert captured_stderr == snapshot("stdio-echo: clean exit\n")
+    # Match the trailing line only: older anyio on py3.14 lowest-direct can emit SyntaxWarning to stderr.
+    assert captured_stderr.endswith("stdio-echo: clean exit\n")
 
 
 @requirement("transport:stdio:stream-purity")

tests/server/mcpserver/resources/test_resource_manager.py

@@ -19,6 +19,16 @@ def temp_file(tmp_path: Path):
     yield tmp_file
 
 
+def greet(name: str) -> str:
+    return f"Hello, {name}!"
+
+
+def test_init_with_resource_templates():
+    template = ResourceTemplate.from_function(fn=greet, uri_template="greet://{name}", name="greeter")
+    manager = ResourceManager(resource_templates=[template])
+    assert manager.list_templates() == [template]
+
+
 def test_init_with_resources(temp_file: Path, caplog: pytest.LogCaptureFixture):
     resource = FileResource(uri=f"file://{temp_file}", name="test", path=temp_file)
     manager = ResourceManager(resources=[resource])
@@ -85,11 +95,8 @@ async def test_get_resource_from_template():
     """Test getting a resource through a template."""
     manager = ResourceManager()
 
-    def greet(name: str) -> str:
-        return f"Hello, {name}!"
-
     template = ResourceTemplate.from_function(fn=greet, uri_template="greet://{name}", name="greeter")
-    manager._templates[template.uri_template] = template
+    manager.add_resource_template(template)
 
     resource = await manager.get_resource(AnyUrl("greet://world"), Context())
     assert isinstance(resource, FunctionResource)
@@ -122,6 +129,47 @@ def test_list_resources(temp_file: Path):
 def get_item(id: str) -> str: ...
 
 
+def test_add_resource_template():
+    """Test adding a resource template."""
+    manager = ResourceManager()
+
+    template = ResourceTemplate.from_function(fn=greet, uri_template="greet://{name}", name="greeter")
+    added = manager.add_resource_template(template)
+    assert added == template
+    assert manager.list_templates() == [template]
+
+
+def test_add_duplicate_resource_template():
+    """Test adding the same resource template twice."""
+    manager = ResourceManager()
+
+    template = ResourceTemplate.from_function(fn=greet, uri_template="greet://{name}", name="greeter")
+    first = manager.add_resource_template(template)
+    second = manager.add_resource_template(template)
+    assert first == second
+    assert manager.list_templates() == [template]
+
+
+def test_warn_on_duplicate_resource_templates(caplog: pytest.LogCaptureFixture):
+    """Test warning on duplicate resource templates."""
+    manager = ResourceManager()
+
+    template = ResourceTemplate.from_function(fn=greet, uri_template="greet://{name}", name="greeter")
+    manager.add_resource_template(template)
+    manager.add_resource_template(template)
+    assert "Resource template already exists" in caplog.text
+
+
+def test_disable_warn_on_duplicate_resource_templates(caplog: pytest.LogCaptureFixture):
+    """Test disabling warning on duplicate resource templates."""
+    manager = ResourceManager(warn_on_duplicate_resources=False)
+
+    template = ResourceTemplate.from_function(fn=greet, uri_template="greet://{name}", name="greeter")
+    manager.add_resource_template(template)
+    manager.add_resource_template(template)
+    assert "Resource template already exists" not in caplog.text
+
+
 def test_add_template_with_metadata():
     """Test that ResourceManager.add_template() accepts and passes meta parameter."""
     manager = ResourceManager()

tests/server/mcpserver/test_server.py

@@ -15,6 +15,7 @@
 from mcp.server.mcpserver.exceptions import ToolError
 from mcp.server.mcpserver.prompts.base import Message, UserMessage
 from mcp.server.mcpserver.resources import FileResource, FunctionResource
+from mcp.server.mcpserver.resources import ResourceTemplate as ServerResourceTemplate
 from mcp.server.mcpserver.utilities.types import Audio, Image
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
@@ -710,6 +711,58 @@ def get_text() -> str:
             assert isinstance(content, TextResourceContents)
             assert content.text == "Hello from init!"
 
+    async def test_init_with_resource_templates(self):
+        def get_weather(city: str) -> str:
+            """Seeded template."""
+            return f"Weather for {city}"
+
+        template = ServerResourceTemplate.from_function(
+            fn=get_weather,
+            uri_template="weather://{city}",
+            name="weather",
+            description="Seeded template.",
+        )
+
+        mcp = MCPServer(resource_templates=[template])
+
+        async with Client(mcp) as client:
+            templates = await client.list_resource_templates()
+            assert len(templates.resource_templates) == 1
+            listed = templates.resource_templates[0]
+            assert listed.uri_template == "weather://{city}"
+            assert listed.name == "weather"
+            assert listed.description == "Seeded template."
+
+            result = await client.read_resource("weather://london")
+
+            assert len(result.contents) == 1
+            content = result.contents[0]
+            assert isinstance(content, TextResourceContents)
+            assert content.text == "Weather for london"
+
+    async def test_add_resource_template(self):
+        mcp = MCPServer()
+
+        def get_weather(city: str) -> str:
+            return f"Weather for {city}"
+
+        template = ServerResourceTemplate.from_function(
+            fn=get_weather,
+            uri_template="weather://{city}",
+            name="weather",
+        )
+        mcp.add_resource_template(template)
+
+        async with Client(mcp) as client:
+            templates = await client.list_resource_templates()
+            assert len(templates.resource_templates) == 1
+            assert templates.resource_templates[0].uri_template == "weather://{city}"
+
+            result = await client.read_resource("weather://paris")
+
+            assert isinstance(result.contents[0], TextResourceContents)
+            assert result.contents[0].text == "Weather for paris"
+
     async def test_text_resource(self):
         mcp = MCPServer()