modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add resources parameter to MCPServer (#2414)

Browse source 
Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>
This commit is contained in:
Matt LeMay 2026-04-12 07:42:12 -05:00 • committed by GitHub
commit c5f12ec1e9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 170 additions and 185 deletions
Download patch file Download diff file Expand all files Collapse all files

16
src/mcp/server/mcpserver/resources/resource_manager.py
View file

@ -22,28 +22,26 @@ logger = get_logger(__name__)
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
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:

5
src/mcp/server/mcpserver/server.py
View file

@ -140,6 +140,7 @@ class MCPServer(Generic[LifespanResultT]):
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,9 @@ class MCPServer(Generic[LifespanResultT]):
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",

16
src/mcp/server/mcpserver/tools/tool_manager.py
View file

@ -18,18 +18,12 @@ logger = get_logger(__name__)
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

256
tests/server/mcpserver/resources/test_resource_manager.py
View file

@ -1,5 +1,5 @@
import logging
from pathlib import Path
from tempfile import NamedTemporaryFile
import pytest
from pydantic import AnyUrl
@ -8,170 +8,134 @@ from mcp.server.mcpserver import Context
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
tmp_file = tmp_path / "file"
tmp_file.touch()
yield tmp_file
class TestResourceManager:
"""Test ResourceManager functionality."""
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]
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]
duplicate_resource = FileResource(uri=f"file://{temp_file}", name="duplicate", path=temp_file)
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]
with caplog.at_level(logging.WARNING):
manager = ResourceManager(True, resources=[resource, duplicate_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]
assert "Resource already exists" in caplog.text
assert manager.list_resources() == [resource]
class TestResourceManagerMetadata:
"""Test ResourceManager Metadata"""
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_template_with_metadata(self):
"""Test that ResourceManager.add_template() accepts and passes meta parameter."""
manager = ResourceManager()
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 get_item(id: str) -> str: # pragma: no cover
return f"Item {id}"
metadata = {"source": "database", "cached": True}
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
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_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
def test_add_template_without_metadata(self):
"""Test that ResourceManager.add_template() works without meta parameter."""
manager = ResourceManager()
@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
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}",
)
@pytest.mark.anyio
async def test_get_resource_from_template():
"""Test getting a resource through a template."""
manager = ResourceManager()
assert template.meta is None
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

26
tests/server/mcpserver/test_server.py
View file

@ -685,6 +685,32 @@ class TestServerTools:
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()