Merge branch 'main' into claude/issue-1625-20250826-0138

This commit is contained in:
William Easton 2025-08-25 22:40:29 -05:00 committed by GitHub
commit 3fc21a2811
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 1046 additions and 360 deletions

View file

@ -165,10 +165,14 @@ These settings leverage standard `uv` arguments for environment creation. When a
```
</ParamField>
<ParamField body="editable" type="string">
Path to a package to install in editable/development mode. Useful for local development when you want changes to be reflected immediately.
<ParamField body="editable" type="list[string]">
List of paths to packages to install in editable/development mode. Useful for local development when you want changes to be reflected immediately. Supports multiple packages for monorepo setups or shared libraries.
```json
"editable": "."
"editable": ["."]
```
Or with multiple packages:
```json
"editable": [".", "../shared-lib", "/path/to/another-package"]
```
</ParamField>
</Expandable>
@ -312,6 +316,20 @@ fastmcp run fastmcp.json --skip-source
fastmcp run fastmcp.json --skip-env --skip-source
```
### Pre-building Environments
You can use `fastmcp project prepare` to create a persistent uv project with all dependencies pre-installed:
```bash
# Create a persistent environment
fastmcp project prepare fastmcp.json --output-dir ./env
# Use the pre-built environment to run the server
fastmcp run fastmcp.json --project ./env
```
This pattern separates environment setup (slow) from server execution (fast), useful for deployment scenarios.
### Using an Existing Environment
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:

View file

@ -22,6 +22,7 @@ fastmcp --help
| `dev` | Run a server with the MCP Inspector for testing | **Supports:** Local files and fastmcp.json configs. **Deps:** Always runs via `uv run` subprocess (never uses your local environment); dependencies must be specified or available in a uv-managed project. With fastmcp.json: Uses configured dependencies |
| `install` | Install a server in MCP client applications | **Supports:** Local files and fastmcp.json configs. **Deps:** Creates an isolated environment; dependencies must be explicitly specified with `--with` and/or `--with-editable`. With fastmcp.json: Uses configured dependencies |
| `inspect` | Generate a JSON report about a FastMCP server | **Supports:** Local files and fastmcp.json configs. **Deps:** Uses your current environment; you are responsible for ensuring all dependencies are available |
| `project prepare` | Create a persistent uv project from fastmcp.json environment config | **Supports:** fastmcp.json configs only. **Deps:** Creates a uv project directory with all dependencies pre-installed for reuse with `--project` flag |
| `version` | Display version information | N/A |
## `fastmcp run`
@ -473,6 +474,37 @@ fastmcp inspect server.py:my_server
fastmcp inspect server.py --output analysis.json
```
## `fastmcp project prepare`
Create a persistent uv project directory from a fastmcp.json file's environment configuration. This allows you to pre-install all dependencies once and reuse them with the `--project` flag.
```bash
fastmcp project prepare fastmcp.json --output-dir ./env
```
### Options
| Option | Flag | Description |
| ------ | ---- | ----------- |
| Output Directory | `--output-dir` | **Required.** Directory where the persistent uv project will be created |
### Usage Pattern
```bash
# Step 1: Prepare the environment (installs dependencies)
fastmcp project prepare fastmcp.json --output-dir ./my-env
# Step 2: Run using the prepared environment (fast, no dependency installation)
fastmcp run fastmcp.json --project ./my-env
```
The prepare command creates a uv project with:
- A `pyproject.toml` containing all dependencies from the fastmcp.json
- A `.venv` with all packages pre-installed
- A `uv.lock` file for reproducible environments
This is useful when you want to separate environment setup from server execution, such as in deployment scenarios where dependencies are installed once and the server is run multiple times.
## `fastmcp version`
Display version information about FastMCP and related components.

View file

@ -101,7 +101,7 @@ def update_claude_config(
# Build uv run command using Environment.build_uv_args()
env_config = Environment(
dependencies=deduplicated_packages,
editable=str(with_editable) if with_editable else None,
editable=[str(with_editable)] if with_editable else None,
)
args = env_config.build_uv_args()

View file

@ -229,9 +229,11 @@ async def dev(
if config.environment.requirements
else None
)
# Note: config.environment.editable is a list, but CLI only supports single path
# Take the first editable path if available
with_editable = with_editable or (
Path(config.environment.editable)
if config.environment.editable
Path(config.environment.editable[0])
if config.environment.editable and config.environment.editable[0]
else None
)
@ -303,7 +305,7 @@ async def dev(
dependencies=with_packages if with_packages else None,
requirements=str(with_requirements) if with_requirements else None,
project=str(project) if project else None,
editable=str(with_editable) if with_editable else None,
editable=[str(with_editable)] if with_editable else None,
)
uv_cmd = ["uv"] + env_config.build_uv_args(["fastmcp", "run", server_spec])
@ -415,14 +417,6 @@ async def run(
help="Requirements file to install dependencies from",
),
] = None,
skip_env: Annotated[
bool,
cyclopts.Parameter(
"--skip-env",
help="Skip environment setup with uv (use when already in a uv environment)",
negative="",
),
] = False,
skip_source: Annotated[
bool,
cyclopts.Parameter(
@ -431,6 +425,14 @@ async def run(
negative="",
),
] = False,
skip_env: Annotated[
bool,
cyclopts.Parameter(
"--skip-env",
help="Skip environment configuration (for internal use when already in a uv environment)",
negative="",
),
] = False,
) -> None:
"""Run an MCP server or connect to a remote one.
@ -509,7 +511,8 @@ async def run(
)
# Merge environment config with CLI values (CLI takes precedence)
if config.environment:
# BUT: Skip this if --skip-env is set
if config.environment and not skip_env:
python = python or config.environment.python
project = project or (
Path(config.environment.project)
@ -552,12 +555,10 @@ async def run(
)
# Check if we need to use uv run (either from CLI args or config)
# Skip if --skip-env flag is set (we're already in a uv environment)
needs_uv = not skip_env and (
python or with_packages or with_requirements or project or editable
)
if not skip_env and not needs_uv and config and config.environment:
# Check if config's environment needs uv
# When --skip-env is set, we ignore config.environment entirely
needs_uv = python or with_packages or with_requirements or project or editable
if not needs_uv and config and config.environment and not skip_env:
# Check if config's environment needs uv (but only if not skipping env)
needs_uv = config.environment.needs_uv()
if needs_uv:
@ -818,6 +819,101 @@ async def inspect(
sys.exit(1)
# Create project subcommand group
project_app = cyclopts.App(name="project", help="Manage FastMCP projects")
@project_app.command
async def prepare(
config_path: Annotated[
str | None,
cyclopts.Parameter(help="Path to fastmcp.json configuration file"),
] = None,
output_dir: Annotated[
str | None,
cyclopts.Parameter(help="Directory to create the persistent environment in"),
] = None,
skip_source: Annotated[
bool,
cyclopts.Parameter(help="Skip source preparation (e.g., git clone)"),
] = False,
) -> None:
"""Prepare a FastMCP project by creating a persistent uv environment.
This command creates a persistent uv project with all dependencies installed:
- Creates a pyproject.toml with dependencies from the config
- Installs all Python packages into a .venv
- Prepares the source (git clone, download, etc.) unless --skip-source
After running this command, you can use:
fastmcp run <config> --project <output-dir>
This is useful for:
- CI/CD pipelines with separate build and run stages
- Docker images where you prepare during build
- Production deployments where you want fast startup times
Example:
fastmcp project prepare myserver.json --output-dir ./prepared-env
fastmcp run myserver.json --project ./prepared-env
"""
from pathlib import Path
from fastmcp.utilities.fastmcp_config import FastMCPConfig
# Require output-dir
if output_dir is None:
logger.error(
"The --output-dir parameter is required.\n"
"Please specify where to create the persistent environment."
)
sys.exit(1)
# Auto-detect fastmcp.json if not provided
if config_path is None:
found_config = FastMCPConfig.find_config()
if found_config:
config_path = str(found_config)
logger.info(f"Using configuration from {config_path}")
else:
logger.error(
"No configuration file specified and no fastmcp.json found.\n"
"Please specify a configuration file or create a fastmcp.json."
)
sys.exit(1)
config_file = Path(config_path)
if not config_file.exists():
logger.error(f"Configuration file not found: {config_path}")
sys.exit(1)
output_path = Path(output_dir)
try:
# Load the configuration
config = FastMCPConfig.from_file(config_file)
# Prepare environment and source
await config.prepare(
skip_source=skip_source,
output_dir=output_path,
)
console.print(
f"[bold green]✓[/bold green] Project prepared successfully in {output_path}!\n"
f"You can now run the server with:\n"
f" [cyan]fastmcp run {config_path} --project {output_dir}[/cyan]"
)
except Exception as e:
logger.error(f"Failed to prepare project: {e}")
console.print(f"[bold red]✗[/bold red] Failed to prepare project: {e}")
sys.exit(1)
# Add project subcommand group
app.command(project_app)
# Add install subcommands using proper Cyclopts pattern
app.command(install_app)

View file

@ -121,7 +121,7 @@ def install_claude_code(
dependencies=deduplicated_packages,
requirements=str(with_requirements) if with_requirements else None,
project=str(project) if project else None,
editable=str(with_editable) if with_editable else None,
editable=[str(with_editable)] if with_editable else None,
)
args = env_config.build_uv_args()

View file

@ -86,7 +86,7 @@ def install_claude_desktop(
dependencies=deduplicated_packages,
requirements=str(with_requirements) if with_requirements else None,
project=str(project) if project else None,
editable=str(with_editable) if with_editable else None,
editable=[str(with_editable)] if with_editable else None,
)
args = env_config.build_uv_args()

View file

@ -120,7 +120,7 @@ def install_cursor_workspace(
dependencies=deduplicated_packages,
requirements=str(with_requirements.resolve()) if with_requirements else None,
project=str(project.resolve()) if project else None,
editable=str(with_editable.resolve()) if with_editable else None,
editable=[str(with_editable.resolve())] if with_editable else None,
)
args = env_config.build_uv_args()
@ -200,7 +200,7 @@ def install_cursor(
dependencies=deduplicated_packages,
requirements=str(with_requirements.resolve()) if with_requirements else None,
project=str(project.resolve()) if project else None,
editable=str(with_editable.resolve()) if with_editable else None,
editable=[str(with_editable.resolve())] if with_editable else None,
)
args = env_config.build_uv_args()

View file

@ -61,7 +61,7 @@ def install_mcp_json(
dependencies=deduplicated_packages,
requirements=str(with_requirements) if with_requirements else None,
project=str(project) if project else None,
editable=str(with_editable) if with_editable else None,
editable=[str(with_editable)] if with_editable else None,
)
args = env_config.build_uv_args()

View file

@ -44,7 +44,7 @@ def run_with_uv(
path: str | None = None,
log_level: LogLevelType | None = None,
show_banner: bool = True,
editable: str | None = None,
editable: str | list[str] | None = None,
) -> None:
"""Run a MCP server using uv run subprocess.
@ -98,7 +98,10 @@ def run_with_uv(
if config.environment.requirements
else None
)
editable = editable or config.environment.editable
# Note: config editable is a list but CLI currently only supports single path
# Just pass through for now - Environment will handle the list
if not editable and config.environment.editable:
editable = config.environment.editable
# Merge packages from both sources
# Only merge if with_packages doesn't already contain them
@ -126,32 +129,33 @@ def run_with_uv(
dependencies=with_packages if with_packages else None,
requirements=str(with_requirements.resolve()) if with_requirements else None,
project=str(project.resolve()) if project else None,
editable=editable,
)
# IMPORTANT: We add --skip-env to prevent infinite recursion.
# When this function executes `uv run ... fastmcp run server.py`, the inner
# `fastmcp run` command will be executed inside the uv environment we're creating.
# Without --skip-env, that inner command would detect it needs uv (due to the same
# CLI args) and try to spawn ANOTHER uv subprocess, creating infinite recursion.
# The --skip-env flag tells the inner fastmcp: "skip environment setup, we're already
# inside the uv environment that was just created for us."
cmd = ["uv"] + env_config.build_uv_args(
["fastmcp", "run", server_spec, "--skip-env"]
editable=editable
if isinstance(editable, list)
else ([editable] if editable else None),
)
# Build the uv command
# Build the inner fastmcp command with --skip-env to prevent infinite recursion
inner_cmd = ["fastmcp", "run", "--skip-env", server_spec]
# Add transport options
# Add transport options to the inner command
if transport:
cmd.extend(["--transport", transport])
if host:
cmd.extend(["--host", host])
if port:
cmd.extend(["--port", str(port)])
if path:
cmd.extend(["--path", path])
inner_cmd.extend(["--transport", transport])
# Only add HTTP-specific options for non-stdio transports
if transport != "stdio":
if host:
inner_cmd.extend(["--host", host])
if port:
inner_cmd.extend(["--port", str(port)])
if path:
inner_cmd.extend(["--path", path])
if log_level:
cmd.extend(["--log-level", log_level])
inner_cmd.extend(["--log-level", log_level])
if not show_banner:
cmd.append("--no-banner")
inner_cmd.append("--no-banner")
# Build the full uv command
uv_args = env_config.build_uv_args(inner_cmd)
cmd = ["uv"] + uv_args
# Run the command
logger.debug(f"Running command: {' '.join(cmd)}")
@ -269,9 +273,8 @@ 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()
# Prepare source only (environment is handled by uv run)
await config.prepare_source() if not skip_source else None
# Load the server using the source
from contextlib import nullcontext
@ -290,9 +293,8 @@ async def run_command(
source = FileSystemSource(path=server_spec)
config = FastMCPConfig(source=source)
# Prepare the source if needed
if not skip_source:
await config.source.prepare()
# Prepare source only (environment is handled by uv run)
await config.prepare_source() if not skip_source else None
# Load the server
from contextlib import nullcontext

View file

@ -474,9 +474,8 @@ class OAuthProxy(OAuthProvider):
query_params["scope"] = " ".join(scopes_to_use)
# Build the upstream authorization URL
upstream_url = (
f"{self._upstream_authorization_endpoint}?{urlencode(query_params)}"
)
separator = "&" if "?" in self._upstream_authorization_endpoint else "?"
upstream_url = f"{self._upstream_authorization_endpoint}{separator}{urlencode(query_params)}"
logger.debug(
"Starting OAuth transaction %s for client %s, redirecting to IdP",

View file

@ -8,6 +8,8 @@ from mcp.server.elicitation import (
DeclinedElicitation,
)
from pydantic import BaseModel
from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue
from pydantic_core import core_schema
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
@ -26,6 +28,60 @@ logger = get_logger(__name__)
T = TypeVar("T")
class ElicitationJsonSchema(GenerateJsonSchema):
"""Custom JSON schema generator for MCP elicitation that always inlines enums.
MCP elicitation requires inline enum schemas without $ref/$defs references.
This generator ensures enums are always generated inline for compatibility.
Optionally adds enumNames for better UI display when available.
"""
def generate_inner(self, schema: core_schema.CoreSchema) -> JsonSchemaValue:
"""Override to prevent ref generation for enums."""
# For enum schemas, bypass the ref mechanism entirely
if schema["type"] == "enum":
# Directly call our custom enum_schema without going through handler
# This prevents the ref/defs mechanism from being invoked
return self.enum_schema(schema)
# For all other types, use the default implementation
return super().generate_inner(schema)
def enum_schema(self, schema: core_schema.EnumSchema) -> JsonSchemaValue:
"""Generate inline enum schema with optional enumNames for better UI.
If enum members have a _display_name_ attribute or custom __str__,
we'll include enumNames for better UI representation.
"""
# Get the base schema from parent
result = super().enum_schema(schema)
# Try to add enumNames if the enum has display-friendly names
enum_cls = schema.get("cls")
if enum_cls:
members = schema.get("members", [])
enum_names = []
has_custom_names = False
for member in members:
# Check if member has a custom display name attribute
if hasattr(member, "_display_name_"):
enum_names.append(member._display_name_)
has_custom_names = True
# Or use the member name with better formatting
else:
# Convert SNAKE_CASE to Title Case for display
display_name = member.name.replace("_", " ").title()
enum_names.append(display_name)
if display_name != member.value:
has_custom_names = True
# Only add enumNames if they differ from the values
if has_custom_names:
result["enumNames"] = enum_names
return result
# we can't use the low-level AcceptedElicitation because it only works with BaseModels
class AcceptedElicitation(BaseModel, Generic[T]):
"""Result when user accepts the elicitation."""
@ -46,7 +102,10 @@ def get_elicitation_schema(response_type: type[T]) -> dict[str, Any]:
response_type: The type of the response
"""
schema = get_cached_typeadapter(response_type).json_schema()
# Use custom schema generator that inlines enums for MCP compatibility
schema = get_cached_typeadapter(response_type).json_schema(
schema_generator=ElicitationJsonSchema
)
schema = compress_schema(schema)
# Validate the schema to ensure it follows MCP elicitation requirements

View file

@ -10,6 +10,8 @@ from __future__ import annotations
import json
import os
import re
import shutil
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, overload
@ -55,10 +57,10 @@ class Environment(BaseModel):
examples=[".", "../my-project"],
)
editable: str | None = Field(
editable: list[str] | None = Field(
default=None,
description="Directory to install in editable mode",
examples=[".", "../my-package"],
description="Directories to install in editable mode",
examples=[[".", "../my-package"], ["/path/to/package"]],
)
def build_uv_args(self, command: str | list[str] | None = None) -> list[str]:
@ -72,34 +74,28 @@ class Environment(BaseModel):
"""
args = ["run"]
# Add Python version if specified
if self.python:
args.extend(["--python", self.python])
# Add project directory if specified
# Add project if specified
if self.project:
args.extend(["--project", str(self.project)])
# 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 Python version if specified (only if no project, as project has its own Python)
if self.python and not self.project:
args.extend(["--python", self.python])
# Add additional dependencies (skip fastmcp if already added)
# Always add dependencies, requirements, and editable packages
# These work with --project to add additional packages on top of the project env
if self.dependencies:
for dep in self.dependencies:
if dep != "fastmcp": # Skip fastmcp since we already added it
args.extend(["--with", dep])
args.extend(["--with", dep])
# Add requirements file
if self.requirements:
args.extend(["--with-requirements", str(self.requirements)])
# Add editable package
# Add editable packages
if self.editable:
args.extend(["--with-editable", str(self.editable)])
for editable_path in self.editable:
args.extend(["--with-editable", str(editable_path)])
# Add the command if provided
if command:
@ -110,34 +106,6 @@ 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.
@ -177,6 +145,170 @@ class Environment(BaseModel):
]
)
async def prepare(self, output_dir: Path | None = None) -> None:
"""Prepare the Python environment using uv.
Args:
output_dir: Directory where the persistent uv project will be created.
If None, creates a temporary directory for ephemeral use.
"""
# Check if uv is available
if not shutil.which("uv"):
raise RuntimeError(
"uv is not installed. Please install it with: "
"curl -LsSf https://astral.sh/uv/install.sh | sh"
)
# Only prepare environment if there are actual settings to apply
if not self.needs_uv():
logger.debug("No environment settings configured, skipping preparation")
return
# Handle None case for ephemeral use
if output_dir is None:
import tempfile
output_dir = Path(tempfile.mkdtemp(prefix="fastmcp-env-"))
logger.info(f"Creating ephemeral environment in {output_dir}")
else:
logger.info(f"Creating persistent environment in {output_dir}")
output_dir = Path(output_dir).resolve()
# Initialize the project
logger.debug(f"Initializing uv project in {output_dir}")
try:
subprocess.run(
[
"uv",
"init",
"--project",
str(output_dir),
"--name",
"fastmcp-env",
],
check=True,
capture_output=True,
text=True,
)
except subprocess.CalledProcessError as e:
# If project already exists, that's fine - continue
if "already initialized" in e.stderr.lower():
logger.debug(
f"Project already initialized at {output_dir}, continuing..."
)
else:
logger.error(f"Failed to initialize project: {e.stderr}")
raise RuntimeError(f"Failed to initialize project: {e.stderr}") from e
# Pin Python version if specified
if self.python:
logger.debug(f"Pinning Python version to {self.python}")
try:
subprocess.run(
[
"uv",
"python",
"pin",
self.python,
"--project",
str(output_dir),
],
check=True,
capture_output=True,
text=True,
)
except subprocess.CalledProcessError as e:
logger.error(f"Failed to pin Python version: {e.stderr}")
raise RuntimeError(f"Failed to pin Python version: {e.stderr}") from e
# Add dependencies with --no-sync to defer installation
# dependencies ALWAYS include fastmcp; this is compatible with
# specific fastmcp versions that might be in the dependencies list
dependencies = (self.dependencies or []) + ["fastmcp"]
logger.debug(f"Adding dependencies: {', '.join(dependencies)}")
try:
subprocess.run(
[
"uv",
"add",
*dependencies,
"--no-sync",
"--project",
str(output_dir),
],
check=True,
capture_output=True,
text=True,
)
except subprocess.CalledProcessError as e:
logger.error(f"Failed to add dependencies: {e.stderr}")
raise RuntimeError(f"Failed to add dependencies: {e.stderr}") from e
# Add requirements file if specified
if self.requirements:
logger.debug(f"Adding requirements from {self.requirements}")
# Resolve requirements path relative to current directory
req_path = Path(self.requirements).resolve()
try:
subprocess.run(
[
"uv",
"add",
"-r",
str(req_path),
"--no-sync",
"--project",
str(output_dir),
],
check=True,
capture_output=True,
text=True,
)
except subprocess.CalledProcessError as e:
logger.error(f"Failed to add requirements: {e.stderr}")
raise RuntimeError(f"Failed to add requirements: {e.stderr}") from e
# Add editable packages if specified
if self.editable:
editable_paths = [str(Path(e).resolve()) for e in self.editable]
logger.debug(f"Adding editable packages: {', '.join(editable_paths)}")
try:
subprocess.run(
[
"uv",
"add",
"--editable",
*editable_paths,
"--no-sync",
"--project",
str(output_dir),
],
check=True,
capture_output=True,
text=True,
)
except subprocess.CalledProcessError as e:
logger.error(f"Failed to add editable packages: {e.stderr}")
raise RuntimeError(
f"Failed to add editable packages: {e.stderr}"
) from e
# Final sync to install everything
logger.info("Installing dependencies...")
try:
subprocess.run(
["uv", "sync", "--project", str(output_dir)],
check=True,
capture_output=True,
text=True,
)
except subprocess.CalledProcessError as e:
logger.error(f"Failed to sync dependencies: {e.stderr}")
raise RuntimeError(f"Failed to sync dependencies: {e.stderr}") from e
logger.info(f"Environment prepared successfully in {output_dir}")
class Deployment(BaseModel):
"""Configuration for server deployment and runtime settings."""
@ -453,7 +585,7 @@ class FastMCPConfig(BaseModel):
dependencies=dependencies,
requirements=requirements,
project=project,
editable=editable,
editable=[editable] if editable else None,
)
# Build deployment config if any deployment args provided
@ -499,6 +631,45 @@ class FastMCPConfig(BaseModel):
return None
async def prepare(
self,
skip_source: bool = False,
output_dir: Path | None = None,
) -> None:
"""Prepare environment and source for execution.
When output_dir is provided, creates a persistent uv project.
When output_dir is None, does ephemeral caching (for backwards compatibility).
Args:
skip_source: Skip source preparation if True
output_dir: Directory to create the persistent uv project in (optional)
"""
# Prepare environment (persistent if output_dir provided, ephemeral otherwise)
if self.environment:
await self.prepare_environment(output_dir=output_dir)
if not skip_source:
await self.prepare_source()
async def prepare_environment(self, output_dir: Path | None = None) -> None:
"""Prepare the Python environment.
Args:
output_dir: If provided, creates a persistent uv project in this directory.
If None, just populates uv's cache for ephemeral use.
Delegates to the environment's prepare() method
"""
await self.environment.prepare(output_dir=output_dir)
async def prepare_source(self) -> None:
"""Prepare the source for loading.
Delegates to the source's prepare() method.
"""
await self.source.prepare()
async def run_server(self, **kwargs: Any) -> None:
"""Load and run the server with this configuration.

View file

@ -243,17 +243,25 @@
"editable": {
"anyOf": [
{
"type": "string"
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"default": null,
"description": "Directory to install in editable mode",
"description": "Directories to install in editable mode",
"examples": [
".",
"../my-package"
[
".",
"../my-package"
],
[
"/path/to/package"
]
],
"title": "Editable"
}

View file

@ -328,18 +328,19 @@ class TestRunCommand:
]
)
def test_run_command_parsing_skip_env_flag(self):
"""Test run command parsing with --skip-env flag."""
def test_run_command_parsing_project_flag(self):
"""Test run command parsing with --project flag."""
command, bound, _ = app.parse_args(
[
"run",
"server.py",
"--skip-env",
"--project",
"./test-env",
]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["skip_env"] is True
assert bound.arguments["project"] == Path("./test-env")
def test_run_command_parsing_skip_source_flag(self):
"""Test run command parsing with --skip-source flag."""
@ -354,19 +355,20 @@ class TestRunCommand:
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."""
def test_run_command_parsing_project_and_skip_source(self):
"""Test run command parsing with --project and --skip-source flags."""
command, bound, _ = app.parse_args(
[
"run",
"server.py",
"--skip-env",
"--project",
"./test-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["project"] == Path("./test-env")
assert bound.arguments["skip_source"] is True

View file

@ -59,7 +59,7 @@ class TestEnvironment:
"dependencies": ["requests", "numpy>=2.0"],
"requirements": "requirements.txt",
"project": ".",
"editable": "../my-package",
"editable": ["../my-package"],
},
)
@ -68,7 +68,7 @@ class TestEnvironment:
assert env.dependencies == ["requests", "numpy>=2.0"]
assert env.requirements == "requirements.txt"
assert env.project == "."
assert env.editable == "../my-package"
assert env.editable == ["../my-package"]
def test_needs_uv(self):
"""Test needs_uv() method."""
@ -107,15 +107,17 @@ class TestEnvironment:
args = config.environment.build_uv_args(["fastmcp", "run", "server.py"])
assert args[0] == "run"
assert "--python" in args
assert "3.12" in args
# Python version not added when project is specified (project defines its own Python)
assert "--python" not in args
assert "3.12" not in args
assert "--project" in args
assert "." in args
assert "--with" in args
assert "fastmcp" in args
assert "requests" in args
assert "numpy" in args
assert "--with-requirements" in args
assert "requirements.txt" in args
# Command args should be at the end
assert "fastmcp" in args[-3:]
assert "run" in args[-2:]
assert "server.py" in args[-1:]

View file

@ -280,11 +280,7 @@ class TestInstallCursor:
# Verify failure message was printed
mock_print.assert_called()
@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):
def test_install_cursor_deduplicate_packages(self):
"""Test that duplicate packages are deduplicated."""
with patch("fastmcp.cli.install.cursor.open_deeplink") as mock_open:
mock_open.return_value = True
@ -305,8 +301,8 @@ class TestInstallCursor:
args_str = " ".join(config_data["args"])
assert args_str.count("numpy") == 1
assert args_str.count("pandas") == 1
# fastmcp appears twice: once as --with fastmcp and once as the command
assert args_str.count("fastmcp") == 2
# fastmcp appears once in the command only (no longer automatically added as --with)
assert args_str.count("fastmcp") == 1
class TestCursorCommand:

View file

@ -0,0 +1,302 @@
"""Tests for the fastmcp project prepare command."""
import subprocess
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastmcp.utilities.fastmcp_config import Environment, FastMCPConfig
from fastmcp.utilities.fastmcp_config.v1.sources.filesystem import FileSystemSource
class TestFastMCPConfigPrepare:
"""Test the FastMCPConfig.prepare() method."""
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.FastMCPConfig.prepare_source",
new_callable=AsyncMock,
)
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.FastMCPConfig.prepare_environment",
new_callable=AsyncMock,
)
async def test_prepare_calls_both_methods(self, mock_env, mock_src):
"""Test that prepare() calls both prepare_environment and prepare_source."""
config = FastMCPConfig(
source=FileSystemSource(path="server.py"),
environment=Environment(python="3.10"),
)
await config.prepare()
mock_env.assert_called_once()
mock_src.assert_called_once()
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.FastMCPConfig.prepare_source",
new_callable=AsyncMock,
)
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.FastMCPConfig.prepare_environment",
new_callable=AsyncMock,
)
async def test_prepare_with_output_dir(self, mock_env, mock_src):
"""Test that prepare() with output_dir calls prepare_environment with it."""
config = FastMCPConfig(
source=FileSystemSource(path="server.py"),
environment=Environment(python="3.10"),
)
output_path = Path("/tmp/test-env")
await config.prepare(skip_source=False, output_dir=output_path)
mock_env.assert_called_once_with(output_dir=output_path)
mock_src.assert_called_once()
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.FastMCPConfig.prepare_source",
new_callable=AsyncMock,
)
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.FastMCPConfig.prepare_environment",
new_callable=AsyncMock,
)
async def test_prepare_skip_source(self, mock_env, mock_src):
"""Test that prepare() skips source when skip_source=True."""
config = FastMCPConfig(
source=FileSystemSource(path="server.py"),
environment=Environment(python="3.10"),
)
await config.prepare(skip_source=True)
mock_env.assert_called_once_with(output_dir=None)
mock_src.assert_not_called()
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.FastMCPConfig.prepare_source",
new_callable=AsyncMock,
)
@patch(
"fastmcp.utilities.fastmcp_config.v1.fastmcp_config.Environment.prepare",
new_callable=AsyncMock,
)
async def test_prepare_no_environment_settings(self, mock_env_prepare, mock_src):
"""Test that prepare() works with default empty environment config."""
config = FastMCPConfig(
source=FileSystemSource(path="server.py"),
# environment defaults to empty Environment()
)
await config.prepare(skip_source=False)
# Environment prepare should be called even with empty config
mock_env_prepare.assert_called_once_with(output_dir=None)
mock_src.assert_called_once()
class TestEnvironmentPrepare:
"""Test the Environment.prepare() method."""
@patch("shutil.which")
async def test_prepare_no_uv_installed(self, mock_which, tmp_path):
"""Test that prepare() raises error when uv is not installed."""
mock_which.return_value = None
env = Environment(python="3.10")
with pytest.raises(RuntimeError, match="uv is not installed"):
await env.prepare(tmp_path / "test-env")
@patch("subprocess.run")
@patch("shutil.which")
async def test_prepare_no_settings(self, mock_which, mock_run, tmp_path):
"""Test that prepare() does nothing when no settings are configured."""
mock_which.return_value = "/usr/bin/uv"
env = Environment() # No settings
await env.prepare(tmp_path / "test-env")
# Should not run any commands
mock_run.assert_not_called()
@patch("subprocess.run")
@patch("shutil.which")
async def test_prepare_with_python(self, mock_which, mock_run, tmp_path):
"""Test that prepare() runs uv with python version."""
mock_which.return_value = "/usr/bin/uv"
mock_run.return_value = MagicMock(
returncode=0, stdout="Environment cached", stderr=""
)
env = Environment(python="3.10")
await env.prepare(tmp_path / "test-env")
# Should run multiple uv commands for initializing the project
assert mock_run.call_count > 0
# Check the first call should be uv init
first_call_args = mock_run.call_args_list[0][0][0]
assert first_call_args[0] == "uv"
assert "init" in first_call_args
@patch("subprocess.run")
@patch("shutil.which")
async def test_prepare_with_dependencies(self, mock_which, mock_run, tmp_path):
"""Test that prepare() includes dependencies."""
mock_which.return_value = "/usr/bin/uv"
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
env = Environment(dependencies=["numpy", "pandas"])
await env.prepare(tmp_path / "test-env")
# Should run multiple uv commands, one of which should be uv add
assert mock_run.call_count > 0
# Find the add command call
add_call = None
for call_args, _ in mock_run.call_args_list:
args = call_args[0]
if "add" in args:
add_call = args
break
assert add_call is not None, "Should have called uv add"
assert "numpy" in add_call
assert "pandas" in add_call
assert "fastmcp" in add_call # Always added
@patch("subprocess.run")
@patch("shutil.which")
async def test_prepare_command_fails(self, mock_which, mock_run, tmp_path):
"""Test that prepare() raises error when uv command fails."""
mock_which.return_value = "/usr/bin/uv"
mock_run.side_effect = subprocess.CalledProcessError(
1, ["uv"], stderr="Package not found"
)
env = Environment(python="3.10")
with pytest.raises(RuntimeError, match="Failed to initialize project"):
await env.prepare(tmp_path / "test-env")
class TestProjectPrepareCommand:
"""Test the CLI project prepare command."""
@patch("fastmcp.utilities.fastmcp_config.FastMCPConfig.from_file")
@patch("fastmcp.utilities.fastmcp_config.FastMCPConfig.find_config")
async def test_project_prepare_auto_detect(self, mock_find, mock_from_file):
"""Test project prepare with auto-detected config."""
from fastmcp.cli.cli import prepare
# Setup mocks
mock_find.return_value = Path("fastmcp.json")
mock_config = AsyncMock()
mock_from_file.return_value = mock_config
# Run command with output_dir
with patch("sys.exit"):
with patch("fastmcp.cli.cli.console.print") as mock_print:
await prepare(config_path=None, output_dir="./test-env")
# Should find and load config
mock_find.assert_called_once()
mock_from_file.assert_called_once_with(Path("fastmcp.json"))
# Should call prepare with output_dir
mock_config.prepare.assert_called_once_with(
skip_source=False,
output_dir=Path("./test-env"),
)
# Should print success message
mock_print.assert_called()
success_call = mock_print.call_args_list[-1][0][0]
assert "Project prepared successfully" in success_call
@patch("pathlib.Path.exists")
@patch("fastmcp.utilities.fastmcp_config.FastMCPConfig.from_file")
async def test_project_prepare_explicit_path(self, mock_from_file, mock_exists):
"""Test project prepare with explicit config path."""
from fastmcp.cli.cli import prepare
# Setup mocks
mock_exists.return_value = True
mock_config = AsyncMock()
mock_from_file.return_value = mock_config
# Run command with explicit path
with patch("fastmcp.cli.cli.console.print"):
await prepare(config_path="myconfig.json", output_dir="./test-env")
# Should load specified config
mock_from_file.assert_called_once_with(Path("myconfig.json"))
# Should call prepare
mock_config.prepare.assert_called_once_with(
skip_source=False,
output_dir=Path("./test-env"),
)
@patch("fastmcp.utilities.fastmcp_config.FastMCPConfig.find_config")
async def test_project_prepare_no_config_found(self, mock_find):
"""Test project prepare when no config is found."""
from fastmcp.cli.cli import prepare
# Setup mocks
mock_find.return_value = None
# Run command without output_dir - should exit with error for missing output_dir
with pytest.raises(SystemExit) as exc_info:
with patch("fastmcp.cli.cli.logger.error") as mock_error:
await prepare(config_path=None, output_dir=None)
assert exc_info.value.code == 1
mock_error.assert_called()
error_msg = mock_error.call_args[0][0]
assert "--output-dir parameter is required" in error_msg
@patch("pathlib.Path.exists")
async def test_project_prepare_config_not_exists(self, mock_exists):
"""Test project prepare when specified config doesn't exist."""
from fastmcp.cli.cli import prepare
# Setup mocks
mock_exists.return_value = False
# Run command without output_dir - should exit with error for missing output_dir
with pytest.raises(SystemExit) as exc_info:
with patch("fastmcp.cli.cli.logger.error") as mock_error:
await prepare(config_path="missing.json", output_dir=None)
assert exc_info.value.code == 1
mock_error.assert_called()
error_msg = mock_error.call_args[0][0]
assert "--output-dir parameter is required" in error_msg
@patch("pathlib.Path.exists")
@patch("fastmcp.utilities.fastmcp_config.FastMCPConfig.from_file")
async def test_project_prepare_failure(self, mock_from_file, mock_exists):
"""Test project prepare when prepare() fails."""
from fastmcp.cli.cli import prepare
# Setup mocks
mock_exists.return_value = True
mock_config = AsyncMock()
mock_config.prepare.side_effect = RuntimeError("Preparation failed")
mock_from_file.return_value = mock_config
# Run command - should exit with error
with pytest.raises(SystemExit) as exc_info:
with patch("fastmcp.cli.cli.console.print") as mock_print:
await prepare(config_path="config.json", output_dir="./test-env")
assert exc_info.value.code == 1
# Should print error message
error_call = mock_print.call_args_list[-1][0][0]
assert "Failed to prepare project" in error_call

View file

@ -248,7 +248,7 @@ def test_environment_config_path_resolution(tmp_path):
"environment": {
"requirements": "requirements.txt",
"project": ".",
"editable": "../other-project",
"editable": ["../other-project"],
},
}

View file

@ -12,12 +12,8 @@ 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, mock_find_dev_path):
def test_run_with_uv_basic(self, mock_run):
"""Test basic run_with_uv execution."""
mock_run.return_value = Mock(returncode=0)
@ -33,21 +29,15 @@ class TestRunWithUv:
expected = [
"uv",
"run",
"--with",
"fastmcp",
"fastmcp",
"run",
"server.py",
"--skip-env",
"server.py",
]
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, mock_find_dev_path):
def test_run_with_uv_python_version(self, mock_run):
"""Test run_with_uv with Python version."""
mock_run.return_value = Mock(returncode=0)
@ -62,21 +52,15 @@ class TestRunWithUv:
"run",
"--python",
"3.11",
"--with",
"fastmcp",
"fastmcp",
"run",
"server.py",
"--skip-env",
"server.py",
]
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, mock_find_dev_path):
def test_run_with_uv_project(self, mock_run):
"""Test run_with_uv with project directory."""
mock_run.return_value = Mock(returncode=0)
# Use an absolute path that works on all platforms
@ -94,20 +78,14 @@ class TestRunWithUv:
assert Path(cmd[3]).is_absolute()
# Check the rest of the command
assert cmd[4:] == [
"--with",
"fastmcp",
"fastmcp",
"run",
"server.py",
"--skip-env",
"server.py",
]
@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, mock_find_dev_path):
def test_run_with_uv_with_packages(self, mock_run):
"""Test run_with_uv with additional packages."""
mock_run.return_value = Mock(returncode=0)
@ -121,24 +99,18 @@ class TestRunWithUv:
"uv",
"run",
"--with",
"fastmcp",
"--with",
"pandas", # original order preserved
"--with",
"numpy", # original order preserved
"fastmcp",
"run",
"server.py",
"--skip-env",
"server.py",
]
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, mock_find_dev_path):
def test_run_with_uv_with_requirements(self, mock_run):
"""Test run_with_uv with requirements file."""
mock_run.return_value = Mock(returncode=0)
req_path = Path("requirements.txt")
@ -152,23 +124,17 @@ class TestRunWithUv:
expected = [
"uv",
"run",
"--with",
"fastmcp",
"--with-requirements",
str(req_path.resolve()), # auto-resolved to absolute path
"fastmcp",
"run",
"server.py",
"--skip-env",
"server.py",
]
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, mock_find_dev_path):
def test_run_with_uv_transport_options(self, mock_run):
"""Test run_with_uv with transport-related options."""
mock_run.return_value = Mock(returncode=0)
@ -189,12 +155,10 @@ class TestRunWithUv:
expected = [
"uv",
"run",
"--with",
"fastmcp",
"fastmcp",
"run",
"server.py",
"--skip-env",
"server.py",
"--transport",
"http",
"--host",
@ -209,12 +173,8 @@ 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, mock_find_dev_path):
def test_run_with_uv_all_options(self, mock_run):
"""Test run_with_uv with all options combined."""
mock_run.return_value = Mock(returncode=0)
@ -237,20 +197,31 @@ class TestRunWithUv:
cmd = mock_run.call_args[0][0]
# Check the structure piece by piece to be platform-agnostic
assert cmd[:5] == ["uv", "run", "--python", "3.10", "--project"]
# Check project path is absolute
assert Path(cmd[5]).is_absolute()
assert cmd[6:10] == ["--with", "fastmcp", "--with", "pandas"]
assert cmd[10] == "--with-requirements"
# Check requirements path is now auto-resolved to absolute
assert Path(cmd[11]).is_absolute()
assert Path(cmd[11]).name == "reqs.txt"
assert cmd[12:] == [
# When project is specified, Python version is not included
# Build expected command step by step
expected_start = ["uv", "run", "--project"]
# Check start and that project path is absolute
assert cmd[:3] == expected_start
assert Path(cmd[3]).is_absolute()
# Find the index where packages and requirements start
next_idx = 4
assert cmd[next_idx : next_idx + 2] == ["--with", "pandas"]
next_idx += 2
assert cmd[next_idx : next_idx + 1] == ["--with-requirements"]
next_idx += 1
# Check requirements path is absolute
assert Path(cmd[next_idx]).is_absolute()
assert Path(cmd[next_idx]).name == "reqs.txt"
next_idx += 1
# Rest should be the fastmcp command with options
assert cmd[next_idx:] == [
"fastmcp",
"run",
"server.py",
"--skip-env",
"server.py",
"--transport",
"http",
"--port",
@ -258,12 +229,8 @@ 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, mock_find_dev_path):
def test_run_with_uv_error_handling(self, mock_run):
"""Test run_with_uv error handling."""
mock_run.side_effect = subprocess.CalledProcessError(1, ["uv", "run"])
@ -272,13 +239,9 @@ 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, mock_find_dev_path):
def test_run_with_uv_logging(self, mock_run, mock_logger):
"""Test that run_with_uv logs the command."""
mock_run.return_value = Mock(returncode=0)

View file

@ -15,6 +15,7 @@ from fastmcp.server.elicitation import (
AcceptedElicitation,
CancelledElicitation,
DeclinedElicitation,
get_elicitation_schema,
validate_elicitation_json_schema,
)
from fastmcp.utilities.types import TypeAdapter
@ -639,3 +640,80 @@ async def test_elicitation_implicit_acceptance_must_be_dict(fastmcp_server):
match="Elicitation responses must be serializable as a JSON object",
):
await client.call_tool("ask_for_name")
def test_enum_elicitation_schema_inline():
"""Test that enum schemas are generated inline without $ref/$defs for MCP compatibility."""
class Priority(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
@dataclass
class TaskRequest:
title: str
priority: Priority
# Generate elicitation schema
schema = get_elicitation_schema(TaskRequest)
# Verify no $defs section exists (enums should be inlined)
assert "$defs" not in schema, (
"Schema should not contain $defs - enums must be inline"
)
# Verify no $ref in properties
for prop_name, prop_schema in schema.get("properties", {}).items():
assert "$ref" not in prop_schema, (
f"Property {prop_name} contains $ref - should be inline"
)
# Verify the priority field has inline enum values
priority_schema = schema["properties"]["priority"]
assert "enum" in priority_schema, "Priority should have enum values inline"
assert priority_schema["enum"] == ["low", "medium", "high"]
assert priority_schema.get("type") == "string"
# Verify title field is a simple string
assert schema["properties"]["title"]["type"] == "string"
def test_enum_elicitation_schema_with_enum_names():
"""Test that enum schemas can include enumNames for better UI display."""
class TaskStatus(Enum):
NOT_STARTED = "not_started"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
ON_HOLD = "on_hold"
@dataclass
class TaskUpdate:
task_id: str
status: TaskStatus
# Generate elicitation schema
schema = get_elicitation_schema(TaskUpdate)
# Verify enum is inline
assert "$defs" not in schema
assert "$ref" not in str(schema)
status_schema = schema["properties"]["status"]
assert "enum" in status_schema
assert status_schema["enum"] == [
"not_started",
"in_progress",
"completed",
"on_hold",
]
# Check if enumNames were added for display
assert "enumNames" in status_schema
assert status_schema["enumNames"] == [
"Not Started",
"In Progress",
"Completed",
"On Hold",
]

View file

@ -114,6 +114,46 @@ class TestOAuthProxyComprehensive:
)
assert proxy2._redirect_path == "/auth/callback"
async def test_authorize_url_with_ampersand_separator(self, jwt_verifier):
"""Test that authorize builds URLs with & separator when upstream endpoint has existing query parameters."""
# Test case: upstream endpoint with existing query parameters
proxy = OAuthProxy(
upstream_authorization_endpoint="https://auth.example.com/authorize?version=2.0",
upstream_token_endpoint="https://auth.example.com/token",
upstream_client_id="client-123",
upstream_client_secret="secret-456",
token_verifier=jwt_verifier,
base_url="https://myserver.com",
)
client = OAuthClientInformationFull(
client_id="test-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:54321/callback")],
)
params = AuthorizationParams(
redirect_uri=AnyUrl("http://localhost:54321/callback"),
redirect_uri_provided_explicitly=True,
state="client-state",
code_challenge="challenge",
scopes=["read"],
)
# Should use "&" separator
redirect_url = await proxy.authorize(client, params)
parsed = urlparse(redirect_url)
query_params = parse_qs(parsed.query)
assert parsed.scheme == "https"
assert parsed.netloc == "auth.example.com"
assert parsed.path == "/authorize"
# Params in the original url are kept
assert "version" in query_params
assert query_params["version"] == ["2.0"]
# New params added correctly
assert query_params["response_type"] == ["code"]
def test_dcr_always_enabled(self, jwt_verifier):
"""Test that DCR is always enabled for OAuth Proxy."""
proxy = OAuthProxy(

View file

@ -1,5 +1,4 @@
from pathlib import Path
from unittest.mock import patch
"""Tests for CLI utility functions."""
from fastmcp.utilities.fastmcp_config.v1.fastmcp_config import Environment
@ -7,24 +6,20 @@ from fastmcp.utilities.fastmcp_config.v1.fastmcp_config import Environment
class TestEnvironmentBuildUVArgs:
"""Test the Environment.build_uv_args() method."""
@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."""
def test_build_uv_args_basic(self):
"""Test building basic uv args with no environment config."""
env = Environment()
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = ["run", "--with", "fastmcp", "fastmcp", "run", "server.py"]
expected = ["run", "fastmcp", "run", "server.py"]
assert args == expected
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_editable(self, mock_dev_path):
def test_build_uv_args_with_editable(self):
"""Test building uv args with editable package."""
editable_path = "/path/to/package"
env = Environment(editable=editable_path)
env = Environment(editable=[editable_path])
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--with",
"fastmcp",
"--with-editable",
editable_path,
"fastmcp",
@ -33,16 +28,13 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_packages(self, mock_dev_path):
def test_build_uv_args_with_packages(self):
"""Test building uv args with additional packages."""
env = Environment(dependencies=["pkg1", "pkg2"])
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--with",
"fastmcp",
"--with",
"pkg1",
"--with",
"pkg2",
@ -52,81 +44,58 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_python_version(self, mock_dev_path):
def test_build_uv_args_with_python_version(self):
"""Test building uv args with Python version."""
env = Environment(python="3.11")
env = Environment(python="3.10")
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--python",
"3.11",
"--with",
"fastmcp",
"3.10",
"fastmcp",
"run",
"server.py",
]
assert args == expected
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_project(self, mock_dev_path):
def test_build_uv_args_with_requirements(self):
"""Test building uv args with requirements file."""
requirements_path = "/path/to/requirements.txt"
env = Environment(requirements=requirements_path)
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--with-requirements",
requirements_path,
"fastmcp",
"run",
"server.py",
]
assert args == expected
def test_build_uv_args_with_project(self):
"""Test building uv args with project directory."""
project_path = "/path/to/project"
env = Environment(project=project_path)
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--project",
project_path,
"--with",
"fastmcp",
"fastmcp",
"run",
"server.py",
]
expected = ["run", "--project", project_path, "fastmcp", "run", "server.py"]
assert args == expected
@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)
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--with",
"fastmcp",
"--with-requirements",
req_path,
"fastmcp",
"run",
"server.py",
]
assert args == expected
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_with_all_options(self, mock_dev_path):
def test_build_uv_args_with_everything(self):
"""Test building uv args with all options."""
project_path = "/my/project"
requirements_path = "/path/to/requirements.txt"
editable_path = "/local/pkg"
requirements_path = "reqs.txt"
env = Environment(
python="3.10",
project=project_path,
dependencies=["pandas", "numpy"],
requirements=requirements_path,
editable=editable_path,
editable=[editable_path],
)
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--python",
"3.10",
"--project",
project_path,
"--with",
"fastmcp",
"--with",
"pandas",
"--with",
@ -141,130 +110,79 @@ class TestEnvironmentBuildUVArgs:
]
assert args == expected
@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")
def test_build_uv_args_no_command(self):
"""Test building uv args without command."""
env = Environment(dependencies=["pkg1"])
args = env.build_uv_args()
expected = ["run", "--python", "3.11", "--with", "fastmcp"]
expected = ["run", "--with", "pkg1"]
assert args == expected
@patch.object(Environment, "_find_fastmcp_dev_path", return_value=None)
def test_build_uv_args_string_command(self, mock_dev_path):
def test_build_uv_args_with_string_command(self):
"""Test building uv args with string command."""
env = Environment()
args = env.build_uv_args("python")
expected = ["run", "--with", "fastmcp", "python"]
expected = ["run", "python"]
assert args == expected
def test_needs_uv_true(self):
"""Test that needs_uv returns True when environment settings are present."""
env = Environment(python="3.11")
assert env.needs_uv() is True
env = Environment(dependencies=["pkg"])
assert env.needs_uv() is True
env = Environment(requirements="reqs.txt")
assert env.needs_uv() is True
env = Environment(project="/project")
assert env.needs_uv() is True
env = Environment(editable="/pkg")
assert env.needs_uv() is True
def test_needs_uv_false(self):
"""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()
def test_build_uv_args_project_with_extras(self):
"""Test that project flag works with additional dependencies."""
project_path = "/path/to/project"
env = Environment(
project=project_path,
python="3.10", # Should be ignored with project
dependencies=["pandas"], # Should be added on top of project
editable=["/pkg"], # Should be added on top of project
)
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = [
"run",
"--project",
project_path,
"--with",
"pandas",
"--with-editable",
str(dev_path),
"/pkg",
"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
class TestEnvironmentNeedsUV:
"""Test the Environment.needs_uv() method."""
def test_needs_uv_with_python(self):
"""Test that needs_uv returns True with Python version."""
env = Environment(python="3.10")
assert env.needs_uv() is True
def test_needs_uv_with_dependencies(self):
"""Test that needs_uv returns True with dependencies."""
env = Environment(dependencies=["pandas"])
assert env.needs_uv() is True
def test_needs_uv_with_requirements(self):
"""Test that needs_uv returns True with requirements."""
env = Environment(requirements="/path/to/requirements.txt")
assert env.needs_uv() is True
def test_needs_uv_with_project(self):
"""Test that needs_uv returns True with project."""
env = Environment(project="/path/to/project")
assert env.needs_uv() is True
def test_needs_uv_with_editable(self):
"""Test that needs_uv returns True with editable."""
env = Environment(editable=["/pkg"])
assert env.needs_uv() is True
def test_needs_uv_empty(self):
"""Test that needs_uv returns False with empty config."""
env = Environment()
args = env.build_uv_args(["fastmcp", "run", "server.py"])
expected = ["run", "--with", "fastmcp", "fastmcp", "run", "server.py"]
assert args == expected
assert env.needs_uv() is False
@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
def test_needs_uv_with_empty_lists(self):
"""Test that needs_uv returns False with empty lists."""
env = Environment(dependencies=None, editable=None)
assert env.needs_uv() is False