Skip to content
Open
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
14 changes: 13 additions & 1 deletion src/mcp/server/mcpserver/utilities/func_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
continue

field_info = key_to_field_info[data_key]
if isinstance(data_value, str) and field_info.annotation is not str:
if isinstance(data_value, str) and _should_pre_parse_json(field_info.annotation):
try:
pre_parsed = json.loads(data_value)
except json.JSONDecodeError:
Expand Down Expand Up @@ -420,6 +420,18 @@ def _try_create_model_and_schema(

_no_default = object()

_SIMPLE_TYPES: frozenset[type] = frozenset({str, int, float, bool, type(None)})


def _should_pre_parse_json(annotation: Any) -> bool:
"""Return True if the annotation requires JSON pre-parsing."""
if annotation is str:
return False
origin = get_origin(annotation)
if origin is not None and is_union_origin(origin):
return any(arg not in _SIMPLE_TYPES for arg in get_args(annotation))
return True


def _create_model_from_class(cls: type[Any], type_hints: dict[str, Any]) -> type[BaseModel]:
"""Create a Pydantic model from an ordinary class.
Expand Down
25 changes: 25 additions & 0 deletions tests/server/mcpserver/test_func_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -551,6 +551,31 @@ def handle_json_payload(payload: str, strict_mode: bool = False) -> str:
assert result == f"Handled payload of length {len(json_array_payload)}"


def test_str_union_pre_parse_json():
"""Regression test for #1873: pre_parse_json should not JSON-parse strings
when the annotation is a union of simple types like str | None.
"""

def func_optional_str(value: str | None = None) -> str: # pragma: no cover
return str(value)

meta = func_metadata(func_optional_str)

# str | None: JSON object/array strings should be preserved, not parsed
json_obj = '{"database": "postgres", "port": 5432}'
assert meta.pre_parse_json({"value": json_obj})["value"] == json_obj

json_array = '["item1", "item2"]'
assert meta.pre_parse_json({"value": json_array})["value"] == json_array

# Complex unions like list[str] | None should still pre-parse
def func_optional_list(items: list[str] | None = None) -> str: # pragma: no cover
return str(items)

meta_list = func_metadata(func_optional_list)
assert meta_list.pre_parse_json({"items": '["a", "b", "c"]'})["items"] == ["a", "b", "c"]


# Tests for structured output functionality


Expand Down