feat: add resource seeding to mcpserver

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

@@ -22,10 +22,18 @@
 class ResourceManager:
     """Manages MCPServer resources."""
 
-    def __init__(self, warn_on_duplicate_resources: bool = True):
+    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
+        if resources is not None:
+            for resource in resources:
+                self.add_resource(resource)
 
     def add_resource(self, resource: Resource) -> Resource:
         """Add a resource to the manager.

src/mcp/server/mcpserver/server.py

@@ -140,6 +140,7 @@ def __init__(
         token_verifier: TokenVerifier | None = None,
         *,
         tools: list[Tool] | None = None,
+        resources: list[Resource] | None = None,
         debug: bool = False,
         log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
         warn_on_duplicate_resources: bool = True,
@@ -162,7 +163,10 @@ def __init__(
         self.dependencies = self.settings.dependencies
 
         self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
-        self._resource_manager = ResourceManager(warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources)
+        self._resource_manager = ResourceManager(
+            resources=resources,
+            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(
             name=name or "mcp-server",

tests/server/mcpserver/resources/test_resource_manager.py

@@ -1,11 +1,17 @@
+import logging
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 
 import pytest
 from pydantic import AnyUrl
 
 from mcp.server.mcpserver import Context
-from mcp.server.mcpserver.resources import FileResource, FunctionResource, ResourceManager, ResourceTemplate
+from mcp.server.mcpserver.resources import (
+    FileResource,
+    FunctionResource,
+    ResourceManager,
+    ResourceTemplate,
+)
 
 
 @pytest.fixture
@@ -28,6 +34,27 @@ def temp_file():
 class TestResourceManager:
     """Test ResourceManager functionality."""
 
+    def test_init_with_resources(self, temp_file: Path, caplog: pytest.LogCaptureFixture):
+        resource = FileResource(
+            uri=f"file://{temp_file}",
+            name="test",
+            path=temp_file,
+        )
+        manager = ResourceManager(resources=[resource])
+        assert manager.list_resources() == [resource]
+
+        duplicate_resource = FileResource(
+            uri=f"file://{temp_file}",
+            name="duplicate",
+            path=temp_file,
+        )
+
+        with caplog.at_level(logging.WARNING):
+            manager = ResourceManager(True, resources=[resource, duplicate_resource])
+
+        assert "Resource already exists" in caplog.text
+        assert manager.list_resources() == [resource]
+
     def test_add_resource(self, temp_file: Path):
         """Test adding a resource."""
         manager = ResourceManager()

tests/server/mcpserver/test_server.py

@@ -685,6 +685,36 @@ async def test_remove_tool_and_call(self):
 
 
 class TestServerResources:
+    async def test_init_with_resources(self):
+        def get_text() -> str:
+            """Seeded resource."""
+            return "Hello from init!"
+
+        resource = FunctionResource.from_function(
+            fn=get_text,
+            uri="resource://init",
+            name="init_resource",
+        )
+
+        mcp = MCPServer(resources=[resource])
+
+        async with Client(mcp) as client:
+            assert client.initialize_result.capabilities.resources is not None
+
+            resources = await client.list_resources()
+            assert len(resources.resources) == 1
+            listed = resources.resources[0]
+            assert listed.uri == "resource://init"
+            assert listed.name == "init_resource"
+            assert listed.description == "Seeded resource."
+
+            result = await client.read_resource("resource://init")
+
+            assert len(result.contents) == 1
+            content = result.contents[0]
+            assert isinstance(content, TextResourceContents)
+            assert content.text == "Hello from init!"
+
     async def test_text_resource(self):
         mcp = MCPServer()