From 919d7e35efa6fe531d2a361a2b75fd9e7f1ac95d Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 6 Jul 2025 20:37:35 -0400 Subject: [PATCH] Refactor from typer to cyclopts --- pyproject.toml | 4 +- src/fastmcp/cli/cli.py | 168 ++++---- src/fastmcp/cli/install/__init__.py | 25 +- src/fastmcp/cli/install/claude_code.py | 59 ++- src/fastmcp/cli/install/claude_desktop.py | 73 ++-- src/fastmcp/cli/install/cursor.py | 88 ++-- src/fastmcp/cli/install/mcp_config.py | 65 ++- src/fastmcp/cli/install/shared.py | 6 +- src/fastmcp/cli/run.py | 13 +- tests/cli/__init__.py | 0 tests/cli/test_claude_code.py | 264 ------------ tests/cli/test_cli.py | 472 ---------------------- tests/cli/test_cursor.py | 276 ------------- tests/cli/test_mcp_config.py | 199 --------- tests/cli/test_run.py | 298 -------------- uv.lock | 51 ++- 16 files changed, 271 insertions(+), 1790 deletions(-) delete mode 100644 tests/cli/__init__.py delete mode 100644 tests/cli/test_claude_code.py delete mode 100644 tests/cli/test_cli.py delete mode 100644 tests/cli/test_cursor.py delete mode 100644 tests/cli/test_mcp_config.py delete mode 100644 tests/cli/test_run.py diff --git a/pyproject.toml b/pyproject.toml index f5751427f..9ad398f6c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", ] diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py index 0be4d0ffe..f2dcec314 100644 --- a/src/fastmcp/cli/cli.py +++ b/src/fastmcp/cli/cli.py @@ -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() diff --git a/src/fastmcp/cli/install/__init__.py b/src/fastmcp/cli/install/__init__.py index 0ccd43606..78b788628 100644 --- a/src/fastmcp/cli/install/__init__.py +++ b/src/fastmcp/cli/install/__init__.py @@ -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") diff --git a/src/fastmcp/cli/install/claude_code.py b/src/fastmcp/cli/install/claude_code.py index df3eb5c0c..bec3db35e 100644 --- a/src/fastmcp/cli/install/claude_code.py +++ b/src/fastmcp/cli/install/claude_code.py @@ -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) diff --git a/src/fastmcp/cli/install/claude_desktop.py b/src/fastmcp/cli/install/claude_desktop.py index f60eb8301..f23ecbd11 100644 --- a/src/fastmcp/cli/install/claude_desktop.py +++ b/src/fastmcp/cli/install/claude_desktop.py @@ -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) diff --git a/src/fastmcp/cli/install/cursor.py b/src/fastmcp/cli/install/cursor.py index 48e7c36b9..6bdd49c65 100644 --- a/src/fastmcp/cli/install/cursor.py +++ b/src/fastmcp/cli/install/cursor.py @@ -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) diff --git a/src/fastmcp/cli/install/mcp_config.py b/src/fastmcp/cli/install/mcp_config.py index 8eee79a86..f96eed10e 100644 --- a/src/fastmcp/cli/install/mcp_config.py +++ b/src/fastmcp/cli/install/mcp_config.py @@ -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) diff --git a/src/fastmcp/cli/install/shared.py b/src/fastmcp/cli/install/shared.py index a6553c4f9..a56bc595b 100644 --- a/src/fastmcp/cli/install/shared.py +++ b/src/fastmcp/cli/install/shared.py @@ -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 diff --git a/src/fastmcp/cli/run.py b/src/fastmcp/cli/run.py index ea139b636..4a544347b 100644 --- a/src/fastmcp/cli/run.py +++ b/src/fastmcp/cli/run.py @@ -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 diff --git a/tests/cli/__init__.py b/tests/cli/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/cli/test_claude_code.py b/tests/cli/test_claude_code.py deleted file mode 100644 index d39ba3732..000000000 --- a/tests/cli/test_claude_code.py +++ /dev/null @@ -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 diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py deleted file mode 100644 index 788851ceb..000000000 --- a/tests/cli/test_cli.py +++ /dev/null @@ -1,472 +0,0 @@ -"""Tests for the CLI module.""" - -import subprocess -from pathlib import Path -from unittest.mock import MagicMock, Mock, patch - -import pytest -from typer.testing import CliRunner - -from fastmcp.cli import cli - -# Set up test runner -runner = CliRunner() - - -@pytest.fixture -def mock_console(): - """Mock the rich console to test output.""" - with patch("fastmcp.cli.cli.console") as mock_console: - yield mock_console - - -@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_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") - - 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"] - - 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 == [ - "uv", - "run", - "--with", - "fastmcp", - "--with-editable", - str(project_path), - "fastmcp", - "run", - "file.py", - ] - - 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 == [ - "uv", - "run", - "--with", - "fastmcp", - "--with", - "pkg1", - "--with", - "pkg2", - "fastmcp", - "run", - "file.py", - ] - - 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 == [ - "uv", - "run", - "--with", - "fastmcp", - "--with-editable", - str(project_path), - "--with", - "pkg1", - "--with", - "pkg2", - "fastmcp", - "run", - "file.py:server", - ] - - -class TestVersionCommand: - """Tests for 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 - - -class TestDevCommand: - """Tests for 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) - - 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" - - -class TestRunCommand: - """Tests for 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 - - 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_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 - - 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") - - 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 = 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") - - # 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 - - 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") - - 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 = 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") - - 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" - ) - - 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, - ) diff --git a/tests/cli/test_cursor.py b/tests/cli/test_cursor.py deleted file mode 100644 index 489d1517f..000000000 --- a/tests/cli/test_cursor.py +++ /dev/null @@ -1,276 +0,0 @@ -"""Tests for Cursor CLI integration.""" - -import base64 -import json -from pathlib import Path -from unittest.mock import patch - -from fastmcp.cli.install.cursor import ( - generate_cursor_deeplink, - install_cursor, - open_deeplink, -) -from fastmcp.mcp_config import StdioMCPServer - - -class TestGenerateCursorDeeplink: - """Test generate_cursor_deeplink function.""" - - def test_generates_valid_deeplink(self): - """Should generate a valid Cursor deeplink with base64 encoded config.""" - 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) - - assert deeplink.startswith("cursor://anysphere.cursor-deeplink/mcp/install?") - 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()) - config_data = json.loads(decoded) - - assert config_data["command"] == "test" - assert config_data["args"] == ["arg1", "arg2"] - - def test_excludes_none_values(self): - """Should exclude None values from the configuration.""" - server_config = StdioMCPServer( - command="test", - args=["arg1"], - timeout=None, # This should be excluded - ) - - deeplink = generate_cursor_deeplink("test", server_config) - config_param = deeplink.split("config=")[1] - decoded = base64.urlsafe_b64decode(config_param.encode()) - config_data = json.loads(decoded) - - assert "timeout" not in config_data - - -class TestOpenDeeplink: - """Test open_deeplink function.""" - - @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 - - result = open_deeplink("cursor://test") - - 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 - - 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 - ) - - @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 - - result = open_deeplink("cursor://test") - - 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 - - mock_run.side_effect = 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.""" - mock_run.side_effect = FileNotFoundError() - - result = open_deeplink("cursor://test") - - assert result is False - - -class TestInstallCursor: - """Test install_cursor function.""" - - @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" - mock_open_deeplink.return_value = True - - result = install_cursor(Path("server.py"), None, "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) - - @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" - 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 - ) - 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) - - @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 - - 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"}, - ) - - # Check that generate_cursor_deeplink was called with correct config - call_args = mock_generate_deeplink.call_args - server_name, server_config = call_args[0] - - 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"} - - @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 - - install_cursor(Path("server.py"), None, "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 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 diff --git a/tests/cli/test_mcp_config.py b/tests/cli/test_mcp_config.py deleted file mode 100644 index 8eecfe861..000000000 --- a/tests/cli/test_mcp_config.py +++ /dev/null @@ -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) diff --git a/tests/cli/test_run.py b/tests/cli/test_run.py deleted file mode 100644 index 33b5906f2..000000000 --- a/tests/cli/test_run.py +++ /dev/null @@ -1,298 +0,0 @@ -"""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() - - -@pytest.fixture -def mock_console(): - """Mock the rich console to test output.""" - with patch("fastmcp.cli.cli.console") as mock_console: - yield mock_console - - -@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: - 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") - - path, obj = fastmcp.cli.run.parse_file_path("file.py") - assert path == Path("file.py") - assert obj is None - - 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") - - path, obj = fastmcp.cli.run.parse_file_path("file.py:server") - assert path == Path("file.py") - assert obj == "server" - - 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") - - path, obj = fastmcp.cli.run.parse_file_path("C:/path/file.py:server") - assert path == Path("C:/path/file.py") - assert obj == "server" - - 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") - - fastmcp.cli.run.parse_file_path("directory") - mock_logger.error.assert_called_once() - mock_exit.assert_called_once_with(1) - - -class TestRunCommand: - """Tests for 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 - - 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_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 - - 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") - - 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 = 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") - - 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 diff --git a/uv.lock b/uv.lock index 2c84c14ea..9fc747b15 100644 --- a/uv.lock +++ b/uv.lock @@ -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"