Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion scripts/test
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ else
echo
fi

export DEFER_PYDANTIC_BUILD=false
export OPENAI_PYDANTIC_DEFER_BUILD=false

if [ "${OPENAI_TEST_HTTP_CLIENT:-httpx}" = "httpx2" ]; then
echo "==> Using HTTPX2 for sync and async API-resource clients (including RESPX-backed cases)"
Expand Down
3 changes: 2 additions & 1 deletion src/openai/_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,8 @@ def __repr_args__(self) -> ReprArgs:
return [arg for arg in super().__repr_args__() if arg[0] not in {"_request_id", "__exclude_fields__"}]
else:
model_config: ClassVar[ConfigDict] = ConfigDict(
extra="allow", defer_build=coerce_boolean(os.environ.get("DEFER_PYDANTIC_BUILD", "true"))
extra="allow",
defer_build=coerce_boolean(os.environ.get("OPENAI_PYDANTIC_DEFER_BUILD", "true")),
)

if TYPE_CHECKING:
Expand Down
3 changes: 2 additions & 1 deletion src/openai/_utils/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -337,7 +337,8 @@ def coerce_float(val: str) -> float:
return float(val)


def coerce_boolean(val: str) -> bool:
def coerce_boolean(obj: object) -> bool:
val = str(obj).lower()
return val == "true" or val == "1" or val == "on"


Expand Down
56 changes: 55 additions & 1 deletion src/openai/lib/streaming/responses/_responses.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,15 @@ def get_final_response(self) -> ParsedResponse[TextFormatT]:

return response

def get_abort_reconciliation_items(self) -> List[Dict[str, Any]]:
"""Returns a list of synthetic function_call_output items for any pending tool calls.

This should be used if the stream is closed before it has been read to completion
to ensure that the conversation state remains consistent.
"""
return self._state.get_abort_reconciliation_items()
Comment on lines +89 to +95

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Remove the unrelated streaming API addition

This environment-variable rename also adds a public ResponseStream method plus substantial reconciliation state machinery that is unrelated to the stated fix and has no focused synchronous or asynchronous tests. Because ResponseStream is re-exported from openai.lib.streaming.responses, merging this commit would silently expand the supported SDK API with unvalidated semantics; move these _responses.py changes to the dedicated feature change instead.

AGENTS.md reference: AGENTS.md:L5-L8

Useful? React with 👍 / 👎.



def until_done(self) -> Self:
"""Blocks until the stream has been consumed."""
consume_sync_iterator(self)
Expand Down Expand Up @@ -188,6 +197,15 @@ async def get_final_response(self) -> ParsedResponse[TextFormatT]:

return response

def get_abort_reconciliation_items(self) -> List[Dict[str, Any]]:
"""Returns a list of synthetic function_call_output items for any pending tool calls.

This should be used if the stream is closed before it has been read to completion
to ensure that the conversation state remains consistent.
"""
return self._state.get_abort_reconciliation_items()


async def until_done(self) -> Self:
"""Blocks until the stream has been consumed."""
await consume_async_iterator(self)
Expand Down Expand Up @@ -340,6 +358,11 @@ def accumulate_event(self, event: RawResponseStreamEvent) -> ParsedResponseSnaps
)
else:
snapshot.output.append(event.item)
elif event.type == "response.output_item.done":
# The JS SDK tracks this to know which tool calls are "complete" from the model's perspective.
# In the Python SDK, ParsedResponseFunctionToolCall is updated via deltas,
# but we can use this to mark it as done if needed.
pass
elif event.type == "response.content_part.added":
output = snapshot.output[event.output_index]
if output.type == "message":
Expand Down Expand Up @@ -369,4 +392,35 @@ def _create_initial_response(self, event: RawResponseStreamEvent) -> ParsedRespo
if event.type != "response.created":
raise RuntimeError(f"Expected to have received `response.created` before `{event.type}`")

return construct_type_unchecked(type_=ParsedResponseSnapshot, value=event.response.to_dict())
snapshot = construct_type_unchecked(type_=ParsedResponseSnapshot, value=event.response.to_dict())
if snapshot.output is None:
snapshot.output = []
return snapshot


def get_abort_reconciliation_items(self) -> List[Dict[str, Any]]:
"""Returns a list of synthetic function_call_output items for any pending tool calls.

This is used to reconcile the conversation state if the stream is aborted
after the model has generated a tool call but before it was completed.
"""
snapshot = self.__current_snapshot
if not snapshot:
return []

items: List[Dict[str, Any]] = []
for output in snapshot.output:
if output.type == "function_call":
Comment on lines +412 to +413

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Skip calls that already have an output

When a streamed response contains both a function_call and its corresponding function_call_output—both are valid ResponseOutputItem variants and the accumulator retains both—this loop still emits another synthetic output for that call ID. Passing the returned reconciliation list into the next request can therefore submit two outputs for an already-satisfied call; collect existing output call IDs and exclude them here.

Useful? React with 👍 / 👎.

# We consider it "pending" if we haven't received a corresponding result yet.
# In the Responses API, results are submitted in a separate turn or turn-part.
# If we are in the middle of a stream that generated a function_call,
# and the stream is aborted, we should mark it as incomplete.
items.append({
"type": "function_call_output",
"call_id": output.call_id,
"name": output.name,
"status": "incomplete",
"output": "aborted"
})
return items