Refactor Environment to support multiple runtime types (#1673)

This commit is contained in:
Jeremiah Lowin 2025-08-29 22:29:35 -04:00 committed by GitHub
commit ea47d232bc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 480 additions and 407 deletions

View file

@ -7,7 +7,7 @@ from pathlib import Path
from typing import Any
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
logger = get_logger(__name__)
@ -99,7 +99,7 @@ def update_claude_config(
deduplicated_packages = None
# Build uv run command using Environment.build_uv_run_command()
env_config = Environment(
env_config = UVEnvironment(
dependencies=deduplicated_packages,
editable=[str(p) for p in with_editable] if with_editable else None,
)
@ -113,7 +113,7 @@ def update_claude_config(
file_spec = str(Path(file_spec).resolve())
# Build the full command
full_command = env_config.build_uv_run_command(["fastmcp", "run", file_spec])
full_command = env_config.build_command(["fastmcp", "run", file_spec])
# Extract command and args for the config
server_config: dict[str, Any] = {

View file

@ -26,7 +26,8 @@ from fastmcp.utilities.inspect import (
inspect_fastmcp,
)
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config import Environment, MCPServerConfig
from fastmcp.utilities.mcp_server_config import MCPServerConfig
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
logger = get_logger("cli")
console = Console()
@ -245,7 +246,7 @@ async def dev(
env_deps = config.environment.dependencies or []
all_deps = list(set(env_deps + server.dependencies))
if not config.environment:
config.environment = Environment(dependencies=all_deps)
config.environment = UVEnvironment(dependencies=all_deps)
else:
config.environment.dependencies = all_deps
@ -269,7 +270,7 @@ async def dev(
inspector_cmd += f"@{inspector_version}"
# Use the environment from config (already has CLI overrides applied)
uv_cmd = config.environment.build_uv_run_command(
uv_cmd = config.environment.build_command(
["fastmcp", "run", server_spec, "--no-banner"]
)
@ -460,7 +461,9 @@ async def run(
)
# Check if we need to use uv run (but skip if we're already in uv or user said to skip)
needs_uv = config.environment.needs_uv() and not skip_env
# We check if the environment would modify the command
test_cmd = ["test"]
needs_uv = config.environment.build_command(test_cmd) != test_cmd and not skip_env
if needs_uv:
# Use uv run subprocess - always use run_with_uv which handles output correctly
@ -627,7 +630,9 @@ async def inspect(
sys.exit(1)
# Check if we need to use uv run (but skip if we're already in uv or user said to skip)
needs_uv = config.environment.needs_uv() and not skip_env
# We check if the environment would modify the command
test_cmd = ["test"]
needs_uv = config.environment.build_command(test_cmd) != test_cmd and not skip_env
if needs_uv:
# Build and run uv command
@ -643,8 +648,14 @@ async def inspect(
inspect_command.extend(["--format", format.value])
if output:
inspect_command.extend(["--output", str(output)])
config.environment.run_with_uv(inspect_command)
return # run_with_uv exits the process
# Run the command using subprocess
import subprocess
cmd = config.environment.build_command(inspect_command)
env = os.environ | {"FASTMCP_UV_SPAWNED": "1"}
process = subprocess.run(cmd, check=True, env=env)
sys.exit(process.returncode)
logger.debug(
"Inspecting server",

View file

@ -10,7 +10,7 @@ import cyclopts
from rich import print
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from .shared import process_common_args
@ -116,7 +116,7 @@ def install_claude_code(
deduplicated_packages = None
# Build uv run command using Environment.build_uv_run_command()
env_config = Environment(
env_config = UVEnvironment(
python=python_version,
dependencies=deduplicated_packages,
requirements=str(with_requirements) if with_requirements else None,
@ -131,7 +131,7 @@ def install_claude_code(
server_spec = str(file.resolve())
# Build the full command
full_command = env_config.build_uv_run_command(["fastmcp", "run", server_spec])
full_command = env_config.build_command(["fastmcp", "run", server_spec])
# Build claude mcp add command
cmd_parts = [claude_cmd, "mcp", "add"]

View file

@ -10,7 +10,7 @@ from rich import print
from fastmcp.mcp_config import StdioMCPServer, update_config_file
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from .shared import process_common_args
@ -81,7 +81,7 @@ def install_claude_desktop(
if not deduplicated_packages:
deduplicated_packages = None
env_config = Environment(
env_config = UVEnvironment(
python=python_version,
dependencies=deduplicated_packages,
requirements=str(with_requirements) if with_requirements else None,
@ -95,7 +95,7 @@ def install_claude_desktop(
server_spec = str(file.resolve())
# Build the full command
full_command = env_config.build_uv_run_command(["fastmcp", "run", server_spec])
full_command = env_config.build_command(["fastmcp", "run", server_spec])
# Create server configuration
server_config = StdioMCPServer(

View file

@ -11,7 +11,7 @@ from rich import print
from fastmcp.mcp_config import StdioMCPServer, update_config_file
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from .shared import process_common_args
@ -115,7 +115,7 @@ def install_cursor_workspace(
if not deduplicated_packages:
deduplicated_packages = None
env_config = Environment(
env_config = UVEnvironment(
python=python_version,
dependencies=deduplicated_packages,
requirements=str(with_requirements.resolve()) if with_requirements else None,
@ -129,7 +129,7 @@ def install_cursor_workspace(
server_spec = str(file.resolve())
# Build the full command
full_command = env_config.build_uv_run_command(["fastmcp", "run", server_spec])
full_command = env_config.build_command(["fastmcp", "run", server_spec])
# Create server configuration
server_config = StdioMCPServer(
@ -193,7 +193,7 @@ def install_cursor(
if not deduplicated_packages:
deduplicated_packages = None
env_config = Environment(
env_config = UVEnvironment(
python=python_version,
dependencies=deduplicated_packages,
requirements=str(with_requirements.resolve()) if with_requirements else None,
@ -207,7 +207,7 @@ def install_cursor(
server_spec = str(file.resolve())
# Build the full command
full_command = env_config.build_uv_run_command(["fastmcp", "run", server_spec])
full_command = env_config.build_command(["fastmcp", "run", server_spec])
# If workspace is specified, install to workspace-specific config
if workspace:

View file

@ -10,7 +10,7 @@ import pyperclip
from rich import print
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from .shared import process_common_args
@ -56,7 +56,7 @@ def install_mcp_json(
if not deduplicated_packages:
deduplicated_packages = None
env_config = Environment(
env_config = UVEnvironment(
python=python_version,
dependencies=deduplicated_packages,
requirements=str(with_requirements) if with_requirements else None,
@ -70,7 +70,7 @@ def install_mcp_json(
server_spec = str(file.resolve())
# Build the full command
full_command = env_config.build_uv_run_command(["fastmcp", "run", server_spec])
full_command = env_config.build_command(["fastmcp", "run", server_spec])
# Build MCP server configuration
server_config = {

View file

@ -13,9 +13,9 @@ from mcp.server.fastmcp import FastMCP as FastMCP1x
from fastmcp.server.server import FastMCP
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config import (
Environment,
MCPServerConfig,
)
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import FileSystemSource
logger = get_logger("cli.run")
@ -67,7 +67,7 @@ def run_with_uv(
"""
# Build uv command using Environment.build_uv_run_command()
env_config = Environment(
env_config = UVEnvironment(
python=python_version,
dependencies=with_packages if with_packages else None,
requirements=str(with_requirements.resolve()) if with_requirements else None,
@ -97,7 +97,7 @@ def run_with_uv(
inner_cmd.append("--no-banner")
# Build the full uv command
cmd = env_config.build_uv_run_command(inner_cmd)
cmd = env_config.build_command(inner_cmd)
# Set marker to prevent infinite loops when subprocess calls FastMCP again
env = os.environ | {"FASTMCP_UV_SPAWNED": "1"}

View file

@ -37,7 +37,7 @@ from fastmcp.mcp_config import MCPConfig, infer_transport_type_from_url
from fastmcp.server.dependencies import get_http_headers
from fastmcp.server.server import FastMCP
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config.v1.mcp_server_config import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
logger = get_logger(__name__)
@ -597,7 +597,7 @@ class UvStdioTransport(StdioTransport):
)
# Create Environment from provided parameters (internal use)
env_config = Environment(
env_config = UVEnvironment(
python=python_version,
dependencies=with_packages,
requirements=with_requirements,

View file

@ -4,9 +4,10 @@ This module provides versioned configuration support for FastMCP servers.
The current version is v1, which is re-exported here for convenience.
"""
from fastmcp.utilities.mcp_server_config.v1.environments.base import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from fastmcp.utilities.mcp_server_config.v1.mcp_server_config import (
Deployment,
Environment,
MCPServerConfig,
generate_schema,
)
@ -17,6 +18,7 @@ __all__ = [
"BaseSource",
"Deployment",
"Environment",
"UVEnvironment",
"MCPServerConfig",
"FileSystemSource",
"generate_schema",

View file

@ -0,0 +1,6 @@
"""Environment configuration for MCP servers."""
from fastmcp.utilities.mcp_server_config.v1.environments.base import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
__all__ = ["Environment", "UVEnvironment"]

View file

@ -0,0 +1,28 @@
from abc import ABC, abstractmethod
from pathlib import Path
from pydantic import BaseModel
class Environment(BaseModel, ABC):
"""Base class for environment configuration."""
@abstractmethod
def build_command(self, command: list[str]) -> list[str]:
"""Build the full command with environment setup.
Args:
command: Base command to wrap with environment setup
Returns:
Full command ready for subprocess execution
"""
pass
async def prepare(self, output_dir: Path | None = None) -> None:
"""Prepare the environment (optional, can be no-op).
Args:
output_dir: Directory for persistent environment setup
"""
pass # Default no-op implementation

View file

@ -0,0 +1,303 @@
import os
import shutil
import subprocess
import sys
from pathlib import Path
from pydantic import Field
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config.v1.environments.base import Environment
logger = get_logger("cli.config")
class UVEnvironment(Environment):
"""Configuration for Python environment setup."""
python: str | None = Field(
default=None,
description="Python version constraint",
examples=["3.10", "3.11", "3.12"],
)
dependencies: list[str] | None = Field(
default=None,
description="Python packages to install with PEP 508 specifiers",
examples=[["fastmcp>=2.0,<3", "httpx", "pandas>=2.0"]],
)
requirements: str | None = Field(
default=None,
description="Path to requirements.txt file",
examples=["requirements.txt", "../requirements/prod.txt"],
)
project: str | None = Field(
default=None,
description="Path to project directory containing pyproject.toml",
examples=[".", "../my-project"],
)
editable: list[str] | None = Field(
default=None,
description="Directories to install in editable mode",
examples=[[".", "../my-package"], ["/path/to/package"]],
)
def build_command(self, command: list[str]) -> list[str]:
"""Build complete uv run command with environment args and command to execute.
Args:
command: Command to execute (e.g., ["fastmcp", "run", "server.py"])
Returns:
Complete command ready for subprocess.run, including "uv" prefix if needed.
If no environment configuration is set, returns the command unchanged.
"""
# If no environment setup is needed, return command as-is
if not self._needs_setup():
return command
args = ["uv", "run"]
# Add project if specified
if self.project:
args.extend(["--project", str(self.project)])
# 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])
# 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:
args.extend(["--with", dep])
# Add requirements file
if self.requirements:
args.extend(["--with-requirements", str(self.requirements)])
# Add editable packages
if self.editable:
for editable_path in self.editable:
args.extend(["--with-editable", str(editable_path)])
# Add the command
args.extend(command)
return args
def run_with_uv(self, command: list[str]) -> None:
"""Execute a command using uv run with this environment configuration.
Args:
command: Command and arguments to execute (e.g., ["fastmcp", "run", "server.py"])
"""
import subprocess
# Build the full uv command
cmd = self.build_command(command)
# Set marker to prevent infinite loops when subprocess calls FastMCP again
env = os.environ | {"FASTMCP_UV_SPAWNED": "1"}
logger.debug(f"Running command: {' '.join(cmd)}")
try:
# Run without capturing output so it flows through naturally
process = subprocess.run(cmd, check=True, env=env)
sys.exit(process.returncode)
except subprocess.CalledProcessError as e:
logger.error(f"Command failed: {e}")
sys.exit(e.returncode)
def _needs_setup(self) -> bool:
"""Check if this environment config requires uv to set up.
Returns:
True if any environment settings require uv run
"""
return any(
[
self.python is not None,
self.dependencies is not None,
self.requirements is not None,
self.project is not None,
self.editable is not None,
]
)
# Backward compatibility aliases
def needs_uv(self) -> bool:
"""Deprecated: Use _needs_setup() internally or check if build_command modifies the command."""
return self._needs_setup()
def build_uv_run_command(self, command: list[str]) -> list[str]:
"""Deprecated: Use build_command() instead."""
return self.build_command(command)
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_setup():
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}")

View file

@ -10,14 +10,13 @@ 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
from typing import TYPE_CHECKING, Any, Literal, TypeAlias, overload
from pydantic import BaseModel, Field, field_validator
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import FileSystemSource
logger = get_logger("cli.config")
@ -27,285 +26,9 @@ FASTMCP_JSON_SCHEMA = "https://gofastmcp.com/public/schemas/fastmcp.json/v1.json
# Type alias for source union (will expand with GitSource, etc in future)
SourceType = FileSystemSource
class Environment(BaseModel):
"""Configuration for Python environment setup."""
python: str | None = Field(
default=None,
description="Python version constraint",
examples=["3.10", "3.11", "3.12"],
)
dependencies: list[str] | None = Field(
default=None,
description="Python packages to install with PEP 508 specifiers",
examples=[["fastmcp>=2.0,<3", "httpx", "pandas>=2.0"]],
)
requirements: str | None = Field(
default=None,
description="Path to requirements.txt file",
examples=["requirements.txt", "../requirements/prod.txt"],
)
project: str | None = Field(
default=None,
description="Path to project directory containing pyproject.toml",
examples=[".", "../my-project"],
)
editable: list[str] | None = Field(
default=None,
description="Directories to install in editable mode",
examples=[[".", "../my-package"], ["/path/to/package"]],
)
def build_uv_run_command(self, command: list[str]) -> list[str]:
"""Build complete uv run command with environment args and command to execute.
Args:
command: Command to execute (e.g., ["fastmcp", "run", "server.py"])
Returns:
Complete command ready for subprocess.run, including "uv" prefix
"""
args = ["uv", "run"]
# Add project if specified
if self.project:
args.extend(["--project", str(self.project)])
# 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])
# 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:
args.extend(["--with", dep])
# Add requirements file
if self.requirements:
args.extend(["--with-requirements", str(self.requirements)])
# Add editable packages
if self.editable:
for editable_path in self.editable:
args.extend(["--with-editable", str(editable_path)])
# Add the command
args.extend(command)
return args
def run_with_uv(self, command: list[str]) -> None:
"""Execute a command using uv run with this environment configuration.
Args:
command: Command and arguments to execute (e.g., ["fastmcp", "run", "server.py"])
"""
import subprocess
import sys
# Build the full uv command
cmd = self.build_uv_run_command(command)
# Set marker to prevent infinite loops when subprocess calls FastMCP again
env = os.environ | {"FASTMCP_UV_SPAWNED": "1"}
logger.debug(f"Running command: {' '.join(cmd)}")
try:
# Run without capturing output so it flows through naturally
process = subprocess.run(cmd, check=True, env=env)
sys.exit(process.returncode)
except subprocess.CalledProcessError as e:
logger.error(f"Command failed: {e}")
sys.exit(e.returncode)
def needs_uv(self) -> bool:
"""Check if this environment config requires uv to set up.
Returns:
True if any environment settings require uv run
"""
return any(
[
self.python is not None,
self.dependencies is not None,
self.requirements is not None,
self.project is not None,
self.editable is not None,
]
)
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}")
SourceType: TypeAlias = FileSystemSource
# Type alias for environment union (will expand with other environments in future)
EnvironmentType: TypeAlias = UVEnvironment
class Deployment(BaseModel):
@ -431,8 +154,8 @@ class MCPServerConfig(BaseModel):
)
# Environment configuration
environment: Environment = Field(
default_factory=lambda: Environment(),
environment: EnvironmentType = Field(
default_factory=lambda: UVEnvironment(),
description="Python environment setup configuration",
)
@ -448,7 +171,7 @@ class MCPServerConfig(BaseModel):
@overload
def __init__(self, *, source: dict | FileSystemSource, **data) -> None: ...
@overload
def __init__(self, *, environment: dict | Environment, **data) -> None: ...
def __init__(self, *, environment: dict | UVEnvironment, **data) -> None: ...
@overload
def __init__(self, *, deployment: dict | Deployment, **data) -> None: ...
def __init__(self, **data) -> None: ...
@ -478,17 +201,17 @@ class MCPServerConfig(BaseModel):
@field_validator("environment", mode="before")
@classmethod
def validate_environment(cls, v: dict | Environment) -> Environment:
def validate_environment(cls, v: dict | UVEnvironment) -> UVEnvironment:
"""Validate and convert environment to Environment.
Accepts:
- Environment instance
- dict that can be converted to Environment
"""
if isinstance(v, Environment):
if isinstance(v, UVEnvironment):
return v
elif isinstance(v, dict):
return Environment(**v) # type: ignore[arg-type]
return UVEnvironment(**v) # type: ignore[arg-type]
else:
raise ValueError("environment must be a dict, Environment instance")
@ -578,7 +301,7 @@ class MCPServerConfig(BaseModel):
# Build environment config if any env args provided
environment = None
if any([python, dependencies, requirements, project, editable]):
environment = Environment(
environment = UVEnvironment(
python=python,
dependencies=dependencies,
requirements=requirements,

View file

@ -162,7 +162,42 @@
"title": "Deployment",
"type": "object"
},
"Environment": {
"FileSystemSource": {
"description": "Source for local Python files.",
"properties": {
"type": {
"const": "filesystem",
"default": "filesystem",
"description": "Source type",
"title": "Type",
"type": "string"
},
"path": {
"description": "Path to Python file containing the server",
"title": "Path",
"type": "string"
},
"entrypoint": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Name of server instance or factory function (a no-arg function that returns a FastMCP server)",
"title": "Entrypoint"
}
},
"required": [
"path"
],
"title": "FileSystemSource",
"type": "object"
},
"UVEnvironment": {
"description": "Configuration for Python environment setup.",
"properties": {
"python": {
@ -266,42 +301,7 @@
"title": "Editable"
}
},
"title": "Environment",
"type": "object"
},
"FileSystemSource": {
"description": "Source for local Python files.",
"properties": {
"type": {
"const": "filesystem",
"default": "filesystem",
"description": "Source type",
"title": "Type",
"type": "string"
},
"path": {
"description": "Path to Python file containing the server",
"title": "Path",
"type": "string"
},
"entrypoint": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Name of server instance or factory function (a no-arg function that returns a FastMCP server)",
"title": "Entrypoint"
}
},
"required": [
"path"
],
"title": "FileSystemSource",
"title": "UVEnvironment",
"type": "object"
}
},
@ -339,7 +339,7 @@
]
},
"environment": {
"$ref": "#/$defs/Environment",
"$ref": "#/$defs/UVEnvironment",
"description": "Python environment setup configuration"
},
"deployment": {

View file

@ -9,9 +9,9 @@ from pydantic import ValidationError
from fastmcp.utilities.mcp_server_config import (
Deployment,
Environment,
MCPServerConfig,
)
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import FileSystemSource
@ -104,7 +104,7 @@ class TestEnvironment:
},
)
cmd = config.environment.build_uv_run_command(["fastmcp", "run", "server.py"])
cmd = config.environment.build_command(["fastmcp", "run", "server.py"])
assert cmd[0] == "uv"
assert cmd[1] == "run"
@ -263,7 +263,7 @@ class TestMCPServerConfig:
assert config.source.path == "server.py"
assert config.source.entrypoint is None
# Environment and deployment are now always present but empty
assert isinstance(config.environment, Environment)
assert isinstance(config.environment, UVEnvironment)
assert isinstance(config.deployment, Deployment)
# Check they have no values set
assert not config.environment.needs_uv()
@ -289,7 +289,7 @@ class TestMCPServerConfig:
assert isinstance(config.source, FileSystemSource)
assert config.source.path == "server.py"
assert config.source.entrypoint is None
assert isinstance(config.environment, Environment)
assert isinstance(config.environment, UVEnvironment)
assert isinstance(config.deployment, Deployment)
def test_from_file(self, tmp_path):
@ -416,7 +416,7 @@ class TestMCPServerConfig:
assert isinstance(config.source, FileSystemSource)
assert config.source.path == "server.py"
# Environment and deployment are now always present but may be empty
assert isinstance(config.environment, Environment)
assert isinstance(config.environment, UVEnvironment)
assert isinstance(config.deployment, Deployment)
# Only environment with values
@ -434,9 +434,9 @@ class TestMCPServerConfig:
config = MCPServerConfig(
source={"path": "server.py"}, deployment={"transport": "http"}
)
assert isinstance(config.environment, Environment)
assert isinstance(config.environment, UVEnvironment)
assert all(
getattr(config.environment, field, None) is None
for field in Environment.model_fields
for field in UVEnvironment.model_fields
)
assert config.deployment.transport == "http"

View file

@ -229,7 +229,7 @@ class TestPathResolution:
# Build UV command
assert config.environment is not None
uv_cmd = config.environment.build_uv_run_command(["fastmcp", "run"])
uv_cmd = config.environment.build_command(["fastmcp", "run"])
# Should include requirements file
assert "--with-requirements" in uv_cmd

View file

@ -6,7 +6,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastmcp.utilities.mcp_server_config import Environment, MCPServerConfig
from fastmcp.utilities.mcp_server_config import MCPServerConfig
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import FileSystemSource
@ -25,7 +26,7 @@ class TestMCPServerConfigPrepare:
"""Test that prepare() calls both prepare_environment and prepare_source."""
config = MCPServerConfig(
source=FileSystemSource(path="server.py"),
environment=Environment(python="3.10"),
environment=UVEnvironment(python="3.10"),
)
await config.prepare()
@ -45,7 +46,7 @@ class TestMCPServerConfigPrepare:
"""Test that prepare() with output_dir calls prepare_environment with it."""
config = MCPServerConfig(
source=FileSystemSource(path="server.py"),
environment=Environment(python="3.10"),
environment=UVEnvironment(python="3.10"),
)
output_path = Path("/tmp/test-env")
@ -66,7 +67,7 @@ class TestMCPServerConfigPrepare:
"""Test that prepare() skips source when skip_source=True."""
config = MCPServerConfig(
source=FileSystemSource(path="server.py"),
environment=Environment(python="3.10"),
environment=UVEnvironment(python="3.10"),
)
await config.prepare(skip_source=True)
@ -79,7 +80,7 @@ class TestMCPServerConfigPrepare:
new_callable=AsyncMock,
)
@patch(
"fastmcp.utilities.mcp_server_config.v1.mcp_server_config.Environment.prepare",
"fastmcp.utilities.mcp_server_config.v1.environments.uv.UVEnvironment.prepare",
new_callable=AsyncMock,
)
async def test_prepare_no_environment_settings(self, mock_env_prepare, mock_src):
@ -104,7 +105,7 @@ class TestEnvironmentPrepare:
"""Test that prepare() raises error when uv is not installed."""
mock_which.return_value = None
env = Environment(python="3.10")
env = UVEnvironment(python="3.10")
with pytest.raises(RuntimeError, match="uv is not installed"):
await env.prepare(tmp_path / "test-env")
@ -115,7 +116,7 @@ class TestEnvironmentPrepare:
"""Test that prepare() does nothing when no settings are configured."""
mock_which.return_value = "/usr/bin/uv"
env = Environment() # No settings
env = UVEnvironment() # No settings
await env.prepare(tmp_path / "test-env")
@ -131,7 +132,7 @@ class TestEnvironmentPrepare:
returncode=0, stdout="Environment cached", stderr=""
)
env = Environment(python="3.10")
env = UVEnvironment(python="3.10")
await env.prepare(tmp_path / "test-env")
@ -150,7 +151,7 @@ class TestEnvironmentPrepare:
mock_which.return_value = "/usr/bin/uv"
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
env = Environment(dependencies=["numpy", "pandas"])
env = UVEnvironment(dependencies=["numpy", "pandas"])
await env.prepare(tmp_path / "test-env")
@ -179,7 +180,7 @@ class TestEnvironmentPrepare:
1, ["uv"], stderr="Package not found"
)
env = Environment(python="3.10")
env = UVEnvironment(python="3.10")
with pytest.raises(RuntimeError, match="Failed to initialize project"):
await env.prepare(tmp_path / "test-env")

View file

@ -9,9 +9,9 @@ import pytest
from fastmcp.cli.run import load_mcp_server_config
from fastmcp.utilities.mcp_server_config import (
Deployment,
Environment,
MCPServerConfig,
)
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import FileSystemSource
@ -56,7 +56,7 @@ def test_load_mcp_server_config(sample_config, monkeypatch):
assert isinstance(config, MCPServerConfig)
assert isinstance(config.source, FileSystemSource)
assert isinstance(config.deployment, Deployment)
assert isinstance(config.environment, Environment)
assert isinstance(config.environment, UVEnvironment)
# Check source - path is not resolved yet, only during load_server
assert config.source.path == "server.py"
@ -262,7 +262,7 @@ def test_environment_config_path_resolution(tmp_path):
config = load_mcp_server_config(config_file)
# Check that UV command is built with resolved paths
uv_cmd = config.environment.build_uv_run_command(["fastmcp", "run", "server.py"])
uv_cmd = config.environment.build_command(["fastmcp", "run", "server.py"])
assert "--with-requirements" in uv_cmd
assert "--project" in uv_cmd

View file

@ -27,9 +27,8 @@ class TestRunWithUv:
cmd = mock_run.call_args[0][0]
env = mock_run.call_args.kwargs.get("env", {})
# With no environment config, the command should be returned unchanged
expected = [
"uv",
"run",
"fastmcp",
"run",
"server.py",
@ -150,9 +149,8 @@ class TestRunWithUv:
assert exc_info.value.code == 0
cmd = mock_run.call_args[0][0]
# With no environment config, no uv run prefix
expected = [
"uv",
"run",
"fastmcp",
"run",
"server.py",

View file

@ -1,6 +1,6 @@
"""Tests for CLI utility functions."""
from fastmcp.utilities.mcp_server_config.v1.mcp_server_config import Environment
from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment
class TestEnvironmentBuildUVRunCommand:
@ -8,16 +8,17 @@ class TestEnvironmentBuildUVRunCommand:
def test_build_uv_run_command_basic(self):
"""Test building basic uv command with no environment config."""
env = Environment()
cmd = env.build_uv_run_command(["fastmcp", "run", "server.py"])
expected = ["uv", "run", "fastmcp", "run", "server.py"]
env = UVEnvironment()
cmd = env.build_command(["fastmcp", "run", "server.py"])
# With no config, the command should be returned unchanged
expected = ["fastmcp", "run", "server.py"]
assert cmd == expected
def test_build_uv_run_command_with_editable(self):
"""Test building uv command with editable package."""
editable_path = "/path/to/package"
env = Environment(editable=[editable_path])
cmd = env.build_uv_run_command(["fastmcp", "run", "server.py"])
env = UVEnvironment(editable=[editable_path])
cmd = env.build_command(["fastmcp", "run", "server.py"])
expected = [
"uv",
"run",
@ -31,8 +32,8 @@ class TestEnvironmentBuildUVRunCommand:
def test_build_uv_run_command_with_packages(self):
"""Test building uv command with additional packages."""
env = Environment(dependencies=["pkg1", "pkg2"])
cmd = env.build_uv_run_command(["fastmcp", "run", "server.py"])
env = UVEnvironment(dependencies=["pkg1", "pkg2"])
cmd = env.build_command(["fastmcp", "run", "server.py"])
expected = [
"uv",
"run",
@ -48,8 +49,8 @@ class TestEnvironmentBuildUVRunCommand:
def test_build_uv_run_command_with_python_version(self):
"""Test building uv command with Python version."""
env = Environment(python="3.10")
cmd = env.build_uv_run_command(["fastmcp", "run", "server.py"])
env = UVEnvironment(python="3.10")
cmd = env.build_command(["fastmcp", "run", "server.py"])
expected = [
"uv",
"run",
@ -64,8 +65,8 @@ class TestEnvironmentBuildUVRunCommand:
def test_build_uv_run_command_with_requirements(self):
"""Test building uv command with requirements file."""
requirements_path = "/path/to/requirements.txt"
env = Environment(requirements=requirements_path)
cmd = env.build_uv_run_command(["fastmcp", "run", "server.py"])
env = UVEnvironment(requirements=requirements_path)
cmd = env.build_command(["fastmcp", "run", "server.py"])
expected = [
"uv",
"run",
@ -80,8 +81,8 @@ class TestEnvironmentBuildUVRunCommand:
def test_build_uv_run_command_with_project(self):
"""Test building uv command with project directory."""
project_path = "/path/to/project"
env = Environment(project=project_path)
cmd = env.build_uv_run_command(["fastmcp", "run", "server.py"])
env = UVEnvironment(project=project_path)
cmd = env.build_command(["fastmcp", "run", "server.py"])
expected = [
"uv",
"run",
@ -97,13 +98,13 @@ class TestEnvironmentBuildUVRunCommand:
"""Test building uv command with all options."""
requirements_path = "/path/to/requirements.txt"
editable_path = "/local/pkg"
env = Environment(
env = UVEnvironment(
python="3.10",
dependencies=["pandas", "numpy"],
requirements=requirements_path,
editable=[editable_path],
)
cmd = env.build_uv_run_command(["fastmcp", "run", "server.py"])
cmd = env.build_command(["fastmcp", "run", "server.py"])
expected = [
"uv",
"run",
@ -129,13 +130,13 @@ class TestEnvironmentBuildUVRunCommand:
def test_build_uv_run_command_project_with_extras(self):
"""Test that project flag works with additional dependencies."""
project_path = "/path/to/project"
env = Environment(
env = UVEnvironment(
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
)
cmd = env.build_uv_run_command(["fastmcp", "run", "server.py"])
cmd = env.build_command(["fastmcp", "run", "server.py"])
expected = [
"uv",
"run",
@ -157,35 +158,35 @@ class TestEnvironmentNeedsUV:
def test_needs_uv_with_python(self):
"""Test that needs_uv returns True with Python version."""
env = Environment(python="3.10")
env = UVEnvironment(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"])
env = UVEnvironment(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")
env = UVEnvironment(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")
env = UVEnvironment(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"])
env = UVEnvironment(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()
env = UVEnvironment()
assert env.needs_uv() is False
def test_needs_uv_with_empty_lists(self):
"""Test that needs_uv returns False with empty lists."""
env = Environment(dependencies=None, editable=None)
env = UVEnvironment(dependencies=None, editable=None)
assert env.needs_uv() is False