diff --git a/src/fastmcp/cli/generate.py b/src/fastmcp/cli/generate.py index 2c9bb86fa..9629c28db 100644 --- a/src/fastmcp/cli/generate.py +++ b/src/fastmcp/cli/generate.py @@ -164,6 +164,7 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: param_lines: list[str] = [] call_args: list[str] = [] json_params: list[tuple[str, str]] = [] # (prop_name, safe_name) + seen_names: dict[str, str] = {} # safe_name -> original prop_name for prop_name, prop_schema in properties.items(): py_type, needs_json = _schema_to_python_type(prop_schema) @@ -171,6 +172,14 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: is_required = prop_name in required safe_name = _to_python_identifier(prop_name) + # Check for name collisions after sanitization + if safe_name in seen_names: + raise ValueError( + f"Parameter name collision: '{prop_name}' and '{seen_names[safe_name]}' " + f"both sanitize to '{safe_name}'" + ) + seen_names[safe_name] = prop_name + # For complex types, add schema to help text if needs_json: schema_help = _format_schema_for_help(prop_schema) @@ -191,14 +200,23 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: else: default = prop_schema.get("default") if default is not None: - annotation = ( - f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' - ) - param_lines.append(f" {safe_name}: {annotation} = {default!r},") + # For complex types with defaults, serialize to JSON string + if needs_json: + import json + + default_str = json.dumps(default) + annotation = f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + param_lines.append( + f" {safe_name}: {annotation} = {default_str!r}," + ) + else: + annotation = f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + param_lines.append(f" {safe_name}: {annotation} = {default!r},") else: # For list types, default to empty list; others default to None if py_type.startswith("list["): - param_lines.append(f" {safe_name}: {py_type} = [],") + annotation = f'Annotated[{py_type}, cyclopts.Parameter(help="{help_escaped}")]' + param_lines.append(f" {safe_name}: {annotation} = [],") else: annotation = f'Annotated[{py_type} | None, cyclopts.Parameter(help="{help_escaped}")]' param_lines.append(f" {safe_name}: {annotation} = None,") @@ -230,7 +248,7 @@ def _tool_function_source(tool: mcp.types.Tool) -> str: lines.append(" # Parse JSON parameters") for _prop_name, safe_name in json_params: lines.append( - f" {safe_name}_parsed = json.loads({safe_name}) if {safe_name} else None" + f" {safe_name}_parsed = json.loads({safe_name}) if isinstance({safe_name}, str) else {safe_name}" ) lines.append("") diff --git a/tests/cli/test_generate_cli.py b/tests/cli/test_generate_cli.py index f86e2460c..c0f1aa43c 100644 --- a/tests/cli/test_generate_cli.py +++ b/tests/cli/test_generate_cli.py @@ -274,8 +274,9 @@ class TestToolFunctionSource: }, ) source = _tool_function_source(tool) - # Should use list[str] type - assert "tags: list[str] = []" in source + # Should use list[str] type with help metadata + assert "tags: Annotated[list[str]" in source + assert "= []" in source # Should not have JSON parsing for simple arrays assert "json.loads" not in source compile(source, "", "exec") @@ -304,8 +305,11 @@ class TestToolFunctionSource: # Should include JSON schema in help (with escaped quotes) assert "JSON Schema:" in source assert '\\"type\\": \\"object\\"' in source - # Should have JSON parsing - assert "metadata_parsed = json.loads(metadata) if metadata else None" in source + # Should have JSON parsing with isinstance check + assert ( + "metadata_parsed = json.loads(metadata) if isinstance(metadata, str) else metadata" + in source + ) # Should use parsed version in call assert "'metadata': metadata_parsed" in source compile(source, "", "exec") @@ -331,9 +335,47 @@ class TestToolFunctionSource: # Nested arrays need JSON parsing assert "batches: Annotated[str" in source assert "JSON Schema:" in source - assert "batches_parsed = json.loads(batches)" in source + assert ( + "batches_parsed = json.loads(batches) if isinstance(batches, str) else batches" + in source + ) compile(source, "", "exec") + def test_complex_type_with_default(self): + """Test that complex types with defaults are JSON-serialized.""" + tool = mcp.types.Tool( + name="configure", + inputSchema={ + "properties": { + "options": { + "type": "object", + "default": {"timeout": 30, "retry": True}, + }, + }, + }, + ) + source = _tool_function_source(tool) + # Default should be JSON string, not Python dict + assert '= \'{"timeout": 30, "retry": true}\'' in source + # Should parse safely even with default + assert "isinstance(options, str)" in source + compile(source, "", "exec") + + def test_name_collision_detection(self): + """Test that parameter name collisions are detected.""" + tool = mcp.types.Tool( + name="test", + inputSchema={ + "properties": { + "content-type": {"type": "string"}, + "content_type": {"type": "string"}, + }, + }, + ) + # Should raise ValueError for collision + with pytest.raises(ValueError, match="both sanitize to 'content_type'"): + _tool_function_source(tool) + # --------------------------------------------------------------------------- # _derive_server_name