OpenAI Responses model, implemented in openai_responses.py, handles request headers, tool parameter validation, namespace grouping, and response streaming for the OpenAI Agents SDK's async Responses API integration. _ResponseStreamWithRequestId wraps WebSocket event streams to attach request IDs and track terminal events, while supporting fallback usage computation and configurable message-size limits for memory-constrained deployments.
openai_responses.py sets a User-Agent header of the form Agents/Python <version> on all Responses API requests via the _HEADERS constant.[1] Per-async-task header injection is supported through _HEADERS_OVERRIDE, a ContextVar[dict[str, str] | None] in openai_responses.py that defaults to None and allows per-context-variable overrides of request headers on Responses API calls.[1]
_require_responses_tool_param in openai_responses.py validates that a tool param payload is a Mapping with a string type key, raising TypeError with a descriptive message if either check fails.[1] _NamespaceToolParam is an internal TypedDict in openai_responses.py representing a namespace-grouped tool parameter with type="namespace", a name, a description, and a list of FunctionToolParam entries.[1] Namespace grouping of function tools is resolved by the _tool_identity module — openai_responses.py imports get_explicit_function_tool_namespace and get_function_tool_namespace_description from .._tool_identity before assembling entries into _NamespaceToolParam.[1] _coerce_response_includables in openai_responses.py deliberately accepts arbitrary strings for ModelSettings.response_include so callers can pass through new server-supported flags before the local SDK updates its enum union.[1]
_ResponseStreamWithRequestId in openai_responses.py wraps an async SDK event stream, retaining the originating request ID and back-propagating it onto each response object in every yielded event via _attach_request_id.[1] _ResponseStreamWithRequestId recognizes four terminal event types — response.completed, response.failed, response.incomplete, and response.error — and sets an internal _yielded_terminal_event flag when one is encountered.[1] For response.completed events whose response carries no usage data, _ResponseStreamWithRequestId.__anext__ calls _mark_transport_request_without_usage to ensure the request counter is preserved.[1] _construct_response_stream_event_from_payload in openai_responses.py parses WebSocket event payloads using the OpenAI SDK's internal construct_type function; if that internal is unavailable, it raises RuntimeError advising an SDK upgrade or a switch back to HTTP transport.[1] OpenAIResponsesWebSocketOptions exposes a max_size field controlling the maximum byte size of an incoming WebSocket message; setting it to None disables the limit, while an explicit value bounds memory usage for long-lived agent processes in memory-constrained containers.[1]
_usage_from_response in openai_responses.py falls back to Usage(requests=_requests_for_response_without_usage(response)) when response.usage is None, preserving the request count even for responses that omit usage data.[1] _json_dumps_default in openai_responses.py handles custom JSON serialization for Pydantic models (via model_dump(mode='json', exclude_none=True)), dataclasses (via asdict), and Enum values; it raises TypeError for all other unrecognized types.[1]
Sources