ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-21 13:04:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Consolidate server loading logic into FileSystemSource (#1614)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-08-24 20:50:25 -04:00 • committed by GitHub
commit 9c5c4f5113
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 1117 additions and 520 deletions
Download patch file Download diff file Expand all files Collapse all files

41
docs/deployment/server-configuration.mdx
View file

@ -304,23 +304,48 @@ fastmcp run prod.fastmcp.json
# Skip environment setup when already in a uv environment
fastmcp run fastmcp.json --skip-env
# Skip source preparation when source is already prepared
fastmcp run fastmcp.json --skip-source
# Skip both environment and source preparation
fastmcp run fastmcp.json --skip-env --skip-source
```
### Environment Setup Control
### Using an Existing Environment
By default, FastMCP uses `uv` to create an isolated environment based on your configuration. The `--skip-env` flag allows you to skip this automatic environment setup:
By default, FastMCP creates an isolated environment with `uv` based on your configuration. When you already have a suitable Python environment, use the `--skip-env` flag to skip environment creation:
```bash
fastmcp run fastmcp.json --skip-env
```
**When to use `--skip-env`:**
- You're already in an activated virtual environment with all dependencies installed
- You're inside a Docker container with pre-installed dependencies
- You're in a uv-managed environment and want to prevent infinite recursion
- You want to test the server without environment setup for debugging purposes
**When you already have an environment:**
- You're in an activated virtual environment with all dependencies installed
- You're inside a Docker container with pre-installed dependencies
- You're in a CI/CD pipeline that pre-builds the environment
- You're using a system-wide installation with all required packages
- You're in a uv-managed environment (prevents infinite recursion)
This flag is particularly useful in CI/CD pipelines, Docker containers, or when you're managing the Python environment yourself.
This flag tells FastMCP: "I already have everything installed, just run the server."
### Using an Existing Source
When working with source types that require preparation (future support for git repositories or cloud sources), use the `--skip-source` flag when you already have the source code available:
```bash
fastmcp run fastmcp.json --skip-source
```
**When you already have the source:**
- You've previously cloned a git repository and don't need to re-fetch
- You have a cached copy of a cloud-hosted server
- You're in a CI/CD pipeline where source checkout is a separate step
- You're iterating locally on already-downloaded code
This flag tells FastMCP: "I already have the source code, skip any download/clone steps."
Note: For filesystem sources (local Python files), this flag has no effect since they don't require preparation.
The configuration file works with all FastMCP commands:
- **`run`** - Start the server in production mode

223
src/fastmcp/cli/cli.py
View file

@ -7,6 +7,7 @@ import os
import platform
import subprocess
import sys
from contextlib import contextmanager
from pathlib import Path
from typing import Annotated, Literal
@ -59,6 +60,29 @@ def _parse_env_var(env_var: str) -> tuple[str, str]:
return key.strip(), value.strip()
@contextmanager
def with_argv(args: list[str] | None):
"""Temporarily replace sys.argv if args provided.
This context manager is used at the CLI boundary to inject
server arguments when needed, without mutating sys.argv deep
in the source loading logic.
Args are provided without the script name, so we preserve sys.argv[0]
and replace the rest.
"""
if args is not None:
original = sys.argv[:]
try:
# Preserve the script name (sys.argv[0]) and replace the rest
sys.argv = [sys.argv[0]] + args
yield
finally:
sys.argv = original
else:
yield
@app.command
def version(
*,
@ -164,12 +188,16 @@ async def dev(
Args:
server_spec: Python file to run, optionally with :object suffix, or None to auto-detect fastmcp.json
"""
from pathlib import Path
from fastmcp.utilities.fastmcp_config import FastMCPConfig
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
config = None
config_path = None
# Auto-detect fastmcp.json if no server_spec provided
if server_spec is None:
from pathlib import Path
from fastmcp.utilities.fastmcp_config import FastMCPConfig
config_path = Path("fastmcp.json")
if not config_path.exists():
# Check if fastmcp.json exists in current directory
@ -182,16 +210,13 @@ async def dev(
"Please specify a server file or create a fastmcp.json configuration."
)
sys.exit(1)
server_spec = str(config_path)
logger.info(f"Using configuration from {config_path}")
# Load the config to get settings
config = FastMCPConfig.from_file(config_path)
entrypoint = config.get_entrypoint(config_path)
# Convert entrypoint to string format for dev command
if entrypoint.object:
server_spec = f"{entrypoint.file}:{entrypoint.object}"
else:
server_spec = entrypoint.file
# Create FastMCPConfig from server_spec
if server_spec.endswith(".json"):
# Load existing config
config = FastMCPConfig.from_file(Path(server_spec))
# Merge environment settings with CLI args (CLI takes precedence)
if config.environment:
@ -220,15 +245,15 @@ async def dev(
# Get server port from deployment config if not specified
if config.deployment and config.deployment.port:
server_port = server_port or config.deployment.port
logger.info(f"Using configuration from {config_path}")
file, server_object = run_module.parse_file_path(server_spec)
else:
# Create config from file path
source = FileSystemSource(path=server_spec)
config = FastMCPConfig(source=source)
logger.debug(
"Starting dev server",
extra={
"file": str(file),
"server_object": server_object,
"server_spec": server_spec,
"with_editable": str(with_editable) if with_editable else None,
"with_packages": with_packages,
"ui_port": ui_port,
@ -237,9 +262,8 @@ async def dev(
)
try:
# Import server to get dependencies
# TODO: Remove dependencies handling (deprecated in v2.11.4)
server: FastMCP = await run_module.import_server(file, server_object)
# Load server to check for deprecated dependencies
server: FastMCP = await config.source.load_server()
if server.dependencies:
import warnings
@ -299,7 +323,7 @@ async def dev(
logger.error(
"Dev server failed",
extra={
"file": str(file),
"file": str(server_spec),
"error": str(e),
"returncode": e.returncode,
},
@ -310,7 +334,7 @@ async def dev(
"npx not found. Please ensure Node.js and npm are properly installed "
"and added to your system PATH. You may need to restart your terminal "
"after installation.",
extra={"file": str(file)},
extra={"file": str(server_spec)},
)
sys.exit(1)
@ -399,6 +423,14 @@ async def run(
negative="",
),
] = False,
skip_source: Annotated[
bool,
cyclopts.Parameter(
"--skip-source",
help="Skip source preparation step (use when source is already prepared)",
negative="",
),
] = False,
) -> None:
"""Run an MCP server or connect to a remote one.
@ -566,6 +598,7 @@ async def run(
log_level=log_level,
server_args=list(server_args),
show_banner=not no_banner,
skip_source=skip_source,
)
except Exception as e:
logger.error(
@ -636,10 +669,10 @@ async def inspect(
Args:
server_spec: Python file to inspect, optionally with :object suffix, or fastmcp.json
"""
# Load configuration if needed
from pathlib import Path
from fastmcp.utilities.fastmcp_config import FastMCPConfig
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
config = None
config_path = None
@ -662,64 +695,53 @@ async def inspect(
server_spec = str(config_path)
logger.info(f"Using configuration from {config_path}")
# Load config if server_spec is a .json file
# Create FastMCPConfig from server_spec
if server_spec.endswith(".json"):
config_path = Path(server_spec)
if config_path.exists():
# Try to load as JSON and discriminate between FastMCPConfig and MCPConfig
try:
with open(config_path) as f:
data = json.load(f)
# Check which type of config it is based on required fields
try:
if "source" in data:
# It's a FastMCPConfig - validate and use it
adapter = get_cached_typeadapter(FastMCPConfig)
config = adapter.validate_python(data)
# Get the actual entrypoint with resolved paths
entrypoint = config.get_entrypoint(config_path)
# Check if it's an MCPConfig (has mcpServers key)
if "mcpServers" in data:
# MCPConfig - we don't process these in inspect
logger.error("MCPConfig files are not supported by inspect command")
sys.exit(1)
else:
# It's a FastMCPConfig
config = FastMCPConfig.from_file(config_path)
if entrypoint.object:
server_spec = f"{entrypoint.file}:{entrypoint.object}"
else:
server_spec = entrypoint.file
# Merge environment settings from config with CLI (CLI takes precedence)
if config.environment:
python = python or config.environment.python
project = project or (
Path(config.environment.project)
if config.environment.project
else None
)
with_requirements = with_requirements or (
Path(config.environment.requirements)
if config.environment.requirements
else None
)
# Merge environment settings from config with CLI (CLI takes precedence)
if config.environment:
python = python or config.environment.python
project = project or (
Path(config.environment.project)
if config.environment.project
else None
)
with_requirements = with_requirements or (
Path(config.environment.requirements)
if config.environment.requirements
else None
)
# Merge packages from both sources
if config.environment.dependencies:
packages = list(config.environment.dependencies)
if with_packages:
packages.extend(with_packages)
with_packages = packages
elif "mcpServers" in data:
# It's an MCPConfig, we don't process these in the run command
# They should be handled through different code paths
config = None
else:
# Not a recognized config format, treat as regular server spec
config = None
except ValidationError:
# Not a valid config, treat as regular server spec
config = None
except (json.JSONDecodeError, FileNotFoundError):
# Not a valid JSON file, treat as regular server spec
config = None
# Merge packages from both sources
if config.environment.dependencies:
packages = list(config.environment.dependencies)
if with_packages:
packages.extend(with_packages)
with_packages = packages
except (json.JSONDecodeError, ValidationError) as e:
logger.error(f"Invalid configuration file: {e}")
sys.exit(1)
else:
config = None
logger.error(f"Configuration file not found: {config_path}")
sys.exit(1)
else:
# Create config from file path
source = FileSystemSource(path=server_spec)
config = FastMCPConfig(source=source)
# Check if we need to use uv run
needs_uv = python or with_packages or with_requirements or project
@ -728,54 +750,37 @@ async def inspect(
if needs_uv:
# Build and run uv command
if config and config.environment:
# Use environment config's run_with_uv method
inspect_command = [
"fastmcp",
"inspect",
server_spec,
"--output",
str(output),
]
config.environment.run_with_uv(inspect_command)
else:
# Build an Environment from CLI args for consistency
from fastmcp.utilities.fastmcp_config import (
Environment,
)
from fastmcp.utilities.fastmcp_config import Environment
env_config = Environment(
python=python,
dependencies=with_packages,
requirements=str(with_requirements) if with_requirements else None,
project=str(project) if project else None,
)
# Create or update environment config
env_config = Environment(
python=python,
dependencies=with_packages if with_packages else None,
requirements=str(with_requirements) if with_requirements else None,
project=str(project) if project else None,
)
inspect_command = [
"fastmcp",
"inspect",
server_spec,
"--output",
str(output),
]
env_config.run_with_uv(inspect_command)
# Direct import path (no uv needed)
# Parse the server specification
file, server_object = run_module.parse_file_path(server_spec)
inspect_command = [
"fastmcp",
"inspect",
server_spec,
"--output",
str(output),
]
env_config.run_with_uv(inspect_command)
return # run_with_uv exits the process
logger.debug(
"Inspecting server",
extra={
"file": str(file),
"server_object": server_object,
"server_spec": server_spec,
"output": str(output),
},
)
try:
# Import the server
server = await run_module.import_server(file, server_object)
# Load the server using the config
server = await config.source.load_server()
# Get server information - using native async support
info = await inspect_fastmcp(server)

61
src/fastmcp/cli/install/shared.py
View file

@ -8,9 +8,9 @@ from dotenv import dotenv_values
from pydantic import ValidationError
from rich import print
from fastmcp.cli.run import import_server, parse_file_path
from fastmcp.utilities.fastmcp_config import FastMCPConfig
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import get_cached_typeadapter
logger = get_logger(__name__)
@ -37,49 +37,46 @@ async def process_common_args(
Handles both fastmcp.json config files and traditional file.py:object syntax.
"""
# Check if server_spec is a .json file
# Create FastMCPConfig from server_spec
config = None
if server_spec.endswith(".json"):
config_path = Path(server_spec).resolve()
if not config_path.exists():
print(f"[red]Configuration file not found: {config_path}[/red]")
sys.exit(1)
# Try to load as JSON and discriminate between FastMCPConfig and MCPConfig
try:
with open(config_path) as f:
data = json.load(f)
# Check if it's an MCPConfig first (has canonical mcpServers key)
from fastmcp.utilities.fastmcp_config import FastMCPConfig
# Check if it's an MCPConfig (has mcpServers key)
if "mcpServers" in data:
# It's an MCPConfig, treat as regular server spec
file, server_object = parse_file_path(server_spec)
# MCPConfig files aren't supported for install
print("[red]MCPConfig files are not supported for installation[/red]")
sys.exit(1)
else:
# Try to parse as FastMCPConfig
try:
adapter = get_cached_typeadapter(FastMCPConfig)
config = adapter.validate_python(data)
entrypoint = config.get_entrypoint(config_path)
# It's a FastMCPConfig
config = FastMCPConfig.from_file(config_path)
# Convert to file and server_object
file = Path(entrypoint.file)
server_object = entrypoint.object
# Merge packages from config if not overridden
if config.environment and config.environment.dependencies:
# Merge with CLI packages (CLI takes precedence)
config_packages = config.environment.dependencies or []
with_packages = list(set(with_packages + config_packages))
except ValidationError:
# Not a valid FastMCPConfig, treat as regular server spec
file, server_object = parse_file_path(server_spec)
except (json.JSONDecodeError, FileNotFoundError):
# Not a valid JSON file, treat as regular server spec
file, server_object = parse_file_path(server_spec)
# Merge packages from config if not overridden
if config.environment and config.environment.dependencies:
# Merge with CLI packages (CLI takes precedence)
config_packages = list(config.environment.dependencies) or []
with_packages = list(set(with_packages + config_packages))
except (json.JSONDecodeError, ValidationError) as e:
print(f"[red]Invalid configuration file: {e}[/red]")
sys.exit(1)
else:
# Parse traditional server spec
file, server_object = parse_file_path(server_spec)
# Create config from file path
source = FileSystemSource(path=server_spec)
config = FastMCPConfig(source=source)
# Extract file and server_object from the source
# The FileSystemSource handles parsing path:object syntax
file = Path(config.source.path).resolve()
server_object = (
config.source.entrypoint if hasattr(config.source, "entrypoint") else None
)
logger.debug(
"Installing server",
@ -96,7 +93,7 @@ async def process_common_args(
server = None
if not name:
try:
server = await import_server(file, server_object)
server = await config.source.load_server()
name = server.name
except (ImportError, ModuleNotFoundError) as e:
logger.debug(

235
src/fastmcp/cli/run.py
View file

@ -1,12 +1,9 @@
"""FastMCP run command implementation with enhanced type hints."""
import importlib.util
import inspect
import json
import re
import subprocess
import sys
from functools import partial
from pathlib import Path
from typing import Any, Literal
@ -17,8 +14,8 @@ from fastmcp.server.server import FastMCP
from fastmcp.utilities.fastmcp_config import (
Environment,
FastMCPConfig,
FileSystemSource,
)
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import get_cached_typeadapter
@ -35,150 +32,6 @@ def is_url(path: str) -> bool:
return bool(url_pattern.match(path))
def parse_file_path(server_spec: str) -> tuple[Path, str | None]:
"""Parse a file path that may include a server object specification.
Args:
server_spec: Path to file, optionally with :object suffix
Returns:
Tuple of (file_path, server_object)
"""
# First check if we have a Windows path (e.g., C:\...)
has_windows_drive = len(server_spec) > 1 and server_spec[1] == ":"
# Split on the last colon, but only if it's not part of the Windows drive letter
# and there's actually another colon in the string after the drive letter
if ":" in (server_spec[2:] if has_windows_drive else server_spec):
file_str, server_object = server_spec.rsplit(":", 1)
else:
file_str, server_object = server_spec, None
# Resolve the file path
file_path = Path(file_str).expanduser().resolve()
if not file_path.exists():
logger.error(f"File not found: {file_path}")
sys.exit(1)
if not file_path.is_file():
logger.error(f"Not a file: {file_path}")
sys.exit(1)
return file_path, server_object
async def import_server(file: Path, server_or_factory: str | None = None) -> Any:
"""Import a MCP server from a file.
Args:
file: Path to the file
server_or_factory: Optional object name in format "module:object" or just "object"
Returns:
The server object (or result of calling a factory function)
"""
# Add parent directory to Python path so imports can be resolved
file_dir = str(file.parent)
if file_dir not in sys.path:
sys.path.insert(0, file_dir)
# Import the module
spec = importlib.util.spec_from_file_location("server_module", file)
if not spec or not spec.loader:
logger.error("Could not load module", extra={"file": str(file)})
sys.exit(1)
module = importlib.util.module_from_spec(spec) # type: ignore[arg-type]
spec.loader.exec_module(module) # type: ignore[union-attr]
# If no object specified, try common server names
if not server_or_factory:
# Look for common server instance names
for name in ["mcp", "server", "app"]:
if hasattr(module, name):
obj = getattr(module, name)
if isinstance(obj, FastMCP | FastMCP1x):
return await _resolve_server_or_factory(obj, file, name)
logger.error(
f"No server object found in {file}. Please either:\n"
"1. Use a standard variable name (mcp, server, or app)\n"
"2. Specify the entrypoint name in fastmcp.json or use `file.py:object` syntax as your path.",
extra={"file": str(file)},
)
sys.exit(1)
# Handle module:object syntax
if server_or_factory and ":" in server_or_factory:
module_name, object_name = server_or_factory.split(":", 1)
try:
server_module = importlib.import_module(module_name)
obj = getattr(server_module, object_name, None)
except ImportError:
logger.error(
f"Could not import module '{module_name}'",
extra={"file": str(file)},
)
sys.exit(1)
else:
# Just object name
obj = getattr(module, server_or_factory, None)
if obj is None:
logger.error(
f"Server object '{server_or_factory}' not found",
extra={"file": str(file)},
)
sys.exit(1)
return await _resolve_server_or_factory(obj, file, server_or_factory) # type: ignore[arg-type]
async def _resolve_server_or_factory(obj: Any, file: Path, name: str) -> Any:
"""Resolve a server object or factory function to a server instance.
Args:
obj: The object that might be a server or factory function
file: Path to the file for error messages
name: Name of the object for error messages
Returns:
A server instance
"""
# Check if it's a function or coroutine function
if inspect.isfunction(obj) or inspect.iscoroutinefunction(obj):
logger.debug(f"Found factory function '{name}' in {file}")
try:
if inspect.iscoroutinefunction(obj):
# Async factory function
server = await obj()
else:
# Sync factory function
server = obj()
# Validate the result is a FastMCP server
if not isinstance(server, FastMCP | FastMCP1x):
logger.error(
f"Factory function '{name}' must return a FastMCP server instance, "
f"got {type(server).__name__}",
extra={"file": str(file)},
)
sys.exit(1)
logger.debug(f"Factory function '{name}' created server: {server.name}")
return server
except Exception as e:
logger.error(
f"Failed to call factory function '{name}': {e}",
extra={"file": str(file)},
)
sys.exit(1)
# Not a function, return as-is (should be a server instance)
return obj
def run_with_uv(
server_spec: str,
python_version: str | None = None,
@ -359,32 +212,6 @@ def load_fastmcp_config(config_path: Path) -> FastMCPConfig:
return config
async def import_server_with_args(
file: Path,
server_or_factory: str | None = None,
server_args: list[str] | None = None,
) -> Any:
"""Import a server with optional command line arguments.
Args:
file: Path to the server file
server_or_factory: Optional server object or factory function name
server_args: Optional command line arguments to inject
Returns:
The imported server object
"""
if server_args:
original_argv = sys.argv[:]
try:
sys.argv = [str(file)] + server_args
return await import_server(file, server_or_factory)
finally:
sys.argv = original_argv
else:
return await import_server(file, server_or_factory)
async def run_command(
server_spec: str,
transport: TransportType | None = None,
@ -395,6 +222,7 @@ async def run_command(
server_args: list[str] | None = None,
show_banner: bool = True,
use_direct_import: bool = False,
skip_source: bool = False,
) -> None:
"""Run a MCP server or connect to a remote one.
@ -408,11 +236,14 @@ async def run_command(
server_args: Additional arguments to pass to the server
show_banner: Whether to show the server banner
use_direct_import: Whether to use direct import instead of subprocess
skip_source: Whether to skip source preparation step
"""
# Special case: URLs
if is_url(server_spec):
# Handle URL case
server = create_client_server(server_spec)
logger.debug(f"Created client proxy server for {server_spec}")
# Special case: MCPConfig files (legacy)
elif server_spec.endswith(".json"):
# Load JSON and check which type of config it is
config_path = Path(server_spec)
@ -424,10 +255,7 @@ async def run_command(
# It's an MCP config
server = create_mcp_config_server(config_path)
else:
# Try to parse as FastMCPConfig
adapter = get_cached_typeadapter(FastMCPConfig)
adapter.validate_python(data) # Validate but don't need to store
# It's a FastMCP config - load it properly with runtime settings
# It's a FastMCP config - load it properly
config = load_fastmcp_config(config_path)
# Merge deployment config with CLI arguments (CLI takes precedence)
@ -441,27 +269,42 @@ async def run_command(
server_args if server_args is not None else config.deployment.args
)
# Prepare the source if needed (e.g., clone git repo, download from cloud)
if not skip_source:
await config.source.prepare()
# Load the server using the source
server = await config.source.load_server(config_path, server_args)
from contextlib import nullcontext
from fastmcp.cli.cli import with_argv
# Use sys.argv context manager if deployment args specified
argv_context = with_argv(server_args) if server_args else nullcontext()
with argv_context:
server = await config.source.load_server()
logger.debug(f'Found server "{server.name}" from config {config_path}')
else:
# Handle file case - parse into FileSystemSource immediately
if ":" in server_spec:
# Check if it's a Windows path (e.g., C:\...)
has_windows_drive = len(server_spec) > 1 and server_spec[1] == ":"
# Only split if colon is not part of Windows drive
if ":" in (server_spec[2:] if has_windows_drive else server_spec):
file_str, obj = server_spec.rsplit(":", 1)
source = FileSystemSource(path=file_str, object=obj)
else:
source = FileSystemSource(path=server_spec)
else:
source = FileSystemSource(path=server_spec)
# Create a temporary config with just the source
# Regular file case - create a FastMCPConfig with FileSystemSource
source = FileSystemSource(path=server_spec)
config = FastMCPConfig(source=source)
server = await config.source.load_server(None, server_args)
# Prepare the source if needed
if not skip_source:
await config.source.prepare()
# Load the server
from contextlib import nullcontext
from fastmcp.cli.cli import with_argv
# Use sys.argv context manager if server_args specified
argv_context = with_argv(server_args) if server_args else nullcontext()
with argv_context:
server = await config.source.load_server()
logger.debug(f'Found server "{server.name}" in {source.path}')
# Run the server
@ -499,6 +342,8 @@ def run_v1_server(
port: int | None = None,
transport: TransportType | None = None,
) -> None:
from functools import partial
if host:
server.settings.host = host
if port:

4
src/fastmcp/utilities/fastmcp_config/__init__.py
View file

@ -5,13 +5,13 @@ The current version is v1, which is re-exported here for convenience.
"""
from fastmcp.utilities.fastmcp_config.v1.fastmcp_config import (
BaseSource,
Deployment,
Environment,
FastMCPConfig,
FileSystemSource,
generate_schema,
)
from fastmcp.utilities.fastmcp_config.v1.sources.base import BaseSource
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
__all__ = [
"BaseSource",

0
src/fastmcp/utilities/fastmcp_config/v1/__init__.py Normal file
View file

136
src/fastmcp/utilities/fastmcp_config/v1/fastmcp_config.py
View file

@ -10,12 +10,12 @@ from __future__ import annotations
import json
import os
import re
from abc import ABC, abstractmethod
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, overload
from pydantic import BaseModel, Field, field_validator
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
from fastmcp.utilities.logging import get_logger
logger = get_logger("cli.config")
@ -24,56 +24,6 @@ logger = get_logger("cli.config")
FASTMCP_JSON_SCHEMA = "https://gofastmcp.com/public/schemas/fastmcp.json/v1.json"
class BaseSource(BaseModel, ABC):
"""Abstract base class for all source types."""
type: str = Field(description="Source type identifier")
async def prepare(self, config_path: Path | None = None) -> Path | None:
"""Prepare the source (download, clone, install, etc).
Returns:
Path to prepared source directory, or None if no preparation needed.
This path may contain a nested fastmcp.json for configuration chaining.
"""
# Default implementation for sources that don't need preparation
return None
@abstractmethod
async def load_server(
self, config_path: Path | None = None, server_args: list[str] | None = None
) -> Any:
"""Load and return the FastMCP server instance.
Must be called after prepare() if the source requires preparation.
"""
...
class FileSystemSource(BaseSource):
"""Source for local Python files."""
type: Literal["filesystem"] = Field(default="filesystem", description="Source type")
path: str = Field(description="Path to Python file containing the server")
entrypoint: str | None = Field(
default=None,
description="Name of server instance or factory function (a no-arg function that returns a FastMCP server)",
)
async def load_server(
self, config_path: Path | None = None, server_args: list[str] | None = None
) -> Any:
"""Load server from filesystem."""
from fastmcp.cli.run import import_server_with_args
# Resolve relative paths if config_path provided
file_path = Path(self.path)
if not file_path.is_absolute() and config_path:
file_path = (config_path.parent / file_path).resolve()
return await import_server_with_args(file_path, self.entrypoint, server_args)
# Type alias for source union (will expand with GitSource, etc in future)
SourceType = FileSystemSource
@ -130,8 +80,12 @@ class Environment(BaseModel):
if self.project:
args.extend(["--project", str(self.project)])
# Add fastmcp as a base dependency
args.extend(["--with", "fastmcp"])
# Add fastmcp dependency - use editable install if in development mode
dev_path = self._find_fastmcp_dev_path()
if dev_path:
args.extend(["--with-editable", str(dev_path)])
else:
args.extend(["--with", "fastmcp"])
# Add additional dependencies (skip fastmcp if already added)
if self.dependencies:
@ -156,6 +110,34 @@ class Environment(BaseModel):
return args
def _find_fastmcp_dev_path(self) -> Path | None:
"""Find the fastmcp development directory by looking for pyproject.toml.
Searches from the current working directory up the directory tree
looking for a pyproject.toml file that contains name = "fastmcp".
Returns:
Path to the fastmcp project directory if found, None otherwise
"""
current_path = Path.cwd()
# Search up the directory tree
for path in [current_path] + list(current_path.parents):
pyproject_path = path / "pyproject.toml"
if pyproject_path.exists():
try:
# Read and check if this is the fastmcp project
content = pyproject_path.read_text(encoding="utf-8")
if 'name = "fastmcp"' in content or "name='fastmcp'" in content:
logger.debug(f"Found fastmcp development project at: {path}")
return path
except (OSError, UnicodeDecodeError):
# Skip files that can't be read
continue
logger.debug("No fastmcp development project found, using PyPI package")
return None
def run_with_uv(self, command: list[str]) -> None:
"""Execute a command using uv run with this environment configuration.
@ -517,47 +499,6 @@ class FastMCPConfig(BaseModel):
return None
async def load_server(
self, config_path: Path | None = None, server_args: list[str] | None = None
) -> Any:
"""Load the server from the configuration.
This handles environment setup, working directory changes,
and delegates to the source's load_server method.
Args:
config_path: Path to the config file (for resolving relative paths)
server_args: Optional arguments to pass to the server
Returns:
The imported server object
"""
import os
from pathlib import Path
# Set environment variables if specified
if self.deployment and self.deployment.env:
for key, value in self.deployment.env.items():
os.environ[key] = value
# Change working directory if specified
if self.deployment and self.deployment.cwd:
cwd_path = Path(self.deployment.cwd)
if not cwd_path.is_absolute():
# If config_path provided, resolve relative to it
if config_path:
cwd_path = (config_path.parent / cwd_path).resolve()
else:
cwd_path = cwd_path.resolve()
os.chdir(cwd_path)
# Use server_args from deployment if not provided
if server_args is None and self.deployment:
server_args = self.deployment.args
# Delegate to the source's load_server method
return await self.source.load_server(config_path, server_args)
async def run_server(self, **kwargs: Any) -> None:
"""Load and run the server with this configuration.
@ -565,7 +506,12 @@ class FastMCPConfig(BaseModel):
**kwargs: Additional arguments to pass to server.run_async()
These override config settings
"""
server = await self.load_server()
# Apply deployment settings (env vars, cwd)
if self.deployment:
self.deployment.apply_runtime_settings()
# Load the server
server = await self.source.load_server()
# Build run arguments from config
run_args = {}

0
src/fastmcp/utilities/fastmcp_config/v1/sources/__init__.py Normal file
View file

30
src/fastmcp/utilities/fastmcp_config/v1/sources/base.py Normal file
View file

@ -0,0 +1,30 @@
from abc import ABC, abstractmethod
from typing import Any
from pydantic import BaseModel, Field
class BaseSource(BaseModel, ABC):
"""Abstract base class for all source types."""
type: str = Field(description="Source type identifier")
async def prepare(self) -> None:
"""Prepare the source (download, clone, install, etc).
For sources that need preparation (e.g., git clone, download),
this method performs that preparation. For sources that don't
need preparation (e.g., local files), this is a no-op.
"""
# Default implementation for sources that don't need preparation
pass
@abstractmethod
async def load_server(self) -> Any:
"""Load and return the FastMCP server instance.
Must be called after prepare() if the source requires preparation.
All information needed to load the server should be available
as attributes on the source instance.
"""
...

215
src/fastmcp/utilities/fastmcp_config/v1/sources/filesystem.py Normal file
View file

@ -0,0 +1,215 @@
import importlib.util
import inspect
import sys
from pathlib import Path
from typing import Any, Literal
from pydantic import Field, field_validator
from fastmcp.utilities.fastmcp_config.v1.sources.base import BaseSource
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class FileSystemSource(BaseSource):
"""Source for local Python files."""
type: Literal["filesystem"] = Field(default="filesystem", description="Source type")
path: str = Field(description="Path to Python file containing the server")
entrypoint: str | None = Field(
default=None,
description="Name of server instance or factory function (a no-arg function that returns a FastMCP server)",
)
@field_validator("path", mode="before")
@classmethod
def parse_path_with_object(cls, v: str) -> str:
"""Parse path:object syntax and extract the object name.
This validator runs before the model is created, allowing us to
handle the "file.py:object" syntax at the model boundary.
"""
if isinstance(v, str) and ":" in v:
# Check if it's a Windows path (e.g., C:\...)
has_windows_drive = len(v) > 1 and v[1] == ":"
# Only split if colon is not part of Windows drive
if ":" in (v[2:] if has_windows_drive else v):
# This path has an object specification
# We'll handle it in __init__ by setting entrypoint
return v
return v
def __init__(self, **data: Any) -> None:
"""Initialize FileSystemSource, handling path:object syntax."""
# Check if path contains an object specification
if "path" in data and isinstance(data["path"], str) and ":" in data["path"]:
path_str = data["path"]
# Check if it's a Windows path (e.g., C:\...)
has_windows_drive = len(path_str) > 1 and path_str[1] == ":"
# Only split if colon is not part of Windows drive
if ":" in (path_str[2:] if has_windows_drive else path_str):
file_str, obj = path_str.rsplit(":", 1)
data["path"] = file_str
# Only set entrypoint if not already provided
if "entrypoint" not in data or data["entrypoint"] is None:
data["entrypoint"] = obj
super().__init__(**data)
async def load_server(self) -> Any:
"""Load server from filesystem."""
# Resolve the file path
file_path = Path(self.path).expanduser().resolve()
if not file_path.exists():
logger.error(f"File not found: {file_path}")
sys.exit(1)
if not file_path.is_file():
logger.error(f"Not a file: {file_path}")
sys.exit(1)
# Import the module
module = self._import_module(file_path)
# Find the server object
server = await self._find_server_object(module, file_path)
return server
def _import_module(self, file_path: Path) -> Any:
"""Import a Python module from a file path.
Args:
file_path: Path to the Python file
Returns:
The imported module
"""
# Add parent directory to Python path so imports can be resolved
file_dir = str(file_path.parent)
if file_dir not in sys.path:
sys.path.insert(0, file_dir)
# Import the module
spec = importlib.util.spec_from_file_location("server_module", file_path)
if not spec or not spec.loader:
logger.error("Could not load module", extra={"file": str(file_path)})
sys.exit(1)
module = importlib.util.module_from_spec(spec) # type: ignore[arg-type]
sys.modules["server_module"] = module # Register in sys.modules
spec.loader.exec_module(module) # type: ignore[union-attr]
return module
async def _find_server_object(self, module: Any, file_path: Path) -> Any:
"""Find the server object in the module.
Args:
module: The imported Python module
file_path: Path to the file (for error messages)
Returns:
The server object (or result of calling a factory function)
"""
# Avoid circular import by importing here
from mcp.server.fastmcp import FastMCP as FastMCP1x
from fastmcp.server.server import FastMCP
# If entrypoint is specified, use it
if self.entrypoint:
# Handle module:object syntax (though this is legacy)
if ":" in self.entrypoint:
module_name, object_name = self.entrypoint.split(":", 1)
try:
import importlib
server_module = importlib.import_module(module_name)
obj = getattr(server_module, object_name, None)
except ImportError:
logger.error(
f"Could not import module '{module_name}'",
extra={"file": str(file_path)},
)
sys.exit(1)
else:
# Just object name
obj = getattr(module, self.entrypoint, None)
if obj is None:
logger.error(
f"Server object '{self.entrypoint}' not found",
extra={"file": str(file_path)},
)
sys.exit(1)
return await self._resolve_factory(obj, file_path, self.entrypoint)
# No entrypoint specified, try common server names
for name in ["mcp", "server", "app"]:
if hasattr(module, name):
obj = getattr(module, name)
if isinstance(obj, FastMCP | FastMCP1x):
return await self._resolve_factory(obj, file_path, name)
# No server found
logger.error(
f"No server object found in {file_path}. Please either:\n"
"1. Use a standard variable name (mcp, server, or app)\n"
"2. Specify the entrypoint name in fastmcp.json or use `file.py:object` syntax as your path.",
extra={"file": str(file_path)},
)
sys.exit(1)
async def _resolve_factory(self, obj: Any, file_path: Path, name: str) -> Any:
"""Resolve a server object or factory function to a server instance.
Args:
obj: The object that might be a server or factory function
file_path: Path to the file for error messages
name: Name of the object for error messages
Returns:
A server instance
"""
# Avoid circular import by importing here
from mcp.server.fastmcp import FastMCP as FastMCP1x
from fastmcp.server.server import FastMCP
# Check if it's a function or coroutine function
if inspect.isfunction(obj) or inspect.iscoroutinefunction(obj):
logger.debug(f"Found factory function '{name}' in {file_path}")
try:
if inspect.iscoroutinefunction(obj):
# Async factory function
server = await obj()
else:
# Sync factory function
server = obj()
# Validate the result is a FastMCP server
if not isinstance(server, FastMCP | FastMCP1x):
logger.error(
f"Factory function '{name}' must return a FastMCP server instance, "
f"got {type(server).__name__}",
extra={"file": str(file_path)},
)
sys.exit(1)
logger.debug(f"Factory function '{name}' created server: {server.name}")
return server
except Exception as e:
logger.error(
f"Failed to call factory function '{name}': {e}",
extra={"file": str(file_path)},
)
sys.exit(1)
# Not a function, return as-is (should be a server instance)
return obj

17
test.fastmcp.json
View file

@ -1,17 +0,0 @@
{
"source": {
"path": "test.py",
"object": "mcp"
},
"environment": {
"python": "3.10",
"dependencies": ["fastmcp", "httpx", "pandas", "httpx23"]
},
"deployment": {
"transport": "http",
"host": "127.0.0.1",
"port": 1234,
"path": "/mcp",
"log_level": "INFO"
}
}

58
tests/cli/test_cli.py
View file

@ -328,6 +328,47 @@ class TestRunCommand:
]
)
def test_run_command_parsing_skip_env_flag(self):
"""Test run command parsing with --skip-env flag."""
command, bound, _ = app.parse_args(
[
"run",
"server.py",
"--skip-env",
]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["skip_env"] is True
def test_run_command_parsing_skip_source_flag(self):
"""Test run command parsing with --skip-source flag."""
command, bound, _ = app.parse_args(
[
"run",
"server.py",
"--skip-source",
]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["skip_source"] is True
def test_run_command_parsing_both_skip_flags(self):
"""Test run command parsing with both --skip-env and --skip-source flags."""
command, bound, _ = app.parse_args(
[
"run",
"server.py",
"--skip-env",
"--skip-source",
]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["skip_env"] is True
assert bound.arguments["skip_source"] is True
class TestWindowsSpecific:
"""Test Windows-specific functionality."""
@ -413,22 +454,27 @@ class TestWindowsSpecific:
def test_windows_path_parsing_with_colon(self, tmp_path):
"""Test parsing Windows paths with drive letters and colons."""
from fastmcp.cli.run import parse_file_path
from pathlib import Path
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import (
FileSystemSource,
)
# Create a real test file to test the logic
test_file = tmp_path / "server.py"
test_file.write_text("# test server")
# Test normal file parsing (works on all platforms)
file_path, obj = parse_file_path(str(test_file))
assert obj is None
source = FileSystemSource(path=str(test_file))
assert source.entrypoint is None
assert Path(source.path).resolve() == test_file.resolve()
# Test file:object parsing
file_path, obj = parse_file_path(f"{test_file}:myapp")
assert obj == "myapp"
source = FileSystemSource(path=f"{test_file}:myapp")
assert source.entrypoint == "myapp"
# Test that the file portion resolves correctly when object is specified
assert file_path == test_file.resolve()
assert Path(source.path).resolve() == test_file.resolve()
class TestInspectCommand:

2
tests/cli/test_config.py
View file

@ -11,8 +11,8 @@ from fastmcp.utilities.fastmcp_config import (
Deployment,
Environment,
FastMCPConfig,
FileSystemSource,
)
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
class TestFileSystemSource:

6
tests/cli/test_cursor.py
View file

@ -280,7 +280,11 @@ class TestInstallCursor:
# Verify failure message was printed
mock_print.assert_called()
def test_install_cursor_deduplicate_packages(self):
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None, # Mock to disable dev mode so "fastmcp" count is predictable
)
def test_install_cursor_deduplicate_packages(self, mock_find_dev):
"""Test that duplicate packages are deduplicated."""
with patch("fastmcp.cli.install.cursor.open_deeplink") as mock_open:
mock_open.return_value = True

197
tests/cli/test_run.py
View file

@ -7,14 +7,13 @@ from pydantic import ValidationError
from fastmcp.cli.run import (
create_mcp_config_server,
import_server,
is_url,
parse_file_path,
)
from fastmcp.client.client import Client
from fastmcp.client.transports import FastMCPTransport
from fastmcp.mcp_config import MCPConfig, StdioMCPServer
from fastmcp.server.server import FastMCP
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
class TestUrlDetection:
@ -41,52 +40,54 @@ class TestUrlDetection:
assert not is_url("file:///path/to/file")
class TestFilePathParsing:
"""Test file path parsing functionality."""
class TestFileSystemSource:
"""Test FileSystemSource path parsing functionality."""
def test_parse_file_path_simple(self, tmp_path):
def test_parse_simple_path(self, tmp_path):
"""Test parsing simple file path without object."""
test_file = tmp_path / "server.py"
test_file.write_text("# test server")
file_path, server_object = parse_file_path(str(test_file))
assert file_path == test_file.resolve()
assert server_object is None
source = FileSystemSource(path=str(test_file))
assert Path(source.path).resolve() == test_file.resolve()
assert source.entrypoint is None
def test_parse_file_path_with_object(self, tmp_path):
def test_parse_path_with_object(self, tmp_path):
"""Test parsing file path with object specification."""
test_file = tmp_path / "server.py"
test_file.write_text("# test server")
file_path, server_object = parse_file_path(f"{test_file}:app")
assert file_path == test_file.resolve()
assert server_object == "app"
source = FileSystemSource(path=f"{test_file}:app")
assert Path(source.path).resolve() == test_file.resolve()
assert source.entrypoint == "app"
def test_parse_file_path_complex_object(self, tmp_path):
def test_parse_complex_object(self, tmp_path):
"""Test parsing file path with complex object specification."""
test_file = tmp_path / "server.py"
test_file.write_text("# test server")
# The current implementation splits on the last colon, so file:module:app
# becomes file_path="file:module" and server_object="app"
# The implementation splits on the last colon, so file:module:app
# becomes file_path="file:module" and entrypoint="app"
# We need to create a file with a colon in the name for this test
complex_file = tmp_path / "server:module.py"
complex_file.write_text("# test server")
file_path, server_object = parse_file_path(f"{complex_file}:app")
assert file_path == complex_file.resolve()
assert server_object == "app"
source = FileSystemSource(path=f"{complex_file}:app")
assert Path(source.path).resolve() == complex_file.resolve()
assert source.entrypoint == "app"
def test_parse_file_path_nonexistent(self):
"""Test parsing nonexistent file path exits."""
async def test_load_server_nonexistent(self):
"""Test loading nonexistent file path exits."""
source = FileSystemSource(path="nonexistent.py")
with pytest.raises(SystemExit) as exc_info:
parse_file_path("nonexistent.py")
await source.load_server()
assert exc_info.value.code == 1
def test_parse_file_path_directory(self, tmp_path):
"""Test parsing directory path exits."""
async def test_load_server_directory(self, tmp_path):
"""Test loading directory path exits."""
source = FileSystemSource(path=str(tmp_path))
with pytest.raises(SystemExit) as exc_info:
parse_file_path(str(tmp_path))
await source.load_server()
assert exc_info.value.code == 1
@ -157,7 +158,8 @@ def greet(name: str) -> str:
return f"Hello, {name}!"
""")
server = await import_server(test_file)
source = FileSystemSource(path=str(test_file))
server = await source.load_server()
assert server.name == "TestServer"
tools = await server.get_tools()
assert "greet" in tools
@ -178,7 +180,8 @@ if __name__ == "__main__":
app.run()
""")
server = await import_server(test_file)
source = FileSystemSource(path=str(test_file))
server = await source.load_server()
assert server.name == "MainServer"
tools = await server.get_tools()
assert "calculate" in tools
@ -192,7 +195,8 @@ import fastmcp
mcp = fastmcp.FastMCP("MCPServer")
""")
server = await import_server(mcp_file)
source = FileSystemSource(path=str(mcp_file))
server = await source.load_server()
assert server.name == "MCPServer"
# Test with 'server' name
@ -202,7 +206,8 @@ import fastmcp
server = fastmcp.FastMCP("ServerServer")
""")
server = await import_server(server_file)
source = FileSystemSource(path=str(server_file))
server = await source.load_server()
assert server.name == "ServerServer"
# Test with 'app' name
@ -212,7 +217,8 @@ import fastmcp
app = fastmcp.FastMCP("AppServer")
""")
server = await import_server(app_file)
source = FileSystemSource(path=str(app_file))
server = await source.load_server()
assert server.name == "AppServer"
async def test_import_server_nonstandard_name(self, tmp_path):
@ -228,7 +234,8 @@ def custom_tool() -> str:
return "custom"
""")
server = await import_server(test_file, "my_custom_server")
source = FileSystemSource(path=f"{test_file}:my_custom_server")
server = await source.load_server()
assert server.name == "CustomServer"
tools = await server.get_tools()
assert "custom_tool" in tools
@ -242,8 +249,9 @@ import fastmcp
other_name = fastmcp.FastMCP("OtherServer")
""")
source = FileSystemSource(path=str(test_file))
with pytest.raises(SystemExit) as exc_info:
await import_server(test_file)
await source.load_server()
assert exc_info.value.code == 1
async def test_import_server_nonexistent_object_fails(self, tmp_path):
@ -255,6 +263,131 @@ import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")
source = FileSystemSource(path=f"{test_file}:nonexistent")
with pytest.raises(SystemExit) as exc_info:
await import_server(test_file, "nonexistent")
await source.load_server()
assert exc_info.value.code == 1
class TestSkipSource:
"""Test the --skip-source functionality."""
async def test_run_command_calls_prepare_by_default(self, tmp_path):
"""Test that run_command calls source.prepare() by default."""
from unittest.mock import AsyncMock, patch
from fastmcp.cli.run import run_command
# Create a test server file
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")
# Create a test config file
config_file = tmp_path / "fastmcp.json"
config_data = {"source": {"path": str(test_file), "entrypoint": "mcp"}}
config_file.write_text(json.dumps(config_data))
# Mock the prepare method and server run
with (
patch.object(
FileSystemSource, "prepare", new_callable=AsyncMock
) as prepare_mock,
patch("fastmcp.server.server.FastMCP.run_async", new_callable=AsyncMock),
):
# Run the command
await run_command(str(config_file))
# Verify prepare was called
prepare_mock.assert_called_once()
async def test_run_command_skips_prepare_with_flag(self, tmp_path):
"""Test that run_command skips source.prepare() when skip_source=True."""
from unittest.mock import AsyncMock, patch
from fastmcp.cli.run import run_command
# Create a test server file
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")
# Create a test config file
config_file = tmp_path / "fastmcp.json"
config_data = {"source": {"path": str(test_file), "entrypoint": "mcp"}}
config_file.write_text(json.dumps(config_data))
# Mock the prepare method and server run
with (
patch.object(
FileSystemSource, "prepare", new_callable=AsyncMock
) as prepare_mock,
patch("fastmcp.server.server.FastMCP.run_async", new_callable=AsyncMock),
):
# Run the command with skip_source=True
await run_command(str(config_file), skip_source=True)
# Verify prepare was NOT called
prepare_mock.assert_not_called()
async def test_filesystem_source_prepare_by_default(self, tmp_path):
"""Test that FileSystemSource is prepared when using direct file spec."""
from unittest.mock import AsyncMock, patch
from fastmcp.cli.run import run_command
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import (
FileSystemSource,
)
# Create a test server file
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")
# Mock the prepare method and server run
with (
patch.object(
FileSystemSource, "prepare", new_callable=AsyncMock
) as prepare_mock,
patch("fastmcp.server.server.FastMCP.run_async", new_callable=AsyncMock),
):
# Run with direct file specification
await run_command(str(test_file))
# Verify prepare was called
prepare_mock.assert_called_once()
async def test_filesystem_source_skip_prepare_with_flag(self, tmp_path):
"""Test that FileSystemSource.prepare() is skipped with skip_source flag."""
from unittest.mock import AsyncMock, patch
from fastmcp.cli.run import run_command
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import (
FileSystemSource,
)
# Create a test server file
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")
# Mock the prepare method and server run
with (
patch.object(
FileSystemSource, "prepare", new_callable=AsyncMock
) as prepare_mock,
patch("fastmcp.server.server.FastMCP.run_async", new_callable=AsyncMock),
):
# Run with direct file specification and skip_source=True
await run_command(str(test_file), skip_source=True)
# Verify prepare was NOT called
prepare_mock.assert_not_called()

2
tests/cli/test_run_config.py
View file

@ -11,8 +11,8 @@ from fastmcp.utilities.fastmcp_config import (
Deployment,
Environment,
FastMCPConfig,
FileSystemSource,
)
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
@pytest.fixture

54
tests/cli/test_run_with_uv.py
View file

@ -12,8 +12,12 @@ from fastmcp.cli.run import run_with_uv
class TestRunWithUv:
"""Test the run_with_uv function."""
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("subprocess.run")
def test_run_with_uv_basic(self, mock_run):
def test_run_with_uv_basic(self, mock_run, mock_find_dev_path):
"""Test basic run_with_uv execution."""
mock_run.return_value = Mock(returncode=0)
@ -38,8 +42,12 @@ class TestRunWithUv:
]
assert cmd == expected
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("subprocess.run")
def test_run_with_uv_python_version(self, mock_run):
def test_run_with_uv_python_version(self, mock_run, mock_find_dev_path):
"""Test run_with_uv with Python version."""
mock_run.return_value = Mock(returncode=0)
@ -63,8 +71,12 @@ class TestRunWithUv:
]
assert cmd == expected
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("subprocess.run")
def test_run_with_uv_project(self, mock_run):
def test_run_with_uv_project(self, mock_run, mock_find_dev_path):
"""Test run_with_uv with project directory."""
mock_run.return_value = Mock(returncode=0)
# Use an absolute path that works on all platforms
@ -90,8 +102,12 @@ class TestRunWithUv:
"--skip-env",
]
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("subprocess.run")
def test_run_with_uv_with_packages(self, mock_run):
def test_run_with_uv_with_packages(self, mock_run, mock_find_dev_path):
"""Test run_with_uv with additional packages."""
mock_run.return_value = Mock(returncode=0)
@ -117,8 +133,12 @@ class TestRunWithUv:
]
assert cmd == expected
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("subprocess.run")
def test_run_with_uv_with_requirements(self, mock_run):
def test_run_with_uv_with_requirements(self, mock_run, mock_find_dev_path):
"""Test run_with_uv with requirements file."""
mock_run.return_value = Mock(returncode=0)
req_path = Path("requirements.txt")
@ -143,8 +163,12 @@ class TestRunWithUv:
]
assert cmd == expected
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("subprocess.run")
def test_run_with_uv_transport_options(self, mock_run):
def test_run_with_uv_transport_options(self, mock_run, mock_find_dev_path):
"""Test run_with_uv with transport-related options."""
mock_run.return_value = Mock(returncode=0)
@ -185,8 +209,12 @@ class TestRunWithUv:
]
assert cmd == expected
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("subprocess.run")
def test_run_with_uv_all_options(self, mock_run):
def test_run_with_uv_all_options(self, mock_run, mock_find_dev_path):
"""Test run_with_uv with all options combined."""
mock_run.return_value = Mock(returncode=0)
@ -230,8 +258,12 @@ class TestRunWithUv:
"--no-banner",
]
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("subprocess.run")
def test_run_with_uv_error_handling(self, mock_run):
def test_run_with_uv_error_handling(self, mock_run, mock_find_dev_path):
"""Test run_with_uv error handling."""
mock_run.side_effect = subprocess.CalledProcessError(1, ["uv", "run"])
@ -240,9 +272,13 @@ class TestRunWithUv:
assert exc_info.value.code == 1
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment._find_fastmcp_dev_path",
return_value=None,
)
@patch("fastmcp.cli.run.logger")
@patch("subprocess.run")
def test_run_with_uv_logging(self, mock_run, mock_logger):
def test_run_with_uv_logging(self, mock_run, mock_logger, mock_find_dev_path):
"""Test that run_with_uv logs the command."""
mock_run.return_value = Mock(returncode=0)

139
tests/cli/test_server_args.py Normal file
View file

@ -0,0 +1,139 @@
"""Test server argument passing functionality."""
from pathlib import Path
import pytest
from fastmcp.utilities.fastmcp_config import FastMCPConfig
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
class TestServerArguments:
"""Test passing arguments to servers."""
@pytest.mark.asyncio
async def test_server_with_argparse(self, tmp_path):
"""Test a server that uses argparse with command line arguments."""
server_file = tmp_path / "argparse_server.py"
server_file.write_text("""
import argparse
from fastmcp import FastMCP
parser = argparse.ArgumentParser()
parser.add_argument("--name", default="DefaultServer")
parser.add_argument("--port", type=int, default=8000)
parser.add_argument("--debug", action="store_true")
args = parser.parse_args()
server_name = f"{args.name}:{args.port}"
if args.debug:
server_name += " (Debug)"
mcp = FastMCP(server_name)
@mcp.tool
def get_config() -> dict:
return {"name": args.name, "port": args.port, "debug": args.debug}
""")
# Test with arguments
source = FileSystemSource(path=str(server_file))
config = FastMCPConfig(source=source)
from fastmcp.cli.cli import with_argv
# Simulate passing arguments
with with_argv(["--name", "TestServer", "--port", "9000", "--debug"]):
server = await config.source.load_server()
assert server.name == "TestServer:9000 (Debug)"
# Test the tool works and can access the parsed args
tools = await server.get_tools()
assert "get_config" in tools
@pytest.mark.asyncio
async def test_server_with_no_args(self, tmp_path):
"""Test a server that uses argparse with no arguments (defaults)."""
server_file = tmp_path / "default_server.py"
server_file.write_text("""
import argparse
from fastmcp import FastMCP
parser = argparse.ArgumentParser()
parser.add_argument("--name", default="DefaultName")
args = parser.parse_args()
mcp = FastMCP(args.name)
""")
source = FileSystemSource(path=str(server_file))
config = FastMCPConfig(source=source)
from fastmcp.cli.cli import with_argv
# Test with empty args list (should use defaults)
with with_argv([]):
server = await config.source.load_server()
assert server.name == "DefaultName"
@pytest.mark.asyncio
async def test_server_with_sys_argv_access(self, tmp_path):
"""Test a server that directly accesses sys.argv."""
server_file = tmp_path / "sysargv_server.py"
server_file.write_text("""
import sys
from fastmcp import FastMCP
# Direct sys.argv access (less common but should work)
name = "DirectServer"
if len(sys.argv) > 1 and sys.argv[1] == "--custom":
name = "CustomServer"
mcp = FastMCP(name)
""")
source = FileSystemSource(path=str(server_file))
config = FastMCPConfig(source=source)
from fastmcp.cli.cli import with_argv
# Test with custom argument
with with_argv(["--custom"]):
server = await config.source.load_server()
assert server.name == "CustomServer"
# Test without argument
with with_argv([]):
server = await config.source.load_server()
assert server.name == "DirectServer"
@pytest.mark.asyncio
async def test_config_server_example(self):
"""Test the actual config_server.py example."""
# Find the examples directory
examples_dir = Path(__file__).parent.parent.parent / "examples"
config_server = examples_dir / "config_server.py"
if not config_server.exists():
pytest.skip("config_server.py example not found")
source = FileSystemSource(path=str(config_server))
config = FastMCPConfig(source=source)
from fastmcp.cli.cli import with_argv
# Test with debug flag
with with_argv(["--name", "TestExample", "--debug"]):
server = await config.source.load_server()
assert server.name == "TestExample (Debug)"
# Verify tools are available
tools = await server.get_tools()
assert "get_status" in tools
assert "echo_message" in tools

91
tests/cli/test_with_argv.py Normal file
View file

@ -0,0 +1,91 @@
"""Test the with_argv context manager."""
import sys
from unittest.mock import patch
import pytest
from fastmcp.cli.cli import with_argv
class TestWithArgv:
"""Test the with_argv context manager."""
def test_with_argv_replaces_args(self):
"""Test that with_argv properly replaces sys.argv."""
original_argv = sys.argv[:]
test_args = ["--name", "TestServer", "--debug"]
with with_argv(test_args):
# Should preserve script name and add new args
assert sys.argv[0] == original_argv[0]
assert sys.argv[1:] == test_args
# Should restore original argv after context
assert sys.argv == original_argv
def test_with_argv_none_does_nothing(self):
"""Test that with_argv(None) doesn't change sys.argv."""
original_argv = sys.argv[:]
with with_argv(None):
assert sys.argv == original_argv
assert sys.argv == original_argv
def test_with_argv_empty_list(self):
"""Test that with_argv([]) clears arguments but keeps script name."""
original_argv = sys.argv[:]
with with_argv([]):
# Should have only the script name (no additional args)
assert sys.argv == [original_argv[0]]
assert len(sys.argv) == 1
assert sys.argv == original_argv
def test_with_argv_restores_on_exception(self):
"""Test that sys.argv is restored even if an exception occurs."""
original_argv = sys.argv[:]
test_args = ["--error"]
with pytest.raises(ValueError):
with with_argv(test_args):
assert sys.argv == [original_argv[0]] + test_args
raise ValueError("Test error")
# Should still restore original argv
assert sys.argv == original_argv
def test_with_argv_nested(self):
"""Test nested with_argv contexts."""
original_argv = sys.argv[:]
args1 = ["--level1"]
args2 = ["--level2", "--debug"]
with with_argv(args1):
assert sys.argv == [original_argv[0]] + args1
with with_argv(args2):
assert sys.argv == [original_argv[0]] + args2
# Should restore to level 1
assert sys.argv == [original_argv[0]] + args1
# Should restore to original
assert sys.argv == original_argv
@patch("sys.argv", ["test_script.py", "existing", "args"])
def test_with_argv_with_existing_args(self):
"""Test with_argv when sys.argv already has arguments."""
original_argv = sys.argv[:]
assert original_argv == ["test_script.py", "existing", "args"]
test_args = ["--new", "args"]
with with_argv(test_args):
# Should replace existing args but keep script name
assert sys.argv == ["test_script.py", "--new", "args"]
# Should restore original
assert sys.argv == original_argv

120
tests/utilities/test_cli.py
View file

@ -1,17 +1,22 @@
from pathlib import Path
from unittest.mock import patch
from fastmcp.utilities.fastmcp_config.v1.fastmcp_config import Environment
class TestEnvironmentBuildUVArgs:
"""Test the Environment.build_uv_args() method."""
def test_build_uv_args_basic(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_basic(self, mock_dev_path):
"""Test building basic uv args."""
env = Environment()
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = ["run", "--with", "fastmcp", "fastmcp", "run", "server.py"]
assert args == expected
def test_build_uv_args_with_editable(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_editable(self, mock_dev_path):
"""Test building uv args with editable package."""
editable_path = "/path/to/package"
env = Environment(editable=editable_path)
@ -28,7 +33,8 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
def test_build_uv_args_with_packages(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_packages(self, mock_dev_path):
"""Test building uv args with additional packages."""
env = Environment(dependencies=["pkg1", "pkg2"])
args = env.build_uv_args(["fastmcp", "run", "server.py"])
@ -46,7 +52,8 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
def test_build_uv_args_with_python_version(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_python_version(self, mock_dev_path):
"""Test building uv args with Python version."""
env = Environment(python="3.11")
args = env.build_uv_args(["fastmcp", "run", "server.py"])
@ -62,7 +69,8 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
def test_build_uv_args_with_project(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_project(self, mock_dev_path):
"""Test building uv args with project directory."""
project_path = "/path/to/project"
env = Environment(project=project_path)
@ -79,7 +87,8 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
def test_build_uv_args_with_requirements(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_requirements(self, mock_dev_path):
"""Test building uv args with requirements file."""
req_path = "requirements.txt"
env = Environment(requirements=req_path)
@ -96,7 +105,8 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
def test_build_uv_args_with_all_options(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_all_options(self, mock_dev_path):
"""Test building uv args with all options."""
project_path = "/my/project"
editable_path = "/local/pkg"
@ -131,14 +141,16 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
def test_build_uv_args_no_command(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_no_command(self, mock_dev_path):
"""Test building uv args with no command."""
env = Environment(python="3.11")
args = env.build_uv_args()
expected = ["run", "--python", "3.11", "--with", "fastmcp"]
assert args == expected
def test_build_uv_args_string_command(self):
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_string_command(self, mock_dev_path):
"""Test building uv args with string command."""
env = Environment()
args = env.build_uv_args("python")
@ -166,3 +178,93 @@ class TestEnvironmentBuildUVArgs:
"""Test that needs_uv returns False when no environment settings are present."""
env = Environment()
assert env.needs_uv() is False
@patch.object(Environment, "_find_fastmcp_dev_path")
def test_build_uv_args_development_mode(self, mock_dev_path):
"""Test building uv args in development mode (when fastmcp project is found)."""
# Mock finding the development path
dev_path = Path("/path/to/fastmcp/dev")
mock_dev_path.return_value = dev_path
env = Environment()
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--with-editable",
str(dev_path),
"fastmcp",
"run",
"server.py",
]
assert args == expected
@patch.object(Environment, "_find_fastmcp_dev_path")
def test_build_uv_args_production_mode(self, mock_dev_path):
"""Test building uv args in production mode (when no fastmcp project is found)."""
# Mock not finding the development path
mock_dev_path.return_value = None
env = Environment()
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = ["run", "--with", "fastmcp", "fastmcp", "run", "server.py"]
assert args == expected
@patch("pathlib.Path.cwd")
@patch("pathlib.Path.exists")
@patch("pathlib.Path.read_text")
def test_find_fastmcp_dev_path_found(self, mock_read_text, mock_exists, mock_cwd):
"""Test finding fastmcp development path when pyproject.toml exists."""
# Set up mock current directory
mock_cwd_path = Path("/path/to/fastmcp")
mock_cwd.return_value = mock_cwd_path
# Mock pyproject.toml exists and contains fastmcp name
mock_exists.return_value = True
mock_read_text.return_value = """[project]
name = "fastmcp"
version = "2.0.0"
"""
env = Environment()
result = env._find_fastmcp_dev_path()
assert result == mock_cwd_path
@patch("pathlib.Path.cwd")
@patch("pathlib.Path.exists")
def test_find_fastmcp_dev_path_not_found(self, mock_exists, mock_cwd):
"""Test not finding fastmcp development path when no pyproject.toml exists."""
# Set up mock current directory
mock_cwd_path = Path("/some/other/directory")
mock_cwd.return_value = mock_cwd_path
# Mock pyproject.toml doesn't exist
mock_exists.return_value = False
env = Environment()
result = env._find_fastmcp_dev_path()
assert result is None
@patch("pathlib.Path.cwd")
@patch("pathlib.Path.exists")
@patch("pathlib.Path.read_text")
def test_find_fastmcp_dev_path_wrong_project(
self, mock_read_text, mock_exists, mock_cwd
):
"""Test not finding fastmcp when pyproject.toml exists but is for different project."""
# Set up mock current directory
mock_cwd_path = Path("/path/to/other/project")
mock_cwd.return_value = mock_cwd_path
# Mock pyproject.toml exists but is for different project
mock_exists.return_value = True
mock_read_text.return_value = """[project]
name = "other-project"
version = "1.0.0"
"""
env = Environment()
result = env._find_fastmcp_dev_path()
assert result is None