Merge pull request #6 from jlowin/templates

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

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.
-
-    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.
-    """
+class Resource(BaseModel, abc.ABC):
+    """Base class for all resources."""
 
-    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
+
+    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
 
-        Returns:
-            Union[str, bytes]: Text content as str for text resources,
-                             binary content as bytes for binary resources
-        """
-        return ""
+
+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.
 
-    func: Union[Callable[[], Any], Callable[[], Awaitable[Any]]]
-    is_async: bool = False
+    The function can return:
+    - str for text content (default)
+    - bytes for binary content
+    - other types will be converted to JSON
+    """
 
-    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."""
-
-    path: Path
-    recursive: bool = False
-    pattern: Optional[str] = None
-    mime_type: Optional[str] = "application/json"
+    """A resource that lists files in a directory."""
+
+    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 == "<lambda>":
+            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
+
+    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})
 
-        if resource := self._resources.get(uri):
+        # 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]:

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()
-
-            resource = FunctionResource(
-                uri=uri,
-                name=name,
-                description=description,
-                mime_type=mime_type or "text/plain",
-                func=wrapper,
-            )
-            self.add_resource(resource)
+            def wrapper(*args: Any, **kwargs: Any) -> Any:
+                return func(*args, **kwargs)
+
+            # 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

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,
+        )
+        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(FileNotFoundError):
+        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

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}"
+        def my_func() -> str:
+            return "test content"
 
         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.mime_type == "text/plain"  # default
         assert resource.func == my_func
 
-    async def test_function_resource_read(self):
-        """Test reading a FunctionResource with no parameters."""
+    async def test_read_text(self):
+        """Test reading text from a FunctionResource."""
 
-        def my_func() -> str:
-            return "test content"
+        def get_data() -> str:
+            return "Hello, world!"
 
         resource = FunctionResource(
-            uri="fn://test",
+            uri="function://test",
             name="test",
-            func=my_func,
+            func=get_data,
+        )
+        content = await resource.read()
+        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 content == "test content"
+        assert isinstance(content, str)

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,204 +27,97 @@ def temp_file():
         pass  # File was already deleted by the test
 
 
-@pytest.fixture
-def temp_file_no_cleanup():
-    """Create a temporary file for testing.
-
-    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
-
+class TestResourceManager:
+    """Test ResourceManager functionality."""
 
-@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
-
-    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_warn_on_duplicate_resources(self, caplog):
-        """Test warning on duplicate resources."""
-        caplog.set_level(logging.WARNING, logger="mcp")
+        assert added == resource
+        assert manager.list_resources() == [resource]
+
+    def test_add_duplicate_resource(self, temp_file: Path):
+        """Test adding the same resource twice."""
         manager = ResourceManager()
         resource = FileResource(
-            uri="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
+        first = manager.add_resource(resource)
+        second = manager.add_resource(resource)
+        assert first == second
+        assert manager.list_resources() == [resource]
 
-    def test_disable_warn_on_duplicate_resources(self, caplog):
-        """Test disabling warning on duplicate resources."""
-        caplog.set_level(logging.WARNING, logger="mcp")
+    def test_warn_on_duplicate_resources(self, temp_file: Path, caplog):
+        """Test warning on duplicate resources."""
         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.warn_on_duplicate_resources = False
         manager.add_resource(resource)
-        assert "Resource already exists: file:///test.txt" not in caplog.text
-
-
-class TestResourceManagerRead:
-    """Test ResourceManager read functionality."""
+        assert "Resource already exists" in caplog.text
 
-    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):
-        """Test getting a resource by URI."""
-        manager = ResourceManager()
+    def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog):
+        """Test disabling warning on duplicate resources."""
+        manager = ResourceManager(warn_on_duplicate_resources=False)
         resource = FileResource(
             uri=f"file://{temp_file}",
             name="test",
             path=temp_file,
         )
-        added = manager.add_resource(resource)
-        retrieved = manager.get_resource(added.uri)
-        assert retrieved == added
+        manager.add_resource(resource)
+        manager.add_resource(resource)
+        assert "Resource already exists" not in caplog.text
 
-    async def test_resource_read_through_manager(self, temp_file: Path):
-        """Test reading a resource through the manager."""
+    async def test_get_resource(self, temp_file: Path):
+        """Test getting a resource by URI."""
         manager = ResourceManager()
         resource = FileResource(
             uri=f"file://{temp_file}",
             name="test",
             path=temp_file,
         )
-        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()
+        manager.add_resource(resource)
+        retrieved = await manager.get_resource(resource.uri)
+        assert retrieved == resource
 
+    async def test_get_resource_from_template(self):
+        """Test getting a resource through a template."""
+        manager = ResourceManager()
 
-class TestResourceManagerList:
-    """Test ResourceManager list functionality."""
+        def greet(name: str) -> str:
+            return f"Hello, {name}!"
 
-    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,
+        template = ResourceTemplate.from_function(
+            func=greet,
+            uri_template="greet://{name}",
+            name="greeter",
         )
-        added = manager.add_resource(resource)
-        resources = manager.list_resources()
-        assert len(resources) == 1
-        assert resources[0] == added
+        manager._templates[template.uri_template] = template
 
-    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)
+        resource = await manager.get_resource("greet://world")
+        assert isinstance(resource, FunctionResource)
+        content = await resource.read()
+        assert content == "Hello, world!"
 
-        resources = manager.list_resources()
-        assert len(resources) == 1
-        assert resources[0] == resource1
-        assert resource1 == resource2
+    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 test_list_multiple_resources(self, temp_file: Path, temp_file_no_cleanup: Path):
-        """Test listing multiple different resources."""
+    def test_list_resources(self, temp_file: Path):
+        """Test listing all resources."""
         manager = ResourceManager()
         resource1 = FileResource(
             uri=f"file://{temp_file}",
@@ -228,15 +125,12 @@ class TestResourceManagerList:
             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]

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!"

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")

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"