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 71e9a9d8e..b1649c9b5 100644 --- a/src/fastmcp/resources.py +++ b/src/fastmcp/resources.py @@ -1,13 +1,14 @@ +import inspect import pydantic.json - import abc import asyncio import json +import re from pathlib import Path -from typing import Dict, Optional, Callable, Any, Union, Awaitable +from typing import Dict, Optional, Callable, Any, Union import httpx -from pydantic import BaseModel, field_validator +from pydantic import BaseModel, Field, TypeAdapter, validate_call, field_validator from pydantic.networks import _BaseUrl from .utilities.logging import get_logger @@ -15,91 +16,122 @@ from .utilities.logging import get_logger logger = get_logger(__name__) -class Resource(BaseModel): - """Base class for all resources. +class Resource(BaseModel, abc.ABC): + """Base class for all resources.""" - Resources can contain either text (UTF-8 encoded) or binary data. - Text resources are suitable for source code, logs, JSON, etc. - Binary resources are suitable for images, PDFs, audio, etc. - """ - - uri: _BaseUrl - name: str - description: Optional[str] = None - mime_type: Optional[str] = None - is_binary: bool = False + uri: _BaseUrl = Field(description="URI of the resource") + 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 async def read(self) -> Union[str, bytes]: - """Read the resource content. + """Read the resource content.""" + pass - Returns: - Union[str, bytes]: Text content as str for text resources, - binary content as bytes for binary resources - """ - return "" + model_config = { + "validate_default": True, + } + + +class TextResource(Resource): + """A resource that reads from a string.""" + + text: str = Field(description="Text content of the resource") + + async def read(self) -> str: + """Read the text content.""" + return self.text + + +class BinaryResource(Resource): + """A resource that reads from bytes.""" + + data: bytes = Field(description="Binary content of the resource") + + async def read(self) -> bytes: + """Read the binary content.""" + return self.data class FunctionResource(Resource): - """A resource that is generated by a function call. + """A resource that defers data loading by wrapping a function. - The function can be sync or async and must return a string, bytes, - or another Resource. + 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: Union[Callable[[], Any], Callable[[], Awaitable[Any]]] - is_async: bool = False - - def __init__(self, **data): - super().__init__(**data) - self.is_async = asyncio.iscoroutinefunction(self.func) + func: Callable[[], Any] = Field(exclude=True) async def read(self) -> Union[str, bytes]: - """Read the resource content by calling the function.""" + """Read the resource by calling the wrapped function.""" try: - result = ( - await self.func() - if self.is_async - else await asyncio.to_thread(self.func) - ) - + result = self.func() if isinstance(result, Resource): return await result.read() if isinstance(result, bytes): return result - if not isinstance(result, str): - try: - return json.dumps(result, default=pydantic.json.pydantic_encoder) - except json.JSONDecodeError: - return str(result) - 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 calling function {self.func.__name__}: {e}") + raise ValueError(f"Error reading resource {self.uri}: {e}") class FileResource(Resource): - """A file resource.""" + """A resource that reads from a file. - path: Path + Set is_binary=True to read file as binary data instead of text. + """ + + path: Path = Field(description="Path to the file") + 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", + ) @field_validator("path") @classmethod def validate_absolute_path(cls, path: Path) -> Path: """Ensure path is absolute.""" if not path.is_absolute(): - raise ValueError(f"Path must be absolute: {path}") + raise ValueError("Path must be absolute") return path async def read(self) -> Union[str, bytes]: @@ -108,47 +140,46 @@ class FileResource(Resource): if self.is_binary: return await asyncio.to_thread(self.path.read_bytes) return await asyncio.to_thread(self.path.read_text) - except FileNotFoundError: - raise FileNotFoundError(f"File not found: {self.path}") - except PermissionError: - raise PermissionError(f"Permission denied: {self.path}") except Exception as e: raise ValueError(f"Error reading file {self.path}: {e}") class HttpResource(Resource): - """An HTTP resource.""" + """A resource that reads from an HTTP endpoint.""" - url: str - headers: Optional[Dict[str, str]] = None + url: str = Field(description="URL to fetch content from") + mime_type: Optional[str] = Field( + default="application/json", description="MIME type of the resource content" + ) async def read(self) -> Union[str, bytes]: - """Read the HTTP resource content.""" - try: - async with httpx.AsyncClient() as client: - response = await client.get(self.url, headers=self.headers) - response.raise_for_status() - return response.content if self.is_binary else response.text - except httpx.HTTPStatusError as e: - raise ValueError(f"HTTP error {e.response.status_code}: {e}") - except httpx.RequestError as e: - raise ValueError(f"Request failed: {e}") + """Read the HTTP content.""" + async with httpx.AsyncClient() as client: + response = await client.get(self.url) + response.raise_for_status() + return response.text class DirectoryResource(Resource): - """A directory resource.""" + """A resource that lists files in a directory.""" - path: Path - recursive: bool = False - pattern: Optional[str] = None - mime_type: Optional[str] = "application/json" + path: Path = Field(description="Path to the directory") + recursive: bool = Field( + default=False, description="Whether to list files recursively" + ) + pattern: Optional[str] = Field( + default=None, description="Optional glob pattern to filter files" + ) + mime_type: Optional[str] = Field( + default="application/json", description="MIME type of the resource content" + ) @field_validator("path") @classmethod def validate_absolute_path(cls, path: Path) -> Path: """Ensure path is absolute.""" if not path.is_absolute(): - raise ValueError(f"Path must be absolute: {path}") + raise ValueError("Path must be absolute") return path def list_files(self) -> list[Path]: @@ -183,21 +214,121 @@ class DirectoryResource(Resource): raise ValueError(f"Error reading directory {self.path}: {e}") +class ResourceTemplate(BaseModel): + """A template for dynamically creating resources.""" + + uri_template: str = Field( + description="URI template with parameters (e.g. weather://{city}/current)" + ) + name: str = Field(description="Name of the resource") + description: Optional[str] = Field( + description="Description of what the resource does" + ) + mime_type: str = Field( + default="text/plain", description="MIME type of the resource content" + ) + func: Callable = Field(exclude=True) + parameters: dict = Field(description="JSON schema for function parameters") + + @classmethod + def from_function( + cls, + func: Callable, + uri_template: str, + name: Optional[str] = None, + description: Optional[str] = None, + mime_type: Optional[str] = None, + ) -> "ResourceTemplate": + """Create a template from a function.""" + func_name = name or func.__name__ + if func_name == "": + raise ValueError("You must provide a name for lambda functions") + + # Get schema from TypeAdapter - will fail if function isn't properly typed + parameters = TypeAdapter(func).json_schema() + + # ensure the arguments are properly cast + func = validate_call(func) + + return cls( + uri_template=uri_template, + name=func_name, + description=description or func.__doc__ or "", + func=func, + parameters=parameters, + ) + + def matches(self, uri: str) -> Optional[Dict[str, Any]]: + """Check if URI matches template and extract parameters.""" + # Convert template to regex pattern + pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)") + match = re.match(f"^{pattern}$", uri) + if match: + return match.groupdict() + return None + + async def create_resource(self, uri: str, params: Dict[str, Any]) -> Resource: + """Create a resource from the template with the given parameters.""" + try: + # Call function and check if result is a coroutine + result = self.func(**params) + if inspect.iscoroutine(result): + result = await result + + return FunctionResource( + uri=uri, + name=self.name, + description=self.description, + func=lambda: result, # Capture result in closure + ) + except Exception as e: + raise ValueError(f"Error creating resource from template: {e}") + + class ResourceManager: """Manages FastMCP resources.""" def __init__(self, warn_on_duplicate_resources: bool = True): self._resources: Dict[str, Resource] = {} + self._templates: Dict[str, ResourceTemplate] = {} self.warn_on_duplicate_resources = warn_on_duplicate_resources - def get_resource(self, uri: Union[_BaseUrl, str]) -> Optional[Resource]: - """Get resource by URI.""" - uri = str(uri) - logger.debug("Getting resource", extra={"uri": uri}) + def add_template( + self, + func: Callable, + uri_template: str, + name: Optional[str] = None, + description: Optional[str] = None, + mime_type: Optional[str] = None, + ) -> ResourceTemplate: + """Add a template from a function.""" + template = ResourceTemplate.from_function( + func, + uri_template=uri_template, + name=name, + description=description, + mime_type=mime_type, + ) + self._templates[template.uri_template] = template + return template - if resource := self._resources.get(uri): + async def get_resource(self, uri: Union[_BaseUrl, str]) -> Optional[Resource]: + """Get resource by URI, checking concrete resources first, then templates.""" + uri_str = str(uri) + logger.debug("Getting resource", extra={"uri": uri_str}) + + # First check concrete resources + if resource := self._resources.get(uri_str): return resource + # Then check templates + for template in self._templates.values(): + if params := template.matches(uri_str): + try: + return await template.create_resource(uri_str, params) + except Exception as e: + raise ValueError(f"Error creating resource from template: {e}") + raise ValueError(f"Unknown resource: {uri}") def list_resources(self) -> list[Resource]: diff --git a/src/fastmcp/server.py b/src/fastmcp/server.py index cd4bfad8c..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): @@ -213,17 +231,42 @@ class FastMCP: def decorator(func: Callable) -> Callable: @functools.wraps(func) - def wrapper() -> Any: - return func() + def wrapper(*args: Any, **kwargs: Any) -> Any: + return func(*args, **kwargs) - resource = FunctionResource( - uri=uri, - name=name, - description=description, - mime_type=mime_type or "text/plain", - func=wrapper, - ) - self.add_resource(resource) + # 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 3bff7c7c2..c63b85935 100644 --- a/tests/resources/test_resource_manager.py +++ b/tests/resources/test_resource_manager.py @@ -1,9 +1,13 @@ -import logging import pytest from pathlib import Path -from tempfile import NamedTemporaryFile, TemporaryDirectory +from tempfile import NamedTemporaryFile -from fastmcp.resources import FileResource, FunctionResource, ResourceManager +from fastmcp.resources import ( + FileResource, + FunctionResource, + ResourceManager, + ResourceTemplate, +) @pytest.fixture @@ -23,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( @@ -148,95 +87,50 @@ class TestResourceManagerRead: name="test", path=temp_file, ) - added = manager.add_resource(resource) - retrieved = manager.get_resource(added.uri) - assert retrieved == added + manager.add_resource(resource) + retrieved = await manager.get_resource(resource.uri) + assert retrieved == resource - async def test_resource_read_through_manager(self, temp_file: Path): - """Test reading a resource through the manager.""" + async def test_get_resource_from_template(self): + """Test getting a resource through a template.""" 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.""" + def greet(name: str) -> str: + return f"Hello, {name}!" + + template = ResourceTemplate.from_function( + func=greet, + uri_template="greet://{name}", + name="greeter", + ) + manager._templates[template.uri_template] = template + + resource = await manager.get_resource("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() - # 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.""" + with pytest.raises(ValueError, match="Unknown resource"): + await manager.get_resource("unknown://test") 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}", + uri=f"file://{temp_file}2", name="test2", - path=temp_file_no_cleanup, + path=temp_file, ) - added1 = manager.add_resource(resource1) - added2 = manager.add_resource(resource2) - + manager.add_resource(resource1) + manager.add_resource(resource2) resources = manager.list_resources() assert len(resources) == 2 - assert resources[0] == added1 - assert resources[1] == added2 - assert added1 != added2 + 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"