fix(mcp): Add schema-driven type conversion (#11796)

* fix: nightly now properly gets 1.9.0 branch (#12215)

before it was attempting to pull release-notes as letters are alphanumerically after numbers when we sort -V then grab tail
now we only look at branch names that follow the pattern '^release-[0-9]+\.[0-9]+\.[0-9]+$'

* fix(mcp): Add schema-driven type conversion

- Add schema-driven type conversion (str→dict, str→int, etc.)
- normalize and unflatten tool arguments for MCP servers
- Unflatten flattened keys (e.g. params.search) into nested objects

* fix(mcp): handle dict type and array type in JSON schema for MCP tools

- Support "type": ["string", "null"] (JSON Schema array type)
- Normalize required to hashable elements (filter non-string entries)
- Add unit tests for create_input_schema_from_json_schema

* fix(mcp): map generic object type to dict for free-form params

When JSON schema has {"type": "object"} with no properties, treat it as
a free-form dict instead of building a nested Pydantic model. This allows
MCP servers expecting arbitrary key-value params to receive proper dicts,
and enables str→dict conversion via _normalize_arguments_for_mcp.

- Add conditional in parse_type: empty properties → dict, else nested model
- Add test_create_input_schema_generic_object_maps_to_dict

* fix(mcp): exclude None from tool arguments sent to MCP servers

* test_mcp: add more tests for datatypes

* fix(mcp): parse JSON strings for nested model params (e.g. foreman-mcp)

* fix(mcp): refactor _try_convert_value reducing repetition

* fix(mcp): add missing tests for remaining use cases

* fix(mcp): Fix mapping of None if expected is list, dict or str

* Update nightly_build.yml

---------

Co-authored-by: Adam-Aghili <149833988+Adam-Aghili@users.noreply.github.com>
Co-authored-by: Eric Hare <ericrhare@gmail.com>

src/backend/tests/unit/base/mcp/test_mcp_util.py

@@ -1108,6 +1108,78 @@ class TestMCPUtilityFunctions:
         with pytest.raises(Exception):  # noqa: B017, PT011
             model_class(bar=1)  # missing required field
 
+    def test_create_input_schema_type_array_string_null(self):
+        """Test schema with type array ['string', 'null'] produces optional string field."""
+        schema = {
+            "type": "object",
+            "properties": {
+                "optional_str": {"type": ["string", "null"], "description": "Optional string"},
+            },
+        }
+        model_class = util.create_input_schema_from_json_schema(schema)
+        # Optional: None and string both valid
+        instance_none = model_class(optional_str=None)
+        assert instance_none.optional_str is None
+        instance_str = model_class(optional_str="hello")
+        assert instance_str.optional_str == "hello"
+
+    def test_create_input_schema_type_array_integer_null(self):
+        """Test schema with type array ['integer', 'null'] produces optional int field."""
+        schema = {
+            "type": "object",
+            "properties": {
+                "optional_int": {"type": ["integer", "null"], "description": "Optional integer"},
+            },
+        }
+        model_class = util.create_input_schema_from_json_schema(schema)
+        instance_none = model_class(optional_int=None)
+        assert instance_none.optional_int is None
+        instance_int = model_class(optional_int=42)
+        assert instance_int.optional_int == 42
+
+    def test_create_input_schema_required_filters_non_strings(self):
+        """Test schema with required containing non-strings does not raise unhashable."""
+        schema = {
+            "type": "object",
+            "properties": {
+                "valid": {"type": "string", "description": "Valid field"},
+            },
+            "required": [["bad"], "valid"],
+        }
+        model_class = util.create_input_schema_from_json_schema(schema)
+        # Only "valid" is required (non-strings filtered out)
+        instance = model_class(valid="ok")
+        assert instance.valid == "ok"
+
+    def test_create_input_schema_empty_properties_no_params(self):
+        """Test schema with empty properties (tool with no params) accepts empty dict."""
+        schema = {"type": "object", "properties": {}}
+        model_class = util.create_input_schema_from_json_schema(schema)
+        instance = model_class()
+        assert instance.model_dump() == {}
+        # Also validate with explicit empty dict
+        instance2 = model_class.model_validate({})
+        assert instance2.model_dump() == {}
+
+    def test_create_input_schema_generic_object_maps_to_dict(self):
+        """Test schema with type object and no properties maps to dict for free-form params."""
+        schema = {
+            "type": "object",
+            "properties": {
+                "params": {"type": "object", "description": "Free-form parameters"},
+            },
+        }
+        model_class = util.create_input_schema_from_json_schema(schema)
+        # Should accept arbitrary key-value dict (not just empty)
+        instance = model_class(params={"search": "test", "per_page": 20})
+        assert instance.params == {"search": "test", "per_page": 20}
+        # Empty dict also valid
+        instance_empty = model_class(params={})
+        assert instance_empty.params == {}
+        # None valid for optional field
+        instance_none = model_class(params=None)
+        assert instance_none.params is None
+
     @pytest.mark.asyncio
     async def test_validate_connection_params(self):
         """Test connection parameter validation."""
@@ -1657,6 +1729,7 @@ class TestMCPStructuredTool:
                 if not input_dict or not isinstance(input_dict, dict):
                     return input_dict
 
+                from lfx.base.agents.utils import maybe_unflatten_dict
                 from lfx.base.mcp.util import _camel_to_snake
 
                 converted_dict = {}
@@ -1675,7 +1748,7 @@ class TestMCPStructuredTool:
                             # Keep original key
                             converted_dict[key] = value
 
-                return converted_dict
+                return maybe_unflatten_dict(converted_dict)
 
         return MCPStructuredTool(
             name="test_tool",
@@ -1730,6 +1803,48 @@ class TestMCPStructuredTool:
         assert mcp_tool._convert_parameters(None) is None
         assert mcp_tool._convert_parameters("not_a_dict") == "not_a_dict"
 
+    def test_convert_parameters_flattened_input_produces_nested(self):
+        """Test flattened keys (params.search, params.per_page) become nested structure."""
+        from lfx.base.agents.utils import maybe_unflatten_dict
+        from lfx.base.mcp.util import _camel_to_snake
+        from pydantic import Field, create_model
+
+        schema = create_model(
+            "ForemanSchema",
+            resource=(str, Field(..., description="Resource")),
+            action=(str, Field(..., description="Action")),
+            params=(dict, Field(default_factory=dict, description="Params")),
+        )
+
+        def _convert_parameters(input_dict, args_schema):
+            if not input_dict or not isinstance(input_dict, dict):
+                return input_dict
+            converted_dict = {}
+            original_fields = set(args_schema.model_fields.keys())
+            for key, value in input_dict.items():
+                if key in original_fields:
+                    converted_dict[key] = value
+                else:
+                    snake_key = _camel_to_snake(key)
+                    if snake_key in original_fields:
+                        converted_dict[snake_key] = value
+                    else:
+                        converted_dict[key] = value
+            return maybe_unflatten_dict(converted_dict)
+
+        input_dict = {
+            "resource": "hosts",
+            "action": "index",
+            "params.search": "name ~ test",
+            "params.per_page": 20,
+        }
+        result = _convert_parameters(input_dict, schema)
+        assert result == {
+            "resource": "hosts",
+            "action": "index",
+            "params": {"search": "name ~ test", "per_page": 20},
+        }
+
     def test_convert_parameters_preserves_value_types(self, mcp_tool):
         """Test _convert_parameters preserves all value types correctly."""
         input_dict = {
@@ -1876,6 +1991,421 @@ class TestMCPStructuredTool:
             await mcp_tool.arun(input_data)
 
 
+class TestNormalizeArgumentsForMcp:
+    """Test _normalize_arguments_for_mcp try-convert on type mismatch."""
+
+    def test_str_to_int_when_int_expected(self):
+        """Test str '42' when int expected -> 42."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            count: int = Field(..., description="Count")
+
+        result = util._normalize_arguments_for_mcp({"count": "42"}, Schema, "test_tool")
+        assert result == {"count": 42}
+        assert isinstance(result["count"], int)
+
+    def test_float_to_int_when_int_expected(self):
+        """Test float 3.0 when int expected -> 3."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            id: int = Field(..., description="ID")
+
+        result = util._normalize_arguments_for_mcp({"id": 3.0}, Schema, "test_tool")
+        assert result == {"id": 3}
+        assert isinstance(result["id"], int)
+
+    def test_str_to_dict_when_dict_expected(self):
+        r"""Test str '{"x":1}' when dict expected -> {"x":1}."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            params: dict = Field(..., description="Params")
+
+        result = util._normalize_arguments_for_mcp({"params": '{"x": 1}'}, Schema, "test_tool")
+        assert result == {"params": {"x": 1}}
+        assert isinstance(result["params"], dict)
+
+    def test_str_to_bool_when_bool_expected(self):
+        """Test str 'true' when bool expected -> True."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            flag: bool = Field(..., description="Flag")
+
+        result = util._normalize_arguments_for_mcp({"flag": "true"}, Schema, "test_tool")
+        assert result == {"flag": True}
+
+    def test_already_correct_types_unchanged(self):
+        """Test value type matches schema -> unchanged."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            count: int = Field(..., description="Count")
+            params: dict = Field(..., description="Params")
+
+        result = util._normalize_arguments_for_mcp({"count": 42, "params": {"search": "x"}}, Schema, "test_tool")
+        assert result == {"count": 42, "params": {"search": "x"}}
+
+    def test_conversion_failure_propagates_error(self):
+        """Test str 'abc' when int expected -> clear error propagated."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            id: int = Field(..., description="ID")
+
+        with pytest.raises(ValueError, match="expects integer") as exc_info:
+            util._normalize_arguments_for_mcp({"id": "abc"}, Schema, "test_tool")
+
+        assert "test_tool" in str(exc_info.value)
+        assert "Parameter 'id'" in str(exc_info.value)
+        assert "could not convert" in str(exc_info.value)
+
+    def test_optional_field_none_unchanged(self):
+        """Test optional field with None -> unchanged."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            count: int = Field(..., description="Count")
+            optional: int | None = Field(default=None, description="Optional")
+
+        result = util._normalize_arguments_for_mcp({"count": 1, "optional": None}, Schema, "test_tool")
+        assert result == {"count": 1, "optional": None}
+
+    def test_required_list_none_maps_to_empty(self):
+        """Test required list field with None -> maps to []."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            items: list = Field(..., description="Items")
+
+        result = util._normalize_arguments_for_mcp({"items": None}, Schema, "test_tool")
+        assert result == {"items": []}
+        assert isinstance(result["items"], list)
+
+    def test_required_dict_none_maps_to_empty(self):
+        """Test required dict field with None -> maps to {}."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            params: dict = Field(..., description="Params")
+
+        result = util._normalize_arguments_for_mcp({"params": None}, Schema, "test_tool")
+        assert result == {"params": {}}
+        assert isinstance(result["params"], dict)
+
+    def test_required_str_none_maps_to_empty(self):
+        """Test required str field with None -> maps to ""."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            name: str = Field(..., description="Name")
+
+        result = util._normalize_arguments_for_mcp({"name": None}, Schema, "test_tool")
+        assert result == {"name": ""}
+        assert result["name"] == ""
+
+    def test_optional_list_none_unchanged(self):
+        """Test optional list field with None -> unchanged."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            items: list[str] | None = Field(default=None, description="Items")
+
+        result = util._normalize_arguments_for_mcp({"items": None}, Schema, "test_tool")
+        assert result == {"items": None}
+
+    def test_optional_dict_none_unchanged(self):
+        """Test optional dict field with None -> unchanged."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            params: dict | None = Field(default=None, description="Params")
+
+        result = util._normalize_arguments_for_mcp({"params": None}, Schema, "test_tool")
+        assert result == {"params": None}
+
+    def test_optional_str_none_unchanged(self):
+        """Test optional str field with None -> unchanged."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            name: str | None = Field(default=None, description="Name")
+
+        result = util._normalize_arguments_for_mcp({"name": None}, Schema, "test_tool")
+        assert result == {"name": None}
+
+    def test_str_to_list_when_optional_list_expected(self):
+        r"""Test str '["a"]' when list[str] | None expected -> ["a"]."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            items: list[str] | None = Field(default=None, description="Items")
+
+        result = util._normalize_arguments_for_mcp({"items": '["a", "b"]'}, Schema, "test_tool")
+        assert result == {"items": ["a", "b"]}
+        assert isinstance(result["items"], list)
+
+    def test_str_to_dict_when_optional_dict_expected(self):
+        r"""Test str '{"x":1}' when dict | None expected -> {"x":1}."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            params: dict | None = Field(default=None, description="Params")
+
+        result = util._normalize_arguments_for_mcp({"params": '{"x": 1}'}, Schema, "test_tool")
+        assert result == {"params": {"x": 1}}
+        assert isinstance(result["params"], dict)
+
+    def test_bool_rejected_when_int_expected(self):
+        """Test bool True/False when int expected -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            count: int = Field(..., description="Count")
+
+        with pytest.raises(ValueError, match=r"expects integer.*bool"):
+            util._normalize_arguments_for_mcp({"count": True}, Schema, "test_tool")
+
+        with pytest.raises(ValueError, match=r"expects integer.*bool"):
+            util._normalize_arguments_for_mcp({"count": False}, Schema, "test_tool")
+
+    def test_extra_keys_preserved_for_pydantic_validation(self):
+        """Test extra keys not in schema are preserved so Pydantic can report them."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            valid_field: int = Field(..., description="Valid")
+
+        result = util._normalize_arguments_for_mcp({"valid_field": 1, "typo_field": 2}, Schema, "test_tool")
+        assert result["valid_field"] == 1
+        assert result["typo_field"] == 2
+
+    def test_str_to_float_when_float_expected(self):
+        """Test str '3.14' when float expected -> 3.14."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            value: float = Field(..., description="Value")
+
+        result = util._normalize_arguments_for_mcp({"value": "3.14"}, Schema, "test_tool")
+        assert result == {"value": 3.14}
+        assert isinstance(result["value"], float)
+
+    def test_invalid_json_for_dict_raises_clear_error(self):
+        """Test invalid JSON string when dict expected raises ValueError with tool and param name."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            params: dict = Field(..., description="Params")
+
+        with pytest.raises(ValueError, match=r"expects object \(dict\)") as exc_info:
+            util._normalize_arguments_for_mcp({"params": "not valid json"}, Schema, "my_tool")
+
+        assert "my_tool" in str(exc_info.value)
+        assert "Parameter 'params'" in str(exc_info.value)
+        assert "invalid JSON" in str(exc_info.value)
+
+    def test_invalid_json_for_list_raises_clear_error(self):
+        """Test invalid JSON string when list expected raises ValueError with tool and param name."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            items: list = Field(..., description="Items")
+
+        with pytest.raises(ValueError, match=r"expects array \(list\)") as exc_info:
+            util._normalize_arguments_for_mcp({"items": "{invalid"}, Schema, "list_tool")
+
+        assert "list_tool" in str(exc_info.value)
+        assert "Parameter 'items'" in str(exc_info.value)
+        assert "invalid JSON" in str(exc_info.value)
+
+    def test_str_to_dict_when_nested_model_expected(self):
+        """Test str JSON when nested Pydantic model expected -> parsed dict (foreman-mcp params case)."""
+        from pydantic import BaseModel, Field
+
+        class ParamsModel(BaseModel):
+            search: str | None = None
+            per_page: int | None = None
+            page: int | None = None
+
+        class Schema(BaseModel):
+            params: ParamsModel = Field(..., description="Params")
+
+        json_str = '{"search":"installed_packages","per_page":50}'
+        result = util._normalize_arguments_for_mcp({"params": json_str}, Schema, "test_tool")
+        assert result["params"] == {"search": "installed_packages", "per_page": 50}
+        assert isinstance(result["params"], dict)
+
+    def test_nested_model_already_dict_unchanged(self):
+        """Test nested model field with dict value -> unchanged."""
+        from pydantic import BaseModel, Field
+
+        class ParamsModel(BaseModel):
+            search: str | None = None
+
+        class Schema(BaseModel):
+            params: ParamsModel = Field(..., description="Params")
+
+        result = util._normalize_arguments_for_mcp({"params": {"search": "test"}}, Schema, "test_tool")
+        assert result == {"params": {"search": "test"}}
+
+    def test_str_to_bool_false_variants(self):
+        """Test str 'false'/'0'/'no' when bool expected -> False."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            flag: bool = Field(..., description="Flag")
+
+        for val in ("false", "0", "no"):
+            result = util._normalize_arguments_for_mcp({"flag": val}, Schema, "test_tool")
+            assert result == {"flag": False}
+            assert result["flag"] is False
+
+    def test_dict_expected_json_parses_to_list_raises(self):
+        """Test dict expected but JSON string parses to list -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            params: dict = Field(..., description="Params")
+
+        with pytest.raises(ValueError, match=r"expects object \(dict\)") as exc_info:
+            util._normalize_arguments_for_mcp({"params": '["a", "b"]'}, Schema, "my_tool")
+
+        assert "my_tool" in str(exc_info.value)
+        assert "Parameter 'params'" in str(exc_info.value)
+        assert "JSON parsed to list" in str(exc_info.value)
+
+    def test_list_expected_json_parses_to_dict_raises(self):
+        """Test list expected but JSON string parses to dict -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            items: list = Field(..., description="Items")
+
+        with pytest.raises(ValueError, match=r"expects array \(list\)") as exc_info:
+            util._normalize_arguments_for_mcp({"items": '{"x": 1}'}, Schema, "list_tool")
+
+        assert "list_tool" in str(exc_info.value)
+        assert "Parameter 'items'" in str(exc_info.value)
+        assert "JSON parsed to dict" in str(exc_info.value)
+
+    def test_int_expected_non_integer_float_raises(self):
+        """Test int expected but float 3.14 (non-integer) -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            count: int = Field(..., description="Count")
+
+        with pytest.raises(ValueError, match=r"expects integer") as exc_info:
+            util._normalize_arguments_for_mcp({"count": 3.14}, Schema, "test_tool")
+
+        assert "test_tool" in str(exc_info.value)
+        assert "Parameter 'count'" in str(exc_info.value)
+        assert "could not convert" in str(exc_info.value)
+
+    def test_int_expected_list_or_dict_raises(self):
+        """Test int expected but list/dict value -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            count: int = Field(..., description="Count")
+
+        for val in ([], {}):
+            with pytest.raises(ValueError, match=r"expects integer"):
+                util._normalize_arguments_for_mcp({"count": val}, Schema, "test_tool")
+
+    def test_float_expected_invalid_str_raises(self):
+        """Test float expected but str 'abc' -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            value: float = Field(..., description="Value")
+
+        with pytest.raises(ValueError, match=r"expects number") as exc_info:
+            util._normalize_arguments_for_mcp({"value": "abc"}, Schema, "test_tool")
+
+        assert "test_tool" in str(exc_info.value)
+        assert "Parameter 'value'" in str(exc_info.value)
+        assert "could not convert" in str(exc_info.value)
+
+    def test_float_expected_list_or_dict_raises(self):
+        """Test float expected but list/dict value -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            value: float = Field(..., description="Value")
+
+        for val in ([], {}):
+            with pytest.raises(ValueError, match=r"expects number"):
+                util._normalize_arguments_for_mcp({"value": val}, Schema, "test_tool")
+
+    def test_bool_expected_invalid_str_raises(self):
+        """Test bool expected but str 'maybe' -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            flag: bool = Field(..., description="Flag")
+
+        with pytest.raises(ValueError, match=r"expects.*bool") as exc_info:
+            util._normalize_arguments_for_mcp({"flag": "maybe"}, Schema, "test_tool")
+
+        assert "test_tool" in str(exc_info.value)
+        assert "Parameter 'flag'" in str(exc_info.value)
+        assert "could not convert" in str(exc_info.value)
+
+    def test_bool_expected_int_raises(self):
+        """Test bool expected but int 1 -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            flag: bool = Field(..., description="Flag")
+
+        with pytest.raises(ValueError, match=r"expects.*bool") as exc_info:
+            util._normalize_arguments_for_mcp({"flag": 1}, Schema, "test_tool")
+
+        assert "test_tool" in str(exc_info.value)
+        assert "Parameter 'flag'" in str(exc_info.value)
+        assert "could not convert" in str(exc_info.value)
+
+    def test_dict_expected_non_str_value_raises(self):
+        """Test dict expected but list value (not str) -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            params: dict = Field(..., description="Params")
+
+        with pytest.raises(ValueError, match=r"expects object \(dict\)") as exc_info:
+            util._normalize_arguments_for_mcp({"params": ["a", "b"]}, Schema, "my_tool")
+
+        assert "my_tool" in str(exc_info.value)
+        assert "Parameter 'params'" in str(exc_info.value)
+        assert "could not convert" in str(exc_info.value)
+
+    def test_list_expected_non_str_value_raises(self):
+        """Test list expected but dict value (not str) -> raises ValueError."""
+        from pydantic import BaseModel, Field
+
+        class Schema(BaseModel):
+            items: list = Field(..., description="Items")
+
+        with pytest.raises(ValueError, match=r"expects array \(list\)") as exc_info:
+            util._normalize_arguments_for_mcp({"items": {"x": 1}}, Schema, "list_tool")
+
+        assert "list_tool" in str(exc_info.value)
+        assert "Parameter 'items'" in str(exc_info.value)
+        assert "could not convert" in str(exc_info.value)
+
+    def test_try_convert_value_none_raises(self):
+        """Test _try_convert_value with None when scalar type expected -> raises ValueError (direct caller)."""
+        with pytest.raises(ValueError, match=r"expects.*but received None") as exc_info:
+            util._try_convert_value(None, int, "count", "test_tool")
+
+        assert "test_tool" in str(exc_info.value)
+        assert "Parameter 'count'" in str(exc_info.value)
+
+
 class TestSnakeToCamelConversion:
     """Test the _snake_to_camel function from json_schema module."""
 

src/backend/tests/unit/components/models_and_agents/test_mcp_component_output.py

@@ -142,3 +142,41 @@ class TestMCPComponentOutputProcessing:
         assert isinstance(result, DataFrame)
         assert len(result) == 1
         assert result.iloc[0]["error"] == "You must select a tool"
+
+    @pytest.mark.asyncio
+    async def test_build_output_adds_page_default_for_empty_kwargs(self, component):
+        """Test that build_output calls coroutine with empty kwargs when no params provided.
+
+        With exclude_none in MCP util, optional params (page, page_size) are omitted from
+        the payload sent to the server, allowing the backend to use its defaults.
+        """
+        from pydantic import Field, create_model
+
+        component.tool = "list_tool"
+        component.tools = []
+
+        mock_tool = MagicMock()
+        mock_tool.name = "list_tool"
+        mock_result = MagicMock()
+        mock_result.content = [
+            MagicMock(model_dump=MagicMock(return_value={"type": "text", "text": '{"results": []}'}))
+        ]
+        mock_tool.coroutine = AsyncMock(return_value=mock_result)
+
+        schema = create_model(
+            "ListSchema",
+            page=(int | None, Field(default=None, description="Page")),
+            page_size=(int | None, Field(default=None, description="Page size")),
+        )
+        mock_tool.args_schema = schema
+        mock_tool.args_schema.model_fields = schema.model_fields
+
+        component._tool_cache = {"list_tool": mock_tool}
+        component.update_tool_list = AsyncMock(return_value=([], None))
+        component.get_inputs_for_all_tools = MagicMock(return_value={"list_tool": []})
+
+        await component.build_output()
+
+        mock_tool.coroutine.assert_called_once()
+        call_kwargs = mock_tool.coroutine.call_args[1]
+        assert call_kwargs == {}  # Empty kwargs; exclude_none sends {} to MCP server

src/lfx/src/lfx/base/mcp/util.py

@@ -10,7 +10,8 @@ import shutil
 import subprocess
 import unicodedata
 from collections.abc import Awaitable, Callable
-from typing import Any
+from types import UnionType
+from typing import Any, Union, get_args, get_origin
 from urllib.parse import urlparse
 from uuid import UUID
 
@@ -22,6 +23,7 @@ from mcp import ClientSession
 from mcp.shared.exceptions import McpError
 from pydantic import BaseModel
 
+from lfx.base.agents.utils import maybe_unflatten_dict
 from lfx.log.logger import logger
 from lfx.schema.json_schema import create_input_schema_from_json_schema
 from lfx.services.deps import get_settings_service
@@ -288,6 +290,169 @@ def _convert_camel_case_to_snake_case(provided_args: dict[str, Any], arg_schema:
     return converted_args
 
 
+def _resolve_expected_type(annotation: Any) -> type | None:
+    """Resolve the effective expected type from a Pydantic field annotation.
+
+    Handles Union, UnionType (X | None), list, list[X]. Returns the primary
+    type (dict, list, int, float, bool, str) or None if not one we normalize.
+    """
+    ann = annotation
+    origin = get_origin(ann)
+    if origin is UnionType or origin is Union:
+        args = get_args(ann)
+        non_none = [a for a in args if a is not type(None)]
+        if non_none:
+            ann = non_none[0]
+            origin = get_origin(ann)
+    if origin is list or ann is list:
+        return list
+    if origin is dict or ann is dict:
+        return dict
+    if ann in (int, float, bool, str):
+        return ann
+    return None
+
+
+def _annotation_accepts_none(annotation: Any) -> bool:
+    """Check if annotation accepts None (e.g. Union[X, None], X | None)."""
+    origin = get_origin(annotation)
+    if origin is UnionType or origin is Union:
+        args = get_args(annotation)
+        return type(None) in args
+    return False
+
+
+def _is_pydantic_model_type(annotation: Any) -> bool:
+    """Check if annotation refers to a Pydantic BaseModel (possibly in Union with None)."""
+    ann = annotation
+    origin = get_origin(ann)
+    if origin is UnionType or origin is Union:
+        args = get_args(ann)
+        non_none = [a for a in args if a is not type(None)]
+        if non_none:
+            ann = non_none[0]
+    return isinstance(ann, type) and issubclass(ann, BaseModel)
+
+
+def _try_convert_value(value: Any, expected_type: type, field_name: str, tool_name: str) -> Any:
+    """Try to convert value to expected type. Raise ValueError with clear message on failure."""
+
+    def _err(type_desc: str, detail: str) -> ValueError:
+        msg = f"Tool '{tool_name}': Parameter '{field_name}' expects {type_desc} {detail}"
+        return ValueError(msg)
+
+    expected_type_desc = expected_type.__name__
+
+    if value is None and expected_type in (int, float, bool, dict, list):
+        raise _err(expected_type_desc, "but received None.")
+
+    # return correctly typed value, but handle the
+    # special case of bool as this is a subclass of int
+    # we'll NOT return but raise an error
+    if isinstance(value, expected_type) and not (expected_type is int and isinstance(value, bool)):
+        return value
+
+    # return custom classes as is
+    if expected_type not in (dict, list, int, float, bool):
+        return value
+
+    if expected_type in (dict, list):
+        expected_type_desc = "object (dict)" if expected_type is dict else "array (list)"
+        if isinstance(value, str):
+            try:
+                parsed = json.loads(value)
+            except json.JSONDecodeError as e:
+                raise _err(expected_type_desc, f"but received invalid JSON string {value!r}; {e}") from e
+            if not isinstance(parsed, expected_type):
+                raise _err(expected_type_desc, f"but JSON parsed to {type(parsed).__name__}.")
+            return parsed
+
+    elif expected_type is int:
+        expected_type_desc = "integer (int)"
+        if isinstance(value, float) and value.is_integer():
+            return int(value)
+        if isinstance(value, str):
+            try:
+                return int(value)
+            except ValueError as e:
+                raise _err(expected_type_desc, f"but received string: {value!r}; could not convert.") from e
+
+    elif expected_type is float:
+        expected_type_desc = "number (float)"
+        if isinstance(value, (int, float)):
+            return float(value)
+        if isinstance(value, str):
+            try:
+                return float(value)
+            except ValueError as e:
+                raise _err(expected_type_desc, f"but received string: {value!r}; could not convert.") from e
+
+    elif expected_type is bool and isinstance(value, str):
+        expected_type_desc = "boolean (bool)"
+        lower = value.strip().lower()
+        if lower in ("true", "1", "yes"):
+            return True
+        if lower in ("false", "0", "no"):
+            return False
+
+    detail = f"but received {type(value).__name__}: {value!r}; could not convert."
+    raise _err(expected_type_desc, detail)
+
+
+def _normalize_arguments_for_mcp(
+    arguments: dict[str, Any], arg_schema: type[BaseModel], tool_name: str
+) -> dict[str, Any]:
+    """Normalize tool arguments for MCP: try-convert when value type != schema expected type.
+
+    Uses schema from MCP server (no guessing). On conversion failure, raises
+    ValueError with clear user-facing message.
+    """
+    result: dict[str, Any] = {}
+    schema_field_names = set(arg_schema.model_fields.keys())
+    for field_name, model_field in arg_schema.model_fields.items():
+        value = arguments.get(field_name)
+        if value is None:
+            if not (model_field.is_required() or field_name in arguments):
+                continue
+            expected = _resolve_expected_type(model_field.annotation)
+            if expected in (list, dict, str) and model_field.is_required():
+                result[field_name] = [] if expected is list else ({} if expected is dict else "")
+            elif expected in (list, dict, str) and _annotation_accepts_none(model_field.annotation):
+                result[field_name] = None
+            else:
+                result[field_name] = value
+            continue
+        expected = _resolve_expected_type(model_field.annotation)
+        if expected is None:
+            # Nested Pydantic model (object with properties): UI/API often sends as JSON string
+            if _is_pydantic_model_type(model_field.annotation) and isinstance(value, str):
+                try:
+                    parsed = json.loads(value)
+                except json.JSONDecodeError as e:
+                    msg = (
+                        f"Tool '{tool_name}': Parameter '{field_name}' expects object "
+                        f"but received invalid JSON string {value!r}; {e}"
+                    )
+                    raise ValueError(msg) from e
+                if not isinstance(parsed, dict):
+                    msg = (
+                        f"Tool '{tool_name}': Parameter '{field_name}' expects object "
+                        f"but JSON parsed to {type(parsed).__name__}."
+                    )
+                    raise ValueError(msg)
+                result[field_name] = parsed
+            else:
+                result[field_name] = value
+            continue
+        if expected is str:
+            result[field_name] = value
+            continue
+        result[field_name] = _try_convert_value(value, expected, field_name, tool_name)
+    # Preserve extra keys so Pydantic validation can report them
+    result.update({k: v for k, v in arguments.items() if k not in schema_field_names})
+    return result
+
+
 def _handle_tool_validation_error(
     e: Exception, tool_name: str, provided_args: dict[str, Any], arg_schema: type[BaseModel]
 ) -> None:
@@ -320,14 +485,16 @@ def create_tool_coroutine(tool_name: str, arg_schema: type[BaseModel], client) -
         # Merge in keyword arguments
         provided_args.update(kwargs)
         provided_args = _convert_camel_case_to_snake_case(provided_args, arg_schema)
+        original_args = provided_args
+        provided_args = _normalize_arguments_for_mcp(provided_args, arg_schema, tool_name)
         # Validate input and fill defaults for missing optional fields
         try:
             validated = arg_schema.model_validate(provided_args)
         except Exception as e:  # noqa: BLE001
-            _handle_tool_validation_error(e, tool_name, provided_args, arg_schema)
+            _handle_tool_validation_error(e, tool_name, original_args, arg_schema)
 
         try:
-            return await client.run_tool(tool_name, arguments=validated.model_dump())
+            return await client.run_tool(tool_name, arguments=validated.model_dump(exclude_none=True))
         except Exception as e:
             await logger.aerror(f"Tool '{tool_name}' execution failed: {e}")
             # Re-raise with more context
@@ -348,13 +515,15 @@ def create_tool_func(tool_name: str, arg_schema: type[BaseModel], client) -> Cal
             provided_args[field_names[i]] = arg
         provided_args.update(kwargs)
         provided_args = _convert_camel_case_to_snake_case(provided_args, arg_schema)
+        original_args = provided_args
+        provided_args = _normalize_arguments_for_mcp(provided_args, arg_schema, tool_name)
         try:
             validated = arg_schema.model_validate(provided_args)
         except Exception as e:  # noqa: BLE001
-            _handle_tool_validation_error(e, tool_name, provided_args, arg_schema)
+            _handle_tool_validation_error(e, tool_name, original_args, arg_schema)
 
         try:
-            return run_until_complete(client.run_tool(tool_name, arguments=validated.model_dump()))
+            return run_until_complete(client.run_tool(tool_name, arguments=validated.model_dump(exclude_none=True)))
         except Exception as e:
             logger.error(f"Tool '{tool_name}' execution failed: {e}")
             # Re-raise with more context
@@ -1713,10 +1882,18 @@ async def update_tools(
                             if snake_key in original_fields:
                                 converted_dict[snake_key] = value
                             else:
-                                # Keep original key
+                                # Keep original key (may be flattened e.g. params.search)
                                 converted_dict[key] = value
 
-                    return converted_dict
+                    unflattened = maybe_unflatten_dict(converted_dict)
+                    # Normalize: convert JSON strings to dict for nested model params
+                    normalized = _normalize_arguments_for_mcp(unflattened, self.args_schema, self.name)
+                    # Preserve extra keys not in schema (e.g. flattened keys)
+                    schema_fields = set(self.args_schema.model_fields.keys())
+                    for key, value in unflattened.items():
+                        if key not in schema_fields and key not in normalized:
+                            normalized[key] = value
+                    return normalized
 
             tool_obj = MCPStructuredTool(
                 name=tool.name,

src/lfx/src/lfx/schema/json_schema.py

@@ -97,13 +97,30 @@ def create_input_schema_from_json_schema(schema: dict[str, Any]) -> type[BaseMod
             return str
 
         t = s.get("type", "any")  # Use string "any" as default instead of Any type
+        if isinstance(t, list):
+            # JSON Schema: "type": ["string", "null"] for nullable
+            non_null = [x for x in t if x != "null" and isinstance(x, str)]
+            if non_null:
+                prim = {
+                    "string": str,
+                    "integer": int,
+                    "number": float,
+                    "boolean": bool,
+                    "object": dict,
+                    "array": list,
+                }.get(non_null[0], Any)
+                return prim | None if "null" in t else prim
+            return Any
         if t == "array":
             item_schema = s.get("items", {})
             schema_type: Any = parse_type(item_schema)
             return list[schema_type]
 
         if t == "object":
-            # inline object not in $defs ⇒ anonymous nested model
+            # Generic object (no properties) ⇒ dict for free-form key-value pairs
+            if not s.get("properties"):
+                return dict
+            # Inline object with defined properties ⇒ nested model
             return _build_model(f"AnonModel{len(model_cache)}", s)
 
         # primitive fallback
@@ -136,7 +153,7 @@ def create_input_schema_from_json_schema(schema: dict[str, Any]) -> type[BaseMod
             return model_cache[name]
 
         props = subschema.get("properties", {})
-        reqs = set(subschema.get("required", []))
+        reqs = {r for r in (subschema.get("required") or []) if isinstance(r, str)}
         fields: dict[str, Any] = {}
 
         for prop_name, prop_schema in props.items():
@@ -163,7 +180,7 @@ def create_input_schema_from_json_schema(schema: dict[str, Any]) -> type[BaseMod
 
     # build the top - level "InputSchema" from the root properties
     top_props = schema.get("properties", {})
-    top_reqs = set(schema.get("required", []))
+    top_reqs = {r for r in (schema.get("required") or []) if isinstance(r, str)}
     top_fields: dict[str, Any] = {}
 
     for fname, fdef in top_props.items():