mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Compare commits
9 commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
24e6a42f68 |
||
|
|
8b285bf33e |
||
|
|
3a5dc22ec0 |
||
|
|
8c8c074d10 |
||
|
|
89c4242034 |
||
|
|
68cd6770a9 |
||
|
|
59f323ebb7 |
||
|
|
d2209059ef |
||
|
|
2c705faa87 |
5 changed files with 1399 additions and 0 deletions
129
docs/clients/generate-cli.mdx
Normal file
129
docs/clients/generate-cli.mdx
Normal 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
|
||||
```
|
||||
|
|
@ -169,6 +169,7 @@
|
|||
"pages": [
|
||||
"clients/client",
|
||||
"clients/cli",
|
||||
"clients/generate-cli",
|
||||
"clients/transports",
|
||||
{
|
||||
"group": "Core Operations",
|
||||
|
|
|
|||
|
|
@ -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__":
|
||||
|
|
|
|||
629
src/fastmcp/cli/generate.py
Normal file
629
src/fastmcp/cli/generate.py
Normal file
|
|
@ -0,0 +1,629 @@
|
|||
"""Generate a standalone CLI script from an MCP server's capabilities."""
|
||||
|
||||
import re
|
||||
import sys
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
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
|
||||
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")
|
||||
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 pydantic_core
|
||||
|
||||
# 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}"
|
||||
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
|
||||
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:
|
||||
import pydantic_core
|
||||
|
||||
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://")):
|
||||
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:
|
||||
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
638
tests/cli/test_generate_cli.py
Normal 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
|
||||
Loading…
Add table
Add a link
Reference in a new issue