ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add fastmcp generate-cli command (#3065)

Browse source 
* 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.

* docs: add generate-cli documentation

* docs: add generate-cli documentation; skip Windows executable test

* 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

* 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.

* 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.

* 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

* 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

* 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)

* Move local imports to module level in generate-cli

* Handle union item types and Python keyword collisions in generate-cli
This commit is contained in:
Jeremiah Lowin 2026-02-03 21:08:51 -05:00 • committed by GitHub
commit 4262cfc16a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 1404 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

129
docs/clients/generate-cli.mdx Normal file
View file

@ -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'
<VersionBadge version="3.0.0" />
`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
```

1
docs/docs.json
View file

@ -169,6 +169,7 @@
"pages": [
"clients/client",
"clients/cli",
"clients/generate-cli",
"clients/transports",
{
"group": "Core Operations",

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

@ -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__":

634
src/fastmcp/cli/generate.py Normal file
View file

@ -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

638
tests/cli/test_generate_cli.py Normal file
View file

@ -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, "<test>", "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, "<test>", "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, "<test>", "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, "<test>", "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, "<test>", "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, "<generated>", "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, "<generated>", "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, "<generated>", "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, "<generated>", "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, "<generated>", "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