fix: return original tool argument when JSON parses but conversion fails

Co-Authored-By: Claude Opus 5.5 <[email protected]>

src/praisonai-agents/praisonaiagents/agent/execution_mixin.py

@@ -1965,6 +1965,8 @@ Write the complete compiled report:"""
                 except ImportError:
                     durable_key = None
                 call_arguments = arguments
+                if hasattr(self, "_cast_arguments"):  # sync-path parity
+                    call_arguments = self._cast_arguments(call_target, arguments)
                 if durable_key:
                     try:
                         parameters = inspect.signature(call_target).parameters.values()
@@ -1975,7 +1977,7 @@ Write the complete compiled report:"""
                         or parameter.kind == inspect.Parameter.VAR_KEYWORD
                         for parameter in parameters
                     ):
-                        call_arguments = dict(arguments)
+                        call_arguments = dict(call_arguments)
                         call_arguments["idempotency_key"] = durable_key
 
                 # Establish the injection context (Injected[T] params, including

src/praisonai-agents/praisonaiagents/agent/tool_execution.py

@@ -7,6 +7,8 @@ for maintainability.
 """
 
 import os
+import dataclasses
+import typing
 import time
 import json
 import logging
@@ -69,6 +71,72 @@ _MAX_SPILL_TOOL_THREADS = 8
 _TOOL_ARGUMENTS_PARSE_FAILED = object()
 
 
+def _coerce_value(value, annotation):
+    """Best-effort cast of an LLM-supplied value to ``annotation``.
+
+    Returns ``value`` unchanged when it cannot be converted.
+    """
+    import collections.abc as cabc
+    import types
+    from typing import get_origin, get_args, Union as _Union
+
+    origin, args = get_origin(annotation), get_args(annotation)
+    if origin is _Union or origin is getattr(types, "UnionType", _Union):  # Optional/Union, X | Y
+        members = [a for a in args if a is not type(None)]
+        if value is None or any(isinstance(a, type) and isinstance(value, a) for a in members):
+            return value
+        for arg in members:
+            coerced = _coerce_value(value, arg)
+            target = get_origin(arg) or arg
+            # Only accept a member's result if it actually produced that type.
+            if coerced is not value and (not isinstance(target, type) or isinstance(coerced, target)):
+                return coerced
+        return value
+    original = value  # returned whenever conversion does not succeed
+    try:
+        if annotation is int and isinstance(value, (str, float)) and not isinstance(value, bool):
+            return int(float(value))
+        if annotation is float and isinstance(value, (str, int)) and not isinstance(value, bool):
+            return float(value)
+        if annotation is bool and isinstance(value, str):
+            return value.lower() in ('true', '1', 'yes', 'on')
+        container = origin or annotation
+        if not isinstance(container, type):
+            return value
+        is_model = hasattr(container, "model_validate")
+        if isinstance(value, str) and container is not str and (
+            is_model or dataclasses.is_dataclass(container)
+            or issubclass(container, (cabc.Iterable, cabc.Mapping))
+        ):
+            stripped = value.strip()
+            if stripped[:1] in ('[', '{'):
+                value = json.loads(stripped)
+        if is_model and isinstance(value, dict):
+            return container.model_validate(value)
+        if isinstance(value, dict) and dataclasses.is_dataclass(container):
+            hints = typing.get_type_hints(container)
+            return container(**{k: _coerce_value(v, hints[k]) if k in hints else v
+                                for k, v in value.items()})
+        if isinstance(value, (list, tuple)) and not isinstance(value, str):
+            if container is tuple and args and Ellipsis not in args:
+                if len(args) != len(value):
+                    return original
+                return tuple(_coerce_value(v, a) for v, a in zip(value, args))
+            if container in (set, frozenset, tuple):
+                item_ann = args[0] if args else None
+                items = [_coerce_value(v, item_ann) for v in value] if item_ann else list(value)
+                return container(items)
+            if container is list and args:
+                return [_coerce_value(v, args[0]) for v in value]
+    except (ValueError, TypeError):
+        return original
+    except Exception as e:  # e.g. pydantic.ValidationError, json.JSONDecodeError
+        logging.debug(f"Could not coerce tool argument to {annotation!r}: {e}")
+        return original
+    # Keep parsed JSON only if it matches the container (e.g. bare ``list``).
+    return value if isinstance(value, container) else original
+
+
 class _CallableGuardrailAdapter:
     """Expose a plain callable guardrail as a chainable guardrail object.
 
@@ -882,7 +950,17 @@ class ToolExecutionMixin:
             return arguments
         
         try:
+            # ``@tool`` wraps the function in FunctionTool whose run() takes
+            # **kwargs; read parameters from the wrapped function instead.
+            from ..tools.decorator import FunctionTool
+            owner = getattr(func, "__self__", func)
+            if isinstance(owner, FunctionTool):
+                func = owner._func
             sig = inspect.signature(func)
+            try:  # resolve postponed (string) annotations
+                hints = typing.get_type_hints(func)
+            except Exception:
+                hints = {}
             valid_params = set(sig.parameters.keys()) - {'self'}
             casted_args = {}
             
@@ -907,21 +985,8 @@ class ToolExecutionMixin:
                 if param_name in sig.parameters:
                     param = sig.parameters[param_name]
                     if param.annotation != inspect.Parameter.empty:
-                        # Handle common type conversions
-                        if param.annotation == int and isinstance(arg_value, (str, float)):
-                            try:
-                                casted_args[param_name] = int(float(arg_value))
-                            except (ValueError, TypeError):
-                                casted_args[param_name] = arg_value
-                        elif param.annotation == float and isinstance(arg_value, (str, int)):
-                            try:
-                                casted_args[param_name] = float(arg_value)
-                            except (ValueError, TypeError):
-                                casted_args[param_name] = arg_value
-                        elif param.annotation == bool and isinstance(arg_value, str):
-                            casted_args[param_name] = arg_value.lower() in ('true', '1', 'yes', 'on')
-                        else:
-                            casted_args[param_name] = arg_value
+                        annotation = hints.get(param_name, param.annotation)
+                        casted_args[param_name] = _coerce_value(arg_value, annotation)
                     else:
                         casted_args[param_name] = arg_value
                 else:

src/praisonai-agents/tests/unit/test_issue_6294_coercion.py

@@ -0,0 +1,83 @@
+"""Regression tests for issue #6294: tool argument coercion."""
+import asyncio
+from dataclasses import dataclass
+from typing import List, Optional, Set, Tuple, Union
+
+from pydantic import BaseModel
+
+from praisonaiagents.agent.tool_execution import _coerce_value
+
+
+class Addr(BaseModel):
+    city: str
+
+
+@dataclass
+class Point:
+    x: int
+
+
+def test_coerce_values():
+    assert _coerce_value("5", Optional[int]) == 5
+    assert _coerce_value("[1,2]", List[int]) == [1, 2]
+    assert _coerce_value(["a", "b"], Set[str]) == {"a", "b"}
+    out = _coerce_value(["1.5", "2"], Tuple[float, int])
+    assert out == (1.5, 2) and type(out[1]) is int
+    out = _coerce_value([{"city": "x"}, "5"], Tuple[Addr, int])
+    assert out == (Addr(city="x"), 5)
+    assert _coerce_value({"x": "3"}, Point) == Point(x=3)
+    assert _coerce_value([{"x": 1}], List[Point]) == [Point(x=1)]
+    assert _coerce_value({"city": "x"}, Addr) == Addr(city="x")
+    assert _coerce_value('{"city": "x"}', Addr) == Addr(city="x")
+    # unchanged when not convertible / already right type
+    assert _coerce_value("abc", Optional[int]) == "abc"
+    assert _coerce_value("hello", Optional[str]) == "hello"
+    assert _coerce_value("[1,", List[int]) == "[1,"
+    assert _coerce_value("5", int) == 5 and _coerce_value("true", bool) is True
+
+
+def _agent_with(tool):
+    from praisonaiagents import Agent
+    return Agent(name="a", instructions="x", llm="gpt-4o-mini", tools=[tool])
+
+
+def test_decorated_tool_and_postponed_annotations_get_models():
+    from praisonaiagents import tool
+
+    def plain(address: "Addr") -> str:
+        return address.city
+
+    @tool
+    def decorated(address: Addr) -> str:
+        return address.city
+
+    agent = _agent_with(plain)
+    assert agent._cast_arguments(plain, {"address": {"city": "x"}}) == {"address": Addr(city="x")}
+    agent = _agent_with(decorated)
+    assert agent._cast_arguments(decorated.run, {"address": {"city": "y"}}) == {"address": Addr(city="y")}
+
+
+def test_async_tool_call_casts_arguments():
+    def where(address: Addr) -> str:
+        return address.city
+
+    agent = _agent_with(where)
+    assert asyncio.run(agent._execute_tool_async_impl("where", {"address": {"city": "z"}})) == "z"
+
+
+def test_union_edge_cases():
+    # PEP 604 unions are unwrapped like typing.Union
+    assert _coerce_value("[1, 2]", list[int] | None) == [1, 2]
+    assert _coerce_value("3", int | None) == 3
+    # A member that yields the wrong type must not win over a later valid one
+    assert _coerce_value('{"city": "x"}', Union[List[int], Addr]) == Addr(city="x")
+    # Dataclasses accept JSON strings, like models
+    assert _coerce_value('{"x": "4"}', Point) == Point(x=4)
+
+
+def test_failed_conversion_returns_original_string():
+    # Parsed JSON must not leak out when the target type is not produced
+    assert _coerce_value('{"town": 1}', Addr) == '{"town": 1}'
+    assert _coerce_value('{"a": 1}', List[int]) == '{"a": 1}'
+    assert _coerce_value("[1, 2]", Tuple[int, int, int]) == "[1, 2]"
+    assert _coerce_value("[1, 2]", list) == [1, 2]