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
18 changes: 17 additions & 1 deletion src/mcp/server/mcpserver/utilities/func_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,22 @@
logger = get_logger(__name__)


def _annotation_accepts_str(annotation: Any) -> bool:
"""Check if an annotation only accepts str (or None), not other complex types.

Returns True for `str` and `Optional[str]` (str | None), where JSON parsing
a string value would be incorrect. Returns False for unions like `str | list[str]`
where the other types benefit from JSON pre-parsing.
"""
if annotation is str:
return True
if is_union_origin(get_origin(annotation)):
args = get_args(annotation)
non_none_args = [a for a in args if a is not type(None)]
return non_none_args == [str]
return False


class StrictJsonSchema(GenerateJsonSchema):
"""A JSON schema generator that raises exceptions instead of emitting warnings.

Expand Down Expand Up @@ -148,7 +164,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 not _annotation_accepts_str(field_info.annotation):
try:
pre_parsed = json.loads(data_value)
except json.JSONDecodeError:
Expand Down
68 changes: 68 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,74 @@ def handle_json_payload(payload: str, strict_mode: bool = False) -> str:
assert result == f"Handled payload of length {len(json_array_payload)}"


def test_optional_str_preserves_json_string():
"""Regression test for issue #381: Ensure that when a parameter is annotated
as Optional[str] or str | None, valid JSON strings are NOT parsed into Python
objects. The pre_parse_json check must recognize that union types containing
str should be treated as string-accepting annotations.
"""
from typing import Optional

def func_optional_str(cid: Optional[str]) -> str: # pragma: no cover
return cid or ""

meta = func_metadata(func_optional_str)

# A numeric string like "1.2" must NOT be parsed into float
result = meta.pre_parse_json({"cid": "1.2"})
assert result["cid"] == "1.2"
assert isinstance(result["cid"], str)

# A JSON object string must NOT be parsed into dict
result = meta.pre_parse_json({"cid": '{"key": "value"}'})
assert result["cid"] == '{"key": "value"}'
assert isinstance(result["cid"], str)

# A JSON array string must NOT be parsed into list
result = meta.pre_parse_json({"cid": '["a", "b"]'})
assert result["cid"] == '["a", "b"]'
assert isinstance(result["cid"], str)


def test_pipe_none_str_preserves_json_string():
"""Same as test_optional_str but using str | None syntax."""

def func_str_or_none(data: str | None) -> str: # pragma: no cover
return data or ""

meta = func_metadata(func_str_or_none)

# Numeric string must stay as str
result = meta.pre_parse_json({"data": "42"})
assert result["data"] == "42"
assert isinstance(result["data"], str)

# JSON dict must stay as str
result = meta.pre_parse_json({"data": '{"x": 1}'})
assert result["data"] == '{"x": 1}'
assert isinstance(result["data"], str)


@pytest.mark.anyio
async def test_optional_str_runtime_validation():
"""End-to-end test: calling a function with Optional[str] param and JSON-like input."""
from typing import Optional

def process(config: Optional[str] = None) -> str:
assert config is None or isinstance(config, str)
return f"got: {config}"

meta = func_metadata(process)

result = await meta.call_fn_with_arg_validation(
process,
fn_is_async=False,
arguments_to_validate={"config": '{"db": "postgres"}'},
arguments_to_pass_directly=None,
)
assert result == 'got: {"db": "postgres"}'


# Tests for structured output functionality


Expand Down
Loading