diff --git a/src/mcp/server/fastmcp/tools/base.py b/src/mcp/server/fastmcp/tools/base.py index 3d8827ed7..f50126081 100644 --- a/src/mcp/server/fastmcp/tools/base.py +++ b/src/mcp/server/fastmcp/tools/base.py @@ -73,7 +73,7 @@ def from_function( skip_names=[context_kwarg] if context_kwarg is not None else [], structured_output=structured_output, ) - parameters = func_arg_metadata.arg_model.model_json_schema() + parameters = func_arg_metadata.arg_model.model_json_schema(by_alias=True) return cls( fn=fn, diff --git a/src/mcp/server/fastmcp/utilities/func_metadata.py b/src/mcp/server/fastmcp/utilities/func_metadata.py index 3c9506fba..70be8796d 100644 --- a/src/mcp/server/fastmcp/utilities/func_metadata.py +++ b/src/mcp/server/fastmcp/utilities/func_metadata.py @@ -47,8 +47,11 @@ def model_dump_one_level(self) -> dict[str, Any]: That is, sub-models etc are not dumped - they are kept as pydantic models. """ kwargs: dict[str, Any] = {} - for field_name in self.__class__.model_fields.keys(): - kwargs[field_name] = getattr(self, field_name) + for field_name, field_info in self.__class__.model_fields.items(): + value = getattr(self, field_name) + # Use the alias if it exists, otherwise use the field name + output_name = field_info.alias if field_info.alias else field_name + kwargs[output_name] = value return kwargs model_config = ConfigDict( @@ -127,12 +130,23 @@ def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]: dicts (JSON objects) as JSON strings, which can be pre-parsed here. """ new_data = data.copy() # Shallow copy - for field_name in self.arg_model.model_fields.keys(): - if field_name not in data.keys(): + + # Build a mapping from input keys (including aliases) to field info + key_to_field_info: dict[str, FieldInfo] = {} + for field_name, field_info in self.arg_model.model_fields.items(): + # Map both the field name and its alias (if any) to the field info + key_to_field_info[field_name] = field_info + if field_info.alias: + key_to_field_info[field_info.alias] = field_info + + for data_key in data.keys(): + if data_key not in key_to_field_info: continue - if isinstance(data[field_name], str) and self.arg_model.model_fields[field_name].annotation is not str: + + field_info = key_to_field_info[data_key] + if isinstance(data[data_key], str) and field_info.annotation is not str: try: - pre_parsed = json.loads(data[field_name]) + pre_parsed = json.loads(data[data_key]) except json.JSONDecodeError: continue # Not JSON - skip if isinstance(pre_parsed, str | int | float): @@ -140,7 +154,7 @@ def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]: # Should really be parsed as '"hello"' in Python - but if we parse # it as JSON it'll turn into just 'hello'. So we skip it. continue - new_data[field_name] = pre_parsed + new_data[data_key] = pre_parsed assert new_data.keys() == data.keys() return new_data @@ -222,7 +236,19 @@ def func_metadata( _get_typed_annotation(annotation, globalns), param.default if param.default is not inspect.Parameter.empty else PydanticUndefined, ) - dynamic_pydantic_model_params[param.name] = (field_info.annotation, field_info) + + # Check if the parameter name conflicts with BaseModel attributes + # This is necessary because Pydantic warns about shadowing parent attributes + if hasattr(BaseModel, param.name) and callable(getattr(BaseModel, param.name)): + # Use an alias to avoid the shadowing warning + field_info.alias = param.name + field_info.validation_alias = param.name + field_info.serialization_alias = param.name + # Use a prefixed internal name + internal_name = f"field_{param.name}" + dynamic_pydantic_model_params[internal_name] = (field_info.annotation, field_info) + else: + dynamic_pydantic_model_params[param.name] = (field_info.annotation, field_info) continue arguments_model = create_model( diff --git a/tests/server/fastmcp/test_func_metadata.py b/tests/server/fastmcp/test_func_metadata.py index 840509906..b5010fcb8 100644 --- a/tests/server/fastmcp/test_func_metadata.py +++ b/tests/server/fastmcp/test_func_metadata.py @@ -884,3 +884,115 @@ def func_with_aliases() -> ModelWithAliases: assert "field_second" not in structured_content_defaults assert structured_content_defaults["first"] is None assert structured_content_defaults["second"] is None + + +def test_basemodel_reserved_names(): + """Test that functions with parameters named after BaseModel methods work correctly""" + + def func_with_reserved_names( + model_dump: str, + model_validate: int, + dict: list[str], + json: dict[str, Any], + validate: bool, + copy: float, + normal_param: str, + ) -> str: + return f"{model_dump}, {model_validate}, {dict}, {json}, {validate}, {copy}, {normal_param}" + + meta = func_metadata(func_with_reserved_names) + + # Check that the schema has all the original parameter names (using aliases) + schema = meta.arg_model.model_json_schema(by_alias=True) + assert "model_dump" in schema["properties"] + assert "model_validate" in schema["properties"] + assert "dict" in schema["properties"] + assert "json" in schema["properties"] + assert "validate" in schema["properties"] + assert "copy" in schema["properties"] + assert "normal_param" in schema["properties"] + + +@pytest.mark.anyio +async def test_basemodel_reserved_names_validation(): + """Test that validation and calling works with reserved parameter names""" + + def func_with_reserved_names( + model_dump: str, + model_validate: int, + dict: list[str], + json: dict[str, Any], + validate: bool, + normal_param: str, + ) -> str: + return f"{model_dump}|{model_validate}|{len(dict)}|{json}|{validate}|{normal_param}" + + meta = func_metadata(func_with_reserved_names) + + # Test validation with reserved names + result = await meta.call_fn_with_arg_validation( + func_with_reserved_names, + fn_is_async=False, + arguments_to_validate={ + "model_dump": "test_dump", + "model_validate": 42, + "dict": ["a", "b", "c"], + "json": {"key": "value"}, + "validate": True, + "normal_param": "normal", + }, + arguments_to_pass_directly=None, + ) + + assert result == "test_dump|42|3|{'key': 'value'}|True|normal" + + # Test that the model can still call its own methods + model_instance = meta.arg_model.model_validate( + { + "model_dump": "dump_value", + "model_validate": 123, + "dict": ["x", "y"], + "json": {"foo": "bar"}, + "validate": False, + "normal_param": "test", + } + ) + + # The model should still have its methods accessible + assert hasattr(model_instance, "model_dump") + assert callable(model_instance.model_dump) + + # model_dump_one_level should return the original parameter names + dumped = model_instance.model_dump_one_level() + assert dumped["model_dump"] == "dump_value" + assert dumped["model_validate"] == 123 + assert dumped["dict"] == ["x", "y"] + assert dumped["json"] == {"foo": "bar"} + assert dumped["validate"] is False + assert dumped["normal_param"] == "test" + + +def test_basemodel_reserved_names_with_json_preparsing(): + """Test that pre_parse_json works correctly with reserved parameter names""" + + def func_with_reserved_json( + json: dict[str, Any], + model_dump: list[int], + normal: str, + ) -> str: + return "ok" + + meta = func_metadata(func_with_reserved_json) + + # Test pre-parsing with reserved names + result = meta.pre_parse_json( + { + "json": '{"nested": "data"}', # JSON string that should be parsed + "model_dump": "[1, 2, 3]", # JSON string that should be parsed + "normal": "plain string", # Should remain as string + } + ) + + assert result["json"] == {"nested": "data"} + assert result["model_dump"] == [1, 2, 3] + assert result["normal"] == "plain string"