diff --git a/src/openai/lib/_parsing/_responses.py b/src/openai/lib/_parsing/_responses.py
index c607587ec1..81e6b2b983 100644
--- a/src/openai/lib/_parsing/_responses.py
+++ b/src/openai/lib/_parsing/_responses.py
@@ -58,7 +58,7 @@ def parse_response(
 ) -> ParsedResponse[TextFormatT]:
     output_list: List[ParsedResponseOutputItem[TextFormatT]] = []
 
-    for output in response.output:
+    for output in response.output or []:
         if output.type == "message":
             content_list: List[ParsedContent[TextFormatT]] = []
             for item in output.content:
diff --git a/src/openai/lib/streaming/responses/_responses.py b/src/openai/lib/streaming/responses/_responses.py
index 6975a9260d..3c4d85b4e3 100644
--- a/src/openai/lib/streaming/responses/_responses.py
+++ b/src/openai/lib/streaming/responses/_responses.py
@@ -357,9 +357,15 @@ def accumulate_event(self, event: RawResponseStreamEvent) -> ParsedResponseSnaps
             if output.type == "function_call":
                 output.arguments += event.delta
         elif event.type == "response.completed":
+            response = event.response
+            if not response.output and snapshot.output:
+                # Some backends send `response.completed` with a null/empty `output`
+                # even though prior `response.output_item.added` / delta events already
+                # populated it on the snapshot; prefer the accumulated snapshot in that case.
+                response = response.model_copy(update={"output": snapshot.output})
             self._completed_response = parse_response(
                 text_format=self._text_format,
-                response=event.response,
+                response=response,
                 input_tools=self._input_tools,
             )
 
diff --git a/tests/lib/responses/test_responses.py b/tests/lib/responses/test_responses.py
index 4ed6dff47d..088c625b4a 100644
--- a/tests/lib/responses/test_responses.py
+++ b/tests/lib/responses/test_responses.py
@@ -72,6 +72,60 @@ def test_parse_response_preserves_program_items(item: dict[str, object]) -> None
     assert parsed.output[0].to_dict() == item
 
 
+def test_parse_response_with_null_output() -> None:
+    # Regression test for https://github.com/openai/openai-python/issues/3325
+    # Some backends (e.g. the chatgpt.com Codex backend) can send `output: null`
+    # in the `response.completed` event, even though the schema declares `output`
+    # as a non-nullable list. `parse_response` should not crash in this case.
+    response = construct_type_unchecked(type_=Response, value={"output": None})
+
+    parsed = parse_response(text_format=omit, input_tools=omit, response=response)
+
+    assert parsed.output == []
+
+
+def test_response_stream_state_preserves_accumulated_output_on_null_completion() -> None:
+    # Regression test for https://github.com/openai/openai-python/issues/3325
+    # A `response.completed` event with a null/empty `output` should not discard
+    # output already accumulated from prior `response.output_item.added` events.
+    from openai.types.responses import ResponseStreamEvent as RawResponseStreamEvent
+    from openai.lib.streaming.responses._responses import ResponseStreamState
+
+    state: ResponseStreamState[object] = ResponseStreamState(input_tools=omit, text_format=omit)
+
+    def make_event(value: dict[str, object]) -> RawResponseStreamEvent:
+        return construct_type_unchecked(type_=RawResponseStreamEvent, value=value)
+
+    created_response = {"id": "resp_123", "output": [], "status": "in_progress"}
+    state.handle_event(make_event({"type": "response.created", "response": created_response, "sequence_number": 0}))
+
+    message_item = {
+        "id": "msg_123",
+        "type": "message",
+        "role": "assistant",
+        "status": "in_progress",
+        "content": [],
+    }
+    state.handle_event(
+        make_event(
+            {
+                "type": "response.output_item.added",
+                "output_index": 0,
+                "item": message_item,
+                "sequence_number": 1,
+            }
+        )
+    )
+
+    completed_response = {"id": "resp_123", "output": None, "status": "completed"}
+    state.handle_event(make_event({"type": "response.completed", "response": completed_response, "sequence_number": 2}))
+
+    final = state._completed_response
+    assert final is not None
+    assert len(final.output) == 1
+    assert final.output[0].id == "msg_123"
+
+
 @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
 def test_stream_method_definition_in_sync(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None:
     checking_client: OpenAI | AsyncOpenAI = client if sync else async_client