From 2c705faa8795ded2248a818de9f81bddd937046a Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 11:11:08 -0500 Subject: [PATCH 1/9] Add `fastmcp generate-cli` command Connects to any MCP server, reads its tool/resource/prompt schemas, and writes a standalone Python CLI script with typed subcommands. --- src/fastmcp/cli/cli.py | 2 + src/fastmcp/cli/generate.py | 497 +++++++++++++++++++++++++++++++++ tests/cli/test_generate_cli.py | 403 ++++++++++++++++++++++++++ 3 files changed, 902 insertions(+) create mode 100644 src/fastmcp/cli/generate.py create mode 100644 tests/cli/test_generate_cli.py 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..ac8a446c9 --- /dev/null +++ b/src/fastmcp/cli/generate.py @@ -0,0 +1,497 @@ +"""Generate a standalone CLI script from an MCP server's capabilities.""" + +import sys +import textwrap +from pathlib import Path +from typing import Annotated, Any + +import cyclopts +import mcp.types +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 +# --------------------------------------------------------------------------- + +_JSON_SCHEMA_TYPE_MAP: dict[str, str] = { + "string": "str", + "integer": "int", + "number": "float", + "boolean": "bool", + "array": "list", + "object": "dict", + "null": "None", +} + + +def _schema_type_to_python(schema: dict[str, Any]) -> str: + """Convert a JSON Schema type fragment to a Python type annotation string.""" + if "anyOf" in schema: + parts = [_schema_type_to_python(s) for s in schema["anyOf"]] + return " | ".join(parts) + + schema_type = schema.get("type", "string") + if isinstance(schema_type, list): + return " | ".join(_JSON_SCHEMA_TYPE_MAP.get(t, "str") for t in schema_type) + + return _JSON_SCHEMA_TYPE_MAP.get(schema_type, "str") + + +# --------------------------------------------------------------------------- +# 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 _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 + param_lines: list[str] = [] + call_args: list[str] = [] + + for prop_name, prop_schema in properties.items(): + py_type = _schema_type_to_python(prop_schema) + help_text = prop_schema.get("description", "") + is_required = prop_name in required + + # Escape quotes in help text + help_escaped = help_text.replace("\\", "\\\\").replace('"', '\\"') + + if is_required: + annotation = ( + f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + ) + param_lines.append(f" {prop_name}: {annotation},") + else: + default = prop_schema.get("default") + if default is not None: + annotation = ( + f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + ) + param_lines.append(f" {prop_name}: {annotation} = {default!r},") + else: + annotation = f'Annotated[{py_type} | None, cyclopts.Parameter(help="{help_escaped}")]' + param_lines.append(f" {prop_name}: {annotation} = None,") + + call_args.append(f"{prop_name!r}: {prop_name}") + + # Function name: use tool name directly (preserve underscores) + fn_name = tool.name.replace("-", "_") + + # Docstring + description = (tool.description or "").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}"""') + dict_items = ", ".join(call_args) + 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 + app_name = server_name.replace(" ", "-").lower() + + # --- 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 --- + lines.append( + f'app = cyclopts.App(name="{app_name}", help="CLI for {server_name} 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: + filtered = {k: v for k, v in arguments.items() if v is not None} + 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 Exception 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://")): + from urllib.parse import urlparse + + 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: + return server_spec.split(":", 1)[1] + + 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..919dc8ee0 --- /dev/null +++ b/tests/cli/test_generate_cli.py @@ -0,0 +1,403 @@ +"""Tests for fastmcp generate-cli command.""" + +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_type_to_python, + _tool_function_source, + generate_cli_command, + generate_cli_script, + serialize_transport, +) +from fastmcp.client.transports.stdio import StdioTransport + +# --------------------------------------------------------------------------- +# _schema_type_to_python +# --------------------------------------------------------------------------- + + +class TestSchemaTypeToPython: + def test_string(self): + assert _schema_type_to_python({"type": "string"}) == "str" + + def test_integer(self): + assert _schema_type_to_python({"type": "integer"}) == "int" + + def test_number(self): + assert _schema_type_to_python({"type": "number"}) == "float" + + def test_boolean(self): + assert _schema_type_to_python({"type": "boolean"}) == "bool" + + def test_array(self): + assert _schema_type_to_python({"type": "array"}) == "list" + + def test_object(self): + assert _schema_type_to_python({"type": "object"}) == "dict" + + def test_null(self): + assert _schema_type_to_python({"type": "null"}) == "None" + + def test_unknown_defaults_to_str(self): + assert _schema_type_to_python({"type": "foobar"}) == "str" + + def test_missing_type_defaults_to_str(self): + assert _schema_type_to_python({}) == "str" + + def test_any_of(self): + result = _schema_type_to_python( + {"anyOf": [{"type": "string"}, {"type": "integer"}]} + ) + assert result == "str | int" + + def test_type_list(self): + result = _schema_type_to_python({"type": ["string", "null"]}) + assert result == "str | None" + + +# --------------------------------------------------------------------------- +# 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_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 + + +# --------------------------------------------------------------------------- +# _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" + + +# --------------------------------------------------------------------------- +# 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_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.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 From d2209059ef73d2610bbef2ca071e1d2c4409ac69 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 11:20:33 -0500 Subject: [PATCH 2/9] docs: add generate-cli documentation --- docs/clients/generate-cli.mdx | 102 ++++++++++++++++++++++++++++++++++ docs/docs.json | 1 + 2 files changed, 103 insertions(+) create mode 100644 docs/clients/generate-cli.mdx diff --git a/docs/clients/generate-cli.mdx b/docs/clients/generate-cli.mdx new file mode 100644 index 000000000..ce8c45e04 --- /dev/null +++ b/docs/clients/generate-cli.mdx @@ -0,0 +1,102 @@ +--- +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: + +```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 + +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. + +Parameters are mapped from JSON Schema to Python annotations: `string` → `str`, `integer` → `int`, `number` → `float`, `boolean` → `bool`, `array` → `list`, `object` → `dict`. Union types (`anyOf`) become Python unions like `str | int`. Required parameters are mandatory flags; optional ones use their schema default or `None`, and `None` values are filtered out before calling the server — so you only pass what matters. + +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` and `cyclopts` as dependencies. diff --git a/docs/docs.json b/docs/docs.json index de6369d10..76f8f7054 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", From 59f323ebb7f7f28815eb9b5d3458b4367f1f6c22 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 11:23:42 -0500 Subject: [PATCH 3/9] docs: add generate-cli documentation; skip Windows executable test --- docs/clients/generate-cli.mdx | 8 ++++++-- tests/cli/test_generate_cli.py | 4 ++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/docs/clients/generate-cli.mdx b/docs/clients/generate-cli.mdx index ce8c45e04..a96fab507 100644 --- a/docs/clients/generate-cli.mdx +++ b/docs/clients/generate-cli.mdx @@ -30,7 +30,7 @@ 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: +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 @@ -99,4 +99,8 @@ The most common edit is changing `CLIENT_SPEC`. If you generated from a local de 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` and `cyclopts` as dependencies. +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/tests/cli/test_generate_cli.py b/tests/cli/test_generate_cli.py index 919dc8ee0..16edf0656 100644 --- a/tests/cli/test_generate_cli.py +++ b/tests/cli/test_generate_cli.py @@ -1,5 +1,6 @@ """Tests for fastmcp generate-cli command.""" +import sys from pathlib import Path from typing import Any from unittest.mock import patch @@ -396,6 +397,9 @@ class TestGenerateCliCommand: 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" From 68cd6770a9c972964da724391561872d74cdfaa1 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 11:37:26 -0500 Subject: [PATCH 4/9] fix: address PR review feedback - Sanitize tool and parameter names to valid Python identifiers - Replace bare except Exception with specific exception types - Escape server name in generated string literals - Handle trailing colon edge case in _derive_server_name - Clarify in docs that generated CLI is a client, not a bundled server --- docs/clients/generate-cli.mdx | 2 + src/fastmcp/cli/generate.py | 31 ++++++++++----- tests/cli/test_generate_cli.py | 72 ++++++++++++++++++++++++++++++++++ 3 files changed, 96 insertions(+), 9 deletions(-) diff --git a/docs/clients/generate-cli.mdx b/docs/clients/generate-cli.mdx index a96fab507..578d5eae0 100644 --- a/docs/clients/generate-cli.mdx +++ b/docs/clients/generate-cli.mdx @@ -87,6 +87,8 @@ Tool names are preserved exactly as the server defines them — underscores stay ## 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. Parameters are mapped from JSON Schema to Python annotations: `string` → `str`, `integer` → `int`, `number` → `float`, `boolean` → `bool`, `array` → `list`, `object` → `dict`. Union types (`anyOf`) become Python unions like `str | int`. Required parameters are mandatory flags; optional ones use their schema default or `None`, and `None` values are filtered out before calling the server — so you only pass what matters. diff --git a/src/fastmcp/cli/generate.py b/src/fastmcp/cli/generate.py index ac8a446c9..fa46f9a3b 100644 --- a/src/fastmcp/cli/generate.py +++ b/src/fastmcp/cli/generate.py @@ -1,5 +1,6 @@ """Generate a standalone CLI script from an MCP server's capabilities.""" +import re import sys import textwrap from pathlib import Path @@ -7,6 +8,7 @@ from typing import Annotated, Any import cyclopts import mcp.types +from mcp import McpError from rich.console import Console from fastmcp.cli.client import _build_client, resolve_server_spec @@ -83,6 +85,14 @@ def serialize_transport( # --------------------------------------------------------------------------- +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}" + return safe or "_unnamed" + + def _tool_function_source(tool: mcp.types.Tool) -> str: """Generate the source for a single ``@call_tool_app.command`` function.""" schema = tool.inputSchema @@ -97,6 +107,7 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: py_type = _schema_type_to_python(prop_schema) help_text = prop_schema.get("description", "") is_required = prop_name in required + safe_name = _to_python_identifier(prop_name) # Escape quotes in help text help_escaped = help_text.replace("\\", "\\\\").replace('"', '\\"') @@ -105,22 +116,22 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: annotation = ( f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' ) - param_lines.append(f" {prop_name}: {annotation},") + param_lines.append(f" {safe_name}: {annotation},") else: default = prop_schema.get("default") if default is not None: annotation = ( f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' ) - param_lines.append(f" {prop_name}: {annotation} = {default!r},") + param_lines.append(f" {safe_name}: {annotation} = {default!r},") else: annotation = f'Annotated[{py_type} | None, cyclopts.Parameter(help="{help_escaped}")]' - param_lines.append(f" {prop_name}: {annotation} = None,") + param_lines.append(f" {safe_name}: {annotation} = None,") - call_args.append(f"{prop_name!r}: {prop_name}") + call_args.append(f"{prop_name!r}: {safe_name}") - # Function name: use tool name directly (preserve underscores) - fn_name = tool.name.replace("-", "_") + # Function name: sanitize to valid Python identifier + fn_name = _to_python_identifier(tool.name) # Docstring description = (tool.description or "").replace('"""', '\\"\\"\\"') @@ -191,8 +202,9 @@ def generate_cli_script( 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} MCP server")' + 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")' @@ -454,7 +466,7 @@ async def generate_cli_command( f"[dim]Discovered {len(tools)} tool(s) from {server_spec}[/dim]" ) - except Exception as exc: + except (RuntimeError, TimeoutError, McpError, OSError) as exc: console.print(f"[bold red]Error:[/bold red] Could not connect: {exc}") sys.exit(1) @@ -492,6 +504,7 @@ def _derive_server_name(server_spec: str) -> str: # Bare name or qualified name if ":" in server_spec: - return server_spec.split(":", 1)[1] + 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 index 16edf0656..98f147a89 100644 --- a/tests/cli/test_generate_cli.py +++ b/tests/cli/test_generate_cli.py @@ -14,6 +14,7 @@ from fastmcp.cli.client import Client from fastmcp.cli.generate import ( _derive_server_name, _schema_type_to_python, + _to_python_identifier, _tool_function_source, generate_cli_command, generate_cli_script, @@ -65,6 +66,31 @@ class TestSchemaTypeToPython: assert result == "str | None" +# --------------------------------------------------------------------------- +# _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 # --------------------------------------------------------------------------- @@ -173,6 +199,27 @@ class TestToolFunctionSource: 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", @@ -204,6 +251,9 @@ class TestDeriveServerName: 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 @@ -293,6 +343,28 @@ class TestGenerateCliScript: compile(script, "", "exec") assert "call_tool_app" 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) From 89c42420344dd14dd4b70834caaca85f6e8cbb3c Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 12:35:22 -0500 Subject: [PATCH 5/9] Fix string escaping issues in generate-cli - Use single-quoted docstrings to avoid triple-quote escaping issues - Escape quotes in app_name derived from server_name - Add tests for descriptions with quotes and server names with quotes Addresses CodeRabbit review comments about insufficient escaping. --- src/fastmcp/cli/generate.py | 12 +++++++----- tests/cli/test_generate_cli.py | 31 ++++++++++++++++++++++++++++++- 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/src/fastmcp/cli/generate.py b/src/fastmcp/cli/generate.py index fa46f9a3b..49f8ad2d9 100644 --- a/src/fastmcp/cli/generate.py +++ b/src/fastmcp/cli/generate.py @@ -133,8 +133,8 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: # Function name: sanitize to valid Python identifier fn_name = _to_python_identifier(tool.name) - # Docstring - description = (tool.description or "").replace('"""', '\\"\\"\\"') + # Docstring - use single-quoted docstrings to avoid triple-quote escaping issues + description = (tool.description or "").replace("\\", "\\\\").replace("'", "\\'") lines = [] lines.append("") @@ -148,7 +148,7 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: lines.extend(param_lines) lines.append(") -> None:") - lines.append(f' """{description}"""') + lines.append(f" '''{description}'''") dict_items = ", ".join(call_args) lines.append(f" await _call_tool({tool.name!r}, {{{dict_items}}})") lines.append("") @@ -170,8 +170,10 @@ def generate_cli_script( ) -> str: """Generate the full CLI script source code.""" - # Determine app name from server_name - app_name = server_name.replace(" ", "-").lower() + # Determine app name from server_name - sanitize for use in string literal + app_name = ( + server_name.replace(" ", "-").lower().replace("\\", "\\\\").replace('"', '\\"') + ) # --- Header --- lines: list[str] = [] diff --git a/tests/cli/test_generate_cli.py b/tests/cli/test_generate_cli.py index 98f147a89..dfae49a08 100644 --- a/tests/cli/test_generate_cli.py +++ b/tests/cli/test_generate_cli.py @@ -230,7 +230,22 @@ class TestToolFunctionSource: }, ) source = _tool_function_source(tool) - assert '"""Say hello to someone."""' in source + 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") # --------------------------------------------------------------------------- @@ -343,6 +358,20 @@ class TestGenerateCliScript: 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( From 8c8c074d107bb36e1cc445945e3b3c878fdc877d Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 12:57:01 -0500 Subject: [PATCH 6/9] Implement smart parameter handling for generate-cli - Simple types (str, int, float, bool): Direct typed flags - Arrays of simple types (list[str], list[int]): Repeatable flags via cyclopts - Complex types (objects, nested arrays): Accept JSON strings with parsing - JSON schema shown in help text for complex parameters - Proper escaping of newlines and quotes in help text - Filter out None and empty list defaults when calling tools This gives typed, discoverable CLIs for common cases while handling complex schemas via JSON input. --- src/fastmcp/cli/generate.py | 151 ++++++++++++++++++++++++++------ tests/cli/test_generate_cli.py | 153 ++++++++++++++++++++++++++------- 2 files changed, 245 insertions(+), 59 deletions(-) diff --git a/src/fastmcp/cli/generate.py b/src/fastmcp/cli/generate.py index 49f8ad2d9..2c9bb86fa 100644 --- a/src/fastmcp/cli/generate.py +++ b/src/fastmcp/cli/generate.py @@ -23,28 +23,89 @@ console = Console() # JSON Schema type → Python type string # --------------------------------------------------------------------------- -_JSON_SCHEMA_TYPE_MAP: dict[str, str] = { - "string": "str", - "integer": "int", - "number": "float", - "boolean": "bool", - "array": "list", - "object": "dict", - "null": "None", -} +_SIMPLE_TYPES = {"string", "integer", "number", "boolean", "null"} -def _schema_type_to_python(schema: dict[str, Any]) -> str: - """Convert a JSON Schema type fragment to a Python type annotation string.""" - if "anyOf" in schema: - parts = [_schema_type_to_python(s) for s in schema["anyOf"]] - return " | ".join(parts) - - schema_type = schema.get("type", "string") +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): - return " | ".join(_JSON_SCHEMA_TYPE_MAP.get(t, "str") for t in schema_type) + # 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 - return _JSON_SCHEMA_TYPE_MAP.get(schema_type, "str") + +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") + type_map = { + "string": "str", + "integer": "int", + "number": "float", + "boolean": "bool", + } + return True, type_map.get(item_type, "str") + + +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.""" + import json + + # Pretty print the schema, indented for help text + schema_str = json.dumps(schema, indent=2) + # Indent each line for help text alignment + lines = schema_str.split("\n") + indented = "\n ".join(lines) + return f"JSON Schema: {indented}" # --------------------------------------------------------------------------- @@ -99,19 +160,29 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: properties: dict[str, Any] = schema.get("properties", {}) required = set(schema.get("required", [])) - # Build parameter lines + # 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) for prop_name, prop_schema in properties.items(): - py_type = _schema_type_to_python(prop_schema) + 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) - # Escape quotes in help text - help_escaped = help_text.replace("\\", "\\\\").replace('"', '\\"') + # 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}")]' @@ -125,8 +196,12 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: ) param_lines.append(f" {safe_name}: {annotation} = {default!r},") else: - annotation = f'Annotated[{py_type} | None, cyclopts.Parameter(help="{help_escaped}")]' - param_lines.append(f" {safe_name}: {annotation} = None,") + # For list types, default to empty list; others default to None + if py_type.startswith("list["): + param_lines.append(f" {safe_name}: {py_type} = [],") + 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}") @@ -149,7 +224,26 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: lines.append(") -> None:") lines.append(f" '''{description}'''") - dict_items = ", ".join(call_args) + + # 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 {safe_name} else None" + ) + 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("") @@ -250,7 +344,12 @@ def generate_cli_script( async def _call_tool(tool_name: str, arguments: dict) -> None: - filtered = {k: v for k, v in arguments.items() if v is not 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) diff --git a/tests/cli/test_generate_cli.py b/tests/cli/test_generate_cli.py index dfae49a08..f86e2460c 100644 --- a/tests/cli/test_generate_cli.py +++ b/tests/cli/test_generate_cli.py @@ -13,7 +13,7 @@ from fastmcp.cli import generate as generate_module from fastmcp.cli.client import Client from fastmcp.cli.generate import ( _derive_server_name, - _schema_type_to_python, + _schema_to_python_type, _to_python_identifier, _tool_function_source, generate_cli_command, @@ -23,47 +23,61 @@ from fastmcp.cli.generate import ( from fastmcp.client.transports.stdio import StdioTransport # --------------------------------------------------------------------------- -# _schema_type_to_python +# _schema_to_python_type # --------------------------------------------------------------------------- -class TestSchemaTypeToPython: - def test_string(self): - assert _schema_type_to_python({"type": "string"}) == "str" +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_integer(self): - assert _schema_type_to_python({"type": "integer"}) == "int" + 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_number(self): - assert _schema_type_to_python({"type": "number"}) == "float" + 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_boolean(self): - assert _schema_type_to_python({"type": "boolean"}) == "bool" + 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(self): - assert _schema_type_to_python({"type": "array"}) == "list" - - def test_object(self): - assert _schema_type_to_python({"type": "object"}) == "dict" - - def test_null(self): - assert _schema_type_to_python({"type": "null"}) == "None" - - def test_unknown_defaults_to_str(self): - assert _schema_type_to_python({"type": "foobar"}) == "str" - - def test_missing_type_defaults_to_str(self): - assert _schema_type_to_python({}) == "str" - - def test_any_of(self): - result = _schema_type_to_python( - {"anyOf": [{"type": "string"}, {"type": "integer"}]} + def test_array_of_strings(self): + py_type, needs_json = _schema_to_python_type( + {"type": "array", "items": {"type": "string"}} ) - assert result == "str | int" + assert py_type == "list[str]" + assert needs_json is False - def test_type_list(self): - result = _schema_type_to_python({"type": ["string", "null"]}) - assert result == "str | None" + 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 # --------------------------------------------------------------------------- @@ -247,6 +261,79 @@ class TestToolFunctionSource: # 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 + assert "tags: list[str] = []" 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 + assert "metadata_parsed = json.loads(metadata) if metadata else None" 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)" in source + compile(source, "", "exec") + # --------------------------------------------------------------------------- # _derive_server_name From 3a5dc22ec0266932138d2427c417b0e9b55da60f Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 13:59:30 -0500 Subject: [PATCH 7/9] Update generate-cli docs to explain smart parameter handling - Document simple types as direct typed flags - Document arrays of simple types as repeatable flags - Document complex types as JSON strings with schema in help - Add examples showing all three patterns --- docs/clients/generate-cli.mdx | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/docs/clients/generate-cli.mdx b/docs/clients/generate-cli.mdx index 578d5eae0..833637423 100644 --- a/docs/clients/generate-cli.mdx +++ b/docs/clients/generate-cli.mdx @@ -91,7 +91,28 @@ The generated script is a client, not a server. It doesn't bundle or embed the M 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. -Parameters are mapped from JSON Schema to Python annotations: `string` → `str`, `integer` → `int`, `number` → `float`, `boolean` → `bool`, `array` → `list`, `object` → `dict`. Union types (`anyOf`) become Python unions like `str | int`. Required parameters are mandatory flags; optional ones use their schema default or `None`, and `None` values are filtered out before calling the server — so you only pass what matters. +### 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. From 8b285bf33ebb0ae33b7ddef4d2f3962fba61d3a3 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 14:13:39 -0500 Subject: [PATCH 8/9] Fix Codex review issues in generate-cli High priority fixes: - Complex type defaults: Serialize dict/list defaults to JSON strings - List params: Preserve help metadata with Annotated wrapper - Name collisions: Detect and error on sanitized name conflicts - JSON parsing: Use isinstance check for safety with defaults Added tests for: - Complex types with default values - Parameter name collision detection - Updated existing tests to match new format --- src/fastmcp/cli/generate.py | 30 ++++++++++++++++---- tests/cli/test_generate_cli.py | 52 ++++++++++++++++++++++++++++++---- 2 files changed, 71 insertions(+), 11 deletions(-) diff --git a/src/fastmcp/cli/generate.py b/src/fastmcp/cli/generate.py index 2c9bb86fa..9629c28db 100644 --- a/src/fastmcp/cli/generate.py +++ b/src/fastmcp/cli/generate.py @@ -164,6 +164,7 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: 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) @@ -171,6 +172,14 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: 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) @@ -191,14 +200,23 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: else: default = prop_schema.get("default") if default is not None: - annotation = ( - f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' - ) - param_lines.append(f" {safe_name}: {annotation} = {default!r},") + # For complex types with defaults, serialize to JSON string + if needs_json: + import json + + default_str = json.dumps(default) + 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["): - param_lines.append(f" {safe_name}: {py_type} = [],") + 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,") @@ -230,7 +248,7 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: lines.append(" # Parse JSON parameters") for _prop_name, safe_name in json_params: lines.append( - f" {safe_name}_parsed = json.loads({safe_name}) if {safe_name} else None" + f" {safe_name}_parsed = json.loads({safe_name}) if isinstance({safe_name}, str) else {safe_name}" ) lines.append("") diff --git a/tests/cli/test_generate_cli.py b/tests/cli/test_generate_cli.py index f86e2460c..c0f1aa43c 100644 --- a/tests/cli/test_generate_cli.py +++ b/tests/cli/test_generate_cli.py @@ -274,8 +274,9 @@ class TestToolFunctionSource: }, ) source = _tool_function_source(tool) - # Should use list[str] type - assert "tags: list[str] = []" in source + # 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") @@ -304,8 +305,11 @@ class TestToolFunctionSource: # Should include JSON schema in help (with escaped quotes) assert "JSON Schema:" in source assert '\\"type\\": \\"object\\"' in source - # Should have JSON parsing - assert "metadata_parsed = json.loads(metadata) if metadata else None" 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") @@ -331,9 +335,47 @@ class TestToolFunctionSource: # Nested arrays need JSON parsing assert "batches: Annotated[str" in source assert "JSON Schema:" in source - assert "batches_parsed = json.loads(batches)" 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 + 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 From 24e6a42f68c25fe696523af57b39ea1c6c8b6f26 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Tue, 3 Feb 2026 14:15:15 -0500 Subject: [PATCH 9/9] Use pydantic_core.to_json for consistency - Generator now uses pydantic_core.to_json() instead of json.dumps() - Consistent with rest of fastmcp codebase - Generated CLI still uses plain json module (standalone script) --- src/fastmcp/cli/generate.py | 8 ++++---- tests/cli/test_generate_cli.py | 3 ++- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/fastmcp/cli/generate.py b/src/fastmcp/cli/generate.py index 9629c28db..559847409 100644 --- a/src/fastmcp/cli/generate.py +++ b/src/fastmcp/cli/generate.py @@ -98,10 +98,10 @@ def _schema_to_python_type(schema: dict[str, Any]) -> tuple[str, bool]: def _format_schema_for_help(schema: dict[str, Any]) -> str: """Format a JSON schema for display in help text.""" - import json + import pydantic_core # Pretty print the schema, indented for help text - schema_str = json.dumps(schema, indent=2) + 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) @@ -202,9 +202,9 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: if default is not None: # For complex types with defaults, serialize to JSON string if needs_json: - import json + import pydantic_core - default_str = json.dumps(default) + 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}," diff --git a/tests/cli/test_generate_cli.py b/tests/cli/test_generate_cli.py index c0f1aa43c..f513338d7 100644 --- a/tests/cli/test_generate_cli.py +++ b/tests/cli/test_generate_cli.py @@ -356,7 +356,8 @@ class TestToolFunctionSource: ) source = _tool_function_source(tool) # Default should be JSON string, not Python dict - assert '= \'{"timeout": 30, "retry": true}\'' in source + # 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")