diff --git a/README.md b/README.md index 0c4b2b0a4..fa9f48e22 100644 --- a/README.md +++ b/README.md @@ -4,13 +4,26 @@ A fast, pythonic way to build Model Context Protocol (MCP) servers. -Anthropic's new [Model Context Protocol](https://modelcontextprotocol.io) is powerful way to give broadcast new functionality and context to LLMs. However, developing MCP servers can be cumbersome. FastMCP provides a simple, intuitive interface for creating MCP servers in Python. +Anthropic's new [Model Context Protocol](https://modelcontextprotocol.io) is a powerful way to give broadcast new functionality and context to LLMs. However, developing MCP servers can be cumbersome. FastMCP provides a simple, intuitive interface for creating MCP servers in Python. + +## Table of Contents + +- [FastMCP](#fastmcp) + - [Table of Contents](#table-of-contents) + - [Installation](#installation) + - [Quick Start](#quick-start) + - [Core Concepts](#core-concepts) + - [Resources](#resources) + - [Tools](#tools) + - [Development](#development) + - [Running the Dev Inspector](#running-the-dev-inspector) + - [Installing in Claude](#installing-in-claude) + - [License](#license) ## Installation MCP servers require you to use [uv](https://github.com/astral-sh/uv) as your dependency manager. - Install uv with brew: ```bash brew install uv @@ -22,8 +35,6 @@ Install FastMCP: uv pip install fastmcp ``` - - ## Quick Start Here's a simple example that exposes your desktop directory as a resource and provides a basic addition tool: @@ -50,35 +61,92 @@ if __name__ == "__main__": mcp.run() ``` -## Features +## Core Concepts + +FastMCP makes it easy to expose two types of functionality to LLMs: Resources and Tools. ### Resources -Resources are data sources that can be accessed by the LLM. They can be files, directories, or any other data source. Resources are defined using the `@resource` decorator: +Resources are data sources that can be accessed by the LLM. They're perfect for providing context like files, API responses, or database queries. + +FastMCP provides a simple `@resource` decorator that handles both static and dynamic resources. While the MCP spec distinguishes between resources and templates, FastMCP automatically handles this distinction based on your function signature: ```python -@mcp.resource("file://config.json") +# Static resource +@mcp.resource("resource://static") +def get_static() -> str: + """Return static content""" + return "Static content" + +# Dynamic resource +@mcp.resource("resource://{city}/weather") +def get_weather(city: str) -> str: + """Get weather for a city""" + return f"Weather for {city}" + +# Multiple parameters are supported +@mcp.resource("db://users/{user_id}/posts/{post_id}") +def get_user_post(user_id: int, post_id: int) -> dict: + """Get a specific post by a user""" + return { + "user_id": user_id, + "post_id": post_id, + "content": "Post content..." + } + +# File resources +@mcp.resource("file://config.json") def get_config() -> str: """Read the config file""" return Path("config.json").read_text() ``` +Resources can return: +- Strings for text content +- Bytes for binary content +- Other types will be converted to JSON + +When your resource URI includes parameters in curly braces (like `{city}`) and your function accepts matching arguments, FastMCP automatically sets up a template resource behind the scenes. This means you don't need to worry about the distinction between resources and templates in the MCP spec - just write your function, and FastMCP handles the rest. + +> **Note**: If you're familiar with the MCP spec, you might notice that dynamic resources are implemented as templates under the hood. FastMCP simplifies this by providing a unified interface through the `@resource` decorator. This is similar to how web frameworks often unify GET and POST handlers under a single route decorator. + + ### Tools -Tools are functions that can be called by the LLM. They are defined using the `@tool` decorator: +Tools are functions that can be called by the LLM to perform actions. They're great for calculations, API calls, or any interactive functionality. Tools are defined using the `@tool` decorator: ```python @mcp.tool() -def calculate(x: int, y: int) -> int: - """Perform a calculation""" - return x + y +def search_docs(query: str, max_results: int = 5) -> list[dict]: + """Search documentation for relevant entries""" + results = perform_search(query, limit=max_results) + return [{"title": r.title, "excerpt": r.excerpt} for r in results] + +@mcp.tool() +def analyze_image(image_path: str) -> dict: + """Analyze an image and return metadata""" + from PIL import Image + img = Image.open(image_path) + return { + "size": img.size, + "mode": img.mode, + "format": img.format + } ``` +Tools support: +- Type hints for parameters +- Default values +- Async functions +- Return value conversion to JSON + ## Development +FastMCP includes developer tools to make testing and debugging easier. + ### Running the Dev Inspector -FastMCP includes a development server with the MCP Inspector for testing your server: +The MCP Inspector helps you test your server during development: ```bash # Basic usage @@ -114,4 +182,6 @@ fastmcp install your_server.py --with pandas --with numpy fastmcp install your_server.py --with-editable . --with pandas --with numpy ``` +## License +Apache 2.0 \ No newline at end of file diff --git a/src/fastmcp/resources.py b/src/fastmcp/resources.py index 5fd7454ab..b1649c9b5 100644 --- a/src/fastmcp/resources.py +++ b/src/fastmcp/resources.py @@ -1,3 +1,4 @@ +import inspect import pydantic.json import abc import asyncio @@ -19,20 +20,30 @@ class Resource(BaseModel, abc.ABC): """Base class for all resources.""" uri: _BaseUrl = Field(description="URI of the resource") - name: str = Field(description="Name of the resource") - description: Optional[str] = Field(description="Description of the resource") - mime_type: Optional[str] = Field(description="MIME type of the resource content") + name: str = Field(description="Name of the resource", default=None) + description: Optional[str] = Field( + description="Description of the resource", default=None + ) + mime_type: str = Field( + default="text/plain", + description="MIME type of the resource content", + pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$", + ) @field_validator("name", mode="before") @classmethod def set_default_name(cls, name: str | None, info) -> str: """Set default name from URI if not provided.""" - if name is not None: + if name: return name # Extract everything after the protocol (e.g., "desktop" from "resource://desktop") uri = info.data.get("uri") if uri: - return str(uri).split("://", 1)[1] + uri_str = str(uri) + if "://" in uri_str: + name = uri_str.split("://", 1)[1] + if name: + return name raise ValueError("Either name or uri must be provided") @abc.abstractmethod @@ -40,14 +51,15 @@ class Resource(BaseModel, abc.ABC): """Read the resource content.""" pass + model_config = { + "validate_default": True, + } + class TextResource(Resource): - """A resource containing text content.""" + """A resource that reads from a string.""" text: str = Field(description="Text content of the resource") - mime_type: Optional[str] = Field( - default="text/plain", description="MIME type of the resource content" - ) async def read(self) -> str: """Read the text content.""" @@ -55,25 +67,62 @@ class TextResource(Resource): class BinaryResource(Resource): - """A resource containing binary content.""" + """A resource that reads from bytes.""" data: bytes = Field(description="Binary content of the resource") - mime_type: Optional[str] = Field( - default="application/octet-stream", - description="MIME type of the resource content", - ) async def read(self) -> bytes: """Read the binary content.""" return self.data +class FunctionResource(Resource): + """A resource that defers data loading by wrapping a function. + + The function is only called when the resource is read, allowing for lazy loading + of potentially expensive data. This is particularly useful when listing resources, + as the function won't be called until the resource is actually accessed. + + The function can return: + - str for text content (default) + - bytes for binary content + - other types will be converted to JSON + """ + + func: Callable[[], Any] = Field(exclude=True) + + async def read(self) -> Union[str, bytes]: + """Read the resource by calling the wrapped function.""" + try: + result = self.func() + if isinstance(result, Resource): + return await result.read() + if isinstance(result, bytes): + return result + if isinstance(result, str): + return result + try: + return json.dumps(result, default=pydantic.json.pydantic_encoder) + except TypeError: + # If JSON serialization fails, try str() + return str(result) + except Exception as e: + raise ValueError(f"Error reading resource {self.uri}: {e}") + + class FileResource(Resource): - """A resource that reads from a file.""" + """A resource that reads from a file. + + Set is_binary=True to read file as binary data instead of text. + """ path: Path = Field(description="Path to the file") - mime_type: Optional[str] = Field( - default="application/octet-stream", + is_binary: bool = Field( + default=False, + description="Whether to read the file as binary data", + ) + mime_type: str = Field( + default="text/plain", description="MIME type of the resource content", ) @@ -87,17 +136,12 @@ class FileResource(Resource): async def read(self) -> Union[str, bytes]: """Read the file content.""" - if self.mime_type and self.mime_type.startswith("text/"): - return await self._read_text() - return await self._read_binary() - - async def _read_text(self) -> str: - """Read file as text.""" - return await asyncio.to_thread(self.path.read_text) - - async def _read_binary(self) -> bytes: - """Read file as binary.""" - return await asyncio.to_thread(self.path.read_bytes) + try: + if self.is_binary: + return await asyncio.to_thread(self.path.read_bytes) + return await asyncio.to_thread(self.path.read_text) + except Exception as e: + raise ValueError(f"Error reading file {self.path}: {e}") class HttpResource(Resource): @@ -180,7 +224,7 @@ class ResourceTemplate(BaseModel): description: Optional[str] = Field( description="Description of what the resource does" ) - mime_type: Optional[str] = Field( + mime_type: str = Field( default="text/plain", description="MIME type of the resource content" ) func: Callable = Field(exclude=True) @@ -210,7 +254,6 @@ class ResourceTemplate(BaseModel): uri_template=uri_template, name=func_name, description=description or func.__doc__ or "", - mime_type=mime_type or "text/plain", func=func, parameters=parameters, ) @@ -226,30 +269,20 @@ class ResourceTemplate(BaseModel): async def create_resource(self, uri: str, params: Dict[str, Any]) -> Resource: """Create a resource from the template with the given parameters.""" - result = await self.func(**params) + try: + # Call function and check if result is a coroutine + result = self.func(**params) + if inspect.iscoroutine(result): + result = await result - if isinstance(result, bytes): - return BinaryResource( + return FunctionResource( uri=uri, name=self.name, description=self.description, - mime_type=self.mime_type, - data=result, - ) - - else: - if not isinstance(result, str): - try: - result = json.dumps(result, default=pydantic.json.pydantic_encoder) - except Exception as e: - raise ValueError(f"Error converting result to JSON: {e}") - return TextResource( - uri=uri, - name=self.name, - description=self.description, - mime_type=self.mime_type, - text=result, + func=lambda: result, # Capture result in closure ) + except Exception as e: + raise ValueError(f"Error creating resource from template: {e}") class ResourceManager: diff --git a/src/fastmcp/server.py b/src/fastmcp/server.py index 7f74b77f2..edd2d25c7 100644 --- a/src/fastmcp/server.py +++ b/src/fastmcp/server.py @@ -4,6 +4,8 @@ import asyncio import functools import json from typing import Any, Callable, Optional, Sequence, Union, Literal +import inspect +import re import pydantic.json from mcp.server import Server as MCPServer @@ -19,7 +21,11 @@ from pydantic_settings import BaseSettings from pydantic.networks import _BaseUrl from .exceptions import ResourceError -from .resources import Resource, FunctionResource, ResourceManager +from .resources import ( + Resource, + FunctionResource, + ResourceManager, +) from .tools import ToolManager, Image from .utilities.logging import get_logger, configure_logging @@ -138,7 +144,7 @@ class FastMCP: async def read_resource(self, uri: _BaseUrl) -> Union[str, bytes]: """Read a resource by URI.""" - resource = self._resource_manager.get_resource(uri) + resource = await self._resource_manager.get_resource(uri) if not resource: raise ResourceError(f"Unknown resource: {uri}") @@ -193,9 +199,17 @@ class FastMCP: """Decorator to register a function as a resource. The function will be called when the resource is read to generate its content. + The function can return: + - str for text content + - bytes for binary content + - other types will be converted to JSON + + If the URI contains parameters (e.g. "resource://{param}") or the function + has parameters, it will be registered as a template resource. Args: - uri: URI for the resource (e.g. "resource://my-resource") + uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}") + name: Optional name for the resource description: Optional description of the resource mime_type: Optional MIME type for the resource @@ -203,6 +217,10 @@ class FastMCP: @server.resource("resource://my-resource") def get_data() -> str: return "Hello, world!" + + @server.resource("resource://{city}/weather") + def get_weather(city: str) -> str: + return f"Weather for {city}" """ # Check if user passed function directly instead of calling decorator if callable(uri): @@ -211,63 +229,44 @@ class FastMCP: "Did you forget to call it? Use @resource('uri') instead of @resource" ) - def decorator(func: Callable) -> Callable: - @functools.wraps(func) - def wrapper() -> Any: - return func() - - resource = FunctionResource( - uri=uri, - name=name, - description=description, - mime_type=mime_type or "text/plain", - func=wrapper, - ) - self.add_resource(resource) - return wrapper - - return decorator - - def template( - self, - uri_template: str, - *, - name: Optional[str] = None, - description: Optional[str] = None, - mime_type: Optional[str] = None, - ) -> Callable: - """Decorator to register a function as a resource template. - - Args: - uri_template: URI template with parameters (e.g. "weather://{city}/current") - name: Optional name for the resource - description: Optional description of the resource - mime_type: Optional MIME type for the resource - - Example: - @server.template("weather://{city}/current") - def get_weather(city: str) -> str: - return f"Weather for {city}" - """ - # Check if user passed function directly instead of calling decorator - if callable(uri_template): - raise TypeError( - "The @template decorator was used incorrectly. " - "Did you forget to call it? Use @template('uri_template') instead of @template" - ) - def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args: Any, **kwargs: Any) -> Any: return func(*args, **kwargs) - self._resource_manager.add_template( - wrapper, - uri_template=uri_template, - name=name, - description=description, - mime_type=mime_type or "text/plain", - ) + # Check if this should be a template + has_uri_params = "{" in uri and "}" in uri + has_func_params = bool(inspect.signature(func).parameters) + + if has_uri_params or has_func_params: + # Validate that URI params match function params + uri_params = set(re.findall(r"{(\w+)}", uri)) + func_params = set(inspect.signature(func).parameters.keys()) + + if uri_params != func_params: + raise ValueError( + f"Mismatch between URI parameters {uri_params} " + f"and function parameters {func_params}" + ) + + # Register as template + self._resource_manager.add_template( + wrapper, + uri_template=uri, + name=name, + description=description, + mime_type=mime_type or "text/plain", + ) + else: + # Register as regular resource + resource = FunctionResource( + uri=uri, + name=name, + description=description, + mime_type=mime_type or "text/plain", + func=wrapper, + ) + self.add_resource(resource) return wrapper return decorator diff --git a/tests/resources/test_file_resources.py b/tests/resources/test_file_resources.py index b9691929d..5b26abe82 100644 --- a/tests/resources/test_file_resources.py +++ b/tests/resources/test_file_resources.py @@ -1,6 +1,6 @@ import pytest from pathlib import Path -from tempfile import NamedTemporaryFile, TemporaryDirectory +from tempfile import NamedTemporaryFile from fastmcp.resources import FileResource @@ -22,20 +22,6 @@ def temp_file(): pass # File was already deleted by the test -@pytest.fixture -def temp_dir_with_files(): - """Create a temporary directory with test files.""" - with TemporaryDirectory() as d: - path = Path(d).resolve() - # Create some test files - (path / "file1.txt").write_text("content1") - (path / "file2.txt").write_text("content2") - (path / "subdir").mkdir() - (path / "subdir/file3.txt").write_text("content3") - (path / "test.json").write_text('{"key": "value"}') - yield path - - class TestFileResource: """Test FileResource functionality.""" @@ -45,23 +31,14 @@ class TestFileResource: uri=f"file://{temp_file}", name="test", description="test file", - mime_type="text/plain", path=temp_file, ) assert str(resource.uri) == f"file://{temp_file}" assert resource.name == "test" assert resource.description == "test file" - assert resource.mime_type == "text/plain" + assert resource.mime_type == "text/plain" # default assert resource.path == temp_file - - def test_file_resource_relative_path_error(self): - """Test FileResource rejects relative paths.""" - with pytest.raises(ValueError, match="Path must be absolute"): - FileResource( - uri="file://test.txt", - name="test", - path=Path("test.txt"), - ) + assert resource.is_binary is False # default def test_file_resource_str_path_conversion(self, temp_file: Path): """Test FileResource handles string paths.""" @@ -73,8 +50,8 @@ class TestFileResource: assert isinstance(resource.path, Path) assert resource.path.is_absolute() - async def test_file_resource_read(self, temp_file: Path): - """Test reading a FileResource.""" + async def test_read_text_file(self, temp_file: Path): + """Test reading a text file.""" resource = FileResource( uri=f"file://{temp_file}", name="test", @@ -82,19 +59,42 @@ class TestFileResource: ) content = await resource.read() assert content == "test content" + assert resource.mime_type == "text/plain" - async def test_file_resource_read_missing_file(self, temp_file: Path): - """Test reading a non-existent file.""" - temp_file.unlink() + async def test_read_binary_file(self, temp_file: Path): + """Test reading a file as binary.""" resource = FileResource( uri=f"file://{temp_file}", name="test", path=temp_file, + is_binary=True, ) - with pytest.raises(FileNotFoundError): + content = await resource.read() + assert isinstance(content, bytes) + assert content == b"test content" + + def test_relative_path_error(self): + """Test error on relative path.""" + with pytest.raises(ValueError, match="Path must be absolute"): + FileResource( + uri="file:///test.txt", + name="test", + path=Path("test.txt"), + ) + + async def test_missing_file_error(self, temp_file: Path): + """Test error when file doesn't exist.""" + # Create path to non-existent file + missing = temp_file.parent / "missing.txt" + resource = FileResource( + uri="file:///missing.txt", + name="test", + path=missing, + ) + with pytest.raises(ValueError, match="Error reading file"): await resource.read() - async def test_file_resource_read_permission_error(self, temp_file: Path): + async def test_permission_error(self, temp_file: Path): """Test reading a file without permissions.""" temp_file.chmod(0o000) # Remove all permissions try: @@ -103,7 +103,7 @@ class TestFileResource: name="test", path=temp_file, ) - with pytest.raises(PermissionError): + with pytest.raises(ValueError, match="Error reading file"): await resource.read() finally: temp_file.chmod(0o644) # Restore permissions diff --git a/tests/resources/test_function_resources.py b/tests/resources/test_function_resources.py index 7de95a0c6..d9333849f 100644 --- a/tests/resources/test_function_resources.py +++ b/tests/resources/test_function_resources.py @@ -1,3 +1,4 @@ +import pytest from fastmcp.resources import FunctionResource @@ -7,32 +8,92 @@ class TestFunctionResource: def test_function_resource_creation(self): """Test creating a FunctionResource.""" - def my_func(x: str = "") -> str: - return f"Content: {x}" - - resource = FunctionResource( - uri="fn://test", - name="test", - description="test function", - mime_type="text/plain", - func=my_func, - ) - assert str(resource.uri) == "fn://test" - assert resource.name == "test" - assert resource.description == "test function" - assert resource.mime_type == "text/plain" - assert resource.func == my_func - - async def test_function_resource_read(self): - """Test reading a FunctionResource with no parameters.""" - def my_func() -> str: return "test content" resource = FunctionResource( uri="fn://test", name="test", + description="test function", func=my_func, ) + assert str(resource.uri) == "fn://test" + assert resource.name == "test" + assert resource.description == "test function" + assert resource.mime_type == "text/plain" # default + assert resource.func == my_func + + async def test_read_text(self): + """Test reading text from a FunctionResource.""" + + def get_data() -> str: + return "Hello, world!" + + resource = FunctionResource( + uri="function://test", + name="test", + func=get_data, + ) content = await resource.read() - assert content == "test content" + assert content == "Hello, world!" + assert resource.mime_type == "text/plain" + + async def test_read_binary(self): + """Test reading binary data from a FunctionResource.""" + + def get_data() -> bytes: + return b"Hello, world!" + + resource = FunctionResource( + uri="function://test", + name="test", + func=get_data, + ) + content = await resource.read() + assert content == b"Hello, world!" + + async def test_json_conversion(self): + """Test automatic JSON conversion of non-string results.""" + + def get_data() -> dict: + return {"key": "value"} + + resource = FunctionResource( + uri="function://test", + name="test", + func=get_data, + ) + content = await resource.read() + assert '"key": "value"' in content + + async def test_error_handling(self): + """Test error handling in FunctionResource.""" + + def failing_func() -> str: + raise ValueError("Test error") + + resource = FunctionResource( + uri="function://test", + name="test", + func=failing_func, + ) + with pytest.raises(ValueError, match="Error reading resource function://test"): + await resource.read() + + async def test_custom_type_conversion(self): + """Test handling of custom types.""" + + class CustomData: + def __str__(self) -> str: + return "custom data" + + def get_data() -> CustomData: + return CustomData() + + resource = FunctionResource( + uri="function://test", + name="test", + func=get_data, + ) + content = await resource.read() + assert isinstance(content, str) diff --git a/tests/resources/test_resource_manager.py b/tests/resources/test_resource_manager.py index 4bcc8de0c..c63b85935 100644 --- a/tests/resources/test_resource_manager.py +++ b/tests/resources/test_resource_manager.py @@ -1,14 +1,11 @@ -import logging import pytest from pathlib import Path -from tempfile import NamedTemporaryFile, TemporaryDirectory +from tempfile import NamedTemporaryFile from fastmcp.resources import ( FileResource, FunctionResource, ResourceManager, - TextResource, - BinaryResource, ResourceTemplate, ) @@ -30,124 +27,59 @@ def temp_file(): pass # File was already deleted by the test -@pytest.fixture -def temp_file_no_cleanup(): - """Create a temporary file for testing. +class TestResourceManager: + """Test ResourceManager functionality.""" - File is NOT automatically cleaned up - tests must handle cleanup. - """ - content = "test content" - with NamedTemporaryFile(mode="w", delete=False) as f: - f.write(content) - path = Path(f.name).resolve() - return path - - -@pytest.fixture -def temp_dir(): - """Create a temporary directory for testing.""" - with TemporaryDirectory() as d: - yield Path(d).resolve() - - -class TestResourceValidation: - def test_resource_uri_validation(self): - def dummy_func() -> str: - return "data" - - # Valid URI - resource = FunctionResource( - uri="http://example.com/data", - name="test", - func=dummy_func, - ) - assert str(resource.uri) == "http://example.com/data" - - # Missing protocol - with pytest.raises(ValueError, match="Input should be a valid URL"): - FunctionResource( - uri="invalid", - name="test", - func=dummy_func, - ) - - # Missing host - with pytest.raises(ValueError, match="Input should be a valid URL"): - FunctionResource( - uri="http://", - name="test", - func=dummy_func, - ) - - -class TestResourceManagerAdd: - """Test ResourceManager add functionality.""" - - def test_add_file_resource(self, temp_file: Path): - """Test adding a file resource.""" + def test_add_resource(self, temp_file: Path): + """Test adding a resource.""" manager = ResourceManager() resource = FileResource( uri=f"file://{temp_file}", name="test", - description="test file", - mime_type="text/plain", path=temp_file, ) added = manager.add_resource(resource) - assert isinstance(added, FileResource) - assert str(added.uri) == f"file://{temp_file}" - assert added.name == "test" - assert added.description == "test file" - assert added.mime_type == "text/plain" - assert added.path == temp_file + assert added == resource + assert manager.list_resources() == [resource] - def test_add_file_resource_relative_path_error(self): - """Test ResourceManager rejects relative paths.""" - with pytest.raises(ValueError, match="Path must be absolute"): - FileResource( - uri="file:///test.txt", - name="test", - path=Path("test.txt"), - ) + 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, caplog): + def test_warn_on_duplicate_resources(self, temp_file: Path, caplog): """Test warning on duplicate resources.""" - caplog.set_level(logging.WARNING, logger="mcp") manager = ResourceManager() resource = FileResource( - uri="file:///test.txt", + uri=f"file://{temp_file}", name="test", - path=Path("/test.txt"), + path=temp_file, ) manager.add_resource(resource) manager.add_resource(resource) - assert "Resource already exists: file:///test.txt" in caplog.text + assert "Resource already exists" in caplog.text - def test_disable_warn_on_duplicate_resources(self, caplog): + def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog): """Test disabling warning on duplicate resources.""" - caplog.set_level(logging.WARNING, logger="mcp") - manager = ResourceManager() + manager = ResourceManager(warn_on_duplicate_resources=False) resource = FileResource( - uri="file:///test.txt", + uri=f"file://{temp_file}", name="test", - path=Path("/test.txt"), + path=temp_file, ) manager.add_resource(resource) - manager.warn_on_duplicate_resources = False manager.add_resource(resource) - assert "Resource already exists: file:///test.txt" not in caplog.text + assert "Resource already exists" not in caplog.text - -class TestResourceManagerRead: - """Test ResourceManager read functionality.""" - - def test_get_resource_unknown_uri(self): - """Test getting a non-existent resource.""" - manager = ResourceManager() - with pytest.raises(ValueError, match="Unknown resource"): - manager.get_resource("file://unknown") - - def test_get_resource(self, temp_file: Path): + async def test_get_resource(self, temp_file: Path): """Test getting a resource by URI.""" manager = ResourceManager() resource = FileResource( @@ -155,271 +87,9 @@ class TestResourceManagerRead: name="test", path=temp_file, ) - added = manager.add_resource(resource) - retrieved = manager.get_resource(added.uri) - assert retrieved == added - - async def test_resource_read_through_manager(self, temp_file: Path): - """Test reading a resource through the manager.""" - manager = ResourceManager() - resource = FileResource( - uri=f"file://{temp_file}", - name="test", - path=temp_file, - ) - added = manager.add_resource(resource) - retrieved = manager.get_resource(added.uri) - assert retrieved is not None - content = await retrieved.read() - assert content == "test content" - - async def test_resource_read_error_through_manager( - self, temp_file_no_cleanup: Path - ): - """Test error handling when reading through manager.""" - manager = ResourceManager() - # Create resource while file exists - resource = FileResource( - uri=f"file://{temp_file_no_cleanup}", - name="test", - path=temp_file_no_cleanup, - ) - added = manager.add_resource(resource) - retrieved = manager.get_resource(added.uri) - assert retrieved is not None - - # Delete file and verify read fails - temp_file_no_cleanup.unlink() - with pytest.raises(FileNotFoundError): - await retrieved.read() - - -class TestResourceManagerList: - """Test ResourceManager list functionality.""" - - def test_list_resources(self, temp_file: Path): - """Test listing all resources.""" - manager = ResourceManager() - resource = FileResource( - uri=f"file://{temp_file}", - name="test", - path=temp_file, - ) - added = manager.add_resource(resource) - resources = manager.list_resources() - assert len(resources) == 1 - assert resources[0] == added - - def test_list_resources_duplicate(self, temp_file: Path): - """Test that adding the same resource twice only stores it once.""" - manager = ResourceManager() - resource = FileResource( - uri=f"file://{temp_file}", - name="test", - path=temp_file, - ) - resource1 = manager.add_resource(resource) - resource2 = manager.add_resource(resource) - - resources = manager.list_resources() - assert len(resources) == 1 - assert resources[0] == resource1 - assert resource1 == resource2 - - def test_list_multiple_resources(self, temp_file: Path, temp_file_no_cleanup: Path): - """Test listing multiple different resources.""" - manager = ResourceManager() - resource1 = FileResource( - uri=f"file://{temp_file}", - name="test1", - path=temp_file, - ) - resource2 = FileResource( - uri=f"file://{temp_file_no_cleanup}", - name="test2", - path=temp_file_no_cleanup, - ) - added1 = manager.add_resource(resource1) - added2 = manager.add_resource(resource2) - - resources = manager.list_resources() - assert len(resources) == 2 - assert resources[0] == added1 - assert resources[1] == added2 - assert added1 != added2 - - -class TestTextResource: - """Test TextResource functionality.""" - - async def test_text_resource_read(self): - """Test reading from a TextResource.""" - resource = TextResource( - uri="text://test", - name="test", - text="Hello, world!", - ) - content = await resource.read() - assert content == "Hello, world!" - assert resource.mime_type == "text/plain" - - def test_text_resource_custom_mime(self): - """Test TextResource with custom mime type.""" - resource = TextResource( - uri="text://test", - name="test", - text="", - mime_type="text/html", - ) - assert resource.mime_type == "text/html" - - -class TestBinaryResource: - """Test BinaryResource functionality.""" - - async def test_binary_resource_read(self): - """Test reading from a BinaryResource.""" - data = b"Hello, world!" - resource = BinaryResource( - uri="binary://test", - name="test", - data=data, - ) - content = await resource.read() - assert content == data - assert resource.mime_type == "application/octet-stream" - - def test_binary_resource_custom_mime(self): - """Test BinaryResource with custom mime type.""" - resource = BinaryResource( - uri="binary://test", - name="test", - data=b"test", - mime_type="image/png", - ) - assert resource.mime_type == "image/png" - - -class TestResourceTemplate: - """Test ResourceTemplate functionality.""" - - def test_template_from_function(self): - """Test creating a template from a function.""" - - def weather(city: str, units: str = "metric") -> str: - return f"Weather in {city} ({units})" - - template = ResourceTemplate.from_function( - func=weather, - uri_template="weather://{city}/current", - name="weather", - description="Get current weather", - mime_type="text/plain", - ) - - assert template.name == "weather" - assert template.uri_template == "weather://{city}/current" - assert template.mime_type == "text/plain" - assert "city" in template.parameters["properties"] - - def test_template_from_lambda_error(self): - """Test error when creating template from lambda without name.""" - with pytest.raises( - ValueError, match="You must provide a name for lambda functions" - ): - ResourceTemplate.from_function( - func=lambda x: x, - uri_template="test://{x}", - ) - - def test_template_matches(self): - """Test URI matching against template.""" - - def dummy(x: str) -> str: - return x - - template = ResourceTemplate.from_function( - func=dummy, - uri_template="test://{x}/value", - name="test", - ) - - # Test matching URI - params = template.matches("test://hello/value") - assert params == {"x": "hello"} - - # Test non-matching URI - params = template.matches("test://hello/wrong") - assert params is None - - async def test_template_create_text_resource(self): - """Test creating a TextResource from template.""" - - def greet(name: str) -> str: - return f"Hello, {name}!" - - template = ResourceTemplate.from_function( - func=greet, - uri_template="greet://{name}", - name="greeter", - ) - - resource = await template.create_resource( - "greet://world", - {"name": "world"}, - ) - - assert isinstance(resource, TextResource) - content = await resource.read() - assert content == "Hello, world!" - - async def test_template_create_binary_resource(self): - """Test creating a BinaryResource from template.""" - - def get_bytes(value: str) -> bytes: - return value.encode() - - template = ResourceTemplate.from_function( - func=get_bytes, - uri_template="bytes://{value}", - name="bytes", - mime_type="application/octet-stream", - ) - - resource = await template.create_resource( - "bytes://test", - {"value": "test"}, - ) - - assert isinstance(resource, BinaryResource) - content = await resource.read() - assert content == b"test" - - async def test_template_json_conversion(self): - """Test automatic JSON conversion of non-string/bytes results.""" - - def get_data(key: str) -> dict: - return {"key": key, "value": 123} - - template = ResourceTemplate.from_function( - func=get_data, - uri_template="data://{key}", - name="data", - ) - - resource = await template.create_resource( - "data://test", - {"key": "test"}, - ) - - assert isinstance(resource, TextResource) - content = await resource.read() - assert '"key": "test"' in content - assert '"value": 123' in content - - -class TestResourceManagerWithTemplates: - """Test ResourceManager template functionality.""" + 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.""" @@ -436,23 +106,31 @@ class TestResourceManagerWithTemplates: manager._templates[template.uri_template] = template resource = await manager.get_resource("greet://world") - assert isinstance(resource, TextResource) + assert isinstance(resource, FunctionResource) content = await resource.read() assert content == "Hello, world!" - async def test_template_error_handling(self): - """Test error handling in template resource creation.""" + 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("unknown://test") - def failing_func(x: str) -> str: - raise ValueError("Test error") - - template = ResourceTemplate.from_function( - func=failing_func, - uri_template="fail://{x}", - name="fail", + 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, ) - manager._templates[template.uri_template] = template - - with pytest.raises(ValueError, match="Error creating resource from template"): - await manager.get_resource("fail://test") + 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] diff --git a/tests/resources/test_resource_template.py b/tests/resources/test_resource_template.py new file mode 100644 index 000000000..0f72e9a85 --- /dev/null +++ b/tests/resources/test_resource_template.py @@ -0,0 +1,238 @@ +import pytest +from fastmcp.resources import ResourceTemplate, FunctionResource + + +class TestResourceTemplate: + """Test ResourceTemplate functionality.""" + + def test_template_from_function(self): + """Test creating a template from a function.""" + + def weather(city: str, units: str = "metric") -> str: + return f"Weather in {city} ({units})" + + template = ResourceTemplate.from_function( + func=weather, + uri_template="weather://{city}/current", + name="weather", + description="Get current weather", + ) + + assert template.name == "weather" + assert template.uri_template == "weather://{city}/current" + assert template.mime_type == "text/plain" + assert "city" in template.parameters["properties"] + + def test_template_from_lambda_error(self): + """Test error when creating template from lambda without name.""" + with pytest.raises( + ValueError, match="You must provide a name for lambda functions" + ): + ResourceTemplate.from_function( + func=lambda x: x, + uri_template="test://{x}", + ) + + def test_template_matches(self): + """Test URI matching against template.""" + + def dummy(x: str) -> str: + return x + + template = ResourceTemplate.from_function( + func=dummy, + uri_template="test://{x}/value", + name="test", + ) + + # Test matching URI + params = template.matches("test://hello/value") + assert params == {"x": "hello"} + + # Test non-matching URI + params = template.matches("test://hello/wrong") + assert params is None + + async def test_create_text_resource(self): + """Test creating a text resource from template.""" + + def greet(name: str) -> str: + return f"Hello, {name}!" + + template = ResourceTemplate.from_function( + func=greet, + uri_template="greet://{name}", + name="greeter", + ) + + resource = await template.create_resource( + "greet://world", + {"name": "world"}, + ) + + assert isinstance(resource, FunctionResource) + content = await resource.read() + assert content == "Hello, world!" + + async def test_create_binary_resource(self): + """Test creating a binary resource from template.""" + + def get_bytes(value: str) -> bytes: + return value.encode() + + template = ResourceTemplate.from_function( + func=get_bytes, + uri_template="bytes://{value}", + name="bytes", + ) + + resource = await template.create_resource( + "bytes://test", + {"value": "test"}, + ) + + assert isinstance(resource, FunctionResource) + content = await resource.read() + assert content == b"test" + + async def test_json_conversion(self): + """Test automatic JSON conversion of non-string/bytes results.""" + + def get_data(key: str) -> dict: + return {"key": key, "value": 123} + + template = ResourceTemplate.from_function( + func=get_data, + uri_template="data://{key}", + name="data", + ) + + resource = await template.create_resource( + "data://test", + {"key": "test"}, + ) + + assert isinstance(resource, FunctionResource) + content = await resource.read() + assert '"key": "test"' in content + assert '"value": 123' in content + + async def test_template_error(self): + """Test error handling in template resource creation.""" + + def failing_func(x: str) -> str: + raise ValueError("Test error") + + template = ResourceTemplate.from_function( + func=failing_func, + uri_template="fail://{x}", + name="fail", + ) + + with pytest.raises(ValueError, match="Error creating resource from template"): + await template.create_resource("fail://test", {"x": "test"}) + + async def test_async_text_resource(self): + """Test creating a text resource from async function.""" + + async def greet(name: str) -> str: + return f"Hello, {name}!" + + template = ResourceTemplate.from_function( + func=greet, + uri_template="greet://{name}", + name="greeter", + ) + + resource = await template.create_resource( + "greet://world", + {"name": "world"}, + ) + + assert isinstance(resource, FunctionResource) + content = await resource.read() + assert content == "Hello, world!" + + async def test_async_binary_resource(self): + """Test creating a binary resource from async function.""" + + async def get_bytes(value: str) -> bytes: + return value.encode() + + template = ResourceTemplate.from_function( + func=get_bytes, + uri_template="bytes://{value}", + name="bytes", + ) + + resource = await template.create_resource( + "bytes://test", + {"value": "test"}, + ) + + assert isinstance(resource, FunctionResource) + content = await resource.read() + assert content == b"test" + + async def test_async_json_conversion(self): + """Test automatic JSON conversion of async results.""" + + async def get_data(key: str) -> dict: + return {"key": key, "value": 123} + + template = ResourceTemplate.from_function( + func=get_data, + uri_template="data://{key}", + name="data", + ) + + resource = await template.create_resource( + "data://test", + {"key": "test"}, + ) + + assert isinstance(resource, FunctionResource) + content = await resource.read() + assert '"key": "test"' in content + assert '"value": 123' in content + + async def test_async_error(self): + """Test error handling in async template.""" + + async def failing_func(x: str) -> str: + raise ValueError("Test error") + + template = ResourceTemplate.from_function( + func=failing_func, + uri_template="fail://{x}", + name="fail", + ) + + with pytest.raises( + ValueError, match="Error creating resource from template: Test error" + ): + await template.create_resource("fail://test", {"x": "test"}) + + async def test_sync_returning_coroutine(self): + """Test sync function that returns a coroutine.""" + + async def async_helper(name: str) -> str: + return f"Hello, {name}!" + + def get_greeting(name: str) -> str: + return async_helper(name) # Returns coroutine + + template = ResourceTemplate.from_function( + func=get_greeting, + uri_template="greet://{name}", + name="greeter", + ) + + resource = await template.create_resource( + "greet://world", + {"name": "world"}, + ) + + assert isinstance(resource, FunctionResource) + content = await resource.read() + assert content == "Hello, world!" diff --git a/tests/resources/test_resources.py b/tests/resources/test_resources.py new file mode 100644 index 000000000..ffb92e099 --- /dev/null +++ b/tests/resources/test_resources.py @@ -0,0 +1,98 @@ +import pytest +from fastmcp.resources import Resource, FunctionResource + + +class TestResourceValidation: + """Test base Resource validation.""" + + def test_resource_uri_validation(self): + """Test URI validation.""" + + def dummy_func() -> str: + return "data" + + # Valid URI + resource = FunctionResource( + uri="http://example.com/data", + name="test", + func=dummy_func, + ) + assert str(resource.uri) == "http://example.com/data" + + # Missing protocol + with pytest.raises(ValueError, match="Input should be a valid URL"): + FunctionResource( + uri="invalid", + name="test", + func=dummy_func, + ) + + # Missing host + with pytest.raises(ValueError, match="Input should be a valid URL"): + FunctionResource( + uri="http://", + name="test", + func=dummy_func, + ) + + def test_resource_name_from_uri(self): + """Test name is extracted from URI if not provided.""" + + def dummy_func() -> str: + return "data" + + resource = FunctionResource( + uri="resource://my-resource", + func=dummy_func, + ) + assert resource.name == "my-resource" + + def test_resource_name_validation(self): + """Test name validation.""" + + def dummy_func() -> str: + return "data" + + # Must provide either name or URI + with pytest.raises(ValueError, match="Either name or uri must be provided"): + FunctionResource( + func=dummy_func, + ) + + # Explicit name takes precedence over URI + resource = FunctionResource( + uri="resource://uri-name", + name="explicit-name", + func=dummy_func, + ) + assert resource.name == "explicit-name" + + def test_resource_mime_type(self): + """Test mime type handling.""" + + def dummy_func() -> str: + return "data" + + # Default mime type + resource = FunctionResource( + uri="resource://test", + func=dummy_func, + ) + assert resource.mime_type == "text/plain" + + # Custom mime type + resource = FunctionResource( + uri="resource://test", + func=dummy_func, + mime_type="application/json", + ) + assert resource.mime_type == "application/json" + + async def test_resource_read_abstract(self): + """Test that Resource.read() is abstract.""" + + class ConcreteResource(Resource): + pass + + with pytest.raises(TypeError, match="abstract method"): + ConcreteResource(uri="test://test", name="test") diff --git a/tests/test_server.py b/tests/test_server.py index 0a491a2ee..a97405ce6 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -38,11 +38,11 @@ class TestServer: async def test_add_resource_decorator(self): mcp = FastMCP() - @mcp.resource("r://data") + @mcp.resource("r://{x}") def get_data(x: str) -> str: return f"Data: {x}" - assert len(mcp._resource_manager.list_resources()) == 1 + assert len(mcp._resource_manager._templates) == 1 async def test_add_resource_decorator_incorrect_usage(self): mcp = FastMCP() @@ -264,3 +264,87 @@ class TestServerResources: result.contents[0].blob == base64.b64encode(b"Binary file data").decode() ) + + async def test_resource_with_params(self): + """Test that a resource with function parameters is automatically a template""" + mcp = FastMCP() + + with pytest.raises(ValueError, match="Mismatch between URI parameters"): + + @mcp.resource("resource://data") + def get_data(param: str) -> str: + return f"Data: {param}" + + async def test_resource_with_uri_params(self): + """Test that a resource with URI parameters is automatically a template""" + mcp = FastMCP() + + with pytest.raises(ValueError, match="Mismatch between URI parameters"): + + @mcp.resource("resource://{param}") + def get_data() -> str: + return "Data" + + async def test_resource_matching_params(self): + """Test that a resource with matching URI and function parameters works""" + mcp = FastMCP() + + @mcp.resource("resource://{name}/data") + def get_data(name: str) -> str: + return f"Data for {name}" + + async with client_session(mcp._mcp_server) as client: + result = await client.read_resource("resource://test/data") + assert result.contents[0].text == "Data for test" + + async def test_resource_mismatched_params(self): + """Test that mismatched parameters raise an error""" + mcp = FastMCP() + + with pytest.raises(ValueError, match="Mismatch between URI parameters"): + + @mcp.resource("resource://{name}/data") + def get_data(user: str) -> str: + return f"Data for {user}" + + async def test_resource_multiple_params(self): + """Test that multiple parameters work correctly""" + mcp = FastMCP() + + @mcp.resource("resource://{org}/{repo}/data") + def get_data(org: str, repo: str) -> str: + return f"Data for {org}/{repo}" + + async with client_session(mcp._mcp_server) as client: + result = await client.read_resource("resource://cursor/fastmcp/data") + assert result.contents[0].text == "Data for cursor/fastmcp" + + async def test_resource_no_params(self): + """Test that a resource with no parameters works as a regular resource""" + mcp = FastMCP() + + @mcp.resource("resource://static") + def get_data() -> str: + return "Static data" + + async with client_session(mcp._mcp_server) as client: + result = await client.read_resource("resource://static") + assert result.contents[0].text == "Static data" + + async def test_template_to_resource_conversion(self): + """Test that templates are properly converted to resources when accessed""" + mcp = FastMCP() + + @mcp.resource("resource://{name}/data") + def get_data(name: str) -> str: + return f"Data for {name}" + + # Should be registered as a template + assert len(mcp._resource_manager._templates) == 1 + assert len(mcp._resource_manager.list_resources()) == 0 + + # When accessed, should create a concrete resource + resource = await mcp._resource_manager.get_resource("resource://test/data") + assert isinstance(resource, FunctionResource) + result = await resource.read() + assert result == "Data for test"