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 FileSystemProvider for filesystem-based component discovery (#2823)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-01-09 16:31:29 -05:00 • committed by GitHub
commit 87f2edd9c0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 2419 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

30
docs/docs.json
View file

@ -20,7 +20,10 @@
"primary": "#2d00f7"
},
"contextual": {
"options": ["copy", "view"]
"options": [
"copy",
"view"
]
},
"description": "The fast, Pythonic way to build MCP servers and clients.",
"errors": {
@ -106,6 +109,7 @@
"pages": [
"servers/providers/overview",
"servers/providers/local",
"servers/providers/filesystem",
"servers/providers/mounting",
"servers/providers/namespacing",
"servers/providers/proxy",
@ -157,7 +161,10 @@
{
"group": "Essentials",
"icon": "cube",
"pages": ["clients/client", "clients/transports"]
"pages": [
"clients/client",
"clients/transports"
]
},
{
"group": "Core Operations",
@ -184,7 +191,10 @@
{
"group": "Authentication",
"icon": "user-shield",
"pages": ["clients/auth/oauth", "clients/auth/bearer"]
"pages": [
"clients/auth/oauth",
"clients/auth/bearer"
]
}
]
},
@ -194,7 +204,10 @@
{
"group": "Providers",
"icon": "globe",
"pages": ["integrations/fastapi", "integrations/openapi"]
"pages": [
"integrations/fastapi",
"integrations/openapi"
]
},
{
"group": "Authentication",
@ -336,6 +349,15 @@
"python-sdk/fastmcp-client-transports"
]
},
{
"group": "fastmcp.fs",
"pages": [
"python-sdk/fastmcp-fs-__init__",
"python-sdk/fastmcp-fs-decorators",
"python-sdk/fastmcp-fs-discovery",
"python-sdk/fastmcp-fs-provider"
]
},
{
"group": "fastmcp.prompts",
"pages": [

251
docs/servers/providers/filesystem.mdx Normal file
View file

@ -0,0 +1,251 @@
---
title: Filesystem Provider
sidebarTitle: Filesystem
description: Automatic component discovery from Python files
icon: folder-tree
---
import { VersionBadge } from '/snippets/version-badge.mdx'
<VersionBadge version="3.0.0" />
`FileSystemProvider` scans a directory for Python files and automatically registers functions decorated with `@tool`, `@resource`, or `@prompt`. This enables a file-based organization pattern similar to Next.js routing, where your project structure becomes your component registry.
## Why Filesystem Discovery
Traditional FastMCP servers require coordination between files. Either your tool files import the server to call `@server.tool()`, or your server file imports all the tool modules. Both approaches create coupling that some developers prefer to avoid.
`FileSystemProvider` eliminates this coordination. Each file is self-contained—it uses decorators from `fastmcp.fs` that don't require access to a server instance. The provider discovers these files at startup, so you can add new tools without modifying your server file.
This is a convention some teams prefer, not necessarily better for all projects. The tradeoffs:
- **No coordination**: Files don't import the server; server doesn't import files
- **Predictable naming**: Function names become component names (unless overridden)
- **Development mode**: Optionally re-scan files on every request for rapid iteration
## Quick Start
Create a provider pointing to your components directory, then pass it to your server.
```python
from fastmcp import FastMCP
from fastmcp.fs import FileSystemProvider
mcp = FastMCP("MyServer", providers=[FileSystemProvider("mcp/")])
```
In your `mcp/` directory, create Python files with decorated functions.
```python
# mcp/tools/greet.py
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
"""Greet someone by name."""
return f"Hello, {name}!"
```
When the server starts, `FileSystemProvider` scans the directory, imports all Python files, and registers any decorated functions it finds.
## Decorators
The `fastmcp.fs` module provides three decorators that mark functions for discovery: `@tool`, `@resource`, and `@prompt`. These support the full syntax of standard FastMCP decorators—all the same parameters work identically.
### @tool
Mark a function as a tool. The function name becomes the tool name by default.
```python
from fastmcp.fs import tool
@tool
def calculate_sum(a: float, b: float) -> float:
"""Add two numbers together."""
return a + b
```
Customize the tool with optional parameters.
```python
from fastmcp.fs import tool
@tool(
name="add-numbers",
description="Add two numbers together.",
tags={"math", "arithmetic"},
)
def add(a: float, b: float) -> float:
return a + b
```
The decorator supports all standard tool options: `name`, `title`, `description`, `icons`, `tags`, `output_schema`, `annotations`, and `meta`.
### @resource
Mark a function as a resource. Unlike `@tool`, the `@resource` decorator requires a URI argument.
```python
from fastmcp.fs import resource
@resource("config://app")
def get_app_config() -> str:
"""Get application configuration."""
return '{"version": "1.0"}'
```
URIs with template parameters create resource templates. The provider automatically detects whether to register a static resource or a template based on whether the URI contains `{parameters}` or the function has arguments.
```python
from fastmcp.fs import resource
@resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
"""Get a user's profile by ID."""
return f'{{"id": "{user_id}", "name": "User"}}'
```
The decorator supports: `uri` (required), `name`, `title`, `description`, `icons`, `mime_type`, `tags`, `annotations`, and `meta`.
### @prompt
Mark a function as a prompt template.
```python
from fastmcp.fs import prompt
@prompt
def code_review(code: str, language: str = "python") -> str:
"""Generate a code review prompt."""
return f"Please review this {language} code:\n\n```{language}\n{code}\n```"
```
```python
from fastmcp.fs import prompt
@prompt(name="explain-concept", tags={"education"})
def explain(topic: str) -> str:
"""Generate an explanation prompt."""
return f"Explain {topic} using clear examples and analogies."
```
The decorator supports: `name`, `title`, `description`, `icons`, `tags`, and `meta`.
## Directory Structure
The directory structure is purely organizational. The provider recursively scans all `.py` files regardless of which subdirectory they're in. Subdirectories like `tools/`, `resources/`, and `prompts/` are optional conventions that help you organize code.
```
mcp/
├── tools/
│ ├── greeting.py # @tool functions
│ └── calculator.py # @tool functions
├── resources/
│ └── config.py # @resource functions
└── prompts/
└── assistant.py # @prompt functions
```
You can also put all components in a single file or organize by feature rather than type.
```
mcp/
├── user_management.py # @tool, @resource, @prompt for users
├── billing.py # @tool, @resource for billing
└── analytics.py # @tool for analytics
```
## Discovery Rules
The provider follows these rules when scanning:
| Rule | Behavior |
|------|----------|
| File extensions | Only `.py` files are scanned |
| `__init__.py` | Skipped (used for package structure, not components) |
| `__pycache__` | Skipped |
| Private functions | Functions starting with `_` are ignored, even if decorated |
| No decorators | Files without `@tool`, `@resource`, or `@prompt` are silently skipped |
| Multiple components | A single file can contain any number of decorated functions |
### Package Imports
If your directory contains an `__init__.py` file, the provider imports files as proper Python package members. This means relative imports work correctly within your components directory.
```python
# mcp/__init__.py exists
# mcp/tools/greeting.py
from ..helpers import format_name # Relative imports work
@tool
def greet(name: str) -> str:
return f"Hello, {format_name(name)}!"
```
Without `__init__.py`, files are imported directly using `importlib.util.spec_from_file_location`.
## Reload Mode
During development, you may want changes to component files to take effect without restarting the server. Enable reload mode to re-scan the directory on every request.
```python
from fastmcp.fs import FileSystemProvider
provider = FileSystemProvider("mcp/", reload=True)
```
With `reload=True`, the provider:
1. Re-discovers all Python files on each request
2. Re-imports modules that have changed
3. Updates the component registry with any new, modified, or removed components
<Warning>
Reload mode adds overhead to every request. Use it only during development, not in production.
</Warning>
## Error Handling
When a file fails to import (syntax error, missing dependency, etc.), the provider logs a warning and continues scanning other files. Failed imports don't prevent the server from starting.
```
WARNING - Failed to import /path/to/broken.py: No module named 'missing_dep'
```
The provider tracks which files have failed and only re-logs warnings when the file's modification time changes. This prevents log spam when a broken file is repeatedly scanned in reload mode.
## Example Project
A complete example is available in the repository at `examples/filesystem-provider/`. The structure demonstrates the recommended organization.
```
examples/filesystem-provider/
├── server.py # Server entry point
└── mcp/
├── tools/
│ ├── greeting.py # greet, farewell tools
│ └── calculator.py # add, multiply tools
├── resources/
│ └── config.py # Static and templated resources
└── prompts/
└── assistant.py # code_review, explain prompts
```
The server entry point is minimal.
```python
from pathlib import Path
from fastmcp import FastMCP
from fastmcp.fs import FileSystemProvider
provider = FileSystemProvider(
root=Path(__file__).parent / "mcp",
reload=True,
)
mcp = FastMCP("FilesystemDemo", providers=[provider])
```
Run with `fastmcp run examples/filesystem-provider/server.py` or inspect with `fastmcp inspect examples/filesystem-provider/server.py`.

39
examples/filesystem-provider/mcp/prompts/assistant.py Normal file
View file

@ -0,0 +1,39 @@
"""Assistant prompts."""
from fastmcp.fs import prompt
@prompt
def code_review(code: str, language: str = "python") -> str:
"""Generate a code review prompt.
Args:
code: The code to review.
language: Programming language (default: python).
"""
return f"""Please review this {language} code:
```{language}
{code}
```
Focus on:
- Code quality and readability
- Potential bugs or issues
- Performance considerations
- Best practices"""
@prompt(
name="explain-concept",
description="Generate a prompt to explain a technical concept.",
tags={"education", "explanation"},
)
def explain(topic: str, audience: str = "developer") -> str:
"""Generate an explanation prompt.
Args:
topic: The concept to explain.
audience: Target audience level.
"""
return f"Explain {topic} to a {audience}. Use clear examples and analogies."

55
examples/filesystem-provider/mcp/resources/config.py Normal file
View file

@ -0,0 +1,55 @@
"""Configuration resources - static and templated."""
import json
from fastmcp.fs import resource
# Static resource - no parameters in URI
@resource("config://app")
def get_app_config() -> str:
"""Get application configuration."""
return json.dumps(
{
"name": "FilesystemDemo",
"version": "1.0.0",
"features": ["tools", "resources", "prompts"],
},
indent=2,
)
# Resource template - {env} is a parameter
@resource("config://env/{env}")
def get_env_config(env: str) -> str:
"""Get environment-specific configuration.
Args:
env: Environment name (dev, staging, prod).
"""
configs = {
"dev": {"debug": True, "log_level": "DEBUG", "database": "localhost"},
"staging": {"debug": True, "log_level": "INFO", "database": "staging-db"},
"prod": {"debug": False, "log_level": "WARNING", "database": "prod-db"},
}
config = configs.get(env, {"error": f"Unknown environment: {env}"})
return json.dumps(config, indent=2)
# Resource with custom metadata
@resource(
"config://features",
name="feature-flags",
mime_type="application/json",
tags={"config", "features"},
)
def get_feature_flags() -> str:
"""Get feature flags configuration."""
return json.dumps(
{
"dark_mode": True,
"beta_features": False,
"max_upload_size_mb": 100,
},
indent=2,
)

24
examples/filesystem-provider/mcp/tools/calculator.py Normal file
View file

@ -0,0 +1,24 @@
"""Math tools with custom metadata."""
from fastmcp.fs import tool
@tool(
name="add-numbers", # Custom name (default would be "add")
description="Add two numbers together.",
tags={"math", "arithmetic"},
)
def add(a: float, b: float) -> float:
"""Add two numbers."""
return a + b
@tool(tags={"math", "arithmetic"})
def multiply(a: float, b: float) -> float:
"""Multiply two numbers.
Args:
a: First number.
b: Second number.
"""
return a * b

28
examples/filesystem-provider/mcp/tools/greeting.py Normal file
View file

@ -0,0 +1,28 @@
"""Greeting tools - multiple tools in one file."""
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
"""Greet someone by name.
Args:
name: The person's name.
"""
return f"Hello, {name}!"
@tool
def farewell(name: str) -> str:
"""Say goodbye to someone.
Args:
name: The person's name.
"""
return f"Goodbye, {name}!"
# Helper functions without decorators are ignored
def _format_message(msg: str) -> str:
return msg.strip().capitalize()

29
examples/filesystem-provider/server.py Normal file
View file

@ -0,0 +1,29 @@
"""Filesystem-based MCP server using FileSystemProvider.
This example demonstrates how to use FileSystemProvider to automatically
discover and register tools, resources, and prompts from the filesystem.
Run:
fastmcp run examples/filesystem-provider/server.py
Inspect:
fastmcp inspect examples/filesystem-provider/server.py
Dev mode (re-scan files on every request):
Change reload=True below, then modify files while the server runs.
"""
from pathlib import Path
from fastmcp import FastMCP
from fastmcp.fs import FileSystemProvider
# The provider scans all .py files in the directory recursively.
# Functions decorated with @tool, @resource, or @prompt are registered.
# Directory structure is purely organizational - decorators determine type.
provider = FileSystemProvider(
root=Path(__file__).parent / "mcp",
reload=True, # Set True for dev mode (re-scan on every request)
)
mcp = FastMCP("FilesystemDemo", providers=[provider])

45
src/fastmcp/fs/__init__.py Normal file
View file

@ -0,0 +1,45 @@
"""Filesystem-based component discovery for FastMCP.
This module provides decorators and a provider for discovering MCP components
from the filesystem. Files are scanned for functions decorated with @tool,
@resource, or @prompt, and automatically registered with the server.
Example:
```python
# server.py
from fastmcp import FastMCP
from fastmcp.fs import FileSystemProvider
mcp = FastMCP("MyServer", providers=[FileSystemProvider("mcp/")])
```
```python
# mcp/tools/greet.py
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
'''Greet someone by name.'''
return f"Hello, {name}!"
```
```python
# mcp/resources/config.py
from fastmcp.fs import resource
@resource("config://app")
def get_config() -> dict:
'''Get application configuration.'''
return {"version": "1.0"}
```
"""
from fastmcp.fs.decorators import prompt, resource, tool
from fastmcp.fs.provider import FileSystemProvider
__all__ = [
"FileSystemProvider",
"prompt",
"resource",
"tool",
]

394
src/fastmcp/fs/decorators.py Normal file
View file

@ -0,0 +1,394 @@
"""Decorators for marking functions in filesystem-based discovery.
These decorators mark functions with metadata so that FileSystemProvider
can discover and register them. Unlike LocalProvider's decorators, these
do NOT register components immediately - they just store metadata on the
function for later discovery.
Example:
```python
# mcp/tools/greet.py
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
'''Greet someone by name.'''
return f"Hello, {name}!"
@tool(name="custom-greet", tags={"greeting"})
def my_greet(name: str) -> str:
return f"Hi, {name}!"
```
"""
from __future__ import annotations
import inspect
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal, overload
from mcp.types import Annotations, AnyFunction, ToolAnnotations
if TYPE_CHECKING:
import mcp.types
# Attribute name used to store metadata on decorated functions
FS_META_ATTR = "_fastmcp_fs_meta"
@dataclass
class ToolMeta:
"""Metadata stored on functions decorated with @tool."""
type: Literal["tool"] = "tool"
name: str | None = None
title: str | None = None
description: str | None = None
icons: list[mcp.types.Icon] | None = None
tags: set[str] | None = None
output_schema: dict[str, Any] | None = None
annotations: ToolAnnotations | None = None
meta: dict[str, Any] | None = None
@dataclass
class ResourceMeta:
"""Metadata stored on functions decorated with @resource."""
type: Literal["resource"] = "resource"
uri: str = ""
name: str | None = None
title: str | None = None
description: str | None = None
icons: list[mcp.types.Icon] | None = None
mime_type: str | None = None
tags: set[str] | None = None
annotations: Annotations | None = None
meta: dict[str, Any] | None = None
@dataclass
class PromptMeta:
"""Metadata stored on functions decorated with @prompt."""
type: Literal["prompt"] = "prompt"
name: str | None = None
title: str | None = None
description: str | None = None
icons: list[mcp.types.Icon] | None = None
tags: set[str] | None = None
meta: dict[str, Any] | None = None
FSMeta = ToolMeta | ResourceMeta | PromptMeta
def get_fs_meta(fn: Any) -> FSMeta | None:
"""Get filesystem metadata from a function if it has been decorated."""
return getattr(fn, FS_META_ATTR, None)
def has_fs_meta(fn: Any) -> bool:
"""Check if a function has filesystem metadata."""
return hasattr(fn, FS_META_ATTR)
# =============================================================================
# @tool decorator
# =============================================================================
@overload
def tool(fn: AnyFunction) -> AnyFunction: ...
@overload
def tool(
fn: None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | None = None,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
) -> Any: ...
@overload
def tool(
fn: str,
*,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | None = None,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
) -> Any: ...
def tool(
fn: AnyFunction | str | None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | None = None,
annotations: ToolAnnotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
) -> Any:
"""Mark a function as a tool for filesystem-based discovery.
This decorator stores metadata on the function but does NOT register it.
FileSystemProvider discovers marked functions when scanning directories.
Supports multiple calling patterns:
- @tool (without parentheses)
- @tool() (with empty parentheses)
- @tool("custom_name") (with name as first argument)
- @tool(name="custom_name") (with keyword arguments)
Args:
fn: The function to decorate, or a name string, or None
name: Optional name for the tool (defaults to function name)
title: Optional title for display
description: Optional description (defaults to docstring)
icons: Optional icons for the tool
tags: Optional tags for categorization
output_schema: Optional JSON schema for output
annotations: Optional tool annotations
meta: Optional metadata dict
Example:
```python
@tool
def greet(name: str) -> str:
'''Greet someone.'''
return f"Hello, {name}!"
@tool(name="custom-greet", tags={"greeting"})
def my_greet(name: str) -> str:
return f"Hi, {name}!"
```
"""
if isinstance(annotations, dict):
annotations = ToolAnnotations(**annotations)
def decorator(func: AnyFunction) -> AnyFunction:
tool_meta = ToolMeta(
name=name,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
)
setattr(func, FS_META_ATTR, tool_meta)
return func
if inspect.isroutine(fn):
# @tool without parentheses
return decorator(fn)
elif isinstance(fn, str):
# @tool("custom_name")
return tool(
name=fn,
title=title,
description=description,
icons=icons,
tags=tags,
output_schema=output_schema,
annotations=annotations,
meta=meta,
)
else:
# @tool() or @tool(name="...") - return decorator
return decorator
# =============================================================================
# @resource decorator
# =============================================================================
def resource(
uri: str,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
annotations: Annotations | dict[str, Any] | None = None,
meta: dict[str, Any] | None = None,
) -> Any:
"""Mark a function as a resource for filesystem-based discovery.
This decorator stores metadata on the function but does NOT register it.
FileSystemProvider discovers marked functions when scanning directories.
Unlike @tool and @prompt, @resource REQUIRES a URI argument.
Args:
uri: URI for the resource (e.g., "config://app" or "users://{user_id}")
name: Optional name for the resource
title: Optional title for display
description: Optional description (defaults to docstring)
icons: Optional icons for the resource
mime_type: Optional MIME type
tags: Optional tags for categorization
annotations: Optional resource annotations
meta: Optional metadata dict
Example:
```python
@resource("config://app")
def get_config() -> dict:
return {"setting": "value"}
@resource("users://{user_id}/profile")
def get_profile(user_id: str) -> dict:
return {"id": user_id, "name": "User"}
```
"""
if inspect.isroutine(uri):
raise TypeError(
"The @resource decorator requires a URI. "
"Use @resource('uri://...') instead of @resource"
)
if isinstance(annotations, dict):
annotations = Annotations(**annotations)
def decorator(func: AnyFunction) -> AnyFunction:
resource_meta = ResourceMeta(
uri=uri,
name=name,
title=title,
description=description,
icons=icons,
mime_type=mime_type,
tags=tags,
annotations=annotations,
meta=meta,
)
setattr(func, FS_META_ATTR, resource_meta)
return func
return decorator
# =============================================================================
# @prompt decorator
# =============================================================================
@overload
def prompt(fn: AnyFunction) -> AnyFunction: ...
@overload
def prompt(
fn: None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
) -> Any: ...
@overload
def prompt(
fn: str,
*,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
) -> Any: ...
def prompt(
fn: AnyFunction | str | None = None,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
icons: list[mcp.types.Icon] | None = None,
tags: set[str] | None = None,
meta: dict[str, Any] | None = None,
) -> Any:
"""Mark a function as a prompt for filesystem-based discovery.
This decorator stores metadata on the function but does NOT register it.
FileSystemProvider discovers marked functions when scanning directories.
Supports multiple calling patterns:
- @prompt (without parentheses)
- @prompt() (with empty parentheses)
- @prompt("custom_name") (with name as first argument)
- @prompt(name="custom_name") (with keyword arguments)
Args:
fn: The function to decorate, or a name string, or None
name: Optional name for the prompt (defaults to function name)
title: Optional title for display
description: Optional description (defaults to docstring)
icons: Optional icons for the prompt
tags: Optional tags for categorization
meta: Optional metadata dict
Example:
```python
@prompt
def analyze(topic: str) -> list:
'''Analyze a topic.'''
return [{"role": "user", "content": f"Analyze: {topic}"}]
@prompt(name="custom-analyze")
def my_analyze(topic: str) -> list:
return [{"role": "user", "content": topic}]
```
"""
def decorator(func: AnyFunction) -> AnyFunction:
prompt_meta = PromptMeta(
name=name,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
)
setattr(func, FS_META_ATTR, prompt_meta)
return func
if inspect.isroutine(fn):
# @prompt without parentheses
return decorator(fn)
elif isinstance(fn, str):
# @prompt("custom_name")
return prompt(
name=fn,
title=title,
description=description,
icons=icons,
tags=tags,
meta=meta,
)
else:
# @prompt() or @prompt(name="...") - return decorator
return decorator

242
src/fastmcp/fs/discovery.py Normal file
View file

@ -0,0 +1,242 @@
"""File discovery and module import utilities for filesystem-based routing.
This module provides functions to:
1. Discover Python files in a directory tree
2. Import modules (as packages if __init__.py exists, else directly)
3. Extract decorated functions from imported modules
"""
from __future__ import annotations
import importlib.util
import sys
from dataclasses import dataclass, field
from pathlib import Path
from types import ModuleType
from typing import Any
from fastmcp.fs.decorators import (
FSMeta,
get_fs_meta,
)
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
@dataclass
class DiscoveryResult:
"""Result of filesystem discovery."""
components: list[tuple[Path, Any, FSMeta]] = field(default_factory=list)
failed_files: dict[Path, str] = field(default_factory=dict) # path -> error message
def discover_files(root: Path) -> list[Path]:
"""Recursively discover all Python files under a directory.
Excludes __init__.py files (they're for package structure, not components).
Args:
root: Root directory to scan.
Returns:
List of .py file paths, sorted for deterministic order.
"""
if not root.exists():
return []
if not root.is_dir():
# If root is a file, just return it (if it's a .py file)
if root.suffix == ".py" and root.name != "__init__.py":
return [root]
return []
files: list[Path] = []
for path in root.rglob("*.py"):
# Skip __init__.py files
if path.name == "__init__.py":
continue
# Skip __pycache__ directories
if "__pycache__" in path.parts:
continue
files.append(path)
# Sort for deterministic discovery order
return sorted(files)
def _is_package_dir(directory: Path) -> bool:
"""Check if a directory is a Python package (has __init__.py)."""
return (directory / "__init__.py").exists()
def _find_package_root(file_path: Path) -> Path | None:
"""Find the root of the package containing this file.
Walks up the directory tree until we find a directory without __init__.py.
Returns:
The package root directory, or None if not in a package.
"""
current = file_path.parent
package_root = None
while current != current.parent: # Stop at filesystem root
if _is_package_dir(current):
package_root = current
current = current.parent
else:
break
return package_root
def _compute_module_name(file_path: Path, package_root: Path) -> str:
"""Compute the dotted module name for a file within a package.
Args:
file_path: Path to the Python file.
package_root: Root directory of the package.
Returns:
Dotted module name (e.g., "mcp.tools.greet").
"""
relative = file_path.relative_to(package_root.parent)
parts = list(relative.parts)
# Remove .py extension from last part
parts[-1] = parts[-1].removesuffix(".py")
return ".".join(parts)
def import_module_from_file(file_path: Path) -> ModuleType:
"""Import a Python file as a module.
If the file is part of a package (directory has __init__.py), imports
it as a proper package member (relative imports work). Otherwise,
imports directly using spec_from_file_location.
Args:
file_path: Path to the Python file.
Returns:
The imported module.
Raises:
ImportError: If the module cannot be imported.
"""
file_path = file_path.resolve()
# Check if this file is part of a package
package_root = _find_package_root(file_path)
if package_root is not None:
# Import as part of a package
module_name = _compute_module_name(file_path, package_root)
# Ensure package root's parent is in sys.path
package_parent = str(package_root.parent)
if package_parent not in sys.path:
sys.path.insert(0, package_parent)
# Import using standard import machinery
# If already imported, reload to pick up changes (for reload mode)
try:
if module_name in sys.modules:
return importlib.reload(sys.modules[module_name])
return importlib.import_module(module_name)
except ImportError as e:
raise ImportError(
f"Failed to import {module_name} from {file_path}: {e}"
) from e
else:
# Import directly using spec_from_file_location
module_name = file_path.stem
# Ensure parent directory is in sys.path for imports
parent_dir = str(file_path.parent)
if parent_dir not in sys.path:
sys.path.insert(0, parent_dir)
spec = importlib.util.spec_from_file_location(module_name, file_path)
if spec is None or spec.loader is None:
raise ImportError(f"Cannot load spec for {file_path}")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
try:
spec.loader.exec_module(module)
except Exception as e:
# Clean up sys.modules on failure
sys.modules.pop(module_name, None)
raise ImportError(f"Failed to execute module {file_path}: {e}") from e
return module
def extract_components(module: ModuleType) -> list[tuple[Any, FSMeta]]:
"""Extract all decorated functions from a module.
Scans all module attributes for functions that have been decorated
with @tool, @resource, or @prompt.
Args:
module: The imported module to scan.
Returns:
List of (function, metadata) tuples for each decorated function.
"""
components: list[tuple[Any, FSMeta]] = []
for name in dir(module):
# Skip private/magic attributes
if name.startswith("_"):
continue
try:
obj = getattr(module, name)
except AttributeError:
continue
# Check if this object has our marker
meta = get_fs_meta(obj)
if meta is not None:
components.append((obj, meta))
return components
def discover_and_import(root: Path) -> DiscoveryResult:
"""Discover files, import modules, and extract components.
This is the main entry point for filesystem-based discovery.
Args:
root: Root directory to scan.
Returns:
DiscoveryResult with components and any failed files.
Note:
Files that fail to import are tracked in failed_files, not logged.
The caller is responsible for logging/handling failures.
Files with no decorated functions are silently skipped.
"""
result = DiscoveryResult()
for file_path in discover_files(root):
try:
module = import_module_from_file(file_path)
except ImportError as e:
result.failed_files[file_path] = str(e)
continue
except Exception as e:
result.failed_files[file_path] = str(e)
continue
components = extract_components(module)
for func, meta in components:
result.components.append((file_path, func, meta))
return result

261
src/fastmcp/fs/provider.py Normal file
View file

@ -0,0 +1,261 @@
"""FileSystemProvider for filesystem-based component discovery.
FileSystemProvider scans a directory for Python files, imports them, and
registers any functions decorated with @tool, @resource, or @prompt.
Example:
```python
from fastmcp import FastMCP
from fastmcp.fs import FileSystemProvider
mcp = FastMCP("MyServer", providers=[FileSystemProvider("mcp/")])
```
"""
from __future__ import annotations
import asyncio
import inspect
from collections.abc import Sequence
from pathlib import Path
from typing import Any
from fastmcp.fs.decorators import PromptMeta, ResourceMeta, ToolMeta
from fastmcp.fs.discovery import discover_and_import
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.providers.local_provider import LocalProvider
from fastmcp.tools.tool import Tool
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class FileSystemProvider(LocalProvider):
"""Provider that discovers components from the filesystem.
Scans a directory for Python files and registers functions decorated
with @tool, @resource, or @prompt from fastmcp.fs.
Args:
root: Root directory to scan. Defaults to current directory.
reload: If True, re-scan files on every request (dev mode).
Defaults to False (scan once at init, cache results).
Example:
```python
from fastmcp import FastMCP
from fastmcp.fs import FileSystemProvider
# Basic usage
mcp = FastMCP("MyServer", providers=[FileSystemProvider("mcp/")])
# Dev mode - re-scan on every request
mcp = FastMCP("MyServer", providers=[FileSystemProvider("mcp/", reload=True)])
```
"""
def __init__(
self,
root: str | Path = ".",
reload: bool = False,
) -> None:
super().__init__(on_duplicate="replace")
self._root = Path(root).resolve()
self._reload = reload
self._loaded = False
# Track files we've warned about: path -> mtime when warned
# Re-warn if file changes (mtime differs)
self._warned_files: dict[Path, float] = {}
# Lock for serializing reload operations (created lazily)
self._reload_lock: asyncio.Lock | None = None
# Always load once at init to catch errors early
self._load_components()
def _load_components(self) -> None:
"""Discover and register all components from the filesystem."""
# Clear existing components if reloading
if self._loaded:
self._components.clear()
self._tool_transformations.clear()
result = discover_and_import(self._root)
# Log warnings for failed files (only once per file version)
for file_path, error in result.failed_files.items():
try:
current_mtime = file_path.stat().st_mtime
except OSError:
current_mtime = 0.0
# Warn if we haven't warned about this file, or if it changed
last_warned_mtime = self._warned_files.get(file_path)
if last_warned_mtime is None or last_warned_mtime != current_mtime:
logger.warning(f"Failed to import {file_path}: {error}")
self._warned_files[file_path] = current_mtime
# Clear warnings for files that now import successfully
successful_files = {fp for fp, _, _ in result.components}
for fp in successful_files:
self._warned_files.pop(fp, None)
for file_path, func, meta in result.components:
try:
self._register_component(func, meta)
except Exception as e:
logger.warning(
f"Failed to register {func.__name__} from {file_path}: {e}"
)
self._loaded = True
logger.debug(
f"FileSystemProvider loaded {len(self._components)} components from {self._root}"
)
def _register_component(
self, func: Any, meta: ToolMeta | ResourceMeta | PromptMeta
) -> None:
"""Register a single component based on its metadata type."""
if isinstance(meta, ToolMeta):
self._register_tool(func, meta)
elif isinstance(meta, ResourceMeta):
self._register_resource(func, meta)
elif isinstance(meta, PromptMeta):
self._register_prompt(func, meta)
def _register_tool(self, func: Any, meta: ToolMeta) -> None:
"""Register a tool from a decorated function."""
tool = Tool.from_function(
fn=func,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
tags=meta.tags,
output_schema=meta.output_schema,
annotations=meta.annotations,
meta=meta.meta,
)
self.add_tool(tool)
def _register_resource(self, func: Any, meta: ResourceMeta) -> None:
"""Register a resource or resource template from a decorated function."""
uri = meta.uri
# Check if this should be a template
has_uri_params = "{" in uri and "}" in uri
# Check for function parameters (excluding injected ones)
from fastmcp.server.dependencies import without_injected_parameters
wrapper_fn = without_injected_parameters(func)
has_func_params = bool(inspect.signature(wrapper_fn).parameters)
if has_uri_params or has_func_params:
# Register as template
template = ResourceTemplate.from_function(
fn=func,
uri_template=uri,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
mime_type=meta.mime_type,
tags=meta.tags,
annotations=meta.annotations,
meta=meta.meta,
)
self.add_template(template)
else:
# Register as static resource
resource = Resource.from_function(
fn=func,
uri=uri,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
mime_type=meta.mime_type,
tags=meta.tags,
annotations=meta.annotations,
meta=meta.meta,
)
self.add_resource(resource)
def _register_prompt(self, func: Any, meta: PromptMeta) -> None:
"""Register a prompt from a decorated function."""
prompt = Prompt.from_function(
fn=func,
name=meta.name,
title=meta.title,
description=meta.description,
icons=meta.icons,
tags=meta.tags,
meta=meta.meta,
)
self.add_prompt(prompt)
async def _ensure_loaded(self) -> None:
"""Ensure components are loaded, reloading if in reload mode.
Uses a lock to serialize concurrent reload operations and runs
filesystem I/O off the event loop using asyncio.to_thread.
"""
if not self._reload and self._loaded:
return
# Create lock lazily (can't create in __init__ without event loop)
if self._reload_lock is None:
self._reload_lock = asyncio.Lock()
async with self._reload_lock:
# Double-check after acquiring lock
if self._reload or not self._loaded:
await asyncio.to_thread(self._load_components)
# Override provider methods to support reload mode
async def list_tools(self) -> Sequence[Tool]:
"""Return all tools, reloading if in reload mode."""
await self._ensure_loaded()
return await super().list_tools()
async def get_tool(self, name: str) -> Tool | None:
"""Get a tool by name, reloading if in reload mode."""
await self._ensure_loaded()
return await super().get_tool(name)
async def list_resources(self) -> Sequence[Resource]:
"""Return all resources, reloading if in reload mode."""
await self._ensure_loaded()
return await super().list_resources()
async def get_resource(self, uri: str) -> Resource | None:
"""Get a resource by URI, reloading if in reload mode."""
await self._ensure_loaded()
return await super().get_resource(uri)
async def list_resource_templates(self) -> Sequence[ResourceTemplate]:
"""Return all resource templates, reloading if in reload mode."""
await self._ensure_loaded()
return await super().list_resource_templates()
async def get_resource_template(self, uri: str) -> ResourceTemplate | None:
"""Get a resource template, reloading if in reload mode."""
await self._ensure_loaded()
return await super().get_resource_template(uri)
async def list_prompts(self) -> Sequence[Prompt]:
"""Return all prompts, reloading if in reload mode."""
await self._ensure_loaded()
return await super().list_prompts()
async def get_prompt(self, name: str) -> Prompt | None:
"""Get a prompt by name, reloading if in reload mode."""
await self._ensure_loaded()
return await super().get_prompt(name)
def __repr__(self) -> str:
return f"FileSystemProvider(root={self._root!r}, reload={self._reload})"

277
tests/fs/test_decorators.py Normal file
View file

@ -0,0 +1,277 @@
"""Tests for fastmcp.fs decorators."""
import pytest
from fastmcp.fs.decorators import (
PromptMeta,
ResourceMeta,
ToolMeta,
get_fs_meta,
has_fs_meta,
prompt,
resource,
tool,
)
class TestToolDecorator:
"""Tests for the @tool decorator."""
def test_tool_without_parens(self):
"""@tool without parentheses should work."""
@tool
def greet(name: str) -> str:
return f"Hello, {name}!"
assert has_fs_meta(greet)
meta = get_fs_meta(greet)
assert isinstance(meta, ToolMeta)
assert meta.type == "tool"
assert meta.name is None # Will use function name
def test_tool_with_empty_parens(self):
"""@tool() with empty parentheses should work."""
@tool()
def greet(name: str) -> str:
return f"Hello, {name}!"
assert has_fs_meta(greet)
meta = get_fs_meta(greet)
assert isinstance(meta, ToolMeta)
def test_tool_with_name_arg(self):
"""@tool("name") with name as first arg should work."""
@tool("custom-greet")
def greet(name: str) -> str:
return f"Hello, {name}!"
meta = get_fs_meta(greet)
assert meta is not None
assert meta.name == "custom-greet"
def test_tool_with_name_kwarg(self):
"""@tool(name="name") with keyword arg should work."""
@tool(name="custom-greet")
def greet(name: str) -> str:
return f"Hello, {name}!"
meta = get_fs_meta(greet)
assert meta is not None
assert meta.name == "custom-greet"
def test_tool_with_all_metadata(self):
"""@tool with all metadata should store it all."""
@tool(
name="custom-greet",
title="Greeting Tool",
description="Greets people",
tags={"greeting", "demo"},
meta={"custom": "value"},
)
def greet(name: str) -> str:
return f"Hello, {name}!"
meta = get_fs_meta(greet)
assert meta is not None
assert meta.name == "custom-greet"
assert meta.title == "Greeting Tool"
assert meta.description == "Greets people"
assert meta.tags == {"greeting", "demo"}
assert meta.meta == {"custom": "value"}
def test_tool_preserves_function(self):
"""@tool should preserve the original function."""
@tool
def greet(name: str) -> str:
"""Greet someone."""
return f"Hello, {name}!"
# Function should still work
assert greet("World") == "Hello, World!"
assert greet.__name__ == "greet"
assert greet.__doc__ == "Greet someone."
class TestResourceDecorator:
"""Tests for the @resource decorator."""
def test_resource_requires_uri(self):
"""@resource should require a URI argument."""
with pytest.raises(TypeError, match="requires a URI"):
@resource # type: ignore[arg-type]
def get_config() -> str:
return "{}"
def test_resource_with_uri(self):
"""@resource("uri") should store the URI."""
@resource("config://app")
def get_config() -> dict:
return {"setting": "value"}
assert has_fs_meta(get_config)
meta = get_fs_meta(get_config)
assert isinstance(meta, ResourceMeta)
assert meta.type == "resource"
assert meta.uri == "config://app"
def test_resource_with_template_uri(self):
"""@resource with template URI should work."""
@resource("users://{user_id}/profile")
def get_profile(user_id: str) -> dict:
return {"id": user_id}
meta = get_fs_meta(get_profile)
assert isinstance(meta, ResourceMeta)
assert meta.uri == "users://{user_id}/profile"
def test_resource_with_all_metadata(self):
"""@resource with all metadata should store it all."""
@resource(
"config://app",
name="app-config",
title="Application Config",
description="Gets app configuration",
mime_type="application/json",
tags={"config"},
meta={"custom": "value"},
)
def get_config() -> dict:
return {"setting": "value"}
meta = get_fs_meta(get_config)
assert isinstance(meta, ResourceMeta)
assert meta.uri == "config://app"
assert meta.name == "app-config"
assert meta.title == "Application Config"
assert meta.description == "Gets app configuration"
assert meta.mime_type == "application/json"
assert meta.tags == {"config"}
assert meta.meta == {"custom": "value"}
def test_resource_preserves_function(self):
"""@resource should preserve the original function."""
@resource("config://app")
def get_config() -> dict:
"""Get config."""
return {"setting": "value"}
# Function should still work
assert get_config() == {"setting": "value"}
assert get_config.__name__ == "get_config"
assert get_config.__doc__ == "Get config."
class TestPromptDecorator:
"""Tests for the @prompt decorator."""
def test_prompt_without_parens(self):
"""@prompt without parentheses should work."""
@prompt
def analyze(topic: str) -> list:
return [{"role": "user", "content": f"Analyze: {topic}"}]
assert has_fs_meta(analyze)
meta = get_fs_meta(analyze)
assert isinstance(meta, PromptMeta)
assert meta.type == "prompt"
assert meta.name is None
def test_prompt_with_empty_parens(self):
"""@prompt() with empty parentheses should work."""
@prompt()
def analyze(topic: str) -> list:
return [{"role": "user", "content": f"Analyze: {topic}"}]
assert has_fs_meta(analyze)
meta = get_fs_meta(analyze)
assert isinstance(meta, PromptMeta)
def test_prompt_with_name_arg(self):
"""@prompt("name") with name as first arg should work."""
@prompt("custom-analyze")
def analyze(topic: str) -> list:
return [{"role": "user", "content": f"Analyze: {topic}"}]
meta = get_fs_meta(analyze)
assert meta is not None
assert meta.name == "custom-analyze"
def test_prompt_with_name_kwarg(self):
"""@prompt(name="name") with keyword arg should work."""
@prompt(name="custom-analyze")
def analyze(topic: str) -> list:
return [{"role": "user", "content": f"Analyze: {topic}"}]
meta = get_fs_meta(analyze)
assert meta is not None
assert meta.name == "custom-analyze"
def test_prompt_with_all_metadata(self):
"""@prompt with all metadata should store it all."""
@prompt(
name="custom-analyze",
title="Analysis Prompt",
description="Analyzes topics",
tags={"analysis"},
meta={"custom": "value"},
)
def analyze(topic: str) -> list:
return [{"role": "user", "content": f"Analyze: {topic}"}]
meta = get_fs_meta(analyze)
assert meta is not None
assert meta.name == "custom-analyze"
assert meta.title == "Analysis Prompt"
assert meta.description == "Analyzes topics"
assert meta.tags == {"analysis"}
assert meta.meta == {"custom": "value"}
def test_prompt_preserves_function(self):
"""@prompt should preserve the original function."""
@prompt
def analyze(topic: str) -> list:
"""Analyze a topic."""
return [{"role": "user", "content": f"Analyze: {topic}"}]
# Function should still work
result = analyze("Python")
assert result == [{"role": "user", "content": "Analyze: Python"}]
assert analyze.__name__ == "analyze"
assert analyze.__doc__ == "Analyze a topic."
class TestHelperFunctions:
"""Tests for helper functions."""
def test_has_fs_meta_false_for_undecorated(self):
"""has_fs_meta should return False for undecorated functions."""
def plain_function():
pass
assert not has_fs_meta(plain_function)
def test_get_fs_meta_none_for_undecorated(self):
"""get_fs_meta should return None for undecorated functions."""
def plain_function():
pass
assert get_fs_meta(plain_function) is None

328
tests/fs/test_discovery.py Normal file
View file

@ -0,0 +1,328 @@
"""Tests for fastmcp.fs discovery module."""
from pathlib import Path
from fastmcp.fs.decorators import ToolMeta
from fastmcp.fs.discovery import (
discover_and_import,
discover_files,
extract_components,
import_module_from_file,
)
class TestDiscoverFiles:
"""Tests for discover_files function."""
def test_discover_files_empty_dir(self, tmp_path: Path):
"""Should return empty list for empty directory."""
files = discover_files(tmp_path)
assert files == []
def test_discover_files_nonexistent_dir(self, tmp_path: Path):
"""Should return empty list for nonexistent directory."""
nonexistent = tmp_path / "does_not_exist"
files = discover_files(nonexistent)
assert files == []
def test_discover_files_single_file(self, tmp_path: Path):
"""Should find a single Python file."""
py_file = tmp_path / "test.py"
py_file.write_text("# test")
files = discover_files(tmp_path)
assert files == [py_file]
def test_discover_files_skips_init(self, tmp_path: Path):
"""Should skip __init__.py files."""
init_file = tmp_path / "__init__.py"
init_file.write_text("# init")
py_file = tmp_path / "test.py"
py_file.write_text("# test")
files = discover_files(tmp_path)
assert files == [py_file]
def test_discover_files_recursive(self, tmp_path: Path):
"""Should find files in subdirectories."""
subdir = tmp_path / "subdir"
subdir.mkdir()
file1 = tmp_path / "a.py"
file2 = subdir / "b.py"
file1.write_text("# a")
file2.write_text("# b")
files = discover_files(tmp_path)
assert sorted(files) == sorted([file1, file2])
def test_discover_files_skips_pycache(self, tmp_path: Path):
"""Should skip __pycache__ directories."""
pycache = tmp_path / "__pycache__"
pycache.mkdir()
cache_file = pycache / "test.py"
cache_file.write_text("# cache")
py_file = tmp_path / "test.py"
py_file.write_text("# test")
files = discover_files(tmp_path)
assert files == [py_file]
def test_discover_files_sorted(self, tmp_path: Path):
"""Files should be returned in sorted order."""
(tmp_path / "z.py").write_text("# z")
(tmp_path / "a.py").write_text("# a")
(tmp_path / "m.py").write_text("# m")
files = discover_files(tmp_path)
names = [f.name for f in files]
assert names == ["a.py", "m.py", "z.py"]
class TestImportModuleFromFile:
"""Tests for import_module_from_file function."""
def test_import_simple_module(self, tmp_path: Path):
"""Should import a simple module."""
py_file = tmp_path / "simple.py"
py_file.write_text("VALUE = 42")
module = import_module_from_file(py_file)
assert module.VALUE == 42
def test_import_module_with_function(self, tmp_path: Path):
"""Should import a module with functions."""
py_file = tmp_path / "funcs.py"
py_file.write_text(
"""\
def greet(name):
return f"Hello, {name}!"
"""
)
module = import_module_from_file(py_file)
assert module.greet("World") == "Hello, World!"
def test_import_module_with_imports(self, tmp_path: Path):
"""Should handle modules with standard library imports."""
py_file = tmp_path / "with_imports.py"
py_file.write_text(
"""\
import os
import sys
def get_cwd():
return os.getcwd()
"""
)
module = import_module_from_file(py_file)
assert callable(module.get_cwd)
def test_import_as_package_with_init(self, tmp_path: Path):
"""Should import as package when __init__.py exists."""
# Create package structure (use unique name to avoid module caching)
pkg = tmp_path / "testpkg_init"
pkg.mkdir()
(pkg / "__init__.py").write_text("PKG_VAR = 'package'")
module_file = pkg / "module.py"
module_file.write_text("MODULE_VAR = 'module'")
module = import_module_from_file(module_file)
assert module.MODULE_VAR == "module"
def test_import_with_relative_import(self, tmp_path: Path):
"""Should support relative imports when in a package."""
# Create package with relative import (use unique name to avoid module caching)
pkg = tmp_path / "testpkg_relative"
pkg.mkdir()
(pkg / "__init__.py").write_text("")
(pkg / "helper.py").write_text("HELPER_VALUE = 123")
(pkg / "main.py").write_text(
"""\
from .helper import HELPER_VALUE
MAIN_VALUE = HELPER_VALUE * 2
"""
)
module = import_module_from_file(pkg / "main.py")
assert module.MAIN_VALUE == 246
def test_import_package_module_reload(self, tmp_path: Path):
"""Re-importing a package module should return updated content."""
# Create package (use unique name to avoid conflicts)
pkg = tmp_path / "testpkg_reload"
pkg.mkdir()
(pkg / "__init__.py").write_text("")
module_file = pkg / "reloadable.py"
module_file.write_text("VALUE = 'original'")
# First import
module = import_module_from_file(module_file)
assert module.VALUE == "original"
# Modify the file
module_file.write_text("VALUE = 'updated'")
# Re-import should see the updated value
module = import_module_from_file(module_file)
assert module.VALUE == "updated"
class TestExtractComponents:
"""Tests for extract_components function."""
def test_extract_no_components(self, tmp_path: Path):
"""Should return empty list for module with no decorated functions."""
py_file = tmp_path / "plain.py"
py_file.write_text(
"""\
def plain_function():
pass
SOME_VAR = 42
"""
)
module = import_module_from_file(py_file)
components = extract_components(module)
assert components == []
def test_extract_tool_component(self, tmp_path: Path):
"""Should extract @tool decorated functions."""
py_file = tmp_path / "tools.py"
py_file.write_text(
"""\
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
return f"Hello, {name}!"
"""
)
module = import_module_from_file(py_file)
components = extract_components(module)
assert len(components) == 1
func, meta = components[0]
assert func.__name__ == "greet"
assert isinstance(meta, ToolMeta)
def test_extract_multiple_components(self, tmp_path: Path):
"""Should extract multiple decorated functions."""
py_file = tmp_path / "multi.py"
py_file.write_text(
"""\
from fastmcp.fs import tool, resource, prompt
@tool
def greet(name: str) -> str:
return f"Hello, {name}!"
@resource("config://app")
def get_config() -> dict:
return {}
@prompt
def analyze(topic: str) -> list:
return []
"""
)
module = import_module_from_file(py_file)
components = extract_components(module)
assert len(components) == 3
names = {func.__name__ for func, _ in components}
assert names == {"greet", "get_config", "analyze"}
def test_extract_skips_private_functions(self, tmp_path: Path):
"""Should skip private functions even if decorated."""
py_file = tmp_path / "private.py"
py_file.write_text(
"""\
from fastmcp.fs import tool
@tool
def public_tool() -> str:
return "public"
@tool
def _private_tool() -> str:
return "private"
"""
)
module = import_module_from_file(py_file)
components = extract_components(module)
# Only public tool should be found (private starts with _)
assert len(components) == 1
func, _ = components[0]
assert func.__name__ == "public_tool"
class TestDiscoverAndImport:
"""Tests for discover_and_import function."""
def test_discover_and_import_empty(self, tmp_path: Path):
"""Should return empty result for empty directory."""
result = discover_and_import(tmp_path)
assert result.components == []
assert result.failed_files == {}
def test_discover_and_import_with_tools(self, tmp_path: Path):
"""Should discover and import tools."""
tools_dir = tmp_path / "tools"
tools_dir.mkdir()
(tools_dir / "greet.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
return f"Hello, {name}!"
"""
)
result = discover_and_import(tmp_path)
assert len(result.components) == 1
file_path, func, meta = result.components[0]
assert file_path.name == "greet.py"
assert func.__name__ == "greet"
assert isinstance(meta, ToolMeta)
def test_discover_and_import_skips_bad_imports(self, tmp_path: Path):
"""Should skip files that fail to import and track them."""
(tmp_path / "good.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def good_tool() -> str:
return "good"
"""
)
(tmp_path / "bad.py").write_text(
"""\
import nonexistent_module_xyz123
def bad_function():
pass
"""
)
result = discover_and_import(tmp_path)
# Only good.py should be imported
assert len(result.components) == 1
_, func, _ = result.components[0]
assert func.__name__ == "good_tool"
# bad.py should be in failed_files
assert len(result.failed_files) == 1
failed_path = tmp_path / "bad.py"
assert failed_path in result.failed_files
assert "nonexistent_module_xyz123" in result.failed_files[failed_path]

420
tests/fs/test_provider.py Normal file
View file

@ -0,0 +1,420 @@
"""Tests for fastmcp.fs FileSystemProvider."""
import time
from pathlib import Path
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.fs import FileSystemProvider
class TestFileSystemProvider:
"""Tests for FileSystemProvider."""
def test_provider_empty_directory(self, tmp_path: Path):
"""Provider should work with empty directory."""
provider = FileSystemProvider(tmp_path)
assert repr(provider).startswith("FileSystemProvider")
def test_provider_discovers_tools(self, tmp_path: Path):
"""Provider should discover @tool decorated functions."""
tools_dir = tmp_path / "tools"
tools_dir.mkdir()
(tools_dir / "greet.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
'''Greet someone by name.'''
return f"Hello, {name}!"
"""
)
provider = FileSystemProvider(tmp_path)
# Check tool was registered
assert len(provider._components) == 1
def test_provider_discovers_resources(self, tmp_path: Path):
"""Provider should discover @resource decorated functions."""
(tmp_path / "config.py").write_text(
"""\
from fastmcp.fs import resource
@resource("config://app")
def get_config() -> dict:
'''Get app config.'''
return {"setting": "value"}
"""
)
provider = FileSystemProvider(tmp_path)
assert len(provider._components) == 1
def test_provider_discovers_resource_templates(self, tmp_path: Path):
"""Provider should discover resource templates."""
(tmp_path / "users.py").write_text(
"""\
from fastmcp.fs import resource
@resource("users://{user_id}/profile")
def get_profile(user_id: str) -> dict:
'''Get user profile.'''
return {"id": user_id}
"""
)
provider = FileSystemProvider(tmp_path)
assert len(provider._components) == 1
def test_provider_discovers_prompts(self, tmp_path: Path):
"""Provider should discover @prompt decorated functions."""
(tmp_path / "analyze.py").write_text(
"""\
from fastmcp.fs import prompt
@prompt
def analyze(topic: str) -> list:
'''Analyze a topic.'''
return [{"role": "user", "content": f"Analyze: {topic}"}]
"""
)
provider = FileSystemProvider(tmp_path)
assert len(provider._components) == 1
def test_provider_discovers_multiple_in_one_file(self, tmp_path: Path):
"""Provider should discover multiple components in one file."""
(tmp_path / "multi.py").write_text(
"""\
from fastmcp.fs import tool, resource, prompt
@tool
def tool1() -> str:
return "tool1"
@tool
def tool2() -> str:
return "tool2"
@resource("config://app")
def get_config() -> dict:
return {}
"""
)
provider = FileSystemProvider(tmp_path)
assert len(provider._components) == 3
def test_provider_skips_undecorated_files(self, tmp_path: Path):
"""Provider should skip files with no decorated functions."""
(tmp_path / "utils.py").write_text(
"""\
def helper_function():
return "helper"
SOME_CONSTANT = 42
"""
)
(tmp_path / "tool.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def my_tool() -> str:
return "tool"
"""
)
provider = FileSystemProvider(tmp_path)
# Only the tool should be registered
assert len(provider._components) == 1
class TestFileSystemProviderReloadMode:
"""Tests for FileSystemProvider reload mode."""
def test_reload_false_caches_at_init(self, tmp_path: Path):
"""With reload=False, components are cached at init."""
(tmp_path / "tool.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def original() -> str:
return "original"
"""
)
provider = FileSystemProvider(tmp_path, reload=False)
assert len(provider._components) == 1
# Add another file - should NOT be picked up
(tmp_path / "tool2.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def added() -> str:
return "added"
"""
)
# Still only one component
assert len(provider._components) == 1
async def test_reload_true_rescans(self, tmp_path: Path):
"""With reload=True, components are rescanned on each request."""
(tmp_path / "tool.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def original() -> str:
return "original"
"""
)
provider = FileSystemProvider(tmp_path, reload=True)
# Always loaded once at init (to catch errors early)
assert provider._loaded
assert len(provider._components) == 1
# Add another file - should be picked up on next _ensure_loaded
(tmp_path / "tool2.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def added() -> str:
return "added"
"""
)
# With reload=True, _ensure_loaded re-scans
await provider._ensure_loaded()
assert len(provider._components) == 2
async def test_warning_deduplication_same_file(self, tmp_path: Path, capsys):
"""Warnings for the same broken file should not repeat."""
bad_file = tmp_path / "bad.py"
bad_file.write_text("1/0 # division by zero")
provider = FileSystemProvider(tmp_path, reload=True)
# First load - should warn
captured = capsys.readouterr()
# Check for warning indicator (rich may truncate long paths)
assert "WARNING" in captured.err and "Failed to import" in captured.err
# Second load (same file, unchanged) - should NOT warn again
await provider._ensure_loaded()
captured = capsys.readouterr()
assert "Failed to import" not in captured.err
async def test_warning_on_file_change(self, tmp_path: Path, capsys):
"""Warnings should reappear when a broken file changes."""
bad_file = tmp_path / "bad.py"
bad_file.write_text("1/0 # division by zero")
provider = FileSystemProvider(tmp_path, reload=True)
# First load - should warn
captured = capsys.readouterr()
# Check for warning indicator (rich may truncate long paths)
assert "WARNING" in captured.err and "Failed to import" in captured.err
# Modify the file (different error) - need to ensure mtime changes
time.sleep(0.01) # Ensure mtime differs
bad_file.write_text("syntax error here !!!")
# Next load - should warn again (file changed)
await provider._ensure_loaded()
captured = capsys.readouterr()
# Check for warning indicator (rich may truncate long paths)
assert "WARNING" in captured.err and "Failed to import" in captured.err
async def test_warning_cleared_when_fixed(self, tmp_path: Path, capsys):
"""Warnings should clear when a file is fixed, and reappear if broken again."""
bad_file = tmp_path / "tool.py"
bad_file.write_text("1/0 # broken")
provider = FileSystemProvider(tmp_path, reload=True)
# First load - should warn
captured = capsys.readouterr()
# Check for warning indicator (rich may truncate long paths)
assert "WARNING" in captured.err and "Failed to import" in captured.err
# Fix the file
time.sleep(0.01)
bad_file.write_text(
"""\
from fastmcp.fs import tool
@tool
def my_tool() -> str:
return "fixed"
"""
)
# Load again - should NOT warn, file is fixed
await provider._ensure_loaded()
captured = capsys.readouterr()
assert "Failed to import" not in captured.err
assert len(provider._components) == 1
# Break it again
time.sleep(0.01)
bad_file.write_text("1/0 # broken again")
# Should warn again
await provider._ensure_loaded()
captured = capsys.readouterr()
# Check for warning indicator (rich may truncate long paths)
assert "WARNING" in captured.err and "Failed to import" in captured.err
class TestFileSystemProviderIntegration:
"""Integration tests with FastMCP server."""
async def test_provider_with_fastmcp_server(self, tmp_path: Path):
"""FileSystemProvider should work with FastMCP server."""
(tmp_path / "greet.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
'''Greet someone.'''
return f"Hello, {name}!"
"""
)
provider = FileSystemProvider(tmp_path)
mcp = FastMCP("TestServer", providers=[provider])
async with Client(mcp) as client:
# List tools
tools = await client.list_tools()
assert len(tools) == 1
assert tools[0].name == "greet"
# Call tool
result = await client.call_tool("greet", {"name": "World"})
assert "Hello, World!" in str(result)
async def test_provider_with_resources(self, tmp_path: Path):
"""FileSystemProvider should work with resources."""
(tmp_path / "config.py").write_text(
"""\
from fastmcp.fs import resource
@resource("config://app")
def get_config() -> str:
'''Get app config.'''
return '{"version": "1.0"}'
"""
)
provider = FileSystemProvider(tmp_path)
mcp = FastMCP("TestServer", providers=[provider])
async with Client(mcp) as client:
# List resources
resources = await client.list_resources()
assert len(resources) == 1
assert str(resources[0].uri) == "config://app"
# Read resource
result = await client.read_resource("config://app")
assert "1.0" in str(result)
async def test_provider_with_resource_templates(self, tmp_path: Path):
"""FileSystemProvider should work with resource templates."""
(tmp_path / "users.py").write_text(
"""\
from fastmcp.fs import resource
@resource("users://{user_id}/profile")
def get_profile(user_id: str) -> str:
'''Get user profile.'''
return f'{{"id": "{user_id}", "name": "User {user_id}"}}'
"""
)
provider = FileSystemProvider(tmp_path)
mcp = FastMCP("TestServer", providers=[provider])
async with Client(mcp) as client:
# List templates
templates = await client.list_resource_templates()
assert len(templates) == 1
# Read with parameter
result = await client.read_resource("users://123/profile")
assert "123" in str(result)
async def test_provider_with_prompts(self, tmp_path: Path):
"""FileSystemProvider should work with prompts."""
(tmp_path / "analyze.py").write_text(
"""\
from fastmcp.fs import prompt
@prompt
def analyze(topic: str) -> str:
'''Analyze a topic.'''
return f"Please analyze: {topic}"
"""
)
provider = FileSystemProvider(tmp_path)
mcp = FastMCP("TestServer", providers=[provider])
async with Client(mcp) as client:
# List prompts
prompts = await client.list_prompts()
assert len(prompts) == 1
assert prompts[0].name == "analyze"
# Get prompt
result = await client.get_prompt("analyze", {"topic": "Python"})
assert "Python" in str(result)
async def test_nested_directory_structure(self, tmp_path: Path):
"""FileSystemProvider should work with nested directories."""
# Create nested structure
tools = tmp_path / "tools"
tools.mkdir()
(tools / "greet.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def greet(name: str) -> str:
return f"Hello, {name}!"
"""
)
payments = tools / "payments"
payments.mkdir()
(payments / "charge.py").write_text(
"""\
from fastmcp.fs import tool
@tool
def charge(amount: float) -> str:
return f"Charged ${amount}"
"""
)
provider = FileSystemProvider(tmp_path)
mcp = FastMCP("TestServer", providers=[provider])
async with Client(mcp) as client:
tools_list = await client.list_tools()
assert len(tools_list) == 2
names = {t.name for t in tools_list}
assert names == {"greet", "charge"}