fix: encode structured programmatic tool errors as JSON

src/agents/run_internal/items.py

@@ -75,6 +75,7 @@
     "deduplicate_input_items",
     "deduplicate_input_items_preferring_latest",
     "strip_internal_input_item_metadata",
+    "function_tool_error_output",
     "function_rejection_item",
     "shell_rejection_item",
     "apply_patch_rejection_item",
@@ -733,22 +734,49 @@ def deduplicate_input_items_preferring_latest(
     return list(reversed(deduplicate_input_items(list(reversed(items)))))
 
 
+def function_tool_error_output(
+    tool_call: Any,
+    output: Any,
+    *,
+    output_json_schema: dict[str, Any] | None,
+) -> Any:
+    """Encode SDK-generated programmatic tool errors as provider-compatible JSON objects."""
+    if output_json_schema is None or not isinstance(output, str):
+        return output
+
+    if isinstance(tool_call, dict):
+        caller = tool_call.get("caller")
+    else:
+        caller = getattr(tool_call, "caller", None)
+    caller_type = caller.get("type") if isinstance(caller, dict) else getattr(caller, "type", None)
+    if caller_type != "program":
+        return output
+
+    return json.dumps({"error": output}, ensure_ascii=False, separators=(",", ":"))
+
+
 def function_rejection_item(
     agent: Any,
     tool_call: Any,
     *,
     rejection_message: str = REJECTION_MESSAGE,
+    output_json_schema: dict[str, Any] | None = None,
     scope_id: str | None = None,
     tool_origin: Any = None,
 ) -> ToolCallOutputItem:
     """Build a ToolCallOutputItem representing a rejected function tool call."""
     if isinstance(tool_call, ResponseFunctionToolCall):
         drop_agent_tool_run_result(tool_call, scope_id=scope_id)
+    provider_output = function_tool_error_output(
+        tool_call,
+        rejection_message,
+        output_json_schema=output_json_schema,
+    )
     return ToolCallOutputItem(
         output=rejection_message,
         raw_item=ItemHelpers.tool_call_output_item(
             tool_call,
-            rejection_message,
+            provider_output,
         ),
         agent=agent,
         tool_origin=tool_origin,

src/agents/run_internal/tool_execution.py

@@ -104,6 +104,7 @@
     extract_mcp_request_id,
     extract_mcp_request_id_from_run,
     function_rejection_item,
+    function_tool_error_output,
 )
 from .run_steps import ToolRunFunction
 from .tool_use_tracker import AgentToolUseTracker
@@ -1729,6 +1730,7 @@ async def _maybe_execute_tool_approval(
                             self.public_agent,
                             tool_call,
                             rejection_message=rejected_message,
+                            output_json_schema=func_tool.output_json_schema,
                             scope_id=self.tool_state_scope_id,
                             tool_origin=get_function_tool_origin(func_tool),
                         ),
@@ -1778,6 +1780,7 @@ async def _maybe_execute_tool_approval(
                 self.public_agent,
                 tool_call,
                 rejection_message=rejection_message,
+                output_json_schema=func_tool.output_json_schema,
                 scope_id=self.tool_state_scope_id,
                 tool_origin=get_function_tool_origin(func_tool),
             ),
@@ -1891,9 +1894,18 @@ async def _invoke_tool_and_run_post_invoke(
         bypass_output_schema = bypass_output_schema or (output_guardrail_result.is_rejection)
         if bypass_output_schema:
             self.schema_bypassed_tool_runs.add(id(task_state.tool_run))
+        provider_result = (
+            function_tool_error_output(
+                tool_call,
+                final_result,
+                output_json_schema=func_tool.output_json_schema,
+            )
+            if bypass_output_schema
+            else final_result
+        )
         raw_output_item = ItemHelpers.tool_call_output_item(
             tool_call,
-            final_result,
+            provider_result,
             output_json_schema=None if bypass_output_schema else func_tool.output_json_schema,
             output_type_adapter=None if bypass_output_schema else func_tool._output_type_adapter,
         )
@@ -2012,11 +2024,20 @@ def _build_function_tool_results(self) -> list[FunctionToolResult]:
 
             run_item: RunItem | None
             if not nested_interruptions:
+                provider_result = (
+                    function_tool_error_output(
+                        tool_run.tool_call,
+                        result,
+                        output_json_schema=tool_run.function_tool.output_json_schema,
+                    )
+                    if bypass_output_schema
+                    else result
+                )
                 run_item = ToolCallOutputItem(
                     output=result,
                     raw_item=ItemHelpers.tool_call_output_item(
                         tool_run.tool_call,
-                        result,
+                        provider_result,
                         output_json_schema=(
                             None
                             if bypass_output_schema

src/agents/run_internal/turn_resolution.py

@@ -1051,6 +1051,7 @@ async def _record_function_rejection(
                 public_agent,
                 tool_call,
                 rejection_message=rejection_message,
+                output_json_schema=function_tool.output_json_schema,
                 scope_id=tool_state_scope_id,
                 tool_origin=get_function_tool_origin(function_tool),
             )

tests/test_programmatic_tool_calling.py

@@ -1934,7 +1934,7 @@ def lookup_inventory(sku: str) -> InventoryOutput:
     assert result.final_output == "request rejected"
     function_outputs = _function_output_raw_items(result)
     assert len(function_outputs) == 1
-    assert function_outputs[0]["output"] == "inventory lookup blocked"
+    assert json.loads(function_outputs[0]["output"]) == {"error": "inventory lookup blocked"}
     assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER
 
 
@@ -1964,8 +1964,9 @@ async def lookup_inventory(sku: str) -> InventoryOutput:
     assert result.final_output == "request timed out"
     function_outputs = _function_output_raw_items(result)
     assert len(function_outputs) == 1
-    assert isinstance(function_outputs[0]["output"], str)
-    assert "timed out" in function_outputs[0]["output"].lower()
+    timeout_output = json.loads(function_outputs[0]["output"])
+    assert isinstance(timeout_output, dict)
+    assert "timed out" in timeout_output["error"].lower()
     assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER
 
 
@@ -2001,12 +2002,23 @@ def lookup_inventory(sku: str) -> InventoryOutput:
     assert result.final_output == "request rejected"
     function_outputs = _function_output_raw_items(result)
     assert len(function_outputs) == 1
-    assert function_outputs[0]["output"] == "inventory result blocked"
+    assert json.loads(function_outputs[0]["output"]) == {"error": "inventory result blocked"}
     assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER
 
 
 @pytest.mark.asyncio
-async def test_typed_programmatic_tool_preserves_approval_rejection() -> None:
+@pytest.mark.parametrize("streaming", [False, True], ids=["nonstreaming", "streaming"])
+@pytest.mark.parametrize("serialize_state", [False, True], ids=["in-memory", "serialized"])
+@pytest.mark.parametrize(
+    "rejection_message",
+    [None, 'Denied: "東京"'],
+    ids=["default-rejection", "custom-rejection"],
+)
+async def test_typed_programmatic_tool_preserves_approval_rejection(
+    streaming: bool,
+    serialize_state: bool,
+    rejection_message: str | None,
+) -> None:
     model = FakeModel()
     model.add_multiple_turn_outputs(
         [
@@ -2024,18 +2036,40 @@ def lookup_inventory(sku: str) -> InventoryOutput:
         model=model,
         tools=[ProgrammaticToolCallingTool(), lookup_inventory],
     )
-    first_result = await Runner.run(agent, "Check inventory")
+    first_result: Any
+    if streaming:
+        first_result = Runner.run_streamed(agent, "Check inventory")
+        async for _event in first_result.stream_events():
+            pass
+    else:
+        first_result = await Runner.run(agent, "Check inventory")
     assert len(first_result.interruptions) == 1
 
     state = first_result.to_state()
-    state.reject(first_result.interruptions[0])
-    result = await Runner.run(agent, state)
+    if serialize_state:
+        state = await RunState.from_json(agent, state.to_json())
+    state.reject(state.get_interruptions()[0], rejection_message=rejection_message)
+    result: Any
+    if streaming:
+        result = Runner.run_streamed(agent, state)
+        async for _event in result.stream_events():
+            pass
+    else:
+        result = await Runner.run(agent, state)
 
     assert result.final_output == "request rejected"
     function_outputs = _function_output_raw_items(result)
     assert len(function_outputs) == 1
-    assert function_outputs[0]["output"] == "Tool execution was not approved."
+    expected_message = rejection_message or "Tool execution was not approved."
+    assert json.loads(function_outputs[0]["output"]) == {"error": expected_message}
     assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER
+    assert model.last_turn_args is not None
+    replayed_output = next(
+        item
+        for item in model.last_turn_args["input"]
+        if isinstance(item, dict) and item.get("type") == "function_call_output"
+    )
+    assert json.loads(replayed_output["output"]) == {"error": expected_message}
 
 
 @pytest.mark.asyncio
@@ -2199,7 +2233,7 @@ def lookup_inventory(sku: str) -> InventoryOutput:
     assert result.final_output == "request rejected"
     function_outputs = _function_output_raw_items(result)
     assert len(function_outputs) == 1
-    assert function_outputs[0]["output"] == "inventory lookup blocked"
+    assert json.loads(function_outputs[0]["output"]) == {"error": "inventory lookup blocked"}
     assert _caller_dict(function_outputs[0]["caller"]) == PROGRAM_CALLER
 
 

tests/test_run_internal_items.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import dataclasses
+import json
 from typing import Any, cast
 
 import pytest
@@ -9,6 +10,7 @@
     ResponseToolSearchCall,
     ResponseToolSearchOutputItem,
 )
+from openai.types.responses.response_function_tool_call import CallerProgram
 from openai.types.responses.response_reasoning_item import ResponseReasoningItem
 
 from agents import Agent
@@ -27,6 +29,51 @@
 from agents.run_internal import items as run_items
 
 
+@pytest.mark.parametrize("mapping_call", [False, True], ids=["typed-call", "mapping-call"])
+def test_programmatic_structured_tool_errors_are_encoded_as_json_objects(
+    mapping_call: bool,
+) -> None:
+    caller = {"type": "program", "caller_id": "program-42"}
+    tool_call: Any
+    if mapping_call:
+        tool_call = {"type": "function_call", "call_id": "call-42", "caller": caller}
+    else:
+        tool_call = ResponseFunctionToolCall(
+            type="function_call",
+            call_id="call-42",
+            name="lookup",
+            arguments="{}",
+            caller=CallerProgram(type="program", caller_id="program-42"),
+        )
+
+    output = run_items.function_tool_error_output(
+        tool_call,
+        'Rejected: "東京"',
+        output_json_schema={"type": "object"},
+    )
+
+    assert json.loads(output) == {"error": 'Rejected: "東京"'}
+
+
+@pytest.mark.parametrize("caller", [None, {"type": "direct"}], ids=["no-caller", "direct"])
+@pytest.mark.parametrize("has_schema", [False, True], ids=["untyped", "typed"])
+def test_direct_function_tool_errors_preserve_plain_text(
+    caller: dict[str, str] | None,
+    has_schema: bool,
+) -> None:
+    tool_call: dict[str, Any] = {"type": "function_call", "call_id": "call-42"}
+    if caller is not None:
+        tool_call["caller"] = caller
+
+    output = run_items.function_tool_error_output(
+        tool_call,
+        "Request rejected.",
+        output_json_schema={"type": "object"} if has_schema else None,
+    )
+
+    assert output == "Request rejected."
+
+
 def test_drop_orphan_function_calls_preserves_non_mapping_entries() -> None:
     payload: list[Any] = [
         cast(TResponseInputItem, "plain-text-input"),