From 0933e6538e4be1014150d368e1990d2669b6cc8b Mon Sep 17 00:00:00 2001 From: Matt LeMay Date: Thu, 9 Apr 2026 13:21:06 -0500 Subject: [PATCH 1/3] feat: allow resources in mcpserver init reported-by: user --- src/mcp/server/mcpserver/server.py | 4 ++++ tests/server/mcpserver/test_server.py | 30 +++++++++++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py index 6f9bb0e28..8ffb29021 100644 --- a/src/mcp/server/mcpserver/server.py +++ b/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, @@ -163,6 +164,9 @@ def __init__( 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) + if resources is not None: + for resource in resources: + self._resource_manager.add_resource(resource) self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts) self._lowlevel_server = Server( name=name or "mcp-server", diff --git a/tests/server/mcpserver/test_server.py b/tests/server/mcpserver/test_server.py index 49b6deb4b..d1f90cac1 100644 --- a/tests/server/mcpserver/test_server.py +++ b/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() From b121760f710f9e6e278cce613e63571a9b1165b8 Mon Sep 17 00:00:00 2001 From: Matt LeMay Date: Thu, 9 Apr 2026 13:27:46 -0500 Subject: [PATCH 2/3] refactor: move resource seeding into resource manager --- .../mcpserver/resources/resource_manager.py | 10 ++++++- src/mcp/server/mcpserver/server.py | 8 ++--- .../resources/test_resource_manager.py | 29 ++++++++++++++++++- 3 files changed, 41 insertions(+), 6 deletions(-) diff --git a/src/mcp/server/mcpserver/resources/resource_manager.py b/src/mcp/server/mcpserver/resources/resource_manager.py index 6bf17376d..a8020d29e 100644 --- a/src/mcp/server/mcpserver/resources/resource_manager.py +++ b/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. diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py index 8ffb29021..7493ac511 100644 --- a/src/mcp/server/mcpserver/server.py +++ b/src/mcp/server/mcpserver/server.py @@ -163,10 +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) - if resources is not None: - for resource in resources: - self._resource_manager.add_resource(resource) + 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", diff --git a/tests/server/mcpserver/resources/test_resource_manager.py b/tests/server/mcpserver/resources/test_resource_manager.py index 724b57997..a95867cc6 100644 --- a/tests/server/mcpserver/resources/test_resource_manager.py +++ b/tests/server/mcpserver/resources/test_resource_manager.py @@ -1,3 +1,4 @@ +import logging from pathlib import Path from tempfile import NamedTemporaryFile @@ -5,7 +6,12 @@ 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() From 28303efcd0ad0860275e6b2d2be9c8bb3703c5ff Mon Sep 17 00:00:00 2001 From: Marcelo Trylesinski Date: Sun, 12 Apr 2026 14:35:49 +0200 Subject: [PATCH 3/3] Clean up --- .../mcpserver/resources/resource_manager.py | 24 +- src/mcp/server/mcpserver/server.py | 3 +- .../server/mcpserver/tools/tool_manager.py | 16 +- .../resources/test_resource_manager.py | 319 +++++++----------- tests/server/mcpserver/test_server.py | 6 +- 5 files changed, 142 insertions(+), 226 deletions(-) diff --git a/src/mcp/server/mcpserver/resources/resource_manager.py b/src/mcp/server/mcpserver/resources/resource_manager.py index a8020d29e..766cf51ae 100644 --- a/src/mcp/server/mcpserver/resources/resource_manager.py +++ b/src/mcp/server/mcpserver/resources/resource_manager.py @@ -22,36 +22,26 @@ 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): 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) + + 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 + resource: A Resource instance to add. Returns: - The added resource. If a resource with the same URI already exists, - returns the existing resource. + 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, - }, + extra={"uri": resource.uri, "type": type(resource).__name__, "resource_name": resource.name}, ) existing = self._resources.get(str(resource.uri)) if existing: diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py index 7493ac511..be77705da 100644 --- a/src/mcp/server/mcpserver/server.py +++ b/src/mcp/server/mcpserver/server.py @@ -164,8 +164,7 @@ 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, 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( diff --git a/src/mcp/server/mcpserver/tools/tool_manager.py b/src/mcp/server/mcpserver/tools/tool_manager.py index 32ed54797..eef4911f9 100644 --- a/src/mcp/server/mcpserver/tools/tool_manager.py +++ b/src/mcp/server/mcpserver/tools/tool_manager.py @@ -18,18 +18,12 @@ class ToolManager: """Manages MCPServer tools.""" - def __init__( - self, - warn_on_duplicate_tools: bool = True, - *, - tools: list[Tool] | None = None, - ): + def __init__(self, warn_on_duplicate_tools: bool = True, *, tools: list[Tool] | None = None): self._tools: dict[str, Tool] = {} - if tools is not None: - for tool in tools: - if warn_on_duplicate_tools and tool.name in self._tools: - logger.warning(f"Tool already exists: {tool.name}") - self._tools[tool.name] = tool + for tool in tools or (): + if warn_on_duplicate_tools and tool.name in self._tools: + logger.warning(f"Tool already exists: {tool.name}") + self._tools[tool.name] = tool self.warn_on_duplicate_tools = warn_on_duplicate_tools diff --git a/tests/server/mcpserver/resources/test_resource_manager.py b/tests/server/mcpserver/resources/test_resource_manager.py index a95867cc6..b91c71581 100644 --- a/tests/server/mcpserver/resources/test_resource_manager.py +++ b/tests/server/mcpserver/resources/test_resource_manager.py @@ -1,204 +1,141 @@ 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 -def temp_file(): +@pytest.fixture() +def temp_file(tmp_path: Path): """Create a temporary file for testing. File is automatically cleaned up after the test if it still exists. """ - content = "test content" - with NamedTemporaryFile(mode="w", delete=False) as f: - f.write(content) - path = Path(f.name).resolve() - yield path - try: # pragma: lax no cover - path.unlink() - except FileNotFoundError: # pragma: lax no cover - pass # File was already deleted by the test - - -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() - resource = FileResource( - uri=f"file://{temp_file}", - name="test", - path=temp_file, - ) - added = manager.add_resource(resource) - assert added == resource - assert manager.list_resources() == [resource] - - def test_add_duplicate_resource(self, temp_file: Path): - """Test adding the same resource twice.""" - manager = ResourceManager() - resource = FileResource( - uri=f"file://{temp_file}", - name="test", - path=temp_file, - ) - first = manager.add_resource(resource) - second = manager.add_resource(resource) - assert first == second - assert manager.list_resources() == [resource] - - def test_warn_on_duplicate_resources(self, temp_file: Path, caplog: pytest.LogCaptureFixture): - """Test warning on duplicate resources.""" - manager = ResourceManager() - resource = FileResource( - uri=f"file://{temp_file}", - name="test", - path=temp_file, - ) - manager.add_resource(resource) - manager.add_resource(resource) - assert "Resource already exists" in caplog.text - - def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog: pytest.LogCaptureFixture): - """Test disabling warning on duplicate resources.""" - manager = ResourceManager(warn_on_duplicate_resources=False) - resource = FileResource( - uri=f"file://{temp_file}", - name="test", - path=temp_file, - ) - manager.add_resource(resource) - manager.add_resource(resource) - assert "Resource already exists" not in caplog.text - - @pytest.mark.anyio - async def test_get_resource(self, temp_file: Path): - """Test getting a resource by URI.""" - manager = ResourceManager() - resource = FileResource( - uri=f"file://{temp_file}", - name="test", - path=temp_file, - ) - manager.add_resource(resource) - retrieved = await manager.get_resource(resource.uri, Context()) - assert retrieved == resource - - @pytest.mark.anyio - async def test_get_resource_from_template(self): - """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 - - resource = await manager.get_resource(AnyUrl("greet://world"), Context()) - assert isinstance(resource, FunctionResource) - content = await resource.read() - assert content == "Hello, world!" - - @pytest.mark.anyio - async def test_get_unknown_resource(self): - """Test getting a non-existent resource.""" - manager = ResourceManager() - with pytest.raises(ValueError, match="Unknown resource"): - await manager.get_resource(AnyUrl("unknown://test"), Context()) - - def test_list_resources(self, temp_file: Path): - """Test listing all resources.""" - manager = ResourceManager() - resource1 = FileResource( - uri=f"file://{temp_file}", - name="test1", - path=temp_file, - ) - resource2 = FileResource( - uri=f"file://{temp_file}2", - name="test2", - path=temp_file, - ) - manager.add_resource(resource1) - manager.add_resource(resource2) - resources = manager.list_resources() - assert len(resources) == 2 - assert resources == [resource1, resource2] - - -class TestResourceManagerMetadata: - """Test ResourceManager Metadata""" - - def test_add_template_with_metadata(self): - """Test that ResourceManager.add_template() accepts and passes meta parameter.""" - - manager = ResourceManager() - - def get_item(id: str) -> str: # pragma: no cover - return f"Item {id}" - - metadata = {"source": "database", "cached": True} - - template = manager.add_template( - fn=get_item, - uri_template="resource://items/{id}", - meta=metadata, - ) - - assert template.meta is not None - assert template.meta == metadata - assert template.meta["source"] == "database" - assert template.meta["cached"] is True - - def test_add_template_without_metadata(self): - """Test that ResourceManager.add_template() works without meta parameter.""" - - manager = ResourceManager() - - def get_item(id: str) -> str: # pragma: no cover - return f"Item {id}" - - template = manager.add_template( - fn=get_item, - uri_template="resource://items/{id}", - ) - - assert template.meta is None + tmp_file = tmp_path / "file" + tmp_file.touch() + yield tmp_file + + +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]) + 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(temp_file: Path): + """Test adding a resource.""" + manager = ResourceManager() + resource = FileResource(uri=f"file://{temp_file}", name="test", path=temp_file) + added = manager.add_resource(resource) + assert added == resource + assert manager.list_resources() == [resource] + + +def test_add_duplicate_resource(temp_file: Path): + """Test adding the same resource twice.""" + manager = ResourceManager() + resource = FileResource(uri=f"file://{temp_file}", name="test", path=temp_file) + first = manager.add_resource(resource) + second = manager.add_resource(resource) + assert first == second + assert manager.list_resources() == [resource] + + +def test_warn_on_duplicate_resources(temp_file: Path, caplog: pytest.LogCaptureFixture): + """Test warning on duplicate resources.""" + manager = ResourceManager() + resource = FileResource(uri=f"file://{temp_file}", name="test", path=temp_file) + manager.add_resource(resource) + manager.add_resource(resource) + assert "Resource already exists" in caplog.text + + +def test_disable_warn_on_duplicate_resources(temp_file: Path, caplog: pytest.LogCaptureFixture): + """Test disabling warning on duplicate resources.""" + manager = ResourceManager(warn_on_duplicate_resources=False) + resource = FileResource(uri=f"file://{temp_file}", name="test", path=temp_file) + manager.add_resource(resource) + manager.add_resource(resource) + assert "Resource already exists" not in caplog.text + + +@pytest.mark.anyio +async def test_get_resource(temp_file: Path): + """Test getting a resource by URI.""" + manager = ResourceManager() + resource = FileResource(uri=f"file://{temp_file}", name="test", path=temp_file) + manager.add_resource(resource) + retrieved = await manager.get_resource(resource.uri, Context()) + assert retrieved == resource + + +@pytest.mark.anyio +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 + + resource = await manager.get_resource(AnyUrl("greet://world"), Context()) + assert isinstance(resource, FunctionResource) + content = await resource.read() + assert content == "Hello, world!" + + +@pytest.mark.anyio +async def test_get_unknown_resource(): + """Test getting a non-existent resource.""" + manager = ResourceManager() + with pytest.raises(ValueError, match="Unknown resource"): + await manager.get_resource(AnyUrl("unknown://test"), Context()) + + +def test_list_resources(temp_file: Path): + """Test listing all resources.""" + manager = ResourceManager() + resource1 = FileResource(uri=f"file://{temp_file}", name="test1", path=temp_file) + resource2 = FileResource(uri=f"file://{temp_file}2", name="test2", path=temp_file) + + manager.add_resource(resource1) + manager.add_resource(resource2) + + resources = manager.list_resources() + assert len(resources) == 2 + assert resources == [resource1, resource2] + + +def get_item(id: str) -> str: ... + + +def test_add_template_with_metadata(): + """Test that ResourceManager.add_template() accepts and passes meta parameter.""" + manager = ResourceManager() + metadata = {"source": "database", "cached": True} + template = manager.add_template(fn=get_item, uri_template="resource://items/{id}", meta=metadata) + + assert template.meta is not None + assert template.meta == metadata + assert template.meta["source"] == "database" + assert template.meta["cached"] is True + + +def test_add_template_without_metadata(): + """Test that ResourceManager.add_template() works without meta parameter.""" + manager = ResourceManager() + template = manager.add_template(fn=get_item, uri_template="resource://items/{id}") + assert template.meta is None diff --git a/tests/server/mcpserver/test_server.py b/tests/server/mcpserver/test_server.py index d1f90cac1..3457ec944 100644 --- a/tests/server/mcpserver/test_server.py +++ b/tests/server/mcpserver/test_server.py @@ -690,11 +690,7 @@ def get_text() -> str: """Seeded resource.""" return "Hello from init!" - resource = FunctionResource.from_function( - fn=get_text, - uri="resource://init", - name="init_resource", - ) + resource = FunctionResource.from_function(fn=get_text, uri="resource://init", name="init_resource") mcp = MCPServer(resources=[resource])