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

Merge pull request #1062 from jlowin/cyclopts

Browse source 
Refactor CLI from typer to cyclopts and add comprehensive tests
This commit is contained in:
Jeremiah Lowin 2025-07-07 10:08:57 -04:00 • committed by GitHub
commit 913b5b95dc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 1230 additions and 1647 deletions
Download patch file Download diff file Expand all files Collapse all files

4
pyproject.toml
View file

@ -1,7 +1,7 @@
[project]
name = "fastmcp"
dynamic = ["version"]
description = "The fast, Pythonic way to build MCP servers."
description = "The fast, Pythonic way to build MCP servers and clients."
authors = [{ name = "Jeremiah Lowin" }]
dependencies = [
"python-dotenv>=1.1.0",
@ -10,7 +10,7 @@ dependencies = [
"mcp>=1.10.0",
"openapi-pydantic>=0.5.1",
"rich>=13.9.4",
"typer>=0.15.2",
"cyclopts>=3.0.0",
"authlib>=1.5.2",
"pydantic[email]>=2.11.7",
]

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

@ -1,6 +1,5 @@
"""FastMCP CLI tools."""
"""FastMCP CLI tools using Cyclopts."""
import asyncio
import importlib.metadata
import importlib.util
import os
@ -8,13 +7,12 @@ import platform
import subprocess
import sys
from pathlib import Path
from typing import Annotated
from typing import Annotated, Literal
import typer
import cyclopts
from pydantic import TypeAdapter
from rich.console import Console
from rich.table import Table
from typer import Context, Exit
import fastmcp
from fastmcp.cli import run as run_module
@ -26,11 +24,10 @@ from fastmcp.utilities.logging import get_logger
logger = get_logger("cli")
console = Console()
app = typer.Typer(
app = cyclopts.App(
name="fastmcp",
help="FastMCP CLI",
add_completion=False,
no_args_is_help=True, # Show help if no args provided
help="FastMCP 2.0 - The fast, Pythonic way to build MCP servers and clients.",
version=fastmcp.__version__,
)
@ -87,11 +84,9 @@ def _build_uv_command(
return cmd
@app.command()
def version(ctx: Context):
if ctx.resilient_parsing:
return
@app.command
def version():
"""Display version information and platform details."""
info = {
"FastMCP version": fastmcp.__version__,
"MCP version": importlib.metadata.version("mcp"),
@ -107,56 +102,55 @@ def version(ctx: Context):
g.add_row(k + ":", str(v).replace("\n", " "))
console.print(g)
raise Exit()
sys.exit(0)
@app.command()
@app.command
def dev(
server_spec: str = typer.Argument(
...,
help="Python file to run, optionally with :object suffix",
),
server_spec: str,
*,
with_editable: Annotated[
Path | None,
typer.Option(
"--with-editable",
"-e",
cyclopts.Parameter(
name=["--with-editable", "-e"],
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
cyclopts.Parameter(
"--with",
help="Additional packages to install",
negative=False,
),
] = [],
inspector_version: Annotated[
str | None,
typer.Option(
cyclopts.Parameter(
"--inspector-version",
help="Version of the MCP Inspector to use",
),
] = None,
ui_port: Annotated[
int | None,
typer.Option(
cyclopts.Parameter(
"--ui-port",
help="Port for the MCP Inspector UI",
),
] = None,
server_port: Annotated[
int | None,
typer.Option(
cyclopts.Parameter(
"--server-port",
help="Port for the MCP Inspector Proxy server",
),
] = None,
) -> None:
"""Run a MCP server with the MCP Inspector."""
"""Run an MCP server with the MCP Inspector for development.
Args:
server_spec: Python file to run, optionally with :object suffix
"""
file, server_object = run_module.parse_file_path(server_spec)
logger.debug(
@ -229,66 +223,62 @@ def dev(
sys.exit(1)
@app.command(context_settings={"allow_extra_args": True})
@app.command
def run(
ctx: typer.Context,
server_spec: str = typer.Argument(
...,
help="Python file, object specification (file:obj), or URL",
),
server_spec: str,
*,
transport: Annotated[
str | None,
typer.Option(
"--transport",
"-t",
help="Transport protocol to use (stdio, http, or sse)",
Literal["stdio", "http", "sse"] | None,
cyclopts.Parameter(
name=["--transport", "-t"],
help="Transport protocol to use",
),
] = None,
host: Annotated[
str | None,
typer.Option(
cyclopts.Parameter(
"--host",
help="Host to bind to when using http transport (default: 127.0.0.1)",
),
] = None,
port: Annotated[
int | None,
typer.Option(
"--port",
"-p",
cyclopts.Parameter(
name=["--port", "-p"],
help="Port to bind to when using http transport (default: 8000)",
),
] = None,
log_level: Annotated[
str | None,
typer.Option(
"--log-level",
"-l",
help="Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)",
Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None,
cyclopts.Parameter(
name=["--log-level", "-l"],
help="Log level",
),
] = None,
no_banner: Annotated[
bool,
typer.Option(
cyclopts.Parameter(
"--no-banner",
help="Don't show the server banner",
negative=False,
),
] = False,
) -> None:
"""Run a MCP server or connect to a remote one.
"""Run an MCP server or connect to a remote one.
The server can be specified in three ways:
1. Module approach: server.py - runs the module directly, looking for an object named mcp/server/app.\n
2. Import approach: server.py:app - imports and runs the specified server object.\n
3. URL approach: http://server-url - connects to a remote server and creates a proxy.\n\n
Note: This command runs the server directly. You are responsible for ensuring
all dependencies are available.
1. Module approach: server.py - runs the module directly, looking for an object named 'mcp', 'server', or 'app'
2. Import approach: server.py:app - imports and runs the specified server object
3. URL approach: http://server-url - connects to a remote server and creates a proxy
Server arguments can be passed after -- :
fastmcp run server.py -- --config config.json --debug
Args:
server_spec: Python file, object specification (file:obj), or URL
"""
server_args = ctx.args # extra args after --
# TODO: Handle server_args from extra context
server_args = [] # Will need to handle this with Cyclopts context
logger.debug(
"Running server or client",
@ -323,39 +313,33 @@ def run(
sys.exit(1)
# Add install subcommands
app.add_typer(install_app)
@app.command()
def inspect(
server_spec: str = typer.Argument(
...,
help="Python file to inspect, optionally with :object suffix",
),
@app.command
async def inspect(
server_spec: str,
*,
output: Annotated[
Path,
typer.Option(
"--output",
"-o",
cyclopts.Parameter(
name=["--output", "-o"],
help="Output file path for the JSON report (default: server-info.json)",
),
] = Path("server-info.json"),
) -> None:
"""Inspect a FastMCP server and generate a JSON report.
"""Inspect an MCP server and generate a JSON report.
This command analyzes a FastMCP server (v1.x or v2.x) and generates
a comprehensive JSON report containing information about the server's
name, instructions, version, tools, prompts, resources, templates,
and capabilities.
This command analyzes an MCP server and generates a comprehensive JSON report
containing information about the server's name, instructions, version, tools,
prompts, resources, templates, and capabilities.
Examples:
fastmcp inspect server.py
fastmcp inspect server.py -o report.json
fastmcp inspect server.py:mcp -o analysis.json
fastmcp inspect path/to/server.py:app -o /tmp/server-info.json
"""
Args:
server_spec: Python file to inspect, optionally with :object suffix
"""
# Parse the server specification
file, server_object = run_module.parse_file_path(server_spec)
@ -372,22 +356,8 @@ def inspect(
# Import the server
server = run_module.import_server(file, server_object)
# Get server information
async def get_info():
return await inspect_fastmcp(server)
try:
# Try to use existing event loop if available
asyncio.get_running_loop()
# If there's already a loop running, we need to run in a thread
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(asyncio.run, get_info())
info = future.result()
except RuntimeError:
# No running loop, safe to use asyncio.run
info = asyncio.run(get_info())
# Get server information - using native async support
info = await inspect_fastmcp(server)
info_json = TypeAdapter(FastMCPInfo).dump_json(info, indent=2)
@ -420,3 +390,11 @@ def inspect(
)
console.print(f"[bold red]✗[/bold red] Failed to inspect server: {e}")
sys.exit(1)
# Add install subcommands using proper Cyclopts pattern
app.command(install_app)
if __name__ == "__main__":
app()

25
src/fastmcp/cli/install/__init__.py
View file

@ -1,27 +1,20 @@
"""Install subcommands for FastMCP CLI."""
"""Install subcommands for FastMCP CLI using Cyclopts."""
import typer
import cyclopts
from .claude_code import claude_code_command
from .claude_desktop import claude_desktop_command
from .cursor import cursor_command
from .mcp_config import mcp_config_command
# Create a typer app for install subcommands
install_app = typer.Typer(
# Create a cyclopts app for install subcommands
install_app = cyclopts.App(
name="install",
help="Install MCP servers in various clients and formats",
no_args_is_help=True,
help="Install MCP servers in various clients and formats.",
)
# Register each command from its respective module
install_app.command("claude-code", help="Install a MCP server in Claude Code")(
claude_code_command
)
install_app.command("claude-desktop", help="Install a MCP server in Claude Desktop")(
claude_desktop_command
)
install_app.command("cursor", help="Install a MCP server in Cursor")(cursor_command)
install_app.command(
"mcp-json", help="Generate MCP JSON configuration for manual installation"
)(mcp_config_command)
install_app.command(claude_code_command, name="claude-code")
install_app.command(claude_desktop_command, name="claude-desktop")
install_app.command(cursor_command, name="cursor")
install_app.command(mcp_config_command, name="mcp-json")

59
src/fastmcp/cli/install/claude_code.py
View file

@ -1,13 +1,11 @@
"""Claude Code integration for FastMCP install."""
from __future__ import annotations
"""Claude Code integration for FastMCP install using Cyclopts."""
import subprocess
import sys
from pathlib import Path
from typing import Annotated
import typer
import cyclopts
from rich import print
from fastmcp.utilities.logging import get_logger
@ -124,54 +122,51 @@ def install_claude_code(
def claude_code_command(
server_spec: Annotated[
str, typer.Argument(help="Python file to run, optionally with :object suffix")
],
server_spec: str,
*,
server_name: Annotated[
str | None,
typer.Option(
"--name",
"-n",
help="Custom name for the server (defaults to server's name attribute or file name)",
cyclopts.Parameter(
name=["--server-name", "-n"],
help="Custom name for the server in Claude Code",
),
] = None,
with_editable: Annotated[
Path | None,
typer.Option(
"--with-editable",
"-e",
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
cyclopts.Parameter(
name=["--with-editable", "-e"],
help="Directory with pyproject.toml to install in editable mode",
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
"--with", help="Additional packages to install, in PEP 508 format"
cyclopts.Parameter(
"--with",
help="Additional packages to install",
negative=False,
),
] = [],
env_vars: Annotated[
list[str],
typer.Option(
"--env-var", "-v", help="Environment variables in KEY=VALUE format"
cyclopts.Parameter(
"--env",
help="Environment variables in KEY=VALUE format",
negative=False,
),
] = [],
env_file: Annotated[
Path | None,
typer.Option(
cyclopts.Parameter(
"--env-file",
"-f",
help="Load environment variables from a .env file",
exists=True,
file_okay=True,
dir_okay=False,
resolve_path=True,
help="Load environment variables from .env file",
),
] = None,
) -> None:
"""Install a MCP server in Claude Code."""
"""Install an MCP server in Claude Code.
Args:
server_spec: Python file to install, optionally with :object suffix
"""
file, server_object, name, packages, env_dict = process_common_args(
server_spec, server_name, with_packages, env_vars, env_file
)
@ -186,8 +181,6 @@ def claude_code_command(
)
if success:
print(
f"[green bold]Successfully installed '[bold]{name}[/bold]' in Claude Code[/green bold]"
)
print(f"[green]Successfully installed '{name}' in Claude Code[/green]")
else:
sys.exit(1)

73
src/fastmcp/cli/install/claude_desktop.py
View file

@ -1,13 +1,11 @@
"""Claude Desktop integration for FastMCP install."""
from __future__ import annotations
"""Claude Desktop integration for FastMCP install using Cyclopts."""
import os
import sys
from pathlib import Path
from typing import Annotated
import typer
import cyclopts
from rich import print
from fastmcp.mcp_config import StdioMCPServer, update_config_file
@ -61,7 +59,7 @@ def install_claude_desktop(
config_dir = get_claude_config_path()
if not config_dir:
print(
"[red]❌ Claude Desktop config directory not found.[/red]\n"
"[red]Claude Desktop config directory not found.[/red]\n"
"[blue]Please ensure Claude Desktop is installed and has been run at least once to initialize its config.[/blue]"
)
return False
@ -116,65 +114,62 @@ def install_claude_desktop(
merged_env = existing_env
server_config.env = merged_env
# Update configuration with correct function signature
update_config_file(config_file, name, server_config)
print(f"[green]Successfully installed '{name}' in Claude Desktop[/green]")
return True
except Exception as e:
print(
f"[red]Failed to install '[bold]{name}[/bold]' in Claude Desktop: {e}[/red]"
)
print(f"[red]Failed to install server: {e}[/red]")
return False
def claude_desktop_command(
server_spec: Annotated[
str, typer.Argument(help="Python file to run, optionally with :object suffix")
],
server_spec: str,
*,
server_name: Annotated[
str | None,
typer.Option(
"--name",
"-n",
help="Custom name for the server (defaults to server's name attribute or file name)",
cyclopts.Parameter(
name=["--server-name", "-n"],
help="Custom name for the server in Claude Desktop's config",
),
] = None,
with_editable: Annotated[
Path | None,
typer.Option(
"--with-editable",
"-e",
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
cyclopts.Parameter(
name=["--with-editable", "-e"],
help="Directory with pyproject.toml to install in editable mode",
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
"--with", help="Additional packages to install, in PEP 508 format"
cyclopts.Parameter(
"--with",
help="Additional packages to install",
negative=False,
),
] = [],
env_vars: Annotated[
list[str],
typer.Option(
"--env-var", "-v", help="Environment variables in KEY=VALUE format"
cyclopts.Parameter(
"--env",
help="Environment variables in KEY=VALUE format",
negative=False,
),
] = [],
env_file: Annotated[
Path | None,
typer.Option(
cyclopts.Parameter(
"--env-file",
"-f",
help="Load environment variables from a .env file",
exists=True,
file_okay=True,
dir_okay=False,
resolve_path=True,
help="Load environment variables from .env file",
),
] = None,
) -> None:
"""Install a MCP server in Claude Desktop."""
file, server_object, name, packages, env_dict = process_common_args(
"""Install an MCP server in Claude Desktop.
Args:
server_spec: Python file to install, optionally with :object suffix
"""
file, server_object, name, with_packages, env_dict = process_common_args(
server_spec, server_name, with_packages, env_vars, env_file
)
@ -183,13 +178,9 @@ def claude_desktop_command(
server_object=server_object,
name=name,
with_editable=with_editable,
with_packages=packages,
with_packages=with_packages,
env_vars=env_dict,
)
if success:
print(
f"[green bold]Successfully installed '[bold]{name}[/bold]' in Claude Desktop[/green bold]"
)
else:
if not success:
sys.exit(1)

88
src/fastmcp/cli/install/cursor.py
View file

@ -1,6 +1,4 @@
"""Cursor integration for FastMCP install."""
from __future__ import annotations
"""Cursor integration for FastMCP install using Cyclopts."""
import base64
import subprocess
@ -8,7 +6,7 @@ import sys
from pathlib import Path
from typing import Annotated
import typer
import cyclopts
from rich import print
from fastmcp.mcp_config import StdioMCPServer
@ -33,7 +31,6 @@ def generate_cursor_deeplink(
Deeplink URL that can be clicked to install the server
"""
# Create the configuration structure expected by Cursor
# Base64 encode the configuration (URL-safe for query parameter)
config_json = server_config.model_dump_json(exclude_none=True)
config_b64 = base64.urlsafe_b64encode(config_json.encode()).decode()
@ -81,7 +78,7 @@ def install_cursor(
Args:
file: Path to the server file
server_object: Optional server object name (for :object suffix)
name: Name for the server in Cursor's config
name: Name for the server in Cursor
with_editable: Optional directory to install in editable mode
with_packages: Optional list of additional packages to install
env_vars: Optional dictionary of environment variables
@ -120,75 +117,69 @@ def install_cursor(
env=env_vars or {},
)
# Generate and open deeplink
try:
deeplink = generate_cursor_deeplink(name, server_config)
# Generate deeplink
deeplink = generate_cursor_deeplink(name, server_config)
if open_deeplink(deeplink):
print(
f"[green]Opening Cursor to install '[bold]{name}[/bold]' - please confirm in Cursor to complete installation[/green]"
)
return True
else:
print("[yellow]Could not open Cursor automatically.[/yellow]")
print(f"[blue]Please open this link to install:[/blue] {deeplink}")
return True
print(f"[blue]Opening Cursor to install '{name}'[/blue]")
except Exception as e:
print(f"[red]Failed to generate Cursor deeplink: {e}[/red]")
if open_deeplink(deeplink):
print("[green]Cursor should now open with the installation dialog[/green]")
return True
else:
print(
"[red]Could not open Cursor automatically.[/red]\n"
f"[blue]Please copy this link and open it in Cursor: {deeplink}[/blue]"
)
return False
def cursor_command(
server_spec: Annotated[
str, typer.Argument(help="Python file to run, optionally with :object suffix")
],
server_spec: str,
*,
server_name: Annotated[
str | None,
typer.Option(
"--name",
"-n",
help="Custom name for the server (defaults to server's name attribute or file name)",
cyclopts.Parameter(
name=["--server-name", "-n"],
help="Custom name for the server in Cursor",
),
] = None,
with_editable: Annotated[
Path | None,
typer.Option(
"--with-editable",
"-e",
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
cyclopts.Parameter(
name=["--with-editable", "-e"],
help="Directory with pyproject.toml to install in editable mode",
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
"--with", help="Additional packages to install, in PEP 508 format"
cyclopts.Parameter(
"--with",
help="Additional packages to install",
negative=False,
),
] = [],
env_vars: Annotated[
list[str],
typer.Option(
"--env-var", "-v", help="Environment variables in KEY=VALUE format"
cyclopts.Parameter(
"--env",
help="Environment variables in KEY=VALUE format",
negative=False,
),
] = [],
env_file: Annotated[
Path | None,
typer.Option(
cyclopts.Parameter(
"--env-file",
"-f",
help="Load environment variables from a .env file",
exists=True,
file_okay=True,
dir_okay=False,
resolve_path=True,
help="Load environment variables from .env file",
),
] = None,
) -> None:
"""Install a MCP server in Cursor."""
file, server_object, name, packages, env_dict = process_common_args(
"""Install an MCP server in Cursor.
Args:
server_spec: Python file to install, optionally with :object suffix
"""
file, server_object, name, with_packages, env_dict = process_common_args(
server_spec, server_name, with_packages, env_vars, env_file
)
@ -197,10 +188,9 @@ def cursor_command(
server_object=server_object,
name=name,
with_editable=with_editable,
with_packages=packages,
with_packages=with_packages,
env_vars=env_dict,
)
# Cursor handles its own messaging, no generic success message needed
if not success:
sys.exit(1)

65
src/fastmcp/cli/install/mcp_config.py
View file

@ -1,13 +1,11 @@
"""MCP configuration JSON generation for FastMCP install."""
from __future__ import annotations
"""MCP configuration JSON generation for FastMCP install using Cyclopts."""
import json
import sys
from pathlib import Path
from typing import Annotated
import typer
import cyclopts
from rich import print
from fastmcp.utilities.logging import get_logger
@ -86,11 +84,11 @@ def install_mcp_config(
pyperclip.copy(json_output)
print(
f"[green]MCP configuration for '[bold]{name}[/bold]' copied to clipboard[/green]"
f"[green]MCP configuration for '{name}' copied to clipboard[/green]"
)
except ImportError:
print(
"[red]The `--copy` flag requires pyperclip. Please install pyperclip and try again: `pip install pyperclip`[/red]"
"[red]The --copy flag requires pyperclip. Please install pyperclip and try again: pip install pyperclip[/red]"
)
return False
else:
@ -100,66 +98,64 @@ def install_mcp_config(
return True
except Exception as e:
print(f"[red]❌ Failed to generate MCP configuration: {e}[/red]")
print(f"[red]Failed to generate MCP configuration: {e}[/red]")
return False
def mcp_config_command(
server_spec: Annotated[
str, typer.Argument(help="Python file to run, optionally with :object suffix")
],
server_spec: str,
*,
server_name: Annotated[
str | None,
typer.Option(
"--name",
"-n",
help="Custom name for the server (defaults to server's name attribute or file name)",
cyclopts.Parameter(
name=["--server-name", "-n"],
help="Custom name for the server in MCP config",
),
] = None,
with_editable: Annotated[
Path | None,
typer.Option(
"--with-editable",
"-e",
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
cyclopts.Parameter(
name=["--with-editable", "-e"],
help="Directory with pyproject.toml to install in editable mode",
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
"--with", help="Additional packages to install, in PEP 508 format"
cyclopts.Parameter(
"--with",
help="Additional packages to install",
negative=False,
),
] = [],
env_vars: Annotated[
list[str],
typer.Option(
"--env-var", "-v", help="Environment variables in KEY=VALUE format"
cyclopts.Parameter(
"--env",
help="Environment variables in KEY=VALUE format",
negative=False,
),
] = [],
env_file: Annotated[
Path | None,
typer.Option(
cyclopts.Parameter(
"--env-file",
"-f",
help="Load environment variables from a .env file",
exists=True,
file_okay=True,
dir_okay=False,
resolve_path=True,
help="Load environment variables from .env file",
),
] = None,
copy: Annotated[
bool,
typer.Option(
cyclopts.Parameter(
"--copy",
help="Copy configuration to clipboard instead of printing to stdout",
negative=False,
),
] = False,
) -> None:
"""Generate MCP configuration JSON for manual installation."""
"""Generate MCP configuration JSON for manual installation.
Args:
server_spec: Python file to install, optionally with :object suffix
"""
file, server_object, name, packages, env_dict = process_common_args(
server_spec, server_name, with_packages, env_vars, env_file
)
@ -174,6 +170,5 @@ def mcp_config_command(
copy=copy,
)
# mcp-config handles its own messaging, no generic success message needed
if not success:
sys.exit(1)

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

@ -1,7 +1,5 @@
"""Shared utilities for install commands."""
from __future__ import annotations
import sys
from pathlib import Path
@ -18,7 +16,7 @@ def parse_env_var(env_var: str) -> tuple[str, str]:
"""Parse environment variable string in format KEY=VALUE."""
if "=" not in env_var:
print(
f"[red]❌ Invalid environment variable format: '[bold]{env_var}[/bold]'. Must be KEY=VALUE[/red]"
f"[red]Invalid environment variable format: '[bold]{env_var}[/bold]'. Must be KEY=VALUE[/red]"
)
sys.exit(1)
key, value = env_var.split("=", 1)
@ -76,7 +74,7 @@ def process_common_args(
k: v for k, v in dotenv_values(env_file).items() if v is not None
}
except Exception as e:
print(f"[red]❌ Failed to load .env file: {e}[/red]")
print(f"[red]Failed to load .env file: {e}[/red]")
sys.exit(1)
# Add command line environment variables

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

@ -1,15 +1,19 @@
"""FastMCP run command implementation."""
"""FastMCP run command implementation with enhanced type hints."""
import importlib.util
import re
import sys
from pathlib import Path
from typing import Any
from typing import Any, Literal
from fastmcp.utilities.logging import get_logger
logger = get_logger("cli.run")
# Type aliases for better type safety
TransportType = Literal["stdio", "http", "sse"]
LogLevelType = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
def is_url(path: str) -> bool:
"""Check if a string is a URL."""
@ -164,10 +168,10 @@ def import_server_with_args(
def run_command(
server_spec: str,
transport: str | None = None,
transport: TransportType | None = None,
host: str | None = None,
port: int | None = None,
log_level: str | None = None,
log_level: LogLevelType | None = None,
server_args: list[str] | None = None,
show_banner: bool = True,
) -> None:
@ -180,6 +184,7 @@ def run_command(
port: Port to bind to when using http transport
log_level: Log level
server_args: Additional arguments to pass to the server
show_banner: Whether to show the server banner
"""
if is_url(server_spec):
# Handle URL case

1
tests/cli/__init__.py
View file

@ -0,0 +1 @@
"""CLI test package."""

264
tests/cli/test_claude_code.py
View file

@ -1,264 +0,0 @@
"""Tests for Claude Code CLI integration."""
from pathlib import Path
from unittest.mock import MagicMock, patch
from fastmcp.cli.install.claude_code import (
check_claude_code_available,
find_claude_command,
install_claude_code,
)
class TestFindClaudeCommand:
"""Test find_claude_command function."""
@patch("subprocess.run")
@patch("pathlib.Path.exists")
def test_finds_command_in_default_location(self, mock_exists, mock_run):
"""Should find claude in default installation location."""
mock_exists.return_value = True
mock_run.return_value = MagicMock(stdout="1.0.43 (Claude Code)")
result = find_claude_command()
expected_path = str(Path.home() / ".claude" / "local" / "claude")
assert result == expected_path
mock_run.assert_called_once_with(
[expected_path, "--version"], check=True, capture_output=True, text=True
)
@patch("subprocess.run")
@patch("pathlib.Path.exists")
def test_rejects_non_claude_code_binary(self, mock_exists, mock_run):
"""Should reject binary that isn't Claude Code."""
mock_exists.return_value = True
mock_run.return_value = MagicMock(stdout="Some other claude 1.0.0")
result = find_claude_command()
assert result is None
@patch("subprocess.run")
@patch("pathlib.Path.exists")
def test_handles_subprocess_error(self, mock_exists, mock_run):
"""Should handle subprocess errors gracefully."""
from subprocess import CalledProcessError
mock_exists.return_value = True
mock_run.side_effect = CalledProcessError(1, "claude")
result = find_claude_command()
assert result is None
@patch("pathlib.Path.exists")
def test_no_command_found(self, mock_exists):
"""Should return None when binary doesn't exist."""
mock_exists.return_value = False
result = find_claude_command()
assert result is None
class TestCheckClaudeCodeAvailable:
"""Test check_claude_code_available function."""
@patch("fastmcp.cli.install.claude_code.find_claude_command")
def test_available_when_command_found(self, mock_find):
"""Should return True when claude command is found."""
mock_find.return_value = "/usr/local/bin/claude"
result = check_claude_code_available()
assert result is True
@patch("fastmcp.cli.install.claude_code.find_claude_command")
def test_not_available_when_command_not_found(self, mock_find):
"""Should return False when claude command is not found."""
mock_find.return_value = None
result = check_claude_code_available()
assert result is False
class TestInstallClaudeCode:
"""Test install_claude_code function."""
@patch("fastmcp.cli.install.claude_code.find_claude_command")
@patch("fastmcp.cli.install.claude_code.print")
def test_fails_when_claude_not_found(self, mock_print, mock_find):
"""Should return False and print error when Claude Code CLI not found."""
mock_find.return_value = None
result = install_claude_code(Path("server.py"), None, "test-server")
assert result is False
mock_print.assert_called_once()
assert "Claude Code CLI not found" in str(mock_print.call_args)
@patch("fastmcp.cli.install.claude_code.find_claude_command")
@patch("subprocess.run")
def test_successful_installation(self, mock_run, mock_find):
"""Should successfully install when command succeeds."""
mock_find.return_value = "/usr/local/bin/claude"
mock_run.return_value = MagicMock()
result = install_claude_code(Path("server.py"), None, "test-server")
assert result is True
mock_run.assert_called_once()
# Check the command that was run
call_args = mock_run.call_args[0][0]
assert call_args[0] == "/usr/local/bin/claude"
assert "mcp" in call_args
assert "add" in call_args
assert "test-server" in call_args
assert "--" in call_args
assert "uv" in call_args
@patch("fastmcp.cli.install.claude_code.find_claude_command")
@patch("subprocess.run")
@patch("fastmcp.cli.install.claude_code.print")
def test_handles_subprocess_error(self, mock_print, mock_run, mock_find):
"""Should handle subprocess errors and return False."""
from subprocess import CalledProcessError
mock_find.return_value = "/usr/local/bin/claude"
mock_run.side_effect = CalledProcessError(
1, "claude", stderr="Permission denied"
)
result = install_claude_code(Path("server.py"), None, "test-server")
assert result is False
mock_print.assert_called_once()
assert "Failed to install" in str(mock_print.call_args)
assert "Permission denied" in str(mock_print.call_args)
@patch("fastmcp.cli.install.claude_code.find_claude_command")
@patch("subprocess.run")
def test_builds_correct_command_with_options(self, mock_run, mock_find):
"""Should build correct command with all options."""
mock_find.return_value = "/usr/local/bin/claude"
mock_run.return_value = MagicMock()
install_claude_code(
file=Path("server.py"),
server_object="custom_server",
name="test-server",
with_editable=Path("/path/to/editable"),
with_packages=["pandas", "requests"],
env_vars={"API_KEY": "secret", "DEBUG": "true"},
)
# Check the command that was run
call_args = mock_run.call_args[0][0]
# Should have claude command
assert call_args[0] == "/usr/local/bin/claude"
assert "mcp" in call_args
assert "add" in call_args
# Should have environment variables
assert "-e" in call_args
env_vars = []
for i, arg in enumerate(call_args):
if arg == "-e" and i + 1 < len(call_args):
env_vars.append(call_args[i + 1])
assert "API_KEY=secret" in env_vars
assert "DEBUG=true" in env_vars
# Should have server name
assert "test-server" in call_args
# Should have separator
assert "--" in call_args
# Should have uv command with packages
assert "uv" in call_args
assert "run" in call_args
assert "--with" in call_args
assert "fastmcp" in call_args
assert "pandas" in call_args
assert "requests" in call_args
assert "--with-editable" in call_args
assert str(Path("/path/to/editable")) in call_args
@patch("fastmcp.cli.install.claude_code.find_claude_command")
@patch("subprocess.run")
def test_resolves_absolute_paths(self, mock_run, mock_find):
"""Should resolve server spec to absolute path."""
mock_find.return_value = "/usr/local/bin/claude"
mock_run.return_value = MagicMock()
install_claude_code(Path("server.py"), None, "test-server")
call_args = mock_run.call_args[0][0]
# Find the server spec after "fastmcp run"
server_spec_in_args = None
for i, arg in enumerate(call_args):
if (
arg == "fastmcp"
and i + 2 < len(call_args)
and call_args[i + 1] == "run"
):
server_spec_in_args = call_args[i + 2]
break
assert server_spec_in_args is not None
assert str(Path("server.py").resolve()) in server_spec_in_args
@patch("fastmcp.cli.install.claude_code.find_claude_command")
@patch("subprocess.run")
def test_handles_server_spec_with_object(self, mock_run, mock_find):
"""Should correctly handle server spec with object notation."""
mock_find.return_value = "/usr/local/bin/claude"
mock_run.return_value = MagicMock()
install_claude_code(Path("server.py"), "custom_object", "test-server")
call_args = mock_run.call_args[0][0]
# Find the server spec after "fastmcp run"
server_spec_in_args = None
for i, arg in enumerate(call_args):
if (
arg == "fastmcp"
and i + 2 < len(call_args)
and call_args[i + 1] == "run"
):
server_spec_in_args = call_args[i + 2]
break
assert server_spec_in_args is not None
assert ":custom_object" in server_spec_in_args
assert str(Path("server.py").resolve()) in server_spec_in_args
@patch("fastmcp.cli.install.claude_code.find_claude_command")
@patch("subprocess.run")
def test_deduplicates_packages(self, mock_run, mock_find):
"""Should deduplicate packages in the command."""
mock_find.return_value = "/usr/local/bin/claude"
mock_run.return_value = MagicMock()
install_claude_code(
file=Path("server.py"),
server_object=None,
name="test-server",
with_packages=["pandas", "fastmcp", "pandas"], # duplicates
)
call_args = mock_run.call_args[0][0]
# Count occurrences of pandas
pandas_count = sum(1 for arg in call_args if arg == "pandas")
fastmcp_count = sum(1 for arg in call_args if arg == "fastmcp")
# Should only appear once each for the package (fastmcp appears twice: once as package, once as command)
assert pandas_count == 1
assert fastmcp_count == 2 # Once in --with fastmcp, once in fastmcp run

695
tests/cli/test_cli.py
View file

@ -1,141 +1,66 @@
"""Tests for the CLI module."""
import subprocess
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from unittest.mock import Mock, patch
import pytest
from typer.testing import CliRunner
from fastmcp.cli import cli
# Set up test runner
runner = CliRunner()
from fastmcp.cli.cli import _build_uv_command, _parse_env_var, app
@pytest.fixture
def mock_console():
"""Mock the rich console to test output."""
with patch("fastmcp.cli.cli.console") as mock_console:
yield mock_console
class TestMainCLI:
"""Test the main CLI application."""
@pytest.fixture
def mock_logger():
"""Mock the logger to test logging."""
with patch("fastmcp.cli.cli.logger") as mock_logger:
yield mock_logger
@pytest.fixture
def mock_exit():
"""Mock sys.exit to prevent tests from exiting."""
with patch("sys.exit") as mock_exit:
yield mock_exit
@pytest.fixture
def temp_python_file(tmp_path):
"""Create a temporary Python file with a test server."""
server_code = """
from mcp import Server
class TestServer(Server):
name = "test_server"
dependencies = ["package1", "package2"]
def run(self, **kwargs):
print("Running server with", kwargs)
mcp = TestServer()
server = TestServer()
app = TestServer()
custom_server = TestServer()
"""
file_path = tmp_path / "test_server.py"
file_path.write_text(server_code)
return file_path
@pytest.fixture
def temp_env_file(tmp_path):
"""Create a temporary .env file."""
env_content = """
TEST_VAR1=value1
TEST_VAR2=value2
"""
env_path = tmp_path / ".env"
env_path.write_text(env_content)
return env_path
class TestHelperFunctions:
"""Tests for helper functions in cli.py."""
def test_get_npx_command_unix(self):
"""Test getting npx command on unix systems."""
with patch("sys.platform", "linux"):
with patch("subprocess.run") as mock_run:
mock_run.return_value = Mock(returncode=0)
assert cli._get_npx_command() == "npx"
def test_get_npx_command_windows(self):
"""Test getting npx command on Windows."""
with patch("sys.platform", "win32"):
with patch("subprocess.run") as mock_run:
# First try fails, second succeeds
mock_run.side_effect = [
subprocess.CalledProcessError(1, "npx.cmd"),
Mock(returncode=0),
]
assert cli._get_npx_command() == "npx.exe"
def test_get_npx_command_not_found(self):
"""Test when npx command is not found."""
with patch("sys.platform", "win32"):
with patch("subprocess.run") as mock_run:
mock_run.side_effect = [
subprocess.CalledProcessError(1, "npx.cmd"),
subprocess.CalledProcessError(1, "npx.exe"),
subprocess.CalledProcessError(1, "npx"),
]
assert cli._get_npx_command() is None
def test_app_exists(self):
"""Test that the main app is properly configured."""
# app.name is a tuple in cyclopts
assert "fastmcp" in app.name
assert "FastMCP 2.0" in app.help
# Just check that version exists, not the specific value
assert hasattr(app, "version")
def test_parse_env_var_valid(self):
"""Test parsing valid environment variables."""
assert cli._parse_env_var("KEY=VALUE") == ("KEY", "VALUE")
assert cli._parse_env_var("KEY=") == ("KEY", "")
assert cli._parse_env_var("KEY=VALUE=WITH=EQUALS") == (
"KEY",
"VALUE=WITH=EQUALS",
)
assert cli._parse_env_var(" KEY = VALUE ") == ("KEY", "VALUE")
key, value = _parse_env_var("KEY=value")
assert key == "KEY"
assert value == "value"
key, value = _parse_env_var("COMPLEX_KEY=complex=value=with=equals")
assert key == "COMPLEX_KEY"
assert value == "complex=value=with=equals"
def test_parse_env_var_invalid(self):
"""Test parsing invalid environment variables exits."""
with pytest.raises(SystemExit) as exc_info:
_parse_env_var("INVALID_FORMAT")
assert exc_info.value.code == 1
def test_build_uv_command_basic(self):
"""Test building basic uv command."""
cmd = cli._build_uv_command("file.py")
assert cmd == ["uv", "run", "--with", "fastmcp", "fastmcp", "run", "file.py"]
cmd = _build_uv_command("server.py")
expected = ["uv", "run", "--with", "fastmcp", "fastmcp", "run", "server.py"]
assert cmd == expected
def test_build_uv_command_with_editable(self):
"""Test building uv command with editable flag."""
project_path = Path("/path/to/project")
cmd = cli._build_uv_command("file.py", with_editable=project_path)
assert cmd == [
"""Test building uv command with editable package."""
editable_path = Path("/path/to/package")
cmd = _build_uv_command("server.py", with_editable=editable_path)
expected = [
"uv",
"run",
"--with",
"fastmcp",
"--with-editable",
str(project_path),
str(editable_path),
"fastmcp",
"run",
"file.py",
"server.py",
]
assert cmd == expected
def test_build_uv_command_with_packages(self):
"""Test building uv command with additional packages."""
cmd = cli._build_uv_command("file.py", with_packages=["pkg1", "pkg2"])
assert cmd == [
cmd = _build_uv_command("server.py", with_packages=["pkg1", "pkg2"])
expected = [
"uv",
"run",
"--with",
@ -146,327 +71,305 @@ class TestHelperFunctions:
"pkg2",
"fastmcp",
"run",
"file.py",
"server.py",
]
assert cmd == expected
def test_build_uv_command_full(self):
"""Test building full uv command with all options."""
project_path = Path("/path/to/project")
cmd = cli._build_uv_command(
"file.py:server",
with_editable=project_path,
with_packages=["pkg1", "pkg2"],
)
assert cmd == [
def test_build_uv_command_no_banner(self):
"""Test building uv command with no banner flag."""
cmd = _build_uv_command("server.py", no_banner=True)
expected = [
"uv",
"run",
"--with",
"fastmcp",
"--with-editable",
str(project_path),
"--with",
"pkg1",
"--with",
"pkg2",
"fastmcp",
"run",
"file.py:server",
"server.py",
"--no-banner",
]
assert cmd == expected
class TestVersionCommand:
"""Tests for the version command."""
"""Test the version command."""
def test_version_early_exit_with_resilient_parsing(self):
"""Test version command exits early with resilient parsing."""
ctx = MagicMock()
ctx.resilient_parsing = True
result = cli.version(ctx)
assert result is None
def test_version_command_parsing(self):
"""Test that version command can be parsed."""
command, bound, _ = app.parse_args(["version"])
assert command is not None
def test_version_command_execution(self):
"""Test that version command executes and exits properly."""
# The version command should exit with code 0 when executed
with pytest.raises(SystemExit) as exc_info:
command, bound, _ = app.parse_args(["version"])
command()
assert exc_info.value.code == 0
class TestDevCommand:
"""Tests for the dev command."""
"""Test the dev command."""
def test_dev_command_success(self, temp_python_file, mock_logger):
"""Test successful dev command execution."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
patch("fastmcp.cli.cli._get_npx_command") as mock_get_npx,
patch("fastmcp.cli.cli._build_uv_command") as mock_build_uv,
patch("subprocess.run") as mock_run,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.dependencies = ["extra_dep"]
mock_import.return_value = mock_server
mock_get_npx.return_value = "npx"
mock_build_uv.return_value = ["uv", "command"]
mock_run.return_value = MagicMock(returncode=0)
def test_dev_command_parsing(self):
"""Test that dev command can be parsed with various options."""
# Test basic parsing
command, bound, _ = app.parse_args(["dev", "server.py"])
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
result = runner.invoke(cli.app, ["dev", str(temp_python_file)])
assert result.exit_code == 0
mock_run.assert_called_once()
# Check dependencies were passed correctly with no_banner=True
mock_build_uv.assert_called_once_with(
str(temp_python_file), None, ["extra_dep"], no_banner=True
)
def test_dev_command_with_ui_port(self, temp_python_file):
"""Test dev command with UI port."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
patch("fastmcp.cli.cli._get_npx_command") as mock_get_npx,
patch("fastmcp.cli.cli._build_uv_command") as mock_build_uv,
patch("subprocess.run") as mock_run,
):
mock_parse.return_value = (temp_python_file, None)
mock_import.return_value = MagicMock(dependencies=[])
mock_get_npx.return_value = "npx"
mock_build_uv.return_value = ["uv", "command"]
mock_run.return_value = MagicMock(returncode=0)
result = runner.invoke(
cli.app, ["dev", str(temp_python_file), "--ui-port", "3000"]
)
assert result.exit_code == 0
# Check environment variables were set
env = mock_run.call_args[1]["env"]
assert "CLIENT_PORT" in env
assert env["CLIENT_PORT"] == "3000"
def test_dev_command_with_server_port(self, temp_python_file):
"""Test dev command with server port."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
patch("fastmcp.cli.cli._get_npx_command") as mock_get_npx,
patch("fastmcp.cli.cli._build_uv_command") as mock_build_uv,
patch("subprocess.run") as mock_run,
):
mock_parse.return_value = (temp_python_file, None)
mock_import.return_value = MagicMock(dependencies=[])
mock_get_npx.return_value = "npx"
mock_build_uv.return_value = ["uv", "command"]
mock_run.return_value = MagicMock(returncode=0)
result = runner.invoke(
cli.app, ["dev", str(temp_python_file), "--server-port", "8080"]
)
assert result.exit_code == 0
# Check environment variables were set
env = mock_run.call_args[1]["env"]
assert "SERVER_PORT" in env
assert env["SERVER_PORT"] == "8080"
def test_dev_command_inspector_version(self, temp_python_file):
"""Test dev command with specific inspector version."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
patch("fastmcp.cli.cli._get_npx_command") as mock_get_npx,
patch("fastmcp.cli.cli._build_uv_command") as mock_build_uv,
patch("subprocess.run") as mock_run,
):
mock_parse.return_value = (temp_python_file, None)
mock_import.return_value = MagicMock(dependencies=[])
mock_get_npx.return_value = "npx"
mock_build_uv.return_value = ["uv", "command"]
mock_run.return_value = MagicMock(returncode=0)
result = runner.invoke(
cli.app, ["dev", str(temp_python_file), "--inspector-version", "1.0.0"]
)
assert result.exit_code == 0
# Check inspector version was used
inspector_cmd = mock_run.call_args[0][0][1]
assert inspector_cmd == "@modelcontextprotocol/inspector@1.0.0"
# Test with options
command, bound, _ = app.parse_args(
[
"dev",
"server.py",
"--with",
"package1",
"--inspector-version",
"1.0.0",
"--ui-port",
"3000",
]
)
assert bound.arguments["with_packages"] == ["package1"]
assert bound.arguments["inspector_version"] == "1.0.0"
assert bound.arguments["ui_port"] == 3000
class TestRunCommand:
"""Tests for the run command."""
"""Test the run command."""
def test_run_command_success(self, temp_python_file):
"""Test successful run command execution."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
patch("fastmcp.cli.run.logger") as mock_logger,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
def test_run_command_parsing_basic(self):
"""Test basic run command parsing."""
command, bound, _ = app.parse_args(["run", "server.py"])
result = runner.invoke(cli.app, ["run", str(temp_python_file)])
assert result.exit_code == 0
mock_server.run.assert_called_once_with()
mock_logger.debug.assert_called_with(
f'Found server "test_server" in {temp_python_file}'
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
# Cyclopts only includes non-default values
assert "transport" not in bound.arguments
assert "host" not in bound.arguments
assert "port" not in bound.arguments
assert "log_level" not in bound.arguments
assert "no_banner" not in bound.arguments
def test_run_command_parsing_with_options(self):
"""Test run command parsing with various options."""
command, bound, _ = app.parse_args(
[
"run",
"server.py",
"--transport",
"http",
"--host",
"localhost",
"--port",
"8080",
"--log-level",
"DEBUG",
"--no-banner",
]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["transport"] == "http"
assert bound.arguments["host"] == "localhost"
assert bound.arguments["port"] == 8080
assert bound.arguments["log_level"] == "DEBUG"
assert bound.arguments["no_banner"] is True
def test_run_command_parsing_partial_options(self):
"""Test run command parsing with only some options."""
command, bound, _ = app.parse_args(
[
"run",
"server.py",
"--transport",
"http",
"--no-banner",
]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["transport"] == "http"
assert bound.arguments["no_banner"] is True
# Other options should not be present
assert "host" not in bound.arguments
assert "port" not in bound.arguments
assert "log_level" not in bound.arguments
class TestWindowsSpecific:
"""Test Windows-specific functionality."""
@patch("subprocess.run")
def test_get_npx_command_windows_cmd(self, mock_run):
"""Test npx command detection on Windows with npx.cmd."""
from fastmcp.cli.cli import _get_npx_command
with patch("sys.platform", "win32"):
# First call succeeds with npx.cmd
mock_run.return_value = Mock(returncode=0)
result = _get_npx_command()
assert result == "npx.cmd"
mock_run.assert_called_once_with(
["npx.cmd", "--version"],
check=True,
capture_output=True,
shell=True,
)
def test_run_command_with_transport(self, temp_python_file):
"""Test run command with transport option."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
@patch("subprocess.run")
def test_get_npx_command_windows_exe(self, mock_run):
"""Test npx command detection on Windows with npx.exe."""
from fastmcp.cli.cli import _get_npx_command
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--transport", "sse"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(transport="sse")
with patch("sys.platform", "win32"):
# First call fails, second succeeds
mock_run.side_effect = [
subprocess.CalledProcessError(1, "npx.cmd"),
Mock(returncode=0),
]
def test_run_command_with_http_transports(self, temp_python_file):
"""Test run command with both http and streamable-http transport options."""
# Test "http" transport
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
result = _get_npx_command()
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--transport", "http"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(transport="http")
assert result == "npx.exe"
assert mock_run.call_count == 2
# Test "streamable-http" transport (alias for http)
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
@patch("subprocess.run")
def test_get_npx_command_windows_fallback(self, mock_run):
"""Test npx command detection on Windows with plain npx."""
from fastmcp.cli.cli import _get_npx_command
result = runner.invoke(
cli.app,
["run", str(temp_python_file), "--transport", "streamable-http"],
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(transport="streamable-http")
with patch("sys.platform", "win32"):
# First two calls fail, third succeeds
mock_run.side_effect = [
subprocess.CalledProcessError(1, "npx.cmd"),
subprocess.CalledProcessError(1, "npx.exe"),
Mock(returncode=0),
]
def test_run_command_with_host(self, temp_python_file):
"""Test run command with host option."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
result = _get_npx_command()
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--host", "0.0.0.0"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(host="0.0.0.0")
assert result == "npx"
assert mock_run.call_count == 3
def test_run_command_with_port(self, temp_python_file):
"""Test run command with port option."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
@patch("subprocess.run")
def test_get_npx_command_windows_not_found(self, mock_run):
"""Test npx command detection on Windows when npx is not found."""
from fastmcp.cli.cli import _get_npx_command
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--port", "8080"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(port=8080)
with patch("sys.platform", "win32"):
# All calls fail
mock_run.side_effect = subprocess.CalledProcessError(1, "npx")
def test_run_command_with_log_level(self, temp_python_file):
"""Test run command with log level option."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
result = _get_npx_command()
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--log-level", "DEBUG"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(log_level="DEBUG")
assert result is None
assert mock_run.call_count == 3
def test_run_command_with_multiple_options(self, temp_python_file):
"""Test run command with multiple options."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
@patch("subprocess.run")
def test_get_npx_command_unix(self, mock_run):
"""Test npx command detection on Unix systems."""
from fastmcp.cli.cli import _get_npx_command
result = runner.invoke(
cli.app,
[
"run",
str(temp_python_file),
"--transport",
"sse",
"--host",
"0.0.0.0",
"--port",
"8080",
"--log-level",
"DEBUG",
],
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(
transport="sse", host="0.0.0.0", port=8080, log_level="DEBUG"
)
with patch("sys.platform", "darwin"):
result = _get_npx_command()
def test_run_command_with_server_args(self, temp_python_file):
"""Test run command with server arguments using -- pattern."""
with (
patch("fastmcp.cli.run.run_command") as mock_run_command,
):
result = runner.invoke(
cli.app,
[
"run",
str(temp_python_file),
"--",
"--config",
"config.json",
],
)
assert result.exit_code == 0
mock_run_command.assert_called_once_with(
server_spec=str(temp_python_file),
transport=None,
host=None,
port=None,
log_level=None,
server_args=["--config", "config.json"],
show_banner=True,
)
assert result == "npx"
mock_run.assert_not_called()
def test_windows_path_parsing_with_colon(self, tmp_path):
"""Test parsing Windows paths with drive letters and colons."""
from fastmcp.cli.run import parse_file_path
# Create a real test file to test the logic
test_file = tmp_path / "server.py"
test_file.write_text("# test server")
# Test normal file parsing (works on all platforms)
file_path, obj = parse_file_path(str(test_file))
assert obj is None
# Test file:object parsing
file_path, obj = parse_file_path(f"{test_file}:myapp")
assert obj == "myapp"
# Test that the file portion resolves correctly when object is specified
assert file_path == test_file.resolve()
class TestInspectCommand:
"""Test the inspect command."""
def test_inspect_command_parsing_basic(self):
"""Test basic inspect command parsing."""
command, bound, _ = app.parse_args(["inspect", "server.py"])
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
# Only explicitly set parameters are in bound.arguments
assert "output" not in bound.arguments
def test_inspect_command_parsing_with_output(self, tmp_path):
"""Test inspect command parsing with output file."""
output_file = tmp_path / "output.json"
command, bound, _ = app.parse_args(
[
"inspect",
"server.py",
"--output",
str(output_file),
]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
# Output is parsed as a Path object
assert bound.arguments["output"] == output_file
async def test_inspect_command_with_real_server(self, tmp_path):
"""Test inspect command with a real server file."""
# Create a real server file
server_file = tmp_path / "test_server.py"
server_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("InspectTestServer")
@mcp.tool
def test_tool(x: int) -> int:
return x * 2
@mcp.prompt
def test_prompt(name: str) -> str:
return f"Hello, {name}!"
""")
output_file = tmp_path / "inspect_output.json"
# Parse and execute the command
command, bound, _ = app.parse_args(
[
"inspect",
str(server_file),
"--output",
str(output_file),
]
)
await command(**bound.arguments)
# Verify the output file was created and contains expected content
assert output_file.exists()
content = output_file.read_text()
# Basic checks that the inspection worked
assert "InspectTestServer" in content
assert "test_tool" in content
assert "test_prompt" in content

451
tests/cli/test_cursor.py
View file

@ -1,11 +1,12 @@
"""Tests for Cursor CLI integration."""
import base64
import json
from pathlib import Path
from unittest.mock import patch
from unittest.mock import Mock, patch
import pytest
from fastmcp.cli.install.cursor import (
cursor_command,
generate_cursor_deeplink,
install_cursor,
open_deeplink,
@ -13,15 +14,14 @@ from fastmcp.cli.install.cursor import (
from fastmcp.mcp_config import StdioMCPServer
class TestGenerateCursorDeeplink:
"""Test generate_cursor_deeplink function."""
class TestCursorDeeplinkGeneration:
"""Test cursor deeplink generation functionality."""
def test_generates_valid_deeplink(self):
"""Should generate a valid Cursor deeplink with base64 encoded config."""
def test_generate_deeplink_basic(self):
"""Test basic deeplink generation."""
server_config = StdioMCPServer(
command="uv",
args=["run", "--with", "fastmcp", "fastmcp", "run", "server.py"],
env={"API_KEY": "secret"},
)
deeplink = generate_cursor_deeplink("test-server", server_config)
@ -30,97 +30,151 @@ class TestGenerateCursorDeeplink:
assert "name=test-server" in deeplink
assert "config=" in deeplink
def test_config_is_url_safe_base64(self):
"""Should use URL-safe base64 encoding for the config."""
server_config = StdioMCPServer(
command="test",
args=["arg1", "arg2"],
)
deeplink = generate_cursor_deeplink("test", server_config)
# Extract the config parameter
config_param = deeplink.split("config=")[1]
# Should be decodable as URL-safe base64
decoded = base64.urlsafe_b64decode(config_param.encode())
# Verify base64 encoding
config_part = deeplink.split("config=")[1]
decoded = base64.urlsafe_b64decode(config_part).decode()
config_data = json.loads(decoded)
assert config_data["command"] == "test"
assert config_data["args"] == ["arg1", "arg2"]
assert config_data["command"] == "uv"
assert config_data["args"] == [
"run",
"--with",
"fastmcp",
"fastmcp",
"run",
"server.py",
]
def test_excludes_none_values(self):
"""Should exclude None values from the configuration."""
def test_generate_deeplink_with_env_vars(self):
"""Test deeplink generation with environment variables."""
server_config = StdioMCPServer(
command="test",
args=["arg1"],
timeout=None, # This should be excluded
command="uv",
args=["run", "--with", "fastmcp", "fastmcp", "run", "server.py"],
env={"API_KEY": "secret123", "DEBUG": "true"},
)
deeplink = generate_cursor_deeplink("test", server_config)
config_param = deeplink.split("config=")[1]
decoded = base64.urlsafe_b64decode(config_param.encode())
deeplink = generate_cursor_deeplink("my-server", server_config)
# Decode and verify
config_part = deeplink.split("config=")[1]
decoded = base64.urlsafe_b64decode(config_part).decode()
config_data = json.loads(decoded)
assert "timeout" not in config_data
assert config_data["env"] == {"API_KEY": "secret123", "DEBUG": "true"}
def test_generate_deeplink_special_characters(self):
"""Test deeplink generation with special characters in server name."""
server_config = StdioMCPServer(
command="uv",
args=["run", "--with", "fastmcp", "fastmcp", "run", "server.py"],
)
# Test with spaces and special chars in name
deeplink = generate_cursor_deeplink("my server (test)", server_config)
assert (
"name=my%20server%20%28test%29" in deeplink
or "name=my server (test)" in deeplink
)
def test_generate_deeplink_empty_config(self):
"""Test deeplink generation with minimal config."""
server_config = StdioMCPServer(command="python", args=["server.py"])
deeplink = generate_cursor_deeplink("minimal", server_config)
config_part = deeplink.split("config=")[1]
decoded = base64.urlsafe_b64decode(config_part).decode()
config_data = json.loads(decoded)
assert config_data["command"] == "python"
assert config_data["args"] == ["server.py"]
assert config_data["env"] == {} # Empty env dict is included
def test_generate_deeplink_complex_args(self):
"""Test deeplink generation with complex arguments."""
server_config = StdioMCPServer(
command="uv",
args=[
"run",
"--with",
"fastmcp",
"--with",
"numpy>=1.20",
"--with-editable",
"/path/to/local/package",
"fastmcp",
"run",
"server.py:CustomServer",
],
)
deeplink = generate_cursor_deeplink("complex-server", server_config)
config_part = deeplink.split("config=")[1]
decoded = base64.urlsafe_b64decode(config_part).decode()
config_data = json.loads(decoded)
assert "--with-editable" in config_data["args"]
assert "server.py:CustomServer" in config_data["args"]
class TestOpenDeeplink:
"""Test open_deeplink function."""
"""Test deeplink opening functionality."""
@patch("subprocess.run")
@patch("fastmcp.cli.install.cursor.sys.platform", "darwin")
def test_opens_on_macos(self, mock_run):
"""Should use 'open' command on macOS."""
mock_run.return_value = None
def test_open_deeplink_macos(self, mock_run):
"""Test opening deeplink on macOS."""
with patch("sys.platform", "darwin"):
mock_run.return_value = Mock(returncode=0)
result = open_deeplink("cursor://test")
result = open_deeplink("cursor://test")
assert result is True
mock_run.assert_called_once_with(
["open", "cursor://test"], check=True, capture_output=True
)
assert result is True
mock_run.assert_called_once_with(
["open", "cursor://test"], check=True, capture_output=True
)
@patch("subprocess.run")
@patch("fastmcp.cli.install.cursor.sys.platform", "win32")
def test_opens_on_windows(self, mock_run):
"""Should use 'start' command on Windows."""
mock_run.return_value = None
def test_open_deeplink_windows(self, mock_run):
"""Test opening deeplink on Windows."""
with patch("sys.platform", "win32"):
mock_run.return_value = Mock(returncode=0)
result = open_deeplink("cursor://test")
result = open_deeplink("cursor://test")
assert result is True
mock_run.assert_called_once_with(
["start", "cursor://test"], shell=True, check=True, capture_output=True
)
assert result is True
mock_run.assert_called_once_with(
["start", "cursor://test"], shell=True, check=True, capture_output=True
)
@patch("subprocess.run")
@patch("fastmcp.cli.install.cursor.sys.platform", "linux")
def test_opens_on_linux(self, mock_run):
"""Should use 'xdg-open' command on Linux."""
mock_run.return_value = None
def test_open_deeplink_linux(self, mock_run):
"""Test opening deeplink on Linux."""
with patch("sys.platform", "linux"):
mock_run.return_value = Mock(returncode=0)
result = open_deeplink("cursor://test")
result = open_deeplink("cursor://test")
assert result is True
mock_run.assert_called_once_with(
["xdg-open", "cursor://test"], check=True, capture_output=True
)
assert result is True
mock_run.assert_called_once_with(
["xdg-open", "cursor://test"], check=True, capture_output=True
)
@patch("subprocess.run")
def test_handles_subprocess_error(self, mock_run):
"""Should return False when subprocess command fails."""
from subprocess import CalledProcessError
def test_open_deeplink_failure(self, mock_run):
"""Test handling of deeplink opening failure."""
import subprocess
mock_run.side_effect = CalledProcessError(1, "open")
mock_run.side_effect = subprocess.CalledProcessError(1, ["open"])
result = open_deeplink("cursor://test")
assert result is False
@patch("subprocess.run")
def test_handles_file_not_found(self, mock_run):
"""Should return False when command is not found."""
def test_open_deeplink_command_not_found(self, mock_run):
"""Test handling when open command is not found."""
mock_run.side_effect = FileNotFoundError()
result = open_deeplink("cursor://test")
@ -129,148 +183,167 @@ class TestOpenDeeplink:
class TestInstallCursor:
"""Test install_cursor function."""
"""Test cursor installation functionality."""
@patch("fastmcp.cli.install.cursor.open_deeplink")
@patch("fastmcp.cli.install.cursor.generate_cursor_deeplink")
@patch("fastmcp.cli.install.cursor.print")
def test_successful_installation(
self, mock_print, mock_generate_deeplink, mock_open_deeplink
):
"""Should successfully install when deeplink opens."""
mock_generate_deeplink.return_value = "cursor://test-deeplink"
def test_install_cursor_success(self, mock_print, mock_open_deeplink):
"""Test successful cursor installation."""
mock_open_deeplink.return_value = True
result = install_cursor(Path("server.py"), None, "test-server")
result = install_cursor(
file=Path("/path/to/server.py"),
server_object=None,
name="test-server",
)
assert result is True
mock_generate_deeplink.assert_called_once()
mock_open_deeplink.assert_called_once_with("cursor://test-deeplink")
mock_print.assert_called_once()
# Check that the success message was printed
assert "Opening Cursor to install" in str(mock_print.call_args)
mock_open_deeplink.assert_called_once()
# Verify the deeplink was generated correctly
call_args = mock_open_deeplink.call_args[0][0]
assert call_args.startswith("cursor://anysphere.cursor-deeplink/mcp/install?")
assert "name=test-server" in call_args
@patch("fastmcp.cli.install.cursor.open_deeplink")
@patch("fastmcp.cli.install.cursor.generate_cursor_deeplink")
@patch("fastmcp.cli.install.cursor.print")
def test_fallback_when_deeplink_fails(
self, mock_print, mock_generate_deeplink, mock_open_deeplink
):
"""Should provide manual link when deeplink fails to open."""
mock_generate_deeplink.return_value = "cursor://test-deeplink"
def test_install_cursor_with_packages(self, mock_print, mock_open_deeplink):
"""Test cursor installation with additional packages."""
mock_open_deeplink.return_value = True
result = install_cursor(
file=Path("/path/to/server.py"),
server_object="app",
name="test-server",
with_packages=["numpy", "pandas"],
env_vars={"API_KEY": "test"},
)
assert result is True
call_args = mock_open_deeplink.call_args[0][0]
# Decode the config to verify packages
config_part = call_args.split("config=")[1]
decoded = base64.urlsafe_b64decode(config_part).decode()
config_data = json.loads(decoded)
# Check that all packages are included
assert "--with" in config_data["args"]
assert "numpy" in config_data["args"]
assert "pandas" in config_data["args"]
assert "fastmcp" in config_data["args"]
assert config_data["env"] == {"API_KEY": "test"}
@patch("fastmcp.cli.install.cursor.open_deeplink")
@patch("fastmcp.cli.install.cursor.print")
def test_install_cursor_with_editable(self, mock_print, mock_open_deeplink):
"""Test cursor installation with editable package."""
mock_open_deeplink.return_value = True
result = install_cursor(
file=Path("/path/to/server.py"),
server_object="custom_app",
name="test-server",
with_editable=Path("/local/package"),
)
assert result is True
call_args = mock_open_deeplink.call_args[0][0]
# Decode and verify editable path
config_part = call_args.split("config=")[1]
decoded = base64.urlsafe_b64decode(config_part).decode()
config_data = json.loads(decoded)
assert "--with-editable" in config_data["args"]
# Check for the editable path in a platform-agnostic way
editable_path_str = str(Path("/local/package"))
assert editable_path_str in config_data["args"]
assert "server.py:custom_app" in " ".join(config_data["args"])
@patch("fastmcp.cli.install.cursor.open_deeplink")
@patch("fastmcp.cli.install.cursor.print")
def test_install_cursor_failure(self, mock_print, mock_open_deeplink):
"""Test cursor installation when deeplink fails to open."""
mock_open_deeplink.return_value = False
result = install_cursor(Path("server.py"), None, "test-server")
assert result is True
assert mock_print.call_count == 2
# Check that both error and manual link messages were printed
print_calls = [str(call) for call in mock_print.call_args_list]
assert any(
"Could not open Cursor automatically" in call for call in print_calls
result = install_cursor(
file=Path("/path/to/server.py"),
server_object=None,
name="test-server",
)
assert any("Please open this link" in call for call in print_calls)
@patch("fastmcp.cli.install.cursor.generate_cursor_deeplink")
@patch("fastmcp.cli.install.cursor.print")
def test_handles_deeplink_generation_error(
self, mock_print, mock_generate_deeplink
):
"""Should return False when deeplink generation fails."""
mock_generate_deeplink.side_effect = Exception("Test error")
result = install_cursor(Path("server.py"), None, "test-server")
assert result is False
mock_print.assert_called_once()
assert "Failed to generate Cursor deeplink" in str(mock_print.call_args)
# Verify failure message was printed
mock_print.assert_called()
@patch("fastmcp.cli.install.cursor.open_deeplink")
@patch("fastmcp.cli.install.cursor.generate_cursor_deeplink")
def test_builds_correct_server_config(
self, mock_generate_deeplink, mock_open_deeplink
):
"""Should build correct server configuration with all options."""
mock_generate_deeplink.return_value = "cursor://test"
mock_open_deeplink.return_value = True
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
install_cursor(
file=Path("server.py"),
server_object="custom_server",
name="test-server",
with_editable=Path("/path/to/editable"),
with_packages=["pandas", "requests"],
env_vars={"API_KEY": "secret", "DEBUG": "true"},
install_cursor(
file=Path("/path/to/server.py"),
server_object=None,
name="test-server",
with_packages=["numpy", "fastmcp", "numpy", "pandas", "fastmcp"],
)
call_args = mock_open.call_args[0][0]
config_part = call_args.split("config=")[1]
decoded = base64.urlsafe_b64decode(config_part).decode()
config_data = json.loads(decoded)
# Count occurrences of each package
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
class TestCursorCommand:
"""Test the cursor CLI command."""
@patch("fastmcp.cli.install.cursor.install_cursor")
@patch("fastmcp.cli.install.cursor.process_common_args")
def test_cursor_command_basic(self, mock_process_args, mock_install):
"""Test basic cursor command execution."""
mock_process_args.return_value = (
Path("server.py"),
None,
"test-server",
[],
{},
)
mock_install.return_value = True
# Check that generate_cursor_deeplink was called with correct config
call_args = mock_generate_deeplink.call_args
server_name, server_config = call_args[0]
with patch("sys.exit") as mock_exit:
cursor_command("server.py")
assert server_name == "test-server"
assert server_config.command == "uv"
assert "run" in server_config.args
assert "--with" in server_config.args
assert "fastmcp" in server_config.args
assert "pandas" in server_config.args
assert "requests" in server_config.args
assert "--with-editable" in server_config.args
assert str(Path("/path/to/editable")) in server_config.args
assert "fastmcp" in server_config.args
assert "run" in server_config.args
assert server_config.env == {"API_KEY": "secret", "DEBUG": "true"}
mock_install.assert_called_once_with(
file=Path("server.py"),
server_object=None,
name="test-server",
with_editable=None,
with_packages=[],
env_vars={},
)
mock_exit.assert_not_called()
@patch("fastmcp.cli.install.cursor.open_deeplink")
@patch("fastmcp.cli.install.cursor.generate_cursor_deeplink")
def test_resolves_absolute_paths(self, mock_generate_deeplink, mock_open_deeplink):
"""Should resolve server spec to absolute path."""
mock_generate_deeplink.return_value = "cursor://test"
mock_open_deeplink.return_value = True
@patch("fastmcp.cli.install.cursor.install_cursor")
@patch("fastmcp.cli.install.cursor.process_common_args")
def test_cursor_command_failure(self, mock_process_args, mock_install):
"""Test cursor command when installation fails."""
mock_process_args.return_value = (
Path("server.py"),
None,
"test-server",
[],
{},
)
mock_install.return_value = False
install_cursor(Path("server.py"), None, "test-server")
with pytest.raises(SystemExit) as exc_info:
cursor_command("server.py")
call_args = mock_generate_deeplink.call_args
_, server_config = call_args[0]
# Find the server spec after "fastmcp run"
server_spec_in_args = None
for i, arg in enumerate(server_config.args):
if (
arg == "fastmcp"
and i + 2 < len(server_config.args)
and server_config.args[i + 1] == "run"
):
server_spec_in_args = server_config.args[i + 2]
break
assert server_spec_in_args is not None
assert str(Path("server.py").resolve()) in server_spec_in_args
@patch("fastmcp.cli.install.cursor.open_deeplink")
@patch("fastmcp.cli.install.cursor.generate_cursor_deeplink")
def test_handles_server_spec_with_object(
self, mock_generate_deeplink, mock_open_deeplink
):
"""Should correctly handle server spec with object notation."""
mock_generate_deeplink.return_value = "cursor://test"
mock_open_deeplink.return_value = True
install_cursor(Path("server.py"), "custom_object", "test-server")
call_args = mock_generate_deeplink.call_args
_, server_config = call_args[0]
# Find the server spec after "fastmcp run"
server_spec_in_args = None
for i, arg in enumerate(server_config.args):
if (
arg == "fastmcp"
and i + 2 < len(server_config.args)
and server_config.args[i + 1] == "run"
):
server_spec_in_args = server_config.args[i + 2]
break
assert server_spec_in_args is not None
assert ":custom_object" in server_spec_in_args
assert str(Path("server.py").resolve()) in server_spec_in_args
assert exc_info.value.code == 1

165
tests/cli/test_install.py Normal file
View file

@ -0,0 +1,165 @@
from fastmcp.cli.install import install_app
class TestInstallApp:
"""Test the install subapp."""
def test_install_app_exists(self):
"""Test that the install app is properly configured."""
# install_app.name is a tuple in cyclopts
assert "install" in install_app.name
assert "Install MCP servers" in install_app.help
def test_install_commands_registered(self):
"""Test that all install commands are registered."""
# Check that the app has the expected help text and structure
# This is a simpler check that doesn't rely on internal methods
assert hasattr(install_app, "help")
assert "Install MCP servers" in install_app.help
# We can test that the commands parse without errors
try:
install_app.parse_args(["claude-code", "--help"])
install_app.parse_args(["claude-desktop", "--help"])
install_app.parse_args(["cursor", "--help"])
install_app.parse_args(["mcp-json", "--help"])
except SystemExit:
# Help commands exit with 0, that's expected
pass
class TestClaudeCodeInstall:
"""Test claude-code install command."""
def test_claude_code_basic(self):
"""Test basic claude-code install command parsing."""
# Parse command with correct parameter names
command, bound, _ = install_app.parse_args(
["claude-code", "server.py", "--server-name", "test-server"]
)
# Verify parsing was successful
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["server_name"] == "test-server"
def test_claude_code_with_options(self):
"""Test claude-code install with various options."""
command, bound, _ = install_app.parse_args(
[
"claude-code",
"server.py",
"--server-name",
"test-server",
"--with",
"package1",
"--with",
"package2",
"--env",
"VAR1=value1",
]
)
assert bound.arguments["with_packages"] == ["package1", "package2"]
assert bound.arguments["env_vars"] == ["VAR1=value1"]
class TestClaudeDesktopInstall:
"""Test claude-desktop install command."""
def test_claude_desktop_basic(self):
"""Test basic claude-desktop install command parsing."""
command, bound, _ = install_app.parse_args(
["claude-desktop", "server.py", "--server-name", "test-server"]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["server_name"] == "test-server"
def test_claude_desktop_with_env_vars(self):
"""Test claude-desktop install with environment variables."""
command, bound, _ = install_app.parse_args(
[
"claude-desktop",
"server.py",
"--server-name",
"test-server",
"--env",
"VAR1=value1",
"--env",
"VAR2=value2",
]
)
assert bound.arguments["env_vars"] == ["VAR1=value1", "VAR2=value2"]
class TestCursorInstall:
"""Test cursor install command."""
def test_cursor_basic(self):
"""Test basic cursor install command parsing."""
command, bound, _ = install_app.parse_args(
["cursor", "server.py", "--server-name", "test-server"]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["server_name"] == "test-server"
def test_cursor_with_options(self):
"""Test cursor install with options."""
command, bound, _ = install_app.parse_args(
["cursor", "server.py", "--server-name", "test-server"]
)
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["server_name"] == "test-server"
class TestMcpJsonInstall:
"""Test mcp-json install command."""
def test_mcp_json_basic(self):
"""Test basic mcp-json install command parsing."""
command, bound, _ = install_app.parse_args(
["mcp-json", "server.py", "--server-name", "test-server"]
)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
assert bound.arguments["server_name"] == "test-server"
def test_mcp_json_with_copy(self):
"""Test mcp-json install with copy to clipboard option."""
command, bound, _ = install_app.parse_args(
["mcp-json", "server.py", "--server-name", "test-server", "--copy"]
)
assert bound.arguments["copy"] is True
class TestInstallCommandParsing:
"""Test command parsing and error handling."""
def test_install_minimal_args(self):
"""Test install commands with minimal required arguments."""
# Each command should work with just a server spec
commands_to_test = [
["claude-code", "server.py"],
["claude-desktop", "server.py"],
["cursor", "server.py"],
]
for cmd_args in commands_to_test:
command, bound, _ = install_app.parse_args(cmd_args)
assert command is not None
assert bound.arguments["server_spec"] == "server.py"
def test_mcp_json_minimal(self):
"""Test that mcp-json works with minimal arguments."""
# Should work with just server spec
command, bound, _ = install_app.parse_args(["mcp-json", "server.py"])
assert command is not None
assert bound.arguments["server_spec"] == "server.py"

199
tests/cli/test_mcp_config.py
View file

@ -1,199 +0,0 @@
"""Tests for MCP configuration JSON generation."""
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
from fastmcp.cli.install.mcp_config import install_mcp_config
class TestInstallMcpConfig:
"""Test install_mcp_config function."""
def test_generates_basic_config(self):
"""Should generate basic MCP configuration with minimal options."""
result = install_mcp_config(
file=Path("server.py"),
server_object=None,
name="test-server",
)
assert result is True
@patch("fastmcp.cli.install.mcp_config.print")
def test_generates_config_with_all_options(self, mock_print):
"""Should generate MCP configuration with all options."""
result = install_mcp_config(
file=Path("server.py"),
server_object="custom_server",
name="test-server",
with_editable=Path("/path/to/editable"),
with_packages=["pandas", "requests"],
env_vars={"API_KEY": "secret", "DEBUG": "true"},
)
assert result is True
mock_print.assert_called_once()
# Get the JSON output from print call
json_output = mock_print.call_args[0][0]
config = json.loads(json_output)
# Verify structure (should be just the server config, not wrapped in mcpServers)
server_config = config
# Verify command and args
assert server_config["command"] == "uv"
assert "run" in server_config["args"]
assert "--with" in server_config["args"]
assert "fastmcp" in server_config["args"]
assert "pandas" in server_config["args"]
assert "requests" in server_config["args"]
assert "--with-editable" in server_config["args"]
assert str(Path("/path/to/editable")) in server_config["args"]
# Verify server spec with object
server_spec_in_args = None
for i, arg in enumerate(server_config["args"]):
if (
arg == "fastmcp"
and i + 2 < len(server_config["args"])
and server_config["args"][i + 1] == "run"
):
server_spec_in_args = server_config["args"][i + 2]
break
assert server_spec_in_args is not None
assert ":custom_server" in server_spec_in_args
# Verify environment variables
assert server_config["env"] == {"API_KEY": "secret", "DEBUG": "true"}
@patch("fastmcp.cli.install.mcp_config.print")
def test_generates_config_without_env_vars(self, mock_print):
"""Should generate MCP configuration without env section when no env vars."""
result = install_mcp_config(
file=Path("server.py"),
server_object=None,
name="test-server",
)
assert result is True
json_output = mock_print.call_args[0][0]
config = json.loads(json_output)
# Should not have env section
assert "env" not in config
@patch("fastmcp.cli.install.mcp_config.print")
def test_deduplicates_packages(self, mock_print):
"""Should deduplicate packages including fastmcp."""
result = install_mcp_config(
file=Path("server.py"),
server_object=None,
name="test-server",
with_packages=["pandas", "fastmcp", "pandas"], # duplicates
)
assert result is True
json_output = mock_print.call_args[0][0]
config = json.loads(json_output)
args = config["args"]
# Count occurrences of packages
pandas_count = sum(1 for arg in args if arg == "pandas")
fastmcp_count = sum(1 for arg in args if arg == "fastmcp")
# Should only appear once each for the package (fastmcp appears twice: once as package, once as command)
assert pandas_count == 1
assert fastmcp_count == 2 # Once in --with fastmcp, once in fastmcp run
@patch("fastmcp.cli.install.mcp_config.print")
def test_resolves_absolute_paths(self, mock_print):
"""Should resolve server file to absolute path."""
result = install_mcp_config(
file=Path("server.py"),
server_object=None,
name="test-server",
)
assert result is True
json_output = mock_print.call_args[0][0]
config = json.loads(json_output)
args = config["args"]
# Find the server spec after "fastmcp run"
server_spec_in_args = None
for i, arg in enumerate(args):
if arg == "fastmcp" and i + 2 < len(args) and args[i + 1] == "run":
server_spec_in_args = args[i + 2]
break
assert server_spec_in_args is not None
assert str(Path("server.py").resolve()) in server_spec_in_args
@patch("fastmcp.cli.install.mcp_config.print")
def test_copy_to_clipboard_success(self, mock_print):
"""Should copy configuration to clipboard when copy=True."""
# Mock the pyperclip module at import time
mock_pyperclip = MagicMock()
mock_copy = MagicMock()
mock_pyperclip.copy = mock_copy
with patch.dict("sys.modules", {"pyperclip": mock_pyperclip}):
result = install_mcp_config(
file=Path("server.py"),
server_object=None,
name="test-server",
copy=True,
)
assert result is True
mock_copy.assert_called_once()
# Verify clipboard content is valid JSON
clipboard_content = mock_copy.call_args[0][0]
config = json.loads(clipboard_content) # Should not raise
assert "command" in config # Should be server config, not wrapped
# Should print success message
mock_print.assert_called_once()
assert "copied to clipboard" in str(mock_print.call_args)
@patch("fastmcp.cli.install.mcp_config.print")
def test_copy_to_clipboard_import_error(self, mock_print):
"""Should handle pyperclip import error gracefully."""
with patch(
"builtins.__import__",
side_effect=ImportError("No module named 'pyperclip'"),
):
result = install_mcp_config(
file=Path("server.py"),
server_object=None,
name="test-server",
copy=True,
)
assert result is False
# Should print error message
mock_print.assert_called_once()
error_call = str(mock_print.call_args)
assert "copy` flag requires pyperclip" in error_call
assert "pip install pyperclip" in error_call
@patch("fastmcp.cli.install.mcp_config.print")
def test_handles_exception_gracefully(self, mock_print):
"""Should handle unexpected exceptions gracefully."""
with patch("json.dumps", side_effect=Exception("JSON error")):
result = install_mcp_config(
file=Path("server.py"),
server_object=None,
name="test-server",
)
assert result is False
mock_print.assert_called_once()
assert "Failed to generate MCP configuration" in str(mock_print.call_args)

431
tests/cli/test_run.py
View file

@ -1,298 +1,199 @@
"""Tests for the CLI module."""
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from typer.testing import CliRunner
import fastmcp.cli.run
from fastmcp.cli import cli
# Set up test runner
runner = CliRunner()
from fastmcp.cli.run import (
import_server,
is_url,
parse_file_path,
)
@pytest.fixture
def mock_console():
"""Mock the rich console to test output."""
with patch("fastmcp.cli.cli.console") as mock_console:
yield mock_console
class TestUrlDetection:
"""Test URL detection functionality."""
def test_is_url_valid_http(self):
"""Test detection of valid HTTP URLs."""
assert is_url("http://example.com")
assert is_url("http://localhost:8080")
assert is_url("http://127.0.0.1:3000/path")
def test_is_url_valid_https(self):
"""Test detection of valid HTTPS URLs."""
assert is_url("https://example.com")
assert is_url("https://api.example.com/mcp")
assert is_url("https://localhost:8443")
def test_is_url_invalid(self):
"""Test detection of non-URLs."""
assert not is_url("server.py")
assert not is_url("/path/to/server.py")
assert not is_url("server.py:app")
assert not is_url("ftp://example.com") # Not http/https
assert not is_url("file:///path/to/file")
@pytest.fixture
def mock_logger():
"""Mock the logger to test logging."""
with patch("fastmcp.cli.cli.logger") as mock_logger:
yield mock_logger
class TestFilePathParsing:
"""Test file path parsing functionality."""
def test_parse_file_path_simple(self, tmp_path):
"""Test parsing simple file path without object."""
test_file = tmp_path / "server.py"
test_file.write_text("# test server")
file_path, server_object = parse_file_path(str(test_file))
assert file_path == test_file.resolve()
assert server_object is None
def test_parse_file_path_with_object(self, tmp_path):
"""Test parsing file path with object specification."""
test_file = tmp_path / "server.py"
test_file.write_text("# test server")
file_path, server_object = parse_file_path(f"{test_file}:app")
assert file_path == test_file.resolve()
assert server_object == "app"
def test_parse_file_path_complex_object(self, tmp_path):
"""Test parsing file path with complex object specification."""
test_file = tmp_path / "server.py"
test_file.write_text("# test server")
# The current implementation splits on the last colon, so file:module:app
# becomes file_path="file:module" and server_object="app"
# We need to create a file with a colon in the name for this test
complex_file = tmp_path / "server:module.py"
complex_file.write_text("# test server")
file_path, server_object = parse_file_path(f"{complex_file}:app")
assert file_path == complex_file.resolve()
assert server_object == "app"
def test_parse_file_path_nonexistent(self):
"""Test parsing nonexistent file path exits."""
with pytest.raises(SystemExit) as exc_info:
parse_file_path("nonexistent.py")
assert exc_info.value.code == 1
def test_parse_file_path_directory(self, tmp_path):
"""Test parsing directory path exits."""
with pytest.raises(SystemExit) as exc_info:
parse_file_path(str(tmp_path))
assert exc_info.value.code == 1
@pytest.fixture
def mock_exit():
"""Mock sys.exit to prevent tests from exiting."""
with patch("sys.exit") as mock_exit:
yield mock_exit
class TestServerImport:
"""Test server import functionality using real files."""
async def test_import_server_basic_mcp(self, tmp_path):
"""Test importing server with basic FastMCP server."""
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
@pytest.fixture
def temp_python_file(tmp_path):
"""Create a temporary Python file with a test server."""
server_code = """
from mcp import Server
mcp = fastmcp.FastMCP("TestServer")
class TestServer(Server):
name = "test_server"
dependencies = ["package1", "package2"]
@mcp.tool
def greet(name: str) -> str:
return f"Hello, {name}!"
""")
def run(self, **kwargs):
print("Running server with", kwargs)
server = import_server(test_file)
assert server.name == "TestServer"
tools = await server.get_tools()
assert "greet" in tools
mcp = TestServer()
server = TestServer()
app = TestServer()
custom_server = TestServer()
"""
file_path = tmp_path / "test_server.py"
file_path.write_text(server_code)
return file_path
async def test_import_server_with_main_block(self, tmp_path):
"""Test importing server with if __name__ == '__main__' block."""
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
app = fastmcp.FastMCP("MainServer")
@pytest.fixture
def temp_env_file(tmp_path):
"""Create a temporary .env file."""
env_content = """
TEST_VAR1=value1
TEST_VAR2=value2
"""
env_path = tmp_path / ".env"
env_path.write_text(env_content)
return env_path
@app.tool
def calculate(x: int, y: int) -> int:
return x + y
if __name__ == "__main__":
app.run()
""")
class TestHelperFunctions:
def test_parse_file_path_simple(self):
"""Test parsing simple file path."""
with (
patch("pathlib.Path.exists") as mock_exists,
patch("pathlib.Path.is_file") as mock_is_file,
patch("pathlib.Path.expanduser") as mock_expanduser,
patch("pathlib.Path.resolve") as mock_resolve,
):
mock_exists.return_value = True
mock_is_file.return_value = True
mock_expanduser.return_value = Path("file.py")
mock_resolve.return_value = Path("file.py")
server = import_server(test_file)
assert server.name == "MainServer"
tools = await server.get_tools()
assert "calculate" in tools
path, obj = fastmcp.cli.run.parse_file_path("file.py")
assert path == Path("file.py")
assert obj is None
def test_import_server_standard_names(self, tmp_path):
"""Test automatic detection of standard names (mcp, server, app)."""
# Test with 'mcp' name
mcp_file = tmp_path / "mcp_server.py"
mcp_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("MCPServer")
""")
def test_parse_file_path_with_object(self):
"""Test parsing file path with object."""
with (
patch("pathlib.Path.exists") as mock_exists,
patch("pathlib.Path.is_file") as mock_is_file,
patch("pathlib.Path.expanduser") as mock_expanduser,
patch("pathlib.Path.resolve") as mock_resolve,
):
mock_exists.return_value = True
mock_is_file.return_value = True
mock_expanduser.return_value = Path("file.py")
mock_resolve.return_value = Path("file.py")
server = import_server(mcp_file)
assert server.name == "MCPServer"
path, obj = fastmcp.cli.run.parse_file_path("file.py:server")
assert path == Path("file.py")
assert obj == "server"
# Test with 'server' name
server_file = tmp_path / "server_server.py"
server_file.write_text("""
import fastmcp
server = fastmcp.FastMCP("ServerServer")
""")
def test_parse_file_path_windows(self):
"""Test parsing Windows file path."""
with (
patch("pathlib.Path.exists") as mock_exists,
patch("pathlib.Path.is_file") as mock_is_file,
patch("pathlib.Path.expanduser") as mock_expanduser,
patch("pathlib.Path.resolve") as mock_resolve,
):
mock_exists.return_value = True
mock_is_file.return_value = True
mock_expanduser.return_value = Path("C:/path/file.py")
mock_resolve.return_value = Path("C:/path/file.py")
server = import_server(server_file)
assert server.name == "ServerServer"
path, obj = fastmcp.cli.run.parse_file_path("C:/path/file.py:server")
assert path == Path("C:/path/file.py")
assert obj == "server"
# Test with 'app' name
app_file = tmp_path / "app_server.py"
app_file.write_text("""
import fastmcp
app = fastmcp.FastMCP("AppServer")
""")
def test_parse_file_path_not_file(self, mock_exit):
"""Test parsing path that is not a file."""
with (
patch("pathlib.Path.exists") as mock_exists,
patch("pathlib.Path.is_file") as mock_is_file,
patch("pathlib.Path.expanduser") as mock_expanduser,
patch("pathlib.Path.resolve") as mock_resolve,
patch("fastmcp.cli.run.logger") as mock_logger,
):
mock_exists.return_value = True
mock_is_file.return_value = False
mock_expanduser.return_value = Path("directory")
mock_resolve.return_value = Path("directory")
server = import_server(app_file)
assert server.name == "AppServer"
fastmcp.cli.run.parse_file_path("directory")
mock_logger.error.assert_called_once()
mock_exit.assert_called_once_with(1)
async def test_import_server_nonstandard_name(self, tmp_path):
"""Test importing server with non-standard object name."""
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
my_custom_server = fastmcp.FastMCP("CustomServer")
class TestRunCommand:
"""Tests for the run command."""
@my_custom_server.tool
def custom_tool() -> str:
return "custom"
""")
def test_run_command_success(self, temp_python_file):
"""Test successful run command execution."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
patch("fastmcp.cli.run.logger") as mock_logger,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
server = import_server(test_file, "my_custom_server")
assert server.name == "CustomServer"
tools = await server.get_tools()
assert "custom_tool" in tools
result = runner.invoke(cli.app, ["run", str(temp_python_file)])
assert result.exit_code == 0
mock_server.run.assert_called_once_with()
mock_logger.debug.assert_called_with(
f'Found server "test_server" in {temp_python_file}'
)
def test_import_server_no_standard_names_fails(self, tmp_path):
"""Test importing server when no standard names exist fails."""
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
def test_run_command_with_transport(self, temp_python_file):
"""Test run command with transport option."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
other_name = fastmcp.FastMCP("OtherServer")
""")
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--transport", "sse"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(transport="sse")
with pytest.raises(SystemExit) as exc_info:
import_server(test_file)
assert exc_info.value.code == 1
def test_run_command_with_host(self, temp_python_file):
"""Test run command with host option."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
def test_import_server_nonexistent_object_fails(self, tmp_path):
"""Test importing nonexistent server object fails."""
test_file = tmp_path / "server.py"
test_file.write_text("""
import fastmcp
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--host", "0.0.0.0"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(host="0.0.0.0")
mcp = fastmcp.FastMCP("TestServer")
""")
def test_run_command_with_port(self, temp_python_file):
"""Test run command with port option."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--port", "8080"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(port=8080)
def test_run_command_with_log_level(self, temp_python_file):
"""Test run command with log level option."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
result = runner.invoke(
cli.app, ["run", str(temp_python_file), "--log-level", "DEBUG"]
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(log_level="DEBUG")
def test_run_command_with_multiple_options(self, temp_python_file):
"""Test run command with multiple options."""
with (
patch("fastmcp.cli.run.parse_file_path") as mock_parse,
patch("fastmcp.cli.run.import_server") as mock_import,
):
mock_parse.return_value = (temp_python_file, None)
mock_server = MagicMock()
mock_server.name = "test_server"
mock_import.return_value = mock_server
result = runner.invoke(
cli.app,
[
"run",
str(temp_python_file),
"--transport",
"sse",
"--host",
"0.0.0.0",
"--port",
"8080",
"--log-level",
"DEBUG",
],
)
assert result.exit_code == 0
mock_server.run.assert_called_once_with(
transport="sse", host="0.0.0.0", port=8080, log_level="DEBUG"
)
class TestImportServerWithArgs:
"""Tests for the import_server_with_args function."""
def test_import_server_with_args_no_args(self, temp_python_file):
"""Test importing server without arguments."""
with patch("fastmcp.cli.run.import_server") as mock_import:
mock_server = MagicMock()
mock_import.return_value = mock_server
result = fastmcp.cli.run.import_server_with_args(
temp_python_file, None, None
)
assert result == mock_server
mock_import.assert_called_once_with(temp_python_file, None)
def test_import_server_with_args_with_args(self, temp_python_file):
"""Test importing server with arguments."""
import sys
with patch("fastmcp.cli.run.import_server") as mock_import:
mock_server = MagicMock()
mock_import.return_value = mock_server
original_argv = sys.argv[:]
result = fastmcp.cli.run.import_server_with_args(
temp_python_file, "custom_server", ["--config", "test.json", "--debug"]
)
assert result == mock_server
mock_import.assert_called_once_with(temp_python_file, "custom_server")
# Verify sys.argv was restored
assert sys.argv == original_argv
with pytest.raises(SystemExit) as exc_info:
import_server(test_file, "nonexistent")
assert exc_info.value.code == 1

29
tests/cli/test_shared.py Normal file
View file

@ -0,0 +1,29 @@
from fastmcp.cli.cli import _parse_env_var
class TestEnvVarParsing:
"""Test environment variable parsing functionality."""
def test_parse_env_var_simple(self):
"""Test parsing simple environment variable."""
key, value = _parse_env_var("API_KEY=secret123")
assert key == "API_KEY"
assert value == "secret123"
def test_parse_env_var_with_equals_in_value(self):
"""Test parsing env var with equals signs in the value."""
key, value = _parse_env_var("DATABASE_URL=postgresql://user:pass@host:5432/db")
assert key == "DATABASE_URL"
assert value == "postgresql://user:pass@host:5432/db"
def test_parse_env_var_with_spaces(self):
"""Test parsing env var with spaces (should be stripped)."""
key, value = _parse_env_var(" API_KEY = secret with spaces ")
assert key == "API_KEY"
assert value == "secret with spaces"
def test_parse_env_var_empty_value(self):
"""Test parsing env var with empty value."""
key, value = _parse_env_var("EMPTY_VAR=")
assert key == "EMPTY_VAR"
assert value == ""

12
tests/contrib/test_component_manager.py
View file

@ -566,7 +566,6 @@ class TestComponentManagerWithPath:
def client_with_path(self, mcp_with_path):
return TestClient(mcp_with_path.http_app())
@pytest.mark.asyncio
async def test_enable_tool_route_with_path(self, client_with_path, mcp_with_path):
tool = await mcp_with_path._tool_manager.get_tool("test_tool")
tool.enabled = False
@ -576,7 +575,6 @@ class TestComponentManagerWithPath:
tool = await mcp_with_path._tool_manager.get_tool("test_tool")
assert tool.enabled is True
@pytest.mark.asyncio
async def test_disable_resource_route_with_path(
self, client_with_path, mcp_with_path
):
@ -592,7 +590,6 @@ class TestComponentManagerWithPath:
)
assert resource.enabled is False
@pytest.mark.asyncio
async def test_enable_prompt_route_with_path(self, client_with_path, mcp_with_path):
prompt = await mcp_with_path._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
@ -646,7 +643,6 @@ class TestComponentManagerWithPathAuth:
self.client = TestClient(self.mcp.http_app())
@pytest.mark.asyncio
async def test_unauthorized_enable_tool(self):
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
@ -654,7 +650,6 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 401
assert tool.enabled is False
@pytest.mark.asyncio
async def test_forbidden_enable_tool(self):
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
@ -665,7 +660,6 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 403
assert tool.enabled is False
@pytest.mark.asyncio
async def test_authorized_enable_tool(self):
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
@ -678,7 +672,6 @@ class TestComponentManagerWithPathAuth:
tool = await self.mcp._tool_manager.get_tool("test_tool")
assert tool.enabled is True
@pytest.mark.asyncio
async def test_unauthorized_disable_resource(self):
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
@ -686,7 +679,6 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 401
assert resource.enabled is True
@pytest.mark.asyncio
async def test_forbidden_disable_resource(self):
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
@ -697,7 +689,6 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 403
assert resource.enabled is True
@pytest.mark.asyncio
async def test_authorized_disable_resource(self):
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
@ -710,7 +701,6 @@ class TestComponentManagerWithPathAuth:
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
assert resource.enabled is False
@pytest.mark.asyncio
async def test_unauthorized_enable_prompt(self):
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
@ -718,7 +708,6 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 401
assert prompt.enabled is False
@pytest.mark.asyncio
async def test_forbidden_enable_prompt(self):
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
@ -729,7 +718,6 @@ class TestComponentManagerWithPathAuth:
assert response.status_code == 403
assert prompt.enabled is False
@pytest.mark.asyncio
async def test_authorized_enable_prompt(self):
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False

4
tests/server/openapi/test_explode_integration.py
View file

@ -7,7 +7,6 @@ specifications and properly applied during HTTP request serialization.
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from fastmcp.server.openapi import OpenAPITool
from fastmcp.utilities.openapi import parse_openapi_to_http_routes
@ -130,7 +129,6 @@ class TestExplodeIntegration:
f"Expected explode=None, got {parameter.explode}"
)
@pytest.mark.asyncio
async def test_explode_false_request_serialization(self):
"""Test that explode=false results in comma-separated query parameters in HTTP requests.
@ -201,7 +199,6 @@ class TestExplodeIntegration:
f"Expected 'red,blue,green', got '{tags_value}'"
)
@pytest.mark.asyncio
async def test_explode_true_request_serialization(self):
"""Test that explode=true results in separate query parameters in HTTP requests."""
openapi_spec = {
@ -262,7 +259,6 @@ class TestExplodeIntegration:
f"Expected ['red', 'blue', 'green'], got {tags_value}"
)
@pytest.mark.asyncio
async def test_explode_default_request_serialization(self):
"""Test that default behavior (no explode) uses explode=true for query parameters."""
openapi_spec = {

51
uv.lock generated
View file

@ -351,6 +351,22 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/99/49/0ab9774f64555a1b50102757811508f5ace451cf5dc0a2d074a4b9deca6a/cryptography-45.0.4-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:bbc505d1dc469ac12a0a064214879eac6294038d6b24ae9f71faae1448a9608d", size = 3337594, upload-time = "2025-06-10T00:03:45.523Z" },
]
[[package]]
name = "cyclopts"
version = "3.22.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "docstring-parser", marker = "python_full_version < '4.0'" },
{ name = "rich" },
{ name = "rich-rst" },
{ name = "typing-extensions", marker = "python_full_version < '3.11'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/4a/d2/3f81aa0852d0a71b8d7614f355cf72655ea26f33dd1ddc01e01ddb41a0d0/cyclopts-3.22.1.tar.gz", hash = "sha256:4f42c9427f1e31f598c8416d88e37040ad783a177fa496541e53a8650bed1261", size = 74470, upload-time = "2025-07-03T00:35:25.094Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c5/f2/0155fe8b06890aec0493ee5862aef2635729997da470d2fe0fb18951131e/cyclopts-3.22.1-py3-none-any.whl", hash = "sha256:1ce307fd835f93dcd5dd5e77fff1d91c9a092bc0126f846b24e9e4740b6ea3c3", size = 84534, upload-time = "2025-07-03T00:35:23.917Z" },
]
[[package]]
name = "decorator"
version = "5.2.1"
@ -387,6 +403,24 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/68/1b/e0a87d256e40e8c888847551b20a017a6b98139178505dc7ffb96f04e954/dnspython-2.7.0-py3-none-any.whl", hash = "sha256:b4c34b7d10b51bcc3a5071e7b8dee77939f1e878477eeecc965e9835f63c6c86", size = 313632, upload-time = "2024-10-05T20:14:57.687Z" },
]
[[package]]
name = "docstring-parser"
version = "0.16"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/08/12/9c22a58c0b1e29271051222d8906257616da84135af9ed167c9e28f85cb3/docstring_parser-0.16.tar.gz", hash = "sha256:538beabd0af1e2db0146b6bd3caa526c35a34d61af9fd2887f3a8a27a739aa6e", size = 26565, upload-time = "2024-03-15T10:39:44.419Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d5/7c/e9fcff7623954d86bdc17782036cbf715ecab1bec4847c008557affe1ca8/docstring_parser-0.16-py3-none-any.whl", hash = "sha256:bf0a1387354d3691d102edef7ec124f219ef639982d096e26e3b60aeffa90637", size = 36533, upload-time = "2024-03-15T10:39:41.527Z" },
]
[[package]]
name = "docutils"
version = "0.21.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/ae/ed/aefcc8cd0ba62a0560c3c18c33925362d46c6075480bfa4df87b28e169a9/docutils-0.21.2.tar.gz", hash = "sha256:3a6b18732edf182daa3cd12775bbb338cf5691468f91eeeb109deff6ebfa986f", size = 2204444, upload-time = "2024-04-23T18:57:18.24Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/8f/d7/9322c609343d929e75e7e5e6255e614fcc67572cfd083959cdef3b7aad79/docutils-0.21.2-py3-none-any.whl", hash = "sha256:dafca5b9e384f0e419294eb4d2ff9fa826435bf15f15b7bd45723e8ad76811b2", size = 587408, upload-time = "2024-04-23T18:57:14.835Z" },
]
[[package]]
name = "email-validator"
version = "2.2.0"
@ -462,6 +496,7 @@ name = "fastmcp"
source = { editable = "." }
dependencies = [
{ name = "authlib" },
{ name = "cyclopts" },
{ name = "exceptiongroup" },
{ name = "httpx" },
{ name = "mcp" },
@ -469,7 +504,6 @@ dependencies = [
{ name = "pydantic", extra = ["email"] },
{ name = "python-dotenv" },
{ name = "rich" },
{ name = "typer" },
]
[package.optional-dependencies]
@ -504,6 +538,7 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "authlib", specifier = ">=1.5.2" },
{ name = "cyclopts", specifier = ">=3.0.0" },
{ name = "exceptiongroup", specifier = ">=1.2.2" },
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "mcp", specifier = ">=1.10.0" },
@ -511,7 +546,6 @@ requires-dist = [
{ name = "pydantic", extras = ["email"], specifier = ">=2.11.7" },
{ name = "python-dotenv", specifier = ">=1.1.0" },
{ name = "rich", specifier = ">=13.9.4" },
{ name = "typer", specifier = ">=0.15.2" },
{ name = "websockets", marker = "extra == 'websockets'", specifier = ">=15.0.1" },
]
provides-extras = ["websockets"]
@ -1459,6 +1493,19 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/0d/9b/63f4c7ebc259242c89b3acafdb37b41d1185c07ff0011164674e9076b491/rich-14.0.0-py3-none-any.whl", hash = "sha256:1c9491e1951aac09caffd42f448ee3d04e58923ffe14993f6e83068dc395d7e0", size = 243229, upload-time = "2025-03-30T14:15:12.283Z" },
]
[[package]]
name = "rich-rst"
version = "1.3.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "docutils" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b0/69/5514c3a87b5f10f09a34bb011bc0927bc12c596c8dae5915604e71abc386/rich_rst-1.3.1.tar.gz", hash = "sha256:fad46e3ba42785ea8c1785e2ceaa56e0ffa32dbe5410dec432f37e4107c4f383", size = 13839, upload-time = "2024-04-30T04:40:38.125Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fd/bc/cc4e3dbc5e7992398dcb7a8eda0cbcf4fb792a0cdb93f857b478bf3cf884/rich_rst-1.3.1-py3-none-any.whl", hash = "sha256:498a74e3896507ab04492d326e794c3ef76e7cda078703aa592d1853d91098c1", size = 11621, upload-time = "2024-04-30T04:40:32.619Z" },
]
[[package]]
name = "rpds-py"
version = "0.25.1"