mirror of
https://github.com/aljazceru/mcp-python-sdk.git
synced 2025-12-19 06:54:18 +01:00
Integrate FastMCP
This commit integrates FastMCP, a high-level MCP server implementation originally written by Jeremiah Lowin, into the official MCP SDK. It also updates dependencies and adds new dev dependencies. It moves the existing SDK into a .lowlevel .
This commit is contained in:
137
tests/server/fastmcp/resources/test_resource_manager.py
Normal file
137
tests/server/fastmcp/resources/test_resource_manager.py
Normal file
@@ -0,0 +1,137 @@
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from tempfile import NamedTemporaryFile
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
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!"
|
||||
|
||||
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]
|
||||
Reference in New Issue
Block a user