Python: Normalize OpenAI function-call arguments at parse time to prevent uni…

python/packages/core/agent_framework/__init__.py

@@ -141,6 +141,7 @@
     detect_media_type_from_base64,
     map_chat_to_agent_update,
     merge_chat_options,
+    normalize_function_call_arguments,
     normalize_messages,
     normalize_tools,
     prepend_instructions_to_messages,

python/packages/core/agent_framework/_types.py

@@ -1511,6 +1511,31 @@ def parse_arguments(self) -> dict[str, Any | None] | None:
         return self.arguments  # type: ignore[return-value]
 
 
+def normalize_function_call_arguments(
+    arguments: str | Mapping[str, Any] | None,
+) -> Mapping[str, Any] | str | None:
+    """Normalize function-call arguments to a mapping when the payload is a JSON object.
+
+    OpenAI chat and responses parsers return function-call arguments as raw JSON
+    strings, while Anthropic already returns parsed dicts.  Eagerly normalizing
+    to a dict avoids a second ``json.loads`` in ``parse_arguments`` that would
+    re-interpret ``\\uXXXX`` escape sequences as Unicode characters — the root
+    cause of escape-corruption bugs when editing source files that contain
+    Python/JS-style ``\\u`` escapes.
+    """
+    if arguments is None or isinstance(arguments, Mapping):
+        return arguments
+    if not isinstance(arguments, str) or not arguments.strip():
+        return arguments
+    try:
+        parsed = json.loads(arguments)
+        if isinstance(parsed, dict):
+            return parsed
+        return arguments
+    except (json.JSONDecodeError, TypeError):
+        return arguments
+
+
 def _combine_additional_props(
     self_additional_properties: dict[str, Any], other_additional_properties: dict[str, Any]
 ) -> dict[str, Any]:

python/packages/core/agent_framework/openai/_chat_client.py

@@ -49,6 +49,7 @@
     Message,
     ResponseStream,
     UsageDetails,
+    normalize_function_call_arguments,
 )
 from ..exceptions import (
     ChatClientException,
@@ -556,7 +557,9 @@ def _parse_tool_calls_from_openai(self, choice: Choice | ChunkChoice) -> list[Co
                     fcc = Content.from_function_call(
                         call_id=tool.id if tool.id else "",
                         name=tool.function.name if tool.function.name else "",
-                        arguments=tool.function.arguments if tool.function.arguments else "",
+                        arguments=normalize_function_call_arguments(
+                            tool.function.arguments if tool.function.arguments else ""
+                        ),
                         raw_representation=tool.function,
                     )
                     resp.append(fcc)

python/packages/core/agent_framework/openai/_responses_client.py

@@ -74,6 +74,7 @@
     UsageDetails,
     detect_media_type_from_base64,
     prepend_instructions_to_messages,
+    normalize_function_call_arguments,
     validate_tool_mode,
 )
 from ..exceptions import (
@@ -1192,12 +1193,17 @@ def _prepare_content_for_openai(
                 if not fc_id.startswith("fc_"):
                     fc_id = f"fc_{fc_id}"
 
+                args = (
+                    json.dumps(content.arguments)
+                    if isinstance(content.arguments, Mapping)
+                    else (content.arguments or "")
+                )
                 function_call_obj = {
                     "call_id": content.call_id,
                     "id": fc_id,
                     "type": "function_call",
                     "name": content.name,
-                    "arguments": content.arguments,
+                    "arguments": args,
                 }
                 if status := content.additional_properties.get("status"):
                     function_call_obj["status"] = status
@@ -1244,10 +1250,12 @@ def _prepare_content_for_openai(
                     "output": output,
                 }
             case "function_approval_request":
+                fc_args = content.function_call.arguments  # type: ignore[union-attr]
+                fc_args_str = json.dumps(fc_args) if isinstance(fc_args, Mapping) else (fc_args or "")
                 return {
                     "type": "mcp_approval_request",
                     "id": content.id,  # type: ignore[union-attr]
-                    "arguments": content.function_call.arguments,  # type: ignore[union-attr]
+                    "arguments": fc_args_str,
                     "name": content.function_call.name,  # type: ignore[union-attr]
                     "server_label": content.function_call.additional_properties.get("server_label")  # type: ignore[union-attr]
                     if content.function_call.additional_properties  # type: ignore[union-attr]
@@ -1523,7 +1531,7 @@ def _parse_response_from_openai(
                         Content.from_function_call(
                             call_id=item.call_id,
                             name=item.name,
-                            arguments=item.arguments,
+                            arguments=normalize_function_call_arguments(item.arguments),
                             additional_properties={"fc_id": item.id, "status": item.status},
                             raw_representation=item,
                         )
@@ -1535,7 +1543,7 @@ def _parse_response_from_openai(
                             function_call=Content.from_function_call(
                                 call_id=item.id,
                                 name=item.name,
-                                arguments=item.arguments,
+                                arguments=normalize_function_call_arguments(item.arguments),
                                 additional_properties={"server_label": item.server_label},
                                 raw_representation=item,
                             ),
@@ -1915,7 +1923,7 @@ def _parse_chunk_from_openai(
                                 function_call=Content.from_function_call(
                                     call_id=event_item.id,
                                     name=event_item.name,
-                                    arguments=event_item.arguments,
+                                    arguments=normalize_function_call_arguments(event_item.arguments),
                                     additional_properties={"server_label": event_item.server_label},
                                     raw_representation=event_item,
                                 ),

python/packages/core/tests/openai/test_openai_responses_client.py

@@ -1037,7 +1037,7 @@ def test_response_content_creation_with_function_call() -> None:
     function_call = response.messages[0].contents[0]
     assert function_call.call_id == "call_123"
     assert function_call.name == "get_weather"
-    assert function_call.arguments == '{"location": "Seattle"}'
+    assert function_call.arguments == {"location": "Seattle"}
 
 
 def test_prepare_content_for_opentool_approval_response() -> None:
@@ -3581,7 +3581,7 @@ def test_parse_response_from_openai_function_call_includes_status() -> None:
     assert function_call.type == "function_call"
     assert function_call.call_id == "call_123"
     assert function_call.name == "get_weather"
-    assert function_call.arguments == '{"location": "Seattle"}'
+    assert function_call.arguments == {"location": "Seattle"}
     # Verify status is included in additional_properties
     assert function_call.additional_properties is not None
     assert function_call.additional_properties.get("status") == "completed"