ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add template support

Browse source
This commit is contained in:
Jeremiah Lowin 2024-11-30 10:29:34 -05:00
commit 55d7cd1488
9 changed files with 796 additions and 535 deletions
Download patch file Download diff file Expand all files Collapse all files

94
README.md
View file

@ -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

131
src/fastmcp/resources.py
View file

@ -1,3 +1,4 @@
import inspect
import pydantic.json
import abc
import asyncio
@ -19,20 +20,30 @@ class Resource(BaseModel, abc.ABC):
"""Base class for all resources."""
uri: _BaseUrl = Field(description="URI of the resource")
name: str = Field(description="Name of the resource")
description: Optional[str] = Field(description="Description of the resource")
mime_type: Optional[str] = Field(description="MIME type of the resource content")
name: str = Field(description="Name of the resource", default=None)
description: Optional[str] = Field(
description="Description of the resource", default=None
)
mime_type: str = Field(
default="text/plain",
description="MIME type of the resource content",
pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$",
)
@field_validator("name", mode="before")
@classmethod
def set_default_name(cls, name: str | None, info) -> str:
"""Set default name from URI if not provided."""
if name is not None:
if name:
return name
# Extract everything after the protocol (e.g., "desktop" from "resource://desktop")
uri = info.data.get("uri")
if uri:
return str(uri).split("://", 1)[1]
uri_str = str(uri)
if "://" in uri_str:
name = uri_str.split("://", 1)[1]
if name:
return name
raise ValueError("Either name or uri must be provided")
@abc.abstractmethod
@ -40,14 +51,15 @@ class Resource(BaseModel, abc.ABC):
"""Read the resource content."""
pass
model_config = {
"validate_default": True,
}
class TextResource(Resource):
"""A resource containing text content."""
"""A resource that reads from a string."""
text: str = Field(description="Text content of the resource")
mime_type: Optional[str] = Field(
default="text/plain", description="MIME type of the resource content"
)
async def read(self) -> str:
"""Read the text content."""
@ -55,25 +67,62 @@ class TextResource(Resource):
class BinaryResource(Resource):
"""A resource containing binary content."""
"""A resource that reads from bytes."""
data: bytes = Field(description="Binary content of the resource")
mime_type: Optional[str] = Field(
default="application/octet-stream",
description="MIME type of the resource content",
)
async def read(self) -> bytes:
"""Read the binary content."""
return self.data
class FunctionResource(Resource):
"""A resource that defers data loading by wrapping a function.
The function is only called when the resource is read, allowing for lazy loading
of potentially expensive data. This is particularly useful when listing resources,
as the function won't be called until the resource is actually accessed.
The function can return:
- str for text content (default)
- bytes for binary content
- other types will be converted to JSON
"""
func: Callable[[], Any] = Field(exclude=True)
async def read(self) -> Union[str, bytes]:
"""Read the resource by calling the wrapped function."""
try:
result = self.func()
if isinstance(result, Resource):
return await result.read()
if isinstance(result, bytes):
return result
if isinstance(result, str):
return result
try:
return json.dumps(result, default=pydantic.json.pydantic_encoder)
except TypeError:
# If JSON serialization fails, try str()
return str(result)
except Exception as e:
raise ValueError(f"Error reading resource {self.uri}: {e}")
class FileResource(Resource):
"""A resource that reads from a file."""
"""A resource that reads from a file.
Set is_binary=True to read file as binary data instead of text.
"""
path: Path = Field(description="Path to the file")
mime_type: Optional[str] = Field(
default="application/octet-stream",
is_binary: bool = Field(
default=False,
description="Whether to read the file as binary data",
)
mime_type: str = Field(
default="text/plain",
description="MIME type of the resource content",
)
@ -87,17 +136,12 @@ class FileResource(Resource):
async def read(self) -> Union[str, bytes]:
"""Read the file content."""
if self.mime_type and self.mime_type.startswith("text/"):
return await self._read_text()
return await self._read_binary()
async def _read_text(self) -> str:
"""Read file as text."""
return await asyncio.to_thread(self.path.read_text)
async def _read_binary(self) -> bytes:
"""Read file as binary."""
return await asyncio.to_thread(self.path.read_bytes)
try:
if self.is_binary:
return await asyncio.to_thread(self.path.read_bytes)
return await asyncio.to_thread(self.path.read_text)
except Exception as e:
raise ValueError(f"Error reading file {self.path}: {e}")
class HttpResource(Resource):
@ -180,7 +224,7 @@ class ResourceTemplate(BaseModel):
description: Optional[str] = Field(
description="Description of what the resource does"
)
mime_type: Optional[str] = Field(
mime_type: str = Field(
default="text/plain", description="MIME type of the resource content"
)
func: Callable = Field(exclude=True)
@ -210,7 +254,6 @@ class ResourceTemplate(BaseModel):
uri_template=uri_template,
name=func_name,
description=description or func.__doc__ or "",
mime_type=mime_type or "text/plain",
func=func,
parameters=parameters,
)
@ -226,30 +269,20 @@ class ResourceTemplate(BaseModel):
async def create_resource(self, uri: str, params: Dict[str, Any]) -> Resource:
"""Create a resource from the template with the given parameters."""
result = await self.func(**params)
try:
# Call function and check if result is a coroutine
result = self.func(**params)
if inspect.iscoroutine(result):
result = await result
if isinstance(result, bytes):
return BinaryResource(
return FunctionResource(
uri=uri,
name=self.name,
description=self.description,
mime_type=self.mime_type,
data=result,
)
else:
if not isinstance(result, str):
try:
result = json.dumps(result, default=pydantic.json.pydantic_encoder)
except Exception as e:
raise ValueError(f"Error converting result to JSON: {e}")
return TextResource(
uri=uri,
name=self.name,
description=self.description,
mime_type=self.mime_type,
text=result,
func=lambda: result, # Capture result in closure
)
except Exception as e:
raise ValueError(f"Error creating resource from template: {e}")
class ResourceManager:

109
src/fastmcp/server.py
View file

@ -4,6 +4,8 @@ import asyncio
import functools
import json
from typing import Any, Callable, Optional, Sequence, Union, Literal
import inspect
import re
import pydantic.json
from mcp.server import Server as MCPServer
@ -19,7 +21,11 @@ from pydantic_settings import BaseSettings
from pydantic.networks import _BaseUrl
from .exceptions import ResourceError
from .resources import Resource, FunctionResource, ResourceManager
from .resources import (
Resource,
FunctionResource,
ResourceManager,
)
from .tools import ToolManager, Image
from .utilities.logging import get_logger, configure_logging
@ -138,7 +144,7 @@ class FastMCP:
async def read_resource(self, uri: _BaseUrl) -> Union[str, bytes]:
"""Read a resource by URI."""
resource = self._resource_manager.get_resource(uri)
resource = await self._resource_manager.get_resource(uri)
if not resource:
raise ResourceError(f"Unknown resource: {uri}")
@ -193,9 +199,17 @@ class FastMCP:
"""Decorator to register a function as a resource.
The function will be called when the resource is read to generate its content.
The function can return:
- str for text content
- bytes for binary content
- other types will be converted to JSON
If the URI contains parameters (e.g. "resource://{param}") or the function
has parameters, it will be registered as a template resource.
Args:
uri: URI for the resource (e.g. "resource://my-resource")
uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}")
name: Optional name for the resource
description: Optional description of the resource
mime_type: Optional MIME type for the resource
@ -203,6 +217,10 @@ class FastMCP:
@server.resource("resource://my-resource")
def get_data() -> str:
return "Hello, world!"
@server.resource("resource://{city}/weather")
def get_weather(city: str) -> str:
return f"Weather for {city}"
"""
# Check if user passed function directly instead of calling decorator
if callable(uri):
@ -211,63 +229,44 @@ class FastMCP:
"Did you forget to call it? Use @resource('uri') instead of @resource"
)
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper() -> Any:
return func()
resource = FunctionResource(
uri=uri,
name=name,
description=description,
mime_type=mime_type or "text/plain",
func=wrapper,
)
self.add_resource(resource)
return wrapper
return decorator
def template(
self,
uri_template: str,
*,
name: Optional[str] = None,
description: Optional[str] = None,
mime_type: Optional[str] = None,
) -> Callable:
"""Decorator to register a function as a resource template.
Args:
uri_template: URI template with parameters (e.g. "weather://{city}/current")
name: Optional name for the resource
description: Optional description of the resource
mime_type: Optional MIME type for the resource
Example:
@server.template("weather://{city}/current")
def get_weather(city: str) -> str:
return f"Weather for {city}"
"""
# Check if user passed function directly instead of calling decorator
if callable(uri_template):
raise TypeError(
"The @template decorator was used incorrectly. "
"Did you forget to call it? Use @template('uri_template') instead of @template"
)
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
return func(*args, **kwargs)
self._resource_manager.add_template(
wrapper,
uri_template=uri_template,
name=name,
description=description,
mime_type=mime_type or "text/plain",
)
# Check if this should be a template
has_uri_params = "{" in uri and "}" in uri
has_func_params = bool(inspect.signature(func).parameters)
if has_uri_params or has_func_params:
# Validate that URI params match function params
uri_params = set(re.findall(r"{(\w+)}", uri))
func_params = set(inspect.signature(func).parameters.keys())
if uri_params != func_params:
raise ValueError(
f"Mismatch between URI parameters {uri_params} "
f"and function parameters {func_params}"
)
# Register as template
self._resource_manager.add_template(
wrapper,
uri_template=uri,
name=name,
description=description,
mime_type=mime_type or "text/plain",
)
else:
# Register as regular resource
resource = FunctionResource(
uri=uri,
name=name,
description=description,
mime_type=mime_type or "text/plain",
func=wrapper,
)
self.add_resource(resource)
return wrapper
return decorator

68
tests/resources/test_file_resources.py
View file

@ -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

101
tests/resources/test_function_resources.py
View file

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

430
tests/resources/test_resource_manager.py
View file

@ -1,14 +1,11 @@
import logging
import pytest
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory
from tempfile import NamedTemporaryFile
from fastmcp.resources import (
FileResource,
FunctionResource,
ResourceManager,
TextResource,
BinaryResource,
ResourceTemplate,
)
@ -30,124 +27,59 @@ def temp_file():
pass # File was already deleted by the test
@pytest.fixture
def temp_file_no_cleanup():
"""Create a temporary file for testing.
class TestResourceManager:
"""Test ResourceManager functionality."""
File is NOT automatically cleaned up - tests must handle cleanup.
"""
content = "test content"
with NamedTemporaryFile(mode="w", delete=False) as f:
f.write(content)
path = Path(f.name).resolve()
return path
@pytest.fixture
def temp_dir():
"""Create a temporary directory for testing."""
with TemporaryDirectory() as d:
yield Path(d).resolve()
class TestResourceValidation:
def test_resource_uri_validation(self):
def dummy_func() -> str:
return "data"
# Valid URI
resource = FunctionResource(
uri="http://example.com/data",
name="test",
func=dummy_func,
)
assert str(resource.uri) == "http://example.com/data"
# Missing protocol
with pytest.raises(ValueError, match="Input should be a valid URL"):
FunctionResource(
uri="invalid",
name="test",
func=dummy_func,
)
# Missing host
with pytest.raises(ValueError, match="Input should be a valid URL"):
FunctionResource(
uri="http://",
name="test",
func=dummy_func,
)
class TestResourceManagerAdd:
"""Test ResourceManager add functionality."""
def test_add_file_resource(self, temp_file: Path):
"""Test adding a file resource."""
def test_add_resource(self, temp_file: Path):
"""Test adding a resource."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
name="test",
description="test file",
mime_type="text/plain",
path=temp_file,
)
added = manager.add_resource(resource)
assert isinstance(added, FileResource)
assert str(added.uri) == f"file://{temp_file}"
assert added.name == "test"
assert added.description == "test file"
assert added.mime_type == "text/plain"
assert added.path == temp_file
assert added == resource
assert manager.list_resources() == [resource]
def test_add_file_resource_relative_path_error(self):
"""Test ResourceManager rejects relative paths."""
with pytest.raises(ValueError, match="Path must be absolute"):
FileResource(
uri="file:///test.txt",
name="test",
path=Path("test.txt"),
)
def test_add_duplicate_resource(self, temp_file: Path):
"""Test adding the same resource twice."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
first = manager.add_resource(resource)
second = manager.add_resource(resource)
assert first == second
assert manager.list_resources() == [resource]
def test_warn_on_duplicate_resources(self, caplog):
def test_warn_on_duplicate_resources(self, temp_file: Path, caplog):
"""Test warning on duplicate resources."""
caplog.set_level(logging.WARNING, logger="mcp")
manager = ResourceManager()
resource = FileResource(
uri="file:///test.txt",
uri=f"file://{temp_file}",
name="test",
path=Path("/test.txt"),
path=temp_file,
)
manager.add_resource(resource)
manager.add_resource(resource)
assert "Resource already exists: file:///test.txt" in caplog.text
assert "Resource already exists" in caplog.text
def test_disable_warn_on_duplicate_resources(self, caplog):
def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog):
"""Test disabling warning on duplicate resources."""
caplog.set_level(logging.WARNING, logger="mcp")
manager = ResourceManager()
manager = ResourceManager(warn_on_duplicate_resources=False)
resource = FileResource(
uri="file:///test.txt",
uri=f"file://{temp_file}",
name="test",
path=Path("/test.txt"),
path=temp_file,
)
manager.add_resource(resource)
manager.warn_on_duplicate_resources = False
manager.add_resource(resource)
assert "Resource already exists: file:///test.txt" not in caplog.text
assert "Resource already exists" not in caplog.text
class TestResourceManagerRead:
"""Test ResourceManager read functionality."""
def test_get_resource_unknown_uri(self):
"""Test getting a non-existent resource."""
manager = ResourceManager()
with pytest.raises(ValueError, match="Unknown resource"):
manager.get_resource("file://unknown")
def test_get_resource(self, temp_file: Path):
async def test_get_resource(self, temp_file: Path):
"""Test getting a resource by URI."""
manager = ResourceManager()
resource = FileResource(
@ -155,271 +87,9 @@ class TestResourceManagerRead:
name="test",
path=temp_file,
)
added = manager.add_resource(resource)
retrieved = manager.get_resource(added.uri)
assert retrieved == added
async def test_resource_read_through_manager(self, temp_file: Path):
"""Test reading a resource through the manager."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
added = manager.add_resource(resource)
retrieved = manager.get_resource(added.uri)
assert retrieved is not None
content = await retrieved.read()
assert content == "test content"
async def test_resource_read_error_through_manager(
self, temp_file_no_cleanup: Path
):
"""Test error handling when reading through manager."""
manager = ResourceManager()
# Create resource while file exists
resource = FileResource(
uri=f"file://{temp_file_no_cleanup}",
name="test",
path=temp_file_no_cleanup,
)
added = manager.add_resource(resource)
retrieved = manager.get_resource(added.uri)
assert retrieved is not None
# Delete file and verify read fails
temp_file_no_cleanup.unlink()
with pytest.raises(FileNotFoundError):
await retrieved.read()
class TestResourceManagerList:
"""Test ResourceManager list functionality."""
def test_list_resources(self, temp_file: Path):
"""Test listing all resources."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
added = manager.add_resource(resource)
resources = manager.list_resources()
assert len(resources) == 1
assert resources[0] == added
def test_list_resources_duplicate(self, temp_file: Path):
"""Test that adding the same resource twice only stores it once."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
resource1 = manager.add_resource(resource)
resource2 = manager.add_resource(resource)
resources = manager.list_resources()
assert len(resources) == 1
assert resources[0] == resource1
assert resource1 == resource2
def test_list_multiple_resources(self, temp_file: Path, temp_file_no_cleanup: Path):
"""Test listing multiple different resources."""
manager = ResourceManager()
resource1 = FileResource(
uri=f"file://{temp_file}",
name="test1",
path=temp_file,
)
resource2 = FileResource(
uri=f"file://{temp_file_no_cleanup}",
name="test2",
path=temp_file_no_cleanup,
)
added1 = manager.add_resource(resource1)
added2 = manager.add_resource(resource2)
resources = manager.list_resources()
assert len(resources) == 2
assert resources[0] == added1
assert resources[1] == added2
assert added1 != added2
class TestTextResource:
"""Test TextResource functionality."""
async def test_text_resource_read(self):
"""Test reading from a TextResource."""
resource = TextResource(
uri="text://test",
name="test",
text="Hello, world!",
)
content = await resource.read()
assert content == "Hello, world!"
assert resource.mime_type == "text/plain"
def test_text_resource_custom_mime(self):
"""Test TextResource with custom mime type."""
resource = TextResource(
uri="text://test",
name="test",
text="<html></html>",
mime_type="text/html",
)
assert resource.mime_type == "text/html"
class TestBinaryResource:
"""Test BinaryResource functionality."""
async def test_binary_resource_read(self):
"""Test reading from a BinaryResource."""
data = b"Hello, world!"
resource = BinaryResource(
uri="binary://test",
name="test",
data=data,
)
content = await resource.read()
assert content == data
assert resource.mime_type == "application/octet-stream"
def test_binary_resource_custom_mime(self):
"""Test BinaryResource with custom mime type."""
resource = BinaryResource(
uri="binary://test",
name="test",
data=b"test",
mime_type="image/png",
)
assert resource.mime_type == "image/png"
class TestResourceTemplate:
"""Test ResourceTemplate functionality."""
def test_template_from_function(self):
"""Test creating a template from a function."""
def weather(city: str, units: str = "metric") -> str:
return f"Weather in {city} ({units})"
template = ResourceTemplate.from_function(
func=weather,
uri_template="weather://{city}/current",
name="weather",
description="Get current weather",
mime_type="text/plain",
)
assert template.name == "weather"
assert template.uri_template == "weather://{city}/current"
assert template.mime_type == "text/plain"
assert "city" in template.parameters["properties"]
def test_template_from_lambda_error(self):
"""Test error when creating template from lambda without name."""
with pytest.raises(
ValueError, match="You must provide a name for lambda functions"
):
ResourceTemplate.from_function(
func=lambda x: x,
uri_template="test://{x}",
)
def test_template_matches(self):
"""Test URI matching against template."""
def dummy(x: str) -> str:
return x
template = ResourceTemplate.from_function(
func=dummy,
uri_template="test://{x}/value",
name="test",
)
# Test matching URI
params = template.matches("test://hello/value")
assert params == {"x": "hello"}
# Test non-matching URI
params = template.matches("test://hello/wrong")
assert params is None
async def test_template_create_text_resource(self):
"""Test creating a TextResource from template."""
def greet(name: str) -> str:
return f"Hello, {name}!"
template = ResourceTemplate.from_function(
func=greet,
uri_template="greet://{name}",
name="greeter",
)
resource = await template.create_resource(
"greet://world",
{"name": "world"},
)
assert isinstance(resource, TextResource)
content = await resource.read()
assert content == "Hello, world!"
async def test_template_create_binary_resource(self):
"""Test creating a BinaryResource from template."""
def get_bytes(value: str) -> bytes:
return value.encode()
template = ResourceTemplate.from_function(
func=get_bytes,
uri_template="bytes://{value}",
name="bytes",
mime_type="application/octet-stream",
)
resource = await template.create_resource(
"bytes://test",
{"value": "test"},
)
assert isinstance(resource, BinaryResource)
content = await resource.read()
assert content == b"test"
async def test_template_json_conversion(self):
"""Test automatic JSON conversion of non-string/bytes results."""
def get_data(key: str) -> dict:
return {"key": key, "value": 123}
template = ResourceTemplate.from_function(
func=get_data,
uri_template="data://{key}",
name="data",
)
resource = await template.create_resource(
"data://test",
{"key": "test"},
)
assert isinstance(resource, TextResource)
content = await resource.read()
assert '"key": "test"' in content
assert '"value": 123' in content
class TestResourceManagerWithTemplates:
"""Test ResourceManager template functionality."""
manager.add_resource(resource)
retrieved = await manager.get_resource(resource.uri)
assert retrieved == resource
async def test_get_resource_from_template(self):
"""Test getting a resource through a template."""
@ -436,23 +106,31 @@ class TestResourceManagerWithTemplates:
manager._templates[template.uri_template] = template
resource = await manager.get_resource("greet://world")
assert isinstance(resource, TextResource)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "Hello, world!"
async def test_template_error_handling(self):
"""Test error handling in template resource creation."""
async def test_get_unknown_resource(self):
"""Test getting a non-existent resource."""
manager = ResourceManager()
with pytest.raises(ValueError, match="Unknown resource"):
await manager.get_resource("unknown://test")
def failing_func(x: str) -> str:
raise ValueError("Test error")
template = ResourceTemplate.from_function(
func=failing_func,
uri_template="fail://{x}",
name="fail",
def test_list_resources(self, temp_file: Path):
"""Test listing all resources."""
manager = ResourceManager()
resource1 = FileResource(
uri=f"file://{temp_file}",
name="test1",
path=temp_file,
)
manager._templates[template.uri_template] = template
with pytest.raises(ValueError, match="Error creating resource from template"):
await manager.get_resource("fail://test")
resource2 = FileResource(
uri=f"file://{temp_file}2",
name="test2",
path=temp_file,
)
manager.add_resource(resource1)
manager.add_resource(resource2)
resources = manager.list_resources()
assert len(resources) == 2
assert resources == [resource1, resource2]

238
tests/resources/test_resource_template.py Normal file
View file

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

98
tests/resources/test_resources.py Normal file
View file

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

88
tests/test_server.py
View file

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