FastMCP
from pathlib import Path
from tempfile import NamedTemporaryFile
import pytest
from pydantic import AnyUrl, FileUrl
from mcp.server.fastmcp.resources import (
FileResource,
FunctionResource,
ResourceManager,
ResourceTemplate,
)
@pytest.fixture
def temp_file():
"""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:
path.unlink()
except FileNotFoundError:
pass # File was already deleted by the test
class TestResourceManager:
"""Test ResourceManager functionality."""
def test_add_resource(self, temp_file: Path):
"""Test adding a resource."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(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=FileUrl(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):
"""Test warning on duplicate resources."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(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):
"""Test disabling warning on duplicate resources."""
manager = ResourceManager(warn_on_duplicate_resources=False)
resource = FileResource(
uri=FileUrl(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=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
manager.add_resource(resource)
retrieved = await manager.get_resource(resource.uri)
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"))
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"))
def test_list_resources(self, temp_file: Path):
"""Test listing all resources."""
manager = ResourceManager()
resource1 = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test1",
path=temp_file,
)
resource2 = FileResource(
uri=FileUrl(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]