Make shallow copies of lists instead of deep copies

src/agents/items.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import abc
-import copy
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar, Union
 
@@ -277,7 +276,7 @@ def input_to_new_input_list(
                     "role": "user",
                 }
             ]
-        return copy.deepcopy(input)
+        return input.copy()
 
     @classmethod
     def text_message_outputs(cls, items: list[RunItem]) -> str:

src/agents/run.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import copy
 import inspect
 from dataclasses import dataclass, field
 from typing import Any, Callable, Generic, cast
@@ -387,7 +386,7 @@ async def run(
             disabled=run_config.tracing_disabled,
         ):
             current_turn = 0
-            original_input: str | list[TResponseInputItem] = copy.deepcopy(prepared_input)
+            original_input: str | list[TResponseInputItem] = _copy_str_or_list(prepared_input)
             generated_items: list[RunItem] = []
             model_responses: list[ModelResponse] = []
 
@@ -446,7 +445,7 @@ async def run(
                                 starting_agent,
                                 starting_agent.input_guardrails
                                 + (run_config.input_guardrails or []),
-                                copy.deepcopy(prepared_input),
+                                _copy_str_or_list(prepared_input),
                                 context_wrapper,
                             ),
                             self._run_single_turn(
@@ -594,7 +593,7 @@ def run_streamed(
         )
 
         streamed_result = RunResultStreaming(
-            input=copy.deepcopy(input),
+            input=_copy_str_or_list(input),
             new_items=[],
             current_agent=starting_agent,
             raw_responses=[],
@@ -647,7 +646,7 @@ async def _maybe_filter_model_input(
 
         try:
             model_input = ModelInputData(
-                input=copy.deepcopy(effective_input),
+                input=effective_input.copy(),
                 instructions=effective_instructions,
             )
             filter_payload: CallModelData[TContext] = CallModelData(
@@ -786,7 +785,7 @@ async def _start_streaming(
                         cls._run_input_guardrails_with_queue(
                             starting_agent,
                             starting_agent.input_guardrails + (run_config.input_guardrails or []),
-                            copy.deepcopy(ItemHelpers.input_to_new_input_list(prepared_input)),
+                            ItemHelpers.input_to_new_input_list(prepared_input),
                             context_wrapper,
                             streamed_result,
                             current_span,
@@ -1376,3 +1375,9 @@ async def _save_result_to_session(
 
 
 DEFAULT_AGENT_RUNNER = AgentRunner()
+
+
+def _copy_str_or_list(input: str | list[TResponseInputItem]) -> str | list[TResponseInputItem]:
+    if isinstance(input, str):
+        return input
+    return input.copy()

tests/test_items_helpers.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import json
+
 from openai.types.responses.response_computer_tool_call import (
     ActionScreenshot,
     ResponseComputerToolCall,
@@ -20,8 +22,10 @@
 from openai.types.responses.response_output_message_param import ResponseOutputMessageParam
 from openai.types.responses.response_output_refusal import ResponseOutputRefusal
 from openai.types.responses.response_output_text import ResponseOutputText
+from openai.types.responses.response_output_text_param import ResponseOutputTextParam
 from openai.types.responses.response_reasoning_item import ResponseReasoningItem, Summary
 from openai.types.responses.response_reasoning_item_param import ResponseReasoningItemParam
+from pydantic import TypeAdapter
 
 from agents import (
     Agent,
@@ -290,3 +294,34 @@ def test_to_input_items_for_reasoning() -> None:
     print(converted_dict)
     print(expected)
     assert converted_dict == expected
+
+
+def test_input_to_new_input_list_copies_the_ones_produced_by_pydantic() -> None:
+    # Given a list of message dictionaries, ensure the returned list is a deep copy.
+    original = ResponseOutputMessageParam(
+        id="a75654dc-7492-4d1c-bce0-89e8312fbdd7",
+        content=[
+            ResponseOutputTextParam(
+                type="output_text",
+                text="Hey, what's up?",
+                annotations=[],
+            )
+        ],
+        role="assistant",
+        status="completed",
+        type="message",
+    )
+    original_json = json.dumps(original)
+    output_item = TypeAdapter(ResponseOutputMessageParam).validate_json(original_json)
+    new_list = ItemHelpers.input_to_new_input_list([output_item])
+    assert len(new_list) == 1
+    assert new_list[0]["id"] == original["id"]  # type: ignore
+    size = 0
+    for i, item in enumerate(original["content"]):
+        size += 1  # pydantic_core._pydantic_core.ValidatorIterator does not support len()
+        assert item["type"] == original["content"][i]["type"]  # type: ignore
+        assert item["text"] == original["content"][i]["text"]  # type: ignore
+    assert size == 1
+    assert new_list[0]["role"] == original["role"]  # type: ignore
+    assert new_list[0]["status"] == original["status"]  # type: ignore
+    assert new_list[0]["type"] == original["type"]