diff --git a/docs/clients/generate-cli.mdx b/docs/clients/generate-cli.mdx new file mode 100644 index 000000000..833637423 --- /dev/null +++ b/docs/clients/generate-cli.mdx @@ -0,0 +1,129 @@ +--- +title: Generate CLI +sidebarTitle: Generate CLI +description: Turn any MCP server into a standalone, typed command-line tool. +icon: wand-magic-sparkles +--- + +import { VersionBadge } from '/snippets/version-badge.mdx' + + + +`fastmcp list` and `fastmcp call` let you poke at a server interactively, but they're developer tools — you always have to spell out the server spec, the tool name, and the arguments. `fastmcp generate-cli` takes the next step: it connects to a server, reads its schemas, and writes a standalone Python script where every tool is a proper subcommand with typed flags, help text, and tab completion. The result is a CLI that feels like it was hand-written for that specific server. + +The key insight is that MCP tool schemas already contain everything a CLI framework needs: parameter names, types, descriptions, required/optional status, and defaults. `generate-cli` maps that schema into [cyclopts](https://cyclopts.readthedocs.io/) commands, so JSON Schema types become Python type annotations, descriptions become `--help` text, and required parameters become mandatory flags. + +## Generating a Script + +Point the command at any server spec — URLs, Python files, discovered server names, MCPConfig JSON — and it writes a CLI script: + +```bash +fastmcp generate-cli weather +fastmcp generate-cli http://localhost:8000/mcp +fastmcp generate-cli server.py my_weather_cli.py +``` + +The second positional argument sets the output path. When omitted, it defaults to `cli.py`. If the file already exists, the command refuses to overwrite unless you pass `-f`: + +```bash +fastmcp generate-cli weather -f +fastmcp generate-cli weather my_cli.py -f +``` + +Name-based resolution works here too, so if you have a server configured in Claude Desktop, Cursor, or any other supported editor, you can reference it by name. Run [`fastmcp discover`](/clients/cli#discovering-configured-servers) to see what's available. + +```bash +fastmcp generate-cli claude-code:my-server output.py +``` + +The `--timeout` and `--auth` flags work the same way they do in `fastmcp list` and `fastmcp call`. + +## What You Get + +The generated script is a regular Python file — executable, editable, and yours. Here's what it looks like in practice: + +``` +$ python cli.py --help +Usage: weather-cli COMMAND + +CLI for weather MCP server + +Commands: + call-tool Call a tool on the server + list-tools List available tools. + list-resources List available resources. + read-resource Read a resource by URI. + list-prompts List available prompts. + get-prompt Get a prompt by name. Pass arguments as key=value pairs. +``` + +The `call-tool` subcommand is where the generated code lives. Each tool on the server becomes its own command: + +``` +$ python cli.py call-tool --help +Usage: weather-cli call-tool COMMAND + +Call a tool on the server + +Commands: + get_forecast Get the weather forecast for a city. + search_city Search for a city by name. +``` + +And each tool has typed parameters with help text pulled directly from the server's schema: + +``` +$ python cli.py call-tool get_forecast --help +Usage: weather-cli call-tool get_forecast [OPTIONS] + +Get the weather forecast for a city. + +Options: + --city [str] City name (required) + --days [int] Number of forecast days (default: 3) +``` + +Tool names are preserved exactly as the server defines them — underscores stay as underscores, so `call-tool get_forecast` matches what the server expects. + +## How It Works + +The generated script is a client, not a server. It doesn't bundle or embed the MCP server — it connects to it on every invocation. For URL-based servers, the server needs to be running. For stdio-based servers, the command specified in `CLIENT_SPEC` must be available on the system's `PATH`. + +At the top of the generated file, a `CLIENT_SPEC` variable holds the resolved transport: either a URL string or a `StdioTransport` with the command and arguments baked in. Every invocation connects through this spec, so the script works without any external configuration. + +### Parameter Handling + +Parameters are mapped intelligently based on their complexity: + +**Simple types** (`string`, `integer`, `number`, `boolean`) become typed Python parameters with clean flags: +```bash +python cli.py call-tool get_forecast --city London --days 3 +``` + +**Arrays of simple types** (`array` with `string`/`integer`/`number`/`boolean` items) become `list[T]` parameters that accept multiple flags: +```bash +python cli.py call-tool tag_items --tags python --tags fastapi --tags mcp +``` + +**Complex types** (objects, nested arrays, or unions) accept JSON strings. The tool's `--help` displays the full JSON schema so you know exactly what structure to pass: +```bash +python cli.py call-tool create_user \ + --name John \ + --metadata '{"role": "admin", "dept": "engineering"}' +``` + +Required parameters are mandatory flags; optional ones default to their schema default or `None`. Empty values are filtered out before calling the server. + +Beyond tool commands, the script includes generic commands that work regardless of what the server exposes: `list-tools`, `list-resources`, `read-resource`, `list-prompts`, and `get-prompt`. These connect to the server at runtime, so they always reflect the server's current state even if the tools have changed since generation. + +## Editing the Output + +The most common edit is changing `CLIENT_SPEC`. If you generated from a local dev server and want to point at production, just change the string. If you generated from a discovered name and want to pin the transport, replace it with an explicit URL or `StdioTransport`. + +Beyond that, it's a regular Python file. You can add commands, change the output formatting, integrate it into a larger application, or strip out the parts you don't need. The helper functions (`_call_tool`, `_print_tool_result`) are thin wrappers around `fastmcp.Client` that are easy to adapt. + +The generated script requires `fastmcp` as a dependency. If the script lives outside a project that already has fastmcp installed, `uv run` is the easiest way to run it without permanent installation: + +```bash +uv run --with fastmcp python cli.py call-tool get_forecast --city London +``` diff --git a/docs/docs.json b/docs/docs.json index b7f26ab6b..916fa2640 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -169,6 +169,7 @@ "pages": [ "clients/client", "clients/cli", + "clients/generate-cli", "clients/transports", { "group": "Core Operations", diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py index 85b9d9ff5..147d9dc2d 100644 --- a/src/fastmcp/cli/cli.py +++ b/src/fastmcp/cli/cli.py @@ -20,6 +20,7 @@ from rich.table import Table import fastmcp from fastmcp.cli import run as run_module from fastmcp.cli.client import call_command, discover_command, list_command +from fastmcp.cli.generate import generate_cli_command from fastmcp.cli.install import install_app from fastmcp.cli.tasks import tasks_app from fastmcp.utilities.cli import is_already_in_uv_subprocess, load_and_merge_config @@ -957,6 +958,7 @@ app.command(tasks_app) app.command(list_command, name="list") app.command(call_command, name="call") app.command(discover_command, name="discover") +app.command(generate_cli_command, name="generate-cli") if __name__ == "__main__": diff --git a/src/fastmcp/cli/generate.py b/src/fastmcp/cli/generate.py new file mode 100644 index 000000000..7fdc6cd8a --- /dev/null +++ b/src/fastmcp/cli/generate.py @@ -0,0 +1,634 @@ +"""Generate a standalone CLI script from an MCP server's capabilities.""" + +import keyword +import re +import sys +import textwrap +from pathlib import Path +from typing import Annotated, Any +from urllib.parse import urlparse + +import cyclopts +import mcp.types +import pydantic_core +from mcp import McpError +from rich.console import Console + +from fastmcp.cli.client import _build_client, resolve_server_spec +from fastmcp.client.transports.base import ClientTransport +from fastmcp.client.transports.stdio import StdioTransport +from fastmcp.utilities.logging import get_logger + +logger = get_logger("cli.generate") +console = Console() + +# --------------------------------------------------------------------------- +# JSON Schema type → Python type string +# --------------------------------------------------------------------------- + +_SIMPLE_TYPES = {"string", "integer", "number", "boolean", "null"} + + +def _is_simple_type(schema: dict[str, Any]) -> bool: + """Check if a schema represents a simple (non-complex) type.""" + schema_type = schema.get("type") + if isinstance(schema_type, list): + # Union of types - simple only if all are simple + return all(t in _SIMPLE_TYPES for t in schema_type) + return schema_type in _SIMPLE_TYPES + + +def _is_simple_array(schema: dict[str, Any]) -> tuple[bool, str | None]: + """Check if schema is an array of simple types. + + Returns (is_simple_array, item_type_str). + """ + if schema.get("type") != "array": + return False, None + + items = schema.get("items", {}) + if not _is_simple_type(items): + return False, None + + # Map JSON Schema type to Python type + item_type = items.get("type", "string") + if isinstance(item_type, list): + return False, None + type_map = { + "string": "str", + "integer": "int", + "number": "float", + "boolean": "bool", + } + py_type = type_map.get(item_type) + if py_type is None: + return False, None + return True, py_type + + +def _schema_to_python_type(schema: dict[str, Any]) -> tuple[str, bool]: + """Convert a JSON Schema to a Python type annotation. + + Returns (type_annotation, needs_json_parsing). + """ + # Check for simple array first + is_simple_arr, item_type = _is_simple_array(schema) + if is_simple_arr: + return f"list[{item_type}]", False + + # Check for simple type + if _is_simple_type(schema): + schema_type = schema.get("type", "string") + if isinstance(schema_type, list): + # Union of simple types + type_map = { + "string": "str", + "integer": "int", + "number": "float", + "boolean": "bool", + "null": "None", + } + parts = [type_map.get(t, "str") for t in schema_type] + return " | ".join(parts), False + + type_map = { + "string": "str", + "integer": "int", + "number": "float", + "boolean": "bool", + "null": "None", + } + return type_map.get(schema_type, "str"), False + + # Complex type - needs JSON parsing + return "str", True + + +def _format_schema_for_help(schema: dict[str, Any]) -> str: + """Format a JSON schema for display in help text.""" + # Pretty print the schema, indented for help text + schema_str = pydantic_core.to_json(schema, indent=2).decode() + # Indent each line for help text alignment + lines = schema_str.split("\n") + indented = "\n ".join(lines) + return f"JSON Schema: {indented}" + + +# --------------------------------------------------------------------------- +# Transport serialization +# --------------------------------------------------------------------------- + + +def serialize_transport( + resolved: str | dict[str, Any] | ClientTransport, +) -> tuple[str, set[str]]: + """Serialize a resolved transport to a Python expression string. + + Returns ``(expression, extra_imports)`` where *extra_imports* is a set of + import lines needed by the expression. + """ + if isinstance(resolved, str): + return repr(resolved), set() + + if isinstance(resolved, StdioTransport): + parts = [f"command={resolved.command!r}", f"args={resolved.args!r}"] + if resolved.env: + parts.append(f"env={resolved.env!r}") + if resolved.cwd: + parts.append(f"cwd={resolved.cwd!r}") + expr = f"StdioTransport({', '.join(parts)})" + imports = {"from fastmcp.client.transports import StdioTransport"} + return expr, imports + + if isinstance(resolved, dict): + return repr(resolved), set() + + # Fallback: try repr + return repr(resolved), set() + + +# --------------------------------------------------------------------------- +# Per-tool code generation +# --------------------------------------------------------------------------- + + +def _to_python_identifier(name: str) -> str: + """Sanitize a string into a valid Python identifier.""" + safe = re.sub(r"[^a-zA-Z0-9_]", "_", name) + if safe and safe[0].isdigit(): + safe = f"_{safe}" + safe = safe or "_unnamed" + if keyword.iskeyword(safe): + safe = f"{safe}_" + return safe + + +def _tool_function_source(tool: mcp.types.Tool) -> str: + """Generate the source for a single ``@call_tool_app.command`` function.""" + schema = tool.inputSchema + properties: dict[str, Any] = schema.get("properties", {}) + required = set(schema.get("required", [])) + + # Build parameter lines and track which need JSON parsing + param_lines: list[str] = [] + call_args: list[str] = [] + json_params: list[tuple[str, str]] = [] # (prop_name, safe_name) + seen_names: dict[str, str] = {} # safe_name -> original prop_name + + for prop_name, prop_schema in properties.items(): + py_type, needs_json = _schema_to_python_type(prop_schema) + help_text = prop_schema.get("description", "") + is_required = prop_name in required + safe_name = _to_python_identifier(prop_name) + + # Check for name collisions after sanitization + if safe_name in seen_names: + raise ValueError( + f"Parameter name collision: '{prop_name}' and '{seen_names[safe_name]}' " + f"both sanitize to '{safe_name}'" + ) + seen_names[safe_name] = prop_name + + # For complex types, add schema to help text + if needs_json: + schema_help = _format_schema_for_help(prop_schema) + help_text = f"{help_text}\\n{schema_help}" if help_text else schema_help + json_params.append((prop_name, safe_name)) + + # Escape special characters in help text + help_escaped = ( + help_text.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") + ) + + # Build parameter annotation + if is_required: + annotation = ( + f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + ) + param_lines.append(f" {safe_name}: {annotation},") + else: + default = prop_schema.get("default") + if default is not None: + # For complex types with defaults, serialize to JSON string + if needs_json: + default_str = pydantic_core.to_json(default, fallback=str).decode() + annotation = f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + param_lines.append( + f" {safe_name}: {annotation} = {default_str!r}," + ) + else: + annotation = f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + param_lines.append(f" {safe_name}: {annotation} = {default!r},") + else: + # For list types, default to empty list; others default to None + if py_type.startswith("list["): + annotation = f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + param_lines.append(f" {safe_name}: {annotation} = [],") + else: + annotation = f'Annotated[{py_type} | None, cyclopts.Parameter(help="{help_escaped}")]' + param_lines.append(f" {safe_name}: {annotation} = None,") + + call_args.append(f"{prop_name!r}: {safe_name}") + + # Function name: sanitize to valid Python identifier + fn_name = _to_python_identifier(tool.name) + + # Docstring - use single-quoted docstrings to avoid triple-quote escaping issues + description = (tool.description or "").replace("\\", "\\\\").replace("'", "\\'") + + lines = [] + lines.append("") + # Always pass name= to preserve the original tool name (cyclopts + # would otherwise convert underscores to hyphens). + lines.append(f"@call_tool_app.command(name={tool.name!r})") + lines.append(f"async def {fn_name}(") + + if param_lines: + lines.append(" *,") + lines.extend(param_lines) + + lines.append(") -> None:") + lines.append(f" '''{description}'''") + + # Add JSON parsing for complex parameters + if json_params: + lines.append(" # Parse JSON parameters") + for _prop_name, safe_name in json_params: + lines.append( + f" {safe_name}_parsed = json.loads({safe_name}) if isinstance({safe_name}, str) else {safe_name}" + ) + lines.append("") + + # Build call arguments, using parsed versions for JSON params + call_arg_parts = [] + for prop_name, _ in properties.items(): + safe_name = _to_python_identifier(prop_name) + if any(pn == prop_name for pn, _ in json_params): + call_arg_parts.append(f"{prop_name!r}: {safe_name}_parsed") + else: + call_arg_parts.append(f"{prop_name!r}: {safe_name}") + + dict_items = ", ".join(call_arg_parts) + lines.append(f" await _call_tool({tool.name!r}, {{{dict_items}}})") + lines.append("") + + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Full script generation +# --------------------------------------------------------------------------- + + +def generate_cli_script( + server_name: str, + server_spec: str, + transport_code: str, + extra_imports: set[str], + tools: list[mcp.types.Tool], +) -> str: + """Generate the full CLI script source code.""" + + # Determine app name from server_name - sanitize for use in string literal + app_name = ( + server_name.replace(" ", "-").lower().replace("\\", "\\\\").replace('"', '\\"') + ) + + # --- Header --- + lines: list[str] = [] + lines.append("#!/usr/bin/env python3") + lines.append(f'"""CLI for {server_name} MCP server.') + lines.append("") + lines.append(f"Generated by: fastmcp generate-cli {server_spec}") + lines.append('"""') + lines.append("") + + # --- Imports --- + lines.append("import json") + lines.append("import sys") + lines.append("from typing import Annotated") + lines.append("") + lines.append("import cyclopts") + lines.append("import mcp.types") + lines.append("from rich.console import Console") + lines.append("") + lines.append("from fastmcp import Client") + for imp in sorted(extra_imports): + lines.append(imp) + lines.append("") + + # --- Transport config --- + lines.append("# Modify this to change how the CLI connects to the MCP server.") + lines.append(f"CLIENT_SPEC = {transport_code}") + lines.append("") + + # --- App setup --- + server_name_escaped = server_name.replace("\\", "\\\\").replace('"', '\\"') + lines.append( + f'app = cyclopts.App(name="{app_name}", help="CLI for {server_name_escaped} MCP server")' + ) + lines.append( + 'call_tool_app = cyclopts.App(name="call-tool", help="Call a tool on the server")' + ) + lines.append("app.command(call_tool_app)") + lines.append("") + lines.append("console = Console()") + lines.append("") + lines.append("") + + # --- Shared helpers --- + lines.append( + textwrap.dedent("""\ + # --------------------------------------------------------------------------- + # Helpers + # --------------------------------------------------------------------------- + + + def _print_tool_result(result): + if result.is_error: + for block in result.content: + if isinstance(block, mcp.types.TextContent): + console.print(f"[bold red]Error:[/bold red] {block.text}") + else: + console.print(f"[bold red]Error:[/bold red] {block}") + sys.exit(1) + + if result.structured_content is not None: + console.print_json(json.dumps(result.structured_content)) + return + + for block in result.content: + if isinstance(block, mcp.types.TextContent): + console.print(block.text) + elif isinstance(block, mcp.types.ImageContent): + size = len(block.data) * 3 // 4 + console.print(f"[dim][Image: {block.mimeType}, ~{size} bytes][/dim]") + elif isinstance(block, mcp.types.AudioContent): + size = len(block.data) * 3 // 4 + console.print(f"[dim][Audio: {block.mimeType}, ~{size} bytes][/dim]") + + + async def _call_tool(tool_name: str, arguments: dict) -> None: + # Filter out None values and empty lists (defaults for optional array params) + filtered = { + k: v + for k, v in arguments.items() + if v is not None and (not isinstance(v, list) or len(v) > 0) + } + async with Client(CLIENT_SPEC) as client: + result = await client.call_tool(tool_name, filtered, raise_on_error=False) + _print_tool_result(result) + if result.is_error: + sys.exit(1)""") + ) + lines.append("") + lines.append("") + + # --- Generic commands --- + lines.append( + textwrap.dedent("""\ + # --------------------------------------------------------------------------- + # List / read commands + # --------------------------------------------------------------------------- + + + @app.command + async def list_tools() -> None: + \"\"\"List available tools.\"\"\" + async with Client(CLIENT_SPEC) as client: + tools = await client.list_tools() + if not tools: + console.print("[dim]No tools found.[/dim]") + return + for tool in tools: + sig_parts = [] + props = tool.inputSchema.get("properties", {}) + required = set(tool.inputSchema.get("required", [])) + for pname, pschema in props.items(): + ptype = pschema.get("type", "string") + if pname in required: + sig_parts.append(f"{pname}: {ptype}") + else: + sig_parts.append(f"{pname}: {ptype} = ...") + sig = f"{tool.name}({', '.join(sig_parts)})" + console.print(f" [cyan]{sig}[/cyan]") + if tool.description: + console.print(f" {tool.description}") + console.print() + + + @app.command + async def list_resources() -> None: + \"\"\"List available resources.\"\"\" + async with Client(CLIENT_SPEC) as client: + resources = await client.list_resources() + if not resources: + console.print("[dim]No resources found.[/dim]") + return + for r in resources: + console.print(f" [cyan]{r.uri}[/cyan]") + desc_parts = [r.name or "", r.description or ""] + desc = " — ".join(p for p in desc_parts if p) + if desc: + console.print(f" {desc}") + console.print() + + + @app.command + async def read_resource(uri: Annotated[str, cyclopts.Parameter(help="Resource URI")]) -> None: + \"\"\"Read a resource by URI.\"\"\" + async with Client(CLIENT_SPEC) as client: + contents = await client.read_resource(uri) + for block in contents: + if isinstance(block, mcp.types.TextResourceContents): + console.print(block.text) + elif isinstance(block, mcp.types.BlobResourceContents): + size = len(block.blob) * 3 // 4 + console.print(f"[dim][Blob: {block.mimeType}, ~{size} bytes][/dim]") + + + @app.command + async def list_prompts() -> None: + \"\"\"List available prompts.\"\"\" + async with Client(CLIENT_SPEC) as client: + prompts = await client.list_prompts() + if not prompts: + console.print("[dim]No prompts found.[/dim]") + return + for p in prompts: + args_str = "" + if p.arguments: + parts = [a.name for a in p.arguments] + args_str = f"({', '.join(parts)})" + console.print(f" [cyan]{p.name}{args_str}[/cyan]") + if p.description: + console.print(f" {p.description}") + console.print() + + + @app.command + async def get_prompt( + name: Annotated[str, cyclopts.Parameter(help="Prompt name")], + *arguments: str, + ) -> None: + \"\"\"Get a prompt by name. Pass arguments as key=value pairs.\"\"\" + parsed: dict[str, str] = {} + for arg in arguments: + if "=" not in arg: + console.print(f"[bold red]Error:[/bold red] Invalid argument {arg!r} — expected key=value") + sys.exit(1) + key, value = arg.split("=", 1) + parsed[key] = value + + async with Client(CLIENT_SPEC) as client: + result = await client.get_prompt(name, parsed or None) + for msg in result.messages: + console.print(f"[bold]{msg.role}:[/bold]") + if isinstance(msg.content, mcp.types.TextContent): + console.print(f" {msg.content.text}") + elif isinstance(msg.content, mcp.types.ImageContent): + size = len(msg.content.data) * 3 // 4 + console.print(f" [dim][Image: {msg.content.mimeType}, ~{size} bytes][/dim]") + else: + console.print(f" {msg.content}") + console.print()""") + ) + lines.append("") + lines.append("") + + # --- Generated tool commands --- + if tools: + lines.append( + "# ---------------------------------------------------------------------------" + ) + lines.append("# Tool commands (generated from server schema)") + lines.append( + "# ---------------------------------------------------------------------------" + ) + + for tool in tools: + lines.append(_tool_function_source(tool)) + + # --- Entry point --- + lines.append("") + lines.append('if __name__ == "__main__":') + lines.append(" app()") + lines.append("") + + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# CLI command +# --------------------------------------------------------------------------- + + +async def generate_cli_command( + server_spec: Annotated[ + str, + cyclopts.Parameter( + help="Server URL, Python file, MCPConfig JSON, discovered name, or .js file", + ), + ], + output: Annotated[ + str, + cyclopts.Parameter( + help="Output file path (default: cli.py)", + ), + ] = "cli.py", + *, + force: Annotated[ + bool, + cyclopts.Parameter( + name=["-f", "--force"], + help="Overwrite output file if it exists", + ), + ] = False, + timeout: Annotated[ + float | None, + cyclopts.Parameter("--timeout", help="Connection timeout in seconds"), + ] = None, + auth: Annotated[ + str | None, + cyclopts.Parameter( + "--auth", + help="Auth method: 'oauth', a bearer token string, or 'none' to disable", + ), + ] = None, +) -> None: + """Generate a standalone CLI script from an MCP server. + + Connects to the server, reads its tools/resources/prompts, and writes + a Python script that can invoke them directly. + + Examples: + fastmcp generate-cli weather + fastmcp generate-cli weather my_cli.py + fastmcp generate-cli http://localhost:8000/mcp + fastmcp generate-cli server.py output.py -f + """ + output_path = Path(output) + if output_path.exists() and not force: + console.print( + f"[bold red]Error:[/bold red] [cyan]{output_path}[/cyan] already exists. " + f"Use [cyan]-f[/cyan] to overwrite." + ) + sys.exit(1) + + # Resolve the server spec to a transport + resolved = resolve_server_spec(server_spec) + transport_code, extra_imports = serialize_transport(resolved) + + # Derive a human-friendly server name from the spec + server_name = _derive_server_name(server_spec) + + # Connect and discover capabilities + client = _build_client(resolved, timeout=timeout, auth=auth) + + try: + async with client: + tools = await client.list_tools() + console.print( + f"[dim]Discovered {len(tools)} tool(s) from {server_spec}[/dim]" + ) + + except (RuntimeError, TimeoutError, McpError, OSError) as exc: + console.print(f"[bold red]Error:[/bold red] Could not connect: {exc}") + sys.exit(1) + + # Generate and write the script + script = generate_cli_script( + server_name=server_name, + server_spec=server_spec, + transport_code=transport_code, + extra_imports=extra_imports, + tools=tools, + ) + + output_path.write_text(script) + output_path.chmod(output_path.stat().st_mode | 0o111) # make executable + + console.print( + f"[green]✓[/green] Wrote [cyan]{output_path}[/cyan] " + f"with {len(tools)} tool command(s)" + ) + console.print(f"[dim]Run: python {output_path} --help[/dim]") + + +def _derive_server_name(server_spec: str) -> str: + """Derive a human-friendly name from a server spec.""" + # URL — use hostname + if server_spec.startswith(("http://", "https://")): + parsed = urlparse(server_spec) + return parsed.hostname or "server" + + # File path — use stem + if server_spec.endswith((".py", ".js", ".json")): + return Path(server_spec).stem + + # Bare name or qualified name + if ":" in server_spec: + name = server_spec.split(":", 1)[1] + return name or server_spec.split(":", 1)[0] + + return server_spec diff --git a/tests/cli/test_generate_cli.py b/tests/cli/test_generate_cli.py new file mode 100644 index 000000000..f513338d7 --- /dev/null +++ b/tests/cli/test_generate_cli.py @@ -0,0 +1,638 @@ +"""Tests for fastmcp generate-cli command.""" + +import sys +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import mcp.types +import pytest + +from fastmcp import FastMCP +from fastmcp.cli import generate as generate_module +from fastmcp.cli.client import Client +from fastmcp.cli.generate import ( + _derive_server_name, + _schema_to_python_type, + _to_python_identifier, + _tool_function_source, + generate_cli_command, + generate_cli_script, + serialize_transport, +) +from fastmcp.client.transports.stdio import StdioTransport + +# --------------------------------------------------------------------------- +# _schema_to_python_type +# --------------------------------------------------------------------------- + + +class TestSchemaToPythonType: + def test_simple_string(self): + py_type, needs_json = _schema_to_python_type({"type": "string"}) + assert py_type == "str" + assert needs_json is False + + def test_simple_integer(self): + py_type, needs_json = _schema_to_python_type({"type": "integer"}) + assert py_type == "int" + assert needs_json is False + + def test_simple_number(self): + py_type, needs_json = _schema_to_python_type({"type": "number"}) + assert py_type == "float" + assert needs_json is False + + def test_simple_boolean(self): + py_type, needs_json = _schema_to_python_type({"type": "boolean"}) + assert py_type == "bool" + assert needs_json is False + + def test_array_of_strings(self): + py_type, needs_json = _schema_to_python_type( + {"type": "array", "items": {"type": "string"}} + ) + assert py_type == "list[str]" + assert needs_json is False + + def test_array_of_integers(self): + py_type, needs_json = _schema_to_python_type( + {"type": "array", "items": {"type": "integer"}} + ) + assert py_type == "list[int]" + assert needs_json is False + + def test_complex_object(self): + py_type, needs_json = _schema_to_python_type({"type": "object"}) + assert py_type == "str" + assert needs_json is True + + def test_complex_nested_array(self): + py_type, needs_json = _schema_to_python_type( + {"type": "array", "items": {"type": "object"}} + ) + assert py_type == "str" + assert needs_json is True + + def test_union_of_simple_types(self): + py_type, needs_json = _schema_to_python_type({"type": ["string", "null"]}) + assert py_type == "str | None" + assert needs_json is False + + +# --------------------------------------------------------------------------- +# _to_python_identifier +# --------------------------------------------------------------------------- + + +class TestToPythonIdentifier: + def test_plain_name(self): + assert _to_python_identifier("hello") == "hello" + + def test_hyphens(self): + assert _to_python_identifier("get-forecast") == "get_forecast" + + def test_dots_and_slashes(self): + assert _to_python_identifier("a.b/c") == "a_b_c" + + def test_leading_digit(self): + assert _to_python_identifier("3d_render") == "_3d_render" + + def test_spaces(self): + assert _to_python_identifier("my tool") == "my_tool" + + def test_empty_string(self): + assert _to_python_identifier("") == "_unnamed" + + +# --------------------------------------------------------------------------- +# serialize_transport +# --------------------------------------------------------------------------- + + +class TestSerializeTransport: + def test_url_string(self): + code, imports = serialize_transport("http://localhost:8000/mcp") + assert code == "'http://localhost:8000/mcp'" + assert imports == set() + + def test_stdio_transport_basic(self): + transport = StdioTransport(command="fastmcp", args=["run", "server.py"]) + code, imports = serialize_transport(transport) + assert "StdioTransport" in code + assert "command='fastmcp'" in code + assert "args=['run', 'server.py']" in code + assert "from fastmcp.client.transports import StdioTransport" in imports + + def test_stdio_transport_with_env(self): + transport = StdioTransport( + command="python", args=["-m", "myserver"], env={"KEY": "val"} + ) + code, imports = serialize_transport(transport) + assert "env={'KEY': 'val'}" in code + + def test_dict_passthrough(self): + d: dict[str, Any] = {"mcpServers": {"test": {"url": "http://localhost"}}} + code, imports = serialize_transport(d) + assert "mcpServers" in code + assert imports == set() + + +# --------------------------------------------------------------------------- +# _tool_function_source +# --------------------------------------------------------------------------- + + +class TestToolFunctionSource: + def test_required_param(self): + tool = mcp.types.Tool( + name="greet", + inputSchema={ + "properties": {"name": {"type": "string", "description": "Who"}}, + "required": ["name"], + }, + ) + source = _tool_function_source(tool) + assert "async def greet(" in source + assert "name: Annotated[str" in source + assert "= None" not in source + assert "_call_tool('greet', {'name': name})" in source + + def test_optional_param(self): + tool = mcp.types.Tool( + name="search", + inputSchema={ + "properties": { + "query": {"type": "string", "description": "Search query"}, + "limit": {"type": "integer", "description": "Max results"}, + }, + "required": ["query"], + }, + ) + source = _tool_function_source(tool) + assert "query: Annotated[str" in source + assert "limit: Annotated[int | None" in source + assert "= None" in source + + def test_param_with_default(self): + tool = mcp.types.Tool( + name="fetch", + inputSchema={ + "properties": { + "url": {"type": "string", "description": "URL"}, + "timeout": { + "type": "integer", + "description": "Timeout", + "default": 30, + }, + }, + "required": ["url"], + }, + ) + source = _tool_function_source(tool) + assert "timeout: Annotated[int" in source + assert "= 30" in source + + def test_no_params(self): + tool = mcp.types.Tool( + name="ping", + inputSchema={"properties": {}}, + ) + source = _tool_function_source(tool) + assert "async def ping(" in source + assert "_call_tool('ping', {})" in source + + def test_preserves_underscores(self): + tool = mcp.types.Tool( + name="get_forecast", + inputSchema={ + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + ) + source = _tool_function_source(tool) + assert "async def get_forecast(" in source + + def test_sanitizes_tool_name(self): + tool = mcp.types.Tool( + name="my.tool/v2", + inputSchema={"properties": {}}, + ) + source = _tool_function_source(tool) + assert "async def my_tool_v2(" in source + assert "name='my.tool/v2'" in source + + def test_sanitizes_param_name(self): + tool = mcp.types.Tool( + name="fetch", + inputSchema={ + "properties": {"content-type": {"type": "string", "description": "CT"}}, + "required": ["content-type"], + }, + ) + source = _tool_function_source(tool) + assert "content_type: Annotated[str" in source + assert "'content-type': content_type" in source + + def test_description_in_docstring(self): + tool = mcp.types.Tool( + name="greet", + description="Say hello to someone.", + inputSchema={ + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, + ) + source = _tool_function_source(tool) + assert "'''Say hello to someone.'''" in source + + def test_description_with_quotes(self): + tool = mcp.types.Tool( + name="fetch", + description="Fetch data from 'source' API.", + inputSchema={ + "properties": {"url": {"type": "string"}}, + "required": ["url"], + }, + ) + source = _tool_function_source(tool) + # Should escape single quotes in the description + assert r"Fetch data from \'source\' API." in source + # Generated code should compile + compile(source, "", "exec") + + def test_array_of_strings_parameter(self): + tool = mcp.types.Tool( + name="tag_items", + description="Tag multiple items.", + inputSchema={ + "properties": { + "item_id": {"type": "string"}, + "tags": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["item_id"], + }, + ) + source = _tool_function_source(tool) + # Should use list[str] type with help metadata + assert "tags: Annotated[list[str]" in source + assert "= []" in source + # Should not have JSON parsing for simple arrays + assert "json.loads" not in source + compile(source, "", "exec") + + def test_complex_object_parameter(self): + tool = mcp.types.Tool( + name="create_user", + description="Create a user.", + inputSchema={ + "properties": { + "name": {"type": "string"}, + "metadata": { + "type": "object", + "properties": { + "role": {"type": "string"}, + "dept": {"type": "string"}, + }, + }, + }, + "required": ["name"], + }, + ) + source = _tool_function_source(tool) + # Should use str type for complex object + assert "metadata: Annotated[str | None" in source + # Should include JSON schema in help (with escaped quotes) + assert "JSON Schema:" in source + assert '\\"type\\": \\"object\\"' in source + # Should have JSON parsing with isinstance check + assert ( + "metadata_parsed = json.loads(metadata) if isinstance(metadata, str) else metadata" + in source + ) + # Should use parsed version in call + assert "'metadata': metadata_parsed" in source + compile(source, "", "exec") + + def test_nested_array_parameter(self): + tool = mcp.types.Tool( + name="batch_process", + description="Process batches.", + inputSchema={ + "properties": { + "batches": { + "type": "array", + "items": { + "type": "object", + "properties": {"id": {"type": "string"}}, + }, + }, + }, + "required": ["batches"], + }, + ) + source = _tool_function_source(tool) + # Nested arrays need JSON parsing + assert "batches: Annotated[str" in source + assert "JSON Schema:" in source + assert ( + "batches_parsed = json.loads(batches) if isinstance(batches, str) else batches" + in source + ) + compile(source, "", "exec") + + def test_complex_type_with_default(self): + """Test that complex types with defaults are JSON-serialized.""" + tool = mcp.types.Tool( + name="configure", + inputSchema={ + "properties": { + "options": { + "type": "object", + "default": {"timeout": 30, "retry": True}, + }, + }, + }, + ) + source = _tool_function_source(tool) + # Default should be JSON string, not Python dict + # pydantic_core.to_json produces compact JSON + assert '= \'{"timeout":30,"retry":true}\'' in source + # Should parse safely even with default + assert "isinstance(options, str)" in source + compile(source, "", "exec") + + def test_name_collision_detection(self): + """Test that parameter name collisions are detected.""" + tool = mcp.types.Tool( + name="test", + inputSchema={ + "properties": { + "content-type": {"type": "string"}, + "content_type": {"type": "string"}, + }, + }, + ) + # Should raise ValueError for collision + with pytest.raises(ValueError, match="both sanitize to 'content_type'"): + _tool_function_source(tool) + + +# --------------------------------------------------------------------------- +# _derive_server_name +# --------------------------------------------------------------------------- + + +class TestDeriveServerName: + def test_bare_name(self): + assert _derive_server_name("weather") == "weather" + + def test_qualified_name(self): + assert _derive_server_name("cursor:weather") == "weather" + + def test_python_file(self): + assert _derive_server_name("server.py") == "server" + + def test_url(self): + assert _derive_server_name("http://localhost:8000/mcp") == "localhost" + + def test_trailing_colon(self): + assert _derive_server_name("source:") == "source" + + +# --------------------------------------------------------------------------- +# generate_cli_script — produces compilable Python +# --------------------------------------------------------------------------- + + +class TestGenerateCliScript: + def _make_tools(self) -> list[mcp.types.Tool]: + return [ + mcp.types.Tool( + name="greet", + description="Say hello", + inputSchema={ + "properties": { + "name": {"type": "string", "description": "Who to greet"}, + }, + "required": ["name"], + }, + ), + mcp.types.Tool( + name="add_numbers", + description="Add two numbers", + inputSchema={ + "properties": { + "a": {"type": "integer", "description": "First number"}, + "b": {"type": "integer", "description": "Second number"}, + }, + "required": ["a", "b"], + }, + ), + ] + + def test_compiles(self): + script = generate_cli_script( + server_name="test", + server_spec="test", + transport_code='"http://localhost:8000/mcp"', + extra_imports=set(), + tools=self._make_tools(), + ) + compile(script, "", "exec") + + def test_contains_tool_functions(self): + script = generate_cli_script( + server_name="test", + server_spec="test", + transport_code='"http://localhost:8000/mcp"', + extra_imports=set(), + tools=self._make_tools(), + ) + assert "async def greet(" in script + assert "async def add_numbers(" in script + + def test_contains_generic_commands(self): + script = generate_cli_script( + server_name="test", + server_spec="test", + transport_code='"http://localhost:8000/mcp"', + extra_imports=set(), + tools=[], + ) + assert "async def list_tools(" in script + assert "async def list_resources(" in script + assert "async def list_prompts(" in script + assert "async def read_resource(" in script + assert "async def get_prompt(" in script + + def test_embeds_transport(self): + script = generate_cli_script( + server_name="test", + server_spec="test", + transport_code="StdioTransport(command='fastmcp', args=['run', 'x.py'])", + extra_imports={"from fastmcp.client.transports import StdioTransport"}, + tools=[], + ) + assert "StdioTransport(command='fastmcp'" in script + assert "from fastmcp.client.transports import StdioTransport" in script + + def test_no_tools_still_valid(self): + script = generate_cli_script( + server_name="empty", + server_spec="empty", + transport_code='"http://localhost"', + extra_imports=set(), + tools=[], + ) + compile(script, "", "exec") + assert "call_tool_app" in script + + def test_server_name_with_quotes(self): + """Test that server names with quotes are properly escaped.""" + script = generate_cli_script( + server_name='Test "Server" Name', + server_spec="test", + transport_code='"http://localhost"', + extra_imports=set(), + tools=[], + ) + # Should compile without syntax errors + compile(script, "", "exec") + # App name should have escaped quotes + assert r'app = cyclopts.App(name="test-\"server\"-name"' in script + + def test_compiles_with_unusual_names(self): + tools = [ + mcp.types.Tool( + name="my.tool/v2", + description="A tool with dots and slashes", + inputSchema={ + "properties": { + "content-type": {"type": "string", "description": "CT"}, + }, + "required": ["content-type"], + }, + ), + ] + script = generate_cli_script( + server_name="test", + server_spec="test", + transport_code='"http://localhost:8000/mcp"', + extra_imports=set(), + tools=tools, + ) + compile(script, "", "exec") + + def test_compiles_with_stdio_transport(self): + transport = StdioTransport(command="fastmcp", args=["run", "server.py"]) + transport_code, extra_imports = serialize_transport(transport) + script = generate_cli_script( + server_name="test", + server_spec="server.py", + transport_code=transport_code, + extra_imports=extra_imports, + tools=self._make_tools(), + ) + compile(script, "", "exec") + + +# --------------------------------------------------------------------------- +# generate_cli_command — integration tests +# --------------------------------------------------------------------------- + + +def _build_test_server() -> FastMCP: + """Create a minimal FastMCP server for integration tests.""" + server = FastMCP("TestServer") + + @server.tool + def greet(name: str) -> str: + """Say hello to someone.""" + return f"Hello, {name}!" + + @server.tool + def add(a: int, b: int) -> int: + """Add two numbers.""" + return a + b + + @server.resource("test://greeting") + def greeting_resource() -> str: + """A static greeting resource.""" + return "Hello from resource!" + + @server.prompt + def ask(topic: str) -> str: + """Ask about a topic.""" + return f"Tell me about {topic}" + + return server + + +@pytest.fixture() +def _patch_client(): + """Patch resolve_server_spec and _build_client to use an in-process server.""" + server = _build_test_server() + + def fake_resolve(server_spec: Any, **kwargs: Any) -> str: + return "fake://server" + + def fake_build_client(resolved: Any, **kwargs: Any) -> Client: + return Client(server) + + with ( + patch.object(generate_module, "resolve_server_spec", side_effect=fake_resolve), + patch.object(generate_module, "_build_client", side_effect=fake_build_client), + ): + yield + + +class TestGenerateCliCommand: + @pytest.mark.usefixtures("_patch_client") + async def test_writes_file(self, tmp_path: Path): + output = tmp_path / "cli.py" + await generate_cli_command("test-server", str(output)) + assert output.exists() + content = output.read_text() + compile(content, str(output), "exec") + + @pytest.mark.usefixtures("_patch_client") + async def test_contains_tools(self, tmp_path: Path): + output = tmp_path / "cli.py" + await generate_cli_command("test-server", str(output)) + content = output.read_text() + assert "async def greet(" in content + assert "async def add(" in content + + @pytest.mark.usefixtures("_patch_client") + async def test_default_output_path( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.chdir(tmp_path) + await generate_cli_command("test-server") + assert (tmp_path / "cli.py").exists() + + @pytest.mark.usefixtures("_patch_client") + async def test_error_if_exists(self, tmp_path: Path): + output = tmp_path / "cli.py" + output.write_text("existing") + with pytest.raises(SystemExit): + await generate_cli_command("test-server", str(output)) + + @pytest.mark.usefixtures("_patch_client") + async def test_force_overwrites(self, tmp_path: Path): + output = tmp_path / "cli.py" + output.write_text("existing") + await generate_cli_command("test-server", str(output), force=True) + content = output.read_text() + assert content != "existing" + assert "async def greet(" in content + + @pytest.mark.skipif( + sys.platform == "win32", reason="Unix executable bits N/A on Windows" + ) + @pytest.mark.usefixtures("_patch_client") + async def test_file_is_executable(self, tmp_path: Path): + output = tmp_path / "cli.py" + await generate_cli_command("test-server", str(output)) + assert output.stat().st_mode & 0o111