use safe json in langgraph python integration

Author: ranst91

python-sdk/ag_ui/core/utils.py

@@ -1,9 +1,8 @@
 """
 This module contains the EventEncoder class
 """
-import json
+
 from ag_ui.core.events import BaseEvent
-from ag_ui.core.utils import make_json_safe
 
 AGUI_MEDIA_TYPE = "application/vnd.ag-ui.event+proto"
 
@@ -30,7 +29,4 @@ def _encode_sse(self, event: BaseEvent) -> str:
         """
         Encodes an event into an SSE string.
         """
-        event_dict = event.model_dump(by_alias=True, exclude_none=True)
-        json_ready = make_json_safe(event_dict)
-        json_string = json.dumps(json_ready)
-        return f"data: {json_string}\n\n"
+        return f"data: {event.model_dump_json(by_alias=True, exclude_none=True)}\n\n"

python-sdk/ag_ui/encoder/encoder.py

@@ -4,7 +4,7 @@
 
 from ag_ui.encoder.encoder import EventEncoder, AGUI_MEDIA_TYPE
 from ag_ui.core.events import BaseEvent, EventType, TextMessageContentEvent, ToolCallStartEvent
-from ag_ui.core.utils import make_json_safe
+
 
 class TestEventEncoder(unittest.TestCase):
     """Test suite for EventEncoder class"""
@@ -29,10 +29,7 @@ def test_encode_method(self):
         encoded = encoder.encode(event)
 
         # The encode method calls encode_sse, so the result should be in SSE format
-        event_dict = event.model_dump(by_alias=True, exclude_none=True)
-        json_ready = make_json_safe(event_dict)
-        json_string = json.dumps(json_ready)
-        expected = f"data: {json_string}\n\n"
+        expected = f"data: {event.model_dump_json(by_alias=True, exclude_none=True)}\n\n"
         self.assertEqual(encoded, expected)
 
         # Verify that camelCase is used in the encoded output
@@ -81,7 +78,7 @@ def test_encode_with_different_event_types(self):
         # Test with a basic BaseEvent
         base_event = BaseEvent(type=EventType.RAW, timestamp=1648214400000)
         encoded_base = encoder.encode(base_event)
-        self.assertIn('"type": "RAW"', encoded_base)
+        self.assertIn('"type":"RAW"', encoded_base)
 
         # Test with a more complex event
         content_event = TextMessageContentEvent(
@@ -92,9 +89,9 @@ def test_encode_with_different_event_types(self):
         encoded_content = encoder.encode(content_event)
 
         # Verify correct encoding and camelCase conversion
-        self.assertIn('"type": "TEXT_MESSAGE_CONTENT"', encoded_content)
-        self.assertIn('"messageId": "msg_456"', encoded_content)  # Check snake_case converted to camelCase
-        self.assertIn('"delta": "Testing different events"', encoded_content)
+        self.assertIn('"type":"TEXT_MESSAGE_CONTENT"', encoded_content)
+        self.assertIn('"messageId":"msg_456"', encoded_content)  # Check snake_case converted to camelCase
+        self.assertIn('"delta":"Testing different events"', encoded_content)
 
         # Extract JSON and verify camelCase conversion
         json_content = encoded_content.split("data: ")[1].rstrip("\n\n")

python-sdk/tests/test_encoder.py

@@ -1,8 +1,6 @@
 import uuid
 import json
 from typing import Optional, List, Any, Union, AsyncGenerator, Generator
-from dataclasses import is_dataclass, asdict
-from datetime import date, datetime
 
 from langgraph.graph.state import CompiledStateGraph
 from langchain.schema import BaseMessage, SystemMessage
@@ -29,7 +27,9 @@
     langchain_messages_to_agui,
     resolve_reasoning_content,
     resolve_message_content,
-    camel_to_snake
+    camel_to_snake,
+    json_safe_stringify,
+    make_json_safe
 )
 
 from ag_ui.core import (
@@ -90,7 +90,7 @@ def __init__(self, *, name: str, graph: CompiledStateGraph, description: Optiona
         self.active_step = None
 
     def _dispatch_event(self, event: ProcessedEvents) -> str:
-        return event  # Fallback if no encoder
+        return make_json_safe(event)  # Fallback if no encoder
 
     async def run(self, input: RunAgentInput) -> AsyncGenerator[str, None]:
         forwarded_props = {}
@@ -224,7 +224,7 @@ async def _handle_stream_events(self, input: RunAgentInput) -> AsyncGenerator[st
                 CustomEvent(
                     type=EventType.CUSTOM,
                     name=LangGraphEventTypes.OnInterrupt.value,
-                    value=json.dumps(interrupt.value, default=make_json_safe) if not isinstance(interrupt.value, str) else interrupt.value,
+                    value=json.dumps(interrupt.value, default=json_safe_stringify) if not isinstance(interrupt.value, str) else interrupt.value,
                     raw_event=interrupt,
                 )
             )
@@ -735,16 +735,3 @@ def end_step(self):
         self.active_run["node_name"] = None
         self.active_step = None
         return dispatch
-
-def make_json_safe(o):
-    if is_dataclass(o):          # dataclasses like Flight(...)
-        return asdict(o)
-    if hasattr(o, "model_dump"): # pydantic v2
-        return o.model_dump()
-    if hasattr(o, "dict"):       # pydantic v1
-        return o.dict()
-    if hasattr(o, "__dict__"):   # plain objects
-        return vars(o)
-    if isinstance(o, (datetime, date)):
-        return o.isoformat()
-    return str(o)                # last resort

typescript-sdk/integrations/langgraph/python/ag_ui_langgraph/agent.py

@@ -1,6 +1,8 @@
 import json
 import re
 from typing import List, Any, Dict, Union
+from dataclasses import is_dataclass, asdict
+from datetime import date, datetime
 
 from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage, ToolMessage
 from ag_ui.core import (
@@ -177,3 +179,60 @@ def resolve_message_content(content: Any) -> str | None:
 
 def camel_to_snake(name):
     return re.sub(r'(?<!^)(?=[A-Z])', '_', name).lower()
+
+def json_safe_stringify(o):
+    if is_dataclass(o):          # dataclasses like Flight(...)
+        return asdict(o)
+    if hasattr(o, "model_dump"): # pydantic v2
+        return o.model_dump()
+    if hasattr(o, "dict"):       # pydantic v1
+        return o.dict()
+    if hasattr(o, "__dict__"):   # plain objects
+        return vars(o)
+    if isinstance(o, (datetime, date)):
+        return o.isoformat()
+    return str(o)                # last resort
+
+def make_json_safe(value: Any) -> Any:
+    """
+    Recursively convert a value into a JSON-serializable structure.
+
+    - Handles Pydantic models via `model_dump`.
+    - Handles LangChain messages via `to_dict`.
+    - Recursively walks dicts, lists, and tuples.
+    - For arbitrary objects, falls back to `__dict__` if available, else `repr()`.
+    """
+    # Pydantic models
+    if hasattr(value, "model_dump"):
+        try:
+            return make_json_safe(value.model_dump(by_alias=True, exclude_none=True))
+        except Exception:
+            pass
+
+    # LangChain-style objects
+    if hasattr(value, "to_dict"):
+        try:
+            return make_json_safe(value.to_dict())
+        except Exception:
+            pass
+
+    # Dict
+    if isinstance(value, dict):
+        return {key: make_json_safe(sub_value) for key, sub_value in value.items()}
+
+    # List / tuple
+    if isinstance(value, (list, tuple)):
+        return [make_json_safe(sub_value) for sub_value in value]
+
+    # Already JSON safe
+    if isinstance(value, (str, int, float, bool)) or value is None:
+        return value
+
+    # Arbitrary object: try __dict__ first, fallback to repr
+    if hasattr(value, "__dict__"):
+        return {
+            "__type__": type(value).__name__,
+            **make_json_safe(value.__dict__),
+        }
+
+    return repr(value)