Python: fix(ag-ui): properly handle json serialize with handoff workflows as agent (#3275)

* fix(ag-ui): properly handle json serialize with handoff workflows as agent

* Other improvements around handling non-serializable objects

python/packages/ag-ui/agent_framework_ag_ui/_events.py

@@ -29,7 +29,7 @@ from agent_framework import (
     prepare_function_call_results,
 )
 
-from ._utils import extract_state_from_tool_args, generate_event_id, safe_json_parse
+from ._utils import extract_state_from_tool_args, generate_event_id, make_json_safe, safe_json_parse
 
 logger = logging.getLogger(__name__)
 
@@ -177,7 +177,11 @@ class AgentFrameworkEventBridge:
             self.current_tool_call_id = tool_call_id
 
         if content.arguments:
-            delta_str = content.arguments if isinstance(content.arguments, str) else json.dumps(content.arguments)
+            delta_str = (
+                content.arguments
+                if isinstance(content.arguments, str)
+                else json.dumps(make_json_safe(content.arguments))
+            )
             logger.info(f"Emitting ToolCallArgsEvent with delta_length={len(delta_str)}, id='{tool_call_id}'")
             args_event = ToolCallArgsEvent(
                 tool_call_id=tool_call_id,
@@ -391,7 +395,7 @@ class AgentFrameworkEventBridge:
             args_dict = {
                 "function_name": function_call.name,
                 "function_call_id": function_call.call_id,
-                "function_arguments": function_call.parse_arguments() or {},
+                "function_arguments": make_json_safe(function_call.parse_arguments() or {}),
                 "steps": [
                     {
                         "description": f"Execute {function_call.name}",
@@ -435,7 +439,7 @@ class AgentFrameworkEventBridge:
         args_dict = {
             "function_name": function_call.name,
             "function_call_id": function_call.call_id,
-            "function_arguments": function_call.parse_arguments() or {},
+            "function_arguments": make_json_safe(function_call.parse_arguments() or {}),
             "steps": [
                 {
                     "description": f"Execute {function_call.name}",

python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py

@@ -17,6 +17,7 @@ from ._utils import (
     AGUI_TO_FRAMEWORK_ROLE,
     FRAMEWORK_TO_AGUI_ROLE,
     get_role_value,
+    make_json_safe,
     normalize_agui_role,
     safe_json_parse,
 )
@@ -265,7 +266,7 @@ def agui_messages_to_agent_framework(messages: list[dict[str, Any]]) -> list[Cha
                 function_payload_dict = cast(dict[str, Any], function_payload)
                 existing_args = function_payload_dict.get("arguments")
                 if isinstance(existing_args, str):
-                    function_payload_dict["arguments"] = json.dumps(modified_args)
+                    function_payload_dict["arguments"] = json.dumps(make_json_safe(modified_args))
                 else:
                     function_payload_dict["arguments"] = modified_args
                 return
@@ -377,7 +378,9 @@ def agui_messages_to_agent_framework(messages: list[dict[str, Any]]) -> list[Cha
                 # a proper FunctionApprovalResponseContent. This enables the agent framework
                 # to execute the approved tool (fix for GitHub issue #3034).
                 accepted = parsed.get("accepted", False) if parsed is not None else False
-                approval_payload_text = result_content if isinstance(result_content, str) else json.dumps(parsed)
+                approval_payload_text = (
+                    result_content if isinstance(result_content, str) else json.dumps(make_json_safe(parsed))
+                )
 
                 # Log the full approval payload to debug modified arguments
                 import logging
@@ -454,7 +457,9 @@ def agui_messages_to_agent_framework(messages: list[dict[str, Any]]) -> list[Cha
 
                         # Keep the original tool call and AG-UI snapshot in sync with approved args.
                         updated_args = (
-                            json.dumps(merged_args) if isinstance(matching_func_call.arguments, str) else merged_args
+                            json.dumps(make_json_safe(merged_args))
+                            if isinstance(matching_func_call.arguments, str)
+                            else merged_args
                         )
                         matching_func_call.arguments = updated_args
                         _update_tool_call_arguments(messages, str(approval_call_id), merged_args)
@@ -462,7 +467,7 @@ def agui_messages_to_agent_framework(messages: list[dict[str, Any]]) -> list[Cha
                         func_call_for_approval = Content.from_function_call(
                             call_id=matching_func_call.call_id,  # type: ignore[arg-type]
                             name=matching_func_call.name,  # type: ignore[arg-type]
-                            arguments=json.dumps(filtered_args),
+                            arguments=json.dumps(make_json_safe(filtered_args)),
                         )
                         logger.info(f"Using modified arguments from approval: {filtered_args}")
                     else:
@@ -767,7 +772,7 @@ def agui_messages_to_snapshot_format(messages: list[dict[str, Any]]) -> list[dic
                 if arguments is None:
                     function_payload_dict["arguments"] = ""
                 elif not isinstance(arguments, str):
-                    function_payload_dict["arguments"] = json.dumps(arguments)
+                    function_payload_dict["arguments"] = json.dumps(make_json_safe(arguments))
 
         # Normalize tool_call_id to toolCallId for tool messages
         normalized_msg["role"] = normalize_agui_role(normalized_msg.get("role"))

python/packages/ag-ui/agent_framework_ag_ui/_orchestration/_helpers.py

@@ -12,7 +12,7 @@ from agent_framework import (
     Content,
 )
 
-from .._utils import get_role_value, safe_json_parse
+from .._utils import get_role_value, make_json_safe, safe_json_parse
 
 if TYPE_CHECKING:
     from .._events import AgentFrameworkEventBridge
@@ -252,7 +252,7 @@ def build_safe_metadata(thread_metadata: dict[str, Any] | None) -> dict[str, Any
         return {}
     safe_metadata: dict[str, Any] = {}
     for key, value in thread_metadata.items():
-        value_str = value if isinstance(value, str) else json.dumps(value)
+        value_str = value if isinstance(value, str) else json.dumps(make_json_safe(value))
         if len(value_str) > 512:
             value_str = value_str[:512]
         safe_metadata[key] = value_str

python/packages/ag-ui/agent_framework_ag_ui/_orchestration/_state_manager.py

@@ -8,6 +8,8 @@ from typing import Any
 from ag_ui.core import CustomEvent, EventType
 from agent_framework import ChatMessage, Content
 
+from .._utils import make_json_safe
+
 
 class StateManager:
     """Coordinates state defaults, snapshots, and structured updates."""
@@ -67,7 +69,7 @@ class StateManager:
         if conversation_has_tool_calls and not self._state_from_input:
             return None
 
-        state_json = json.dumps(self.current_state, indent=2)
+        state_json = json.dumps(make_json_safe(self.current_state), indent=2)
         return ChatMessage(
             role="system",
             contents=[

python/packages/ag-ui/agent_framework_ag_ui/_orchestrators.py

@@ -243,7 +243,7 @@ class HumanInTheLoopOrchestrator(Orchestrator):
         if not msg:
             return False
 
-        return bool(msg.additional_properties.get("is_tool_result", False))
+        return bool((msg.additional_properties or {}).get("is_tool_result", False))
 
     async def run(
         self,
@@ -388,7 +388,7 @@ class DefaultOrchestrator(Orchestrator):
 
         response_format = None
         if isinstance(context.agent, ChatAgent):
-            response_format = context.agent.default_options.get("response_format")
+            response_format = (context.agent.default_options or {}).get("response_format")
         skip_text_content = response_format is not None
 
         client_tools = convert_agui_tools_to_agent_framework(context.input_data.get("tools"))

python/packages/ag-ui/agent_framework_ag_ui/_utils.py

@@ -141,11 +141,14 @@ def make_json_safe(obj: Any) -> Any:  # noqa: ANN401
     if isinstance(obj, (datetime, date)):
         return obj.isoformat()
     if is_dataclass(obj):
-        return asdict(obj)  # type: ignore[arg-type]
+        # asdict may return nested non-dataclass objects, so recursively make them safe
+        return make_json_safe(asdict(obj))  # type: ignore[arg-type]
     if hasattr(obj, "model_dump"):
-        return obj.model_dump()  # type: ignore[no-any-return]
+        return make_json_safe(obj.model_dump())  # type: ignore[no-any-return]
+    if hasattr(obj, "to_dict"):
+        return make_json_safe(obj.to_dict())  # type: ignore[no-any-return]
     if hasattr(obj, "dict"):
-        return obj.dict()  # type: ignore[no-any-return]
+        return make_json_safe(obj.dict())  # type: ignore[no-any-return]
     if hasattr(obj, "__dict__"):
         return {key: make_json_safe(value) for key, value in vars(obj).items()}  # type: ignore[misc]
     if isinstance(obj, (list, tuple)):

python/packages/ag-ui/tests/test_events_comprehensive.py

@@ -825,3 +825,93 @@ async def test_tool_result_with_model_dump_objects():
     assert events[1].type == "TOOL_CALL_RESULT"
     # Should be properly serialized JSON array without double escaping
     assert events[1].content == '[{"value": 1}, {"value": 2}]'
+
+
+async def test_function_call_with_dataclass_arguments():
+    """Test FunctionCallContent with dataclass arguments is serialized correctly.
+
+    This test verifies the fix for the AG-UI JSON serialization error when
+    HandoffAgentUserRequest (a dataclass) is passed as FunctionCallContent.arguments.
+    """
+    from dataclasses import dataclass
+
+    from agent_framework_ag_ui._events import AgentFrameworkEventBridge
+
+    @dataclass
+    class TestRequest:
+        field1: str
+        field2: int
+
+    bridge = AgentFrameworkEventBridge(run_id="test_run", thread_id="test_thread")
+
+    # FunctionCallContent with a dataclass as arguments (not a string)
+    update = AgentResponseUpdate(
+        contents=[
+            Content.from_function_call(
+                name="request_info",
+                call_id="call_dataclass",
+                arguments=TestRequest(field1="value", field2=42),
+            )
+        ]
+    )
+
+    events = await bridge.from_agent_run_update(update)
+
+    # Should have ToolCallStartEvent and ToolCallArgsEvent
+    tool_args_events = [e for e in events if e.type == "TOOL_CALL_ARGS"]
+    assert len(tool_args_events) == 1
+
+    # Verify the delta is valid JSON
+    delta = tool_args_events[0].delta
+    parsed = json.loads(delta)
+    assert parsed == {"field1": "value", "field2": 42}
+
+
+async def test_function_call_with_nested_dataclass_arguments():
+    """Test FunctionCallContent with nested dataclass arguments is serialized correctly.
+
+    This test covers the scenario where HandoffAgentUserRequest contains an AgentResponse
+    with nested content objects.
+    """
+    from dataclasses import dataclass
+
+    from agent_framework_ag_ui._events import AgentFrameworkEventBridge
+
+    @dataclass
+    class InnerContent:
+        text: str
+
+    @dataclass
+    class AgentResponseMock:
+        contents: list[InnerContent]
+
+    @dataclass
+    class HandoffRequest:
+        agent_response: AgentResponseMock
+
+    bridge = AgentFrameworkEventBridge(run_id="test_run", thread_id="test_thread")
+
+    # Simulate a HandoffAgentUserRequest-like structure
+    update = AgentResponseUpdate(
+        contents=[
+            Content.from_function_call(
+                name="request_info",
+                call_id="call_nested",
+                arguments=HandoffRequest(
+                    agent_response=AgentResponseMock(contents=[InnerContent(text="Hello from agent")])
+                ),
+            )
+        ]
+    )
+
+    events = await bridge.from_agent_run_update(update)
+
+    # Should have ToolCallStartEvent and ToolCallArgsEvent
+    tool_args_events = [e for e in events if e.type == "TOOL_CALL_ARGS"]
+    assert len(tool_args_events) == 1
+
+    # Verify the delta is valid JSON and contains nested structure
+    delta = tool_args_events[0].delta
+    parsed = json.loads(delta)
+    assert "agent_response" in parsed
+    assert parsed["agent_response"]["contents"] == [{"text": "Hello from agent"}]

python/packages/ag-ui/tests/test_message_adapters.py

@@ -642,3 +642,51 @@ def test_agent_framework_to_agui_function_result_multiple_text_contents():
     agui_msg = messages[0]
     # Multiple items should return JSON array
     assert agui_msg["content"] == '["First result", "Second result"]'
+
+
+def test_agui_tool_approval_with_dataclass_modified_args():
+    """Test that agui_messages_to_agent_framework handles dataclass in modified args.
+
+    This tests the fix for json.dumps() serialization errors at line 274
+    when modified_args contains non-serializable objects via make_json_safe.
+    """
+    from dataclasses import dataclass
+
+    @dataclass
+    class ModifiedData:
+        field1: str
+        field2: int
+
+    # Create AG-UI format messages that simulate tool approval flow
+    # where modified args could contain a dataclass after parsing
+
+    # First, an assistant message with a tool call (string arguments)
+    assistant_msg = {
+        "id": "msg-1",
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [
+            {
+                "id": "call-test",
+                "type": "function",
+                "function": {
+                    "name": "update_state",
+                    "arguments": '{"data": "original"}',  # String args
+                },
+            }
+        ],
+    }
+
+    # Then a user approval message (the approval path will merge modified args)
+    approval_msg = {
+        "id": "msg-2",
+        "role": "user",
+        "content": '{"approved": true}',
+        "toolCallId": "call-test",
+    }
+
+    # This should NOT raise TypeError
+    result = agui_messages_to_agent_framework([assistant_msg, approval_msg])
+
+    # Should have processed both messages without error
+    assert len(result) == 2

python/packages/ag-ui/tests/test_orchestrators_coverage.py

@@ -870,3 +870,60 @@ async def test_response_format_skip_text_content() -> None:
 
     # Test passes if no errors occur - verifies response_format code path
     assert len(events) > 0
+
+
+async def test_human_in_the_loop_handles_none_additional_properties() -> None:
+    """Test that HumanInTheLoopOrchestrator handles None additional_properties gracefully.
+
+    This test ensures the null safety fix for msg.additional_properties.get() works.
+    """
+    orchestrator = HumanInTheLoopOrchestrator()
+
+    # Create a message with None additional_properties
+    msg = ChatMessage(
+        role="user",
+        contents=[Content.from_text(text="Hello")],
+    )
+    # Explicitly set additional_properties to None
+    msg.additional_properties = None  # type: ignore[assignment]
+
+    agent = StubAgent()  # Use default StubAgent
+    config = AgentConfig()
+    context = TestExecutionContext(
+        input_data={"messages": [{"role": "user", "content": "Hello"}]},
+        agent=agent,
+        config=config,
+    )
+    context.set_messages([msg])
+
+    # can_handle should return False (not crash) when additional_properties is None
+    result = orchestrator.can_handle(context)
+    assert result is False
+
+
+async def test_default_orchestrator_handles_none_default_options() -> None:
+    """Test that DefaultOrchestrator handles None default_options gracefully.
+
+    This test ensures the null safety fix for context.agent.default_options.get() works.
+    """
+    orchestrator = DefaultOrchestrator()
+
+    # Use StubAgent with default_options set to None
+    agent = StubAgent(default_options=None)
+    config = AgentConfig()
+    context = TestExecutionContext(
+        input_data={"messages": [{"role": "user", "content": "Hello"}]},
+        agent=agent,
+        config=config,
+    )
+
+    # This should NOT crash when accessing default_options.get()
+    events: list[Any] = []
+    async for event in orchestrator.run(context):
+        events.append(event)
+        # Just check a few events to verify it's working
+        if len(events) > 3:
+            break
+
+    # Test passes if no AttributeError occurred
+    assert True

python/packages/ag-ui/tests/test_state_manager.py

@@ -49,3 +49,57 @@ def test_state_context_only_when_new_user_turn() -> None:
     assert isinstance(message, ChatMessage)
     assert message.contents[0].type == "text"
     assert "Current state of the application" in message.contents[0].text
+
+
+def test_state_manager_with_dataclass_in_state() -> None:
+    """Test that state containing dataclasses can be serialized without crashing.
+
+    This test ensures the fix for JSON serialization errors when state
+    contains dataclass or other non-JSON-serializable objects.
+    """
+    from dataclasses import dataclass
+
+    @dataclass
+    class UserData:
+        name: str
+        age: int
+
+    state_manager = StateManager(
+        state_schema={"user": {"type": "object"}},
+        predict_state_config=None,
+        require_confirmation=True,
+    )
+    # Initialize with a dataclass object in the state
+    state_manager.initialize({"user": UserData(name="Alice", age=30)})
+
+    # This should NOT raise TypeError when generating the context message
+    message = state_manager.state_context_message(is_new_user_turn=True, conversation_has_tool_calls=False)
+
+    assert message is not None
+    assert isinstance(message, ChatMessage)
+    # The dataclass should be serialized to JSON in the message
+    assert "Alice" in message.contents[0].text
+    assert "30" in message.contents[0].text
+
+
+def test_state_manager_with_pydantic_in_state() -> None:
+    """Test that state containing Pydantic models can be serialized without crashing."""
+    from pydantic import BaseModel
+
+    class UserModel(BaseModel):
+        email: str
+        active: bool
+
+    state_manager = StateManager(
+        state_schema={"user": {"type": "object"}},
+        predict_state_config=None,
+        require_confirmation=True,
+    )
+    # Initialize with a Pydantic model in the state
+    state_manager.initialize({"user": UserModel(email="test@example.com", active=True)})
+
+    # This should NOT raise TypeError
+    message = state_manager.state_context_message(is_new_user_turn=True, conversation_has_tool_calls=False)
+
+    assert message is not None
+    assert "test@example.com" in message.contents[0].text

python/packages/ag-ui/tests/test_utils.py

@@ -122,6 +122,20 @@ def test_make_json_safe_model_dump():
     assert result == {"type": "model", "data": "dump"}
 
 
+class ToDictObject:
+    """Object with to_dict method (like SerializationMixin)."""
+
+    def to_dict(self):
+        return {"type": "serialization_mixin", "method": "to_dict"}
+
+
+def test_make_json_safe_to_dict():
+    """Test object with to_dict method (SerializationMixin pattern)."""
+    obj = ToDictObject()
+    result = make_json_safe(obj)
+    assert result == {"type": "serialization_mixin", "method": "to_dict"}
+
+
 class DictObject:
     """Object with dict method."""
 
@@ -203,6 +217,41 @@ def test_make_json_safe_fallback():
     assert isinstance(result, dict)
 
 
+def test_make_json_safe_dataclass_with_nested_to_dict_object():
+    """Test dataclass containing a to_dict object (like HandoffAgentUserRequest with AgentResponse).
+
+    This test verifies the fix for the AG-UI JSON serialization error when
+    HandoffAgentUserRequest (a dataclass) contains an AgentResponse (SerializationMixin).
+    """
+
+    class NestedToDictObject:
+        """Simulates SerializationMixin objects like AgentResponse."""
+
+        def __init__(self, contents: list[str]):
+            self.contents = contents
+
+        def to_dict(self):
+            return {"type": "response", "contents": self.contents}
+
+    @dataclass
+    class ContainerDataclass:
+        """Simulates HandoffAgentUserRequest dataclass."""
+
+        response: NestedToDictObject
+
+    obj = ContainerDataclass(response=NestedToDictObject(contents=["hello", "world"]))
+    result = make_json_safe(obj)
+
+    # Verify the nested to_dict object was properly serialized
+    assert result == {"response": {"type": "response", "contents": ["hello", "world"]}}
+
+    # Verify the result is actually JSON serializable
+    import json
+
+    json_str = json.dumps(result)
+    assert json_str is not None
+
+
 def test_convert_tools_to_agui_format_with_ai_function():
     """Test converting AIFunction to AG-UI format."""
     from agent_framework import ai_function