Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/mcp/server/fastmcp/tools/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
42 changes: 34 additions & 8 deletions src/mcp/server/fastmcp/utilities/func_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -127,20 +130,31 @@ 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):
# This is likely that the raw value is e.g. `"hello"` which we
# 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

Expand Down Expand Up @@ -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(
Expand Down
112 changes: 112 additions & 0 deletions tests/server/fastmcp/test_func_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Loading