mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-23 05:54:19 +02:00
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
This commit is contained in:
Jeremiah Lowin
parent
59f323ebb7
commit
68cd6770a9
3 changed files with 96 additions and 9 deletions
|
|
@ -87,6 +87,8 @@ Tool names are preserved exactly as the server defines them — underscores stay
|
|||
|
||||
## How It Works
|
||||
|
||||
The generated script is a client, not a server. It doesn't bundle or embed the MCP server — it connects to it on every invocation. For URL-based servers, the server needs to be running. For stdio-based servers, the command specified in `CLIENT_SPEC` must be available on the system's `PATH`.
|
||||
|
||||
At the top of the generated file, a `CLIENT_SPEC` variable holds the resolved transport: either a URL string or a `StdioTransport` with the command and arguments baked in. Every invocation connects through this spec, so the script works without any external configuration.
|
||||
|
||||
Parameters are mapped from JSON Schema to Python annotations: `string` → `str`, `integer` → `int`, `number` → `float`, `boolean` → `bool`, `array` → `list`, `object` → `dict`. Union types (`anyOf`) become Python unions like `str | int`. Required parameters are mandatory flags; optional ones use their schema default or `None`, and `None` values are filtered out before calling the server — so you only pass what matters.
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""Generate a standalone CLI script from an MCP server's capabilities."""
|
||||
|
||||
import re
|
||||
import sys
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
|
@ -7,6 +8,7 @@ from typing import Annotated, Any
|
|||
|
||||
import cyclopts
|
||||
import mcp.types
|
||||
from mcp import McpError
|
||||
from rich.console import Console
|
||||
|
||||
from fastmcp.cli.client import _build_client, resolve_server_spec
|
||||
|
|
@ -83,6 +85,14 @@ def serialize_transport(
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _to_python_identifier(name: str) -> str:
|
||||
"""Sanitize a string into a valid Python identifier."""
|
||||
safe = re.sub(r"[^a-zA-Z0-9_]", "_", name)
|
||||
if safe and safe[0].isdigit():
|
||||
safe = f"_{safe}"
|
||||
return safe or "_unnamed"
|
||||
|
||||
|
||||
def _tool_function_source(tool: mcp.types.Tool) -> str:
|
||||
"""Generate the source for a single ``@call_tool_app.command`` function."""
|
||||
schema = tool.inputSchema
|
||||
|
|
@ -97,6 +107,7 @@ def _tool_function_source(tool: mcp.types.Tool) -> str:
|
|||
py_type = _schema_type_to_python(prop_schema)
|
||||
help_text = prop_schema.get("description", "")
|
||||
is_required = prop_name in required
|
||||
safe_name = _to_python_identifier(prop_name)
|
||||
|
||||
# Escape quotes in help text
|
||||
help_escaped = help_text.replace("\\", "\\\\").replace('"', '\\"')
|
||||
|
|
@ -105,22 +116,22 @@ def _tool_function_source(tool: mcp.types.Tool) -> str:
|
|||
annotation = (
|
||||
f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]'
|
||||
)
|
||||
param_lines.append(f" {prop_name}: {annotation},")
|
||||
param_lines.append(f" {safe_name}: {annotation},")
|
||||
else:
|
||||
default = prop_schema.get("default")
|
||||
if default is not None:
|
||||
annotation = (
|
||||
f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]'
|
||||
)
|
||||
param_lines.append(f" {prop_name}: {annotation} = {default!r},")
|
||||
param_lines.append(f" {safe_name}: {annotation} = {default!r},")
|
||||
else:
|
||||
annotation = f'Annotated[{py_type} | None, cyclopts.Parameter(help="{help_escaped}")]'
|
||||
param_lines.append(f" {prop_name}: {annotation} = None,")
|
||||
param_lines.append(f" {safe_name}: {annotation} = None,")
|
||||
|
||||
call_args.append(f"{prop_name!r}: {prop_name}")
|
||||
call_args.append(f"{prop_name!r}: {safe_name}")
|
||||
|
||||
# Function name: use tool name directly (preserve underscores)
|
||||
fn_name = tool.name.replace("-", "_")
|
||||
# Function name: sanitize to valid Python identifier
|
||||
fn_name = _to_python_identifier(tool.name)
|
||||
|
||||
# Docstring
|
||||
description = (tool.description or "").replace('"""', '\\"\\"\\"')
|
||||
|
|
@ -191,8 +202,9 @@ def generate_cli_script(
|
|||
lines.append("")
|
||||
|
||||
# --- App setup ---
|
||||
server_name_escaped = server_name.replace("\\", "\\\\").replace('"', '\\"')
|
||||
lines.append(
|
||||
f'app = cyclopts.App(name="{app_name}", help="CLI for {server_name} MCP server")'
|
||||
f'app = cyclopts.App(name="{app_name}", help="CLI for {server_name_escaped} MCP server")'
|
||||
)
|
||||
lines.append(
|
||||
'call_tool_app = cyclopts.App(name="call-tool", help="Call a tool on the server")'
|
||||
|
|
@ -454,7 +466,7 @@ async def generate_cli_command(
|
|||
f"[dim]Discovered {len(tools)} tool(s) from {server_spec}[/dim]"
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
except (RuntimeError, TimeoutError, McpError, OSError) as exc:
|
||||
console.print(f"[bold red]Error:[/bold red] Could not connect: {exc}")
|
||||
sys.exit(1)
|
||||
|
||||
|
|
@ -492,6 +504,7 @@ def _derive_server_name(server_spec: str) -> str:
|
|||
|
||||
# Bare name or qualified name
|
||||
if ":" in server_spec:
|
||||
return server_spec.split(":", 1)[1]
|
||||
name = server_spec.split(":", 1)[1]
|
||||
return name or server_spec.split(":", 1)[0]
|
||||
|
||||
return server_spec
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ from fastmcp.cli.client import Client
|
|||
from fastmcp.cli.generate import (
|
||||
_derive_server_name,
|
||||
_schema_type_to_python,
|
||||
_to_python_identifier,
|
||||
_tool_function_source,
|
||||
generate_cli_command,
|
||||
generate_cli_script,
|
||||
|
|
@ -65,6 +66,31 @@ class TestSchemaTypeToPython:
|
|||
assert result == "str | None"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _to_python_identifier
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestToPythonIdentifier:
|
||||
def test_plain_name(self):
|
||||
assert _to_python_identifier("hello") == "hello"
|
||||
|
||||
def test_hyphens(self):
|
||||
assert _to_python_identifier("get-forecast") == "get_forecast"
|
||||
|
||||
def test_dots_and_slashes(self):
|
||||
assert _to_python_identifier("a.b/c") == "a_b_c"
|
||||
|
||||
def test_leading_digit(self):
|
||||
assert _to_python_identifier("3d_render") == "_3d_render"
|
||||
|
||||
def test_spaces(self):
|
||||
assert _to_python_identifier("my tool") == "my_tool"
|
||||
|
||||
def test_empty_string(self):
|
||||
assert _to_python_identifier("") == "_unnamed"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# serialize_transport
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -173,6 +199,27 @@ class TestToolFunctionSource:
|
|||
source = _tool_function_source(tool)
|
||||
assert "async def get_forecast(" in source
|
||||
|
||||
def test_sanitizes_tool_name(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="my.tool/v2",
|
||||
inputSchema={"properties": {}},
|
||||
)
|
||||
source = _tool_function_source(tool)
|
||||
assert "async def my_tool_v2(" in source
|
||||
assert "name='my.tool/v2'" in source
|
||||
|
||||
def test_sanitizes_param_name(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="fetch",
|
||||
inputSchema={
|
||||
"properties": {"content-type": {"type": "string", "description": "CT"}},
|
||||
"required": ["content-type"],
|
||||
},
|
||||
)
|
||||
source = _tool_function_source(tool)
|
||||
assert "content_type: Annotated[str" in source
|
||||
assert "'content-type': content_type" in source
|
||||
|
||||
def test_description_in_docstring(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="greet",
|
||||
|
|
@ -204,6 +251,9 @@ class TestDeriveServerName:
|
|||
def test_url(self):
|
||||
assert _derive_server_name("http://localhost:8000/mcp") == "localhost"
|
||||
|
||||
def test_trailing_colon(self):
|
||||
assert _derive_server_name("source:") == "source"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# generate_cli_script — produces compilable Python
|
||||
|
|
@ -293,6 +343,28 @@ class TestGenerateCliScript:
|
|||
compile(script, "<generated>", "exec")
|
||||
assert "call_tool_app" in script
|
||||
|
||||
def test_compiles_with_unusual_names(self):
|
||||
tools = [
|
||||
mcp.types.Tool(
|
||||
name="my.tool/v2",
|
||||
description="A tool with dots and slashes",
|
||||
inputSchema={
|
||||
"properties": {
|
||||
"content-type": {"type": "string", "description": "CT"},
|
||||
},
|
||||
"required": ["content-type"],
|
||||
},
|
||||
),
|
||||
]
|
||||
script = generate_cli_script(
|
||||
server_name="test",
|
||||
server_spec="test",
|
||||
transport_code='"http://localhost:8000/mcp"',
|
||||
extra_imports=set(),
|
||||
tools=tools,
|
||||
)
|
||||
compile(script, "<generated>", "exec")
|
||||
|
||||
def test_compiles_with_stdio_transport(self):
|
||||
transport = StdioTransport(command="fastmcp", args=["run", "server.py"])
|
||||
transport_code, extra_imports = serialize_transport(transport)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue