Stream responses.create

README.md

@@ -481,6 +481,16 @@ response = client.responses.create(
     }],
 )
 
+# Streaming: events keep their Responses API type and fields
+stream = client.responses.create(
+    model="gpt-5.4",
+    input="Write a haiku about programming",
+    stream=True,
+)
+for event in stream:
+    if event.type == "response.output_text.delta":
+        print(event.delta, end="", flush=True)
+
 # With reasoning
 response = client.responses.create(
     model="gpt-5.4",
@@ -624,6 +634,8 @@ the `publish` confirmation value.
   `ResponseInputImage` (`input_image`) and `ResponseInputFile` (`input_file`). The old
   `text` / `image` part types still type-check.
 - Add `budget_tokens` to `ThinkingConfig`.
+- Add `stream` to `responses.create()`: `stream=True` returns `ResponseStreamEvent`
+  events, and an `error` event raises `APIError`.
 
 ### v0.4.0
 - Add `client.audio`: `speech.create()` (binary audio with `write_to_file()`),

apertis/resources/responses.py

@@ -2,15 +2,51 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Sequence, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Sequence, Union, overload
 
+from apertis._exceptions import _make_api_error
+from apertis._streaming import AsyncStream, Stream
 from apertis.types.chat import CompressionConfig
-from apertis.types.responses import Response, ResponseInputItem
+from apertis.types.responses import Response, ResponseInputItem, ResponseStreamEvent
 
 if TYPE_CHECKING:
+    import httpx
+
     from apertis._base_client import AsyncClient, SyncClient
 
 
+# Error codes and types to the HTTP status the same error has outside a stream.
+_ERROR_STATUS = {
+    "invalid_request_error": 400,
+    "authentication_error": 401,
+    "permission_error": 403,
+    "not_found_error": 404,
+    "rate_limit_exceeded": 429,
+    "rate_limit_error": 429,
+    "insufficient_quota": 429,
+    "server_error": 500,
+    "api_error": 500,
+    "overloaded_error": 529,
+}
+
+
+def _parse_event(data: Dict[str, Any], response: "httpx.Response") -> ResponseStreamEvent:
+    """Map one SSE payload to an event; raise on an error event."""
+    # OpenAI sends {"type": "error", "code", "message"}; the gateway may send {"error": {...}}.
+    if data.get("type") == "error" or ("type" not in data and "error" in data):
+        error = data.get("error")
+        error = error if isinstance(error, dict) else data
+        # The stream itself is HTTP 200; the error code or type says what failed.
+        status = _ERROR_STATUS.get(str(error.get("code"))) or _ERROR_STATUS.get(str(error.get("type")), 500)
+        raise _make_api_error(
+            str(error.get("message") or data),
+            status_code=status,
+            response=response,
+            body=data,
+        )
+    return ResponseStreamEvent.model_validate(data)
+
+
 class Responses:
     """Synchronous responses resource.
 
@@ -20,11 +56,66 @@ class Responses:
     def __init__(self, client: SyncClient) -> None:
         self._client = client
 
+    @overload
+    def create(
+        self,
+        *,
+        model: str,
+        input: Union[str, Sequence[ResponseInputItem]],
+        stream: Literal[True],
+        instructions: Optional[str] = None,
+        max_output_tokens: Optional[int] = None,
+        temperature: Optional[float] = None,
+        top_p: Optional[float] = None,
+        reasoning: Optional[Dict[str, Any]] = None,
+        tools: Optional[List[Dict[str, Any]]] = None,
+        tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        compression: Optional[CompressionConfig] = None,
+    ) -> Stream[ResponseStreamEvent]: ...
+
+    @overload
+    def create(
+        self,
+        *,
+        model: str,
+        input: Union[str, Sequence[ResponseInputItem]],
+        stream: Literal[False] = False,
+        instructions: Optional[str] = None,
+        max_output_tokens: Optional[int] = None,
+        temperature: Optional[float] = None,
+        top_p: Optional[float] = None,
+        reasoning: Optional[Dict[str, Any]] = None,
+        tools: Optional[List[Dict[str, Any]]] = None,
+        tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        compression: Optional[CompressionConfig] = None,
+    ) -> Response: ...
+
+    @overload
+    def create(
+        self,
+        *,
+        model: str,
+        input: Union[str, Sequence[ResponseInputItem]],
+        stream: bool = False,
+        instructions: Optional[str] = None,
+        max_output_tokens: Optional[int] = None,
+        temperature: Optional[float] = None,
+        top_p: Optional[float] = None,
+        reasoning: Optional[Dict[str, Any]] = None,
+        tools: Optional[List[Dict[str, Any]]] = None,
+        tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        compression: Optional[CompressionConfig] = None,
+    ) -> Union[Response, Stream[ResponseStreamEvent]]: ...
+
     def create(
         self,
         *,
         model: str,
         input: Union[str, Sequence[ResponseInputItem]],
+        stream: bool = False,
         instructions: Optional[str] = None,
         max_output_tokens: Optional[int] = None,
         temperature: Optional[float] = None,
@@ -34,12 +125,13 @@ def create(
         tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
         metadata: Optional[Dict[str, str]] = None,
         compression: Optional[CompressionConfig] = None,
-    ) -> Response:
+    ) -> Union[Response, Stream[ResponseStreamEvent]]:
         """Create a response.
 
         Args:
             model: ID of the model to use (e.g., "gpt-5.4").
             input: Input text or structured input items.
+            stream: If True, returns an iterator of ResponseStreamEvent.
             instructions: System instructions for the model.
             max_output_tokens: Maximum tokens to generate.
             temperature: Sampling temperature (0-2).
@@ -51,11 +143,12 @@ def create(
             compression: Context compression configuration.
 
         Returns:
-            Response object with generated content.
+            Response, or a Stream of ResponseStreamEvent when stream is True.
         """
-        body = self._build_request_body(
+        body = _build_request_body(
             model=model,
             input=input,
+            stream=stream,
             instructions=instructions,
             max_output_tokens=max_output_tokens,
             temperature=temperature,
@@ -67,66 +160,79 @@ def create(
             compression=compression,
         )
 
+        if stream:
+            return Stream(self._client.stream("POST", "/responses", json=body), _parse_event)
+
         response = self._client.request("POST", "/responses", json=body)
         return Response.model_validate(response.json())
 
-    def _build_request_body(
-        self,
-        *,
-        model: str,
-        input: Union[str, Sequence[ResponseInputItem]],
-        instructions: Optional[str],
-        max_output_tokens: Optional[int],
-        temperature: Optional[float],
-        top_p: Optional[float],
-        reasoning: Optional[Dict[str, Any]],
-        tools: Optional[List[Dict[str, Any]]],
-        tool_choice: Optional[Union[str, Dict[str, Any]]],
-        metadata: Optional[Dict[str, str]],
-        compression: Optional[CompressionConfig],
-    ) -> Dict[str, Any]:
-        """Build request body for responses."""
-        body: Dict[str, Any] = {"model": model}
-
-        # Handle input - can be string or list of input items
-        if isinstance(input, str):
-            body["input"] = input
-        else:
-            body["input"] = list(input)
-
-        if instructions is not None:
-            body["instructions"] = instructions
-        if max_output_tokens is not None:
-            body["max_output_tokens"] = max_output_tokens
-        if temperature is not None:
-            body["temperature"] = temperature
-        if top_p is not None:
-            body["top_p"] = top_p
-        if reasoning is not None:
-            body["reasoning"] = reasoning
-        if tools is not None:
-            body["tools"] = tools
-        if tool_choice is not None:
-            body["tool_choice"] = tool_choice
-        if metadata is not None:
-            body["metadata"] = metadata
-        if compression is not None:
-            body["compression"] = compression
-
-        return body
-
 
 class AsyncResponses:
     """Asynchronous responses resource."""
 
     def __init__(self, client: AsyncClient) -> None:
         self._client = client
 
+    @overload
+    async def create(
+        self,
+        *,
+        model: str,
+        input: Union[str, Sequence[ResponseInputItem]],
+        stream: Literal[True],
+        instructions: Optional[str] = None,
+        max_output_tokens: Optional[int] = None,
+        temperature: Optional[float] = None,
+        top_p: Optional[float] = None,
+        reasoning: Optional[Dict[str, Any]] = None,
+        tools: Optional[List[Dict[str, Any]]] = None,
+        tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        compression: Optional[CompressionConfig] = None,
+    ) -> AsyncStream[ResponseStreamEvent]: ...
+
+    @overload
+    async def create(
+        self,
+        *,
+        model: str,
+        input: Union[str, Sequence[ResponseInputItem]],
+        stream: Literal[False] = False,
+        instructions: Optional[str] = None,
+        max_output_tokens: Optional[int] = None,
+        temperature: Optional[float] = None,
+        top_p: Optional[float] = None,
+        reasoning: Optional[Dict[str, Any]] = None,
+        tools: Optional[List[Dict[str, Any]]] = None,
+        tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        compression: Optional[CompressionConfig] = None,
+    ) -> Response: ...
+
+    @overload
+    async def create(
+        self,
+        *,
+        model: str,
+        input: Union[str, Sequence[ResponseInputItem]],
+        stream: bool = False,
+        instructions: Optional[str] = None,
+        max_output_tokens: Optional[int] = None,
+        temperature: Optional[float] = None,
+        top_p: Optional[float] = None,
+        reasoning: Optional[Dict[str, Any]] = None,
+        tools: Optional[List[Dict[str, Any]]] = None,
+        tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
+        metadata: Optional[Dict[str, str]] = None,
+        compression: Optional[CompressionConfig] = None,
+    ) -> Union[Response, AsyncStream[ResponseStreamEvent]]: ...
+
     async def create(
         self,
         *,
         model: str,
         input: Union[str, Sequence[ResponseInputItem]],
+        stream: bool = False,
         instructions: Optional[str] = None,
         max_output_tokens: Optional[int] = None,
         temperature: Optional[float] = None,
@@ -136,14 +242,15 @@ async def create(
         tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
         metadata: Optional[Dict[str, str]] = None,
         compression: Optional[CompressionConfig] = None,
-    ) -> Response:
+    ) -> Union[Response, AsyncStream[ResponseStreamEvent]]:
         """Create a response asynchronously.
 
         See Responses.create() for parameter documentation.
         """
         body = _build_request_body(
             model=model,
             input=input,
+            stream=stream,
             instructions=instructions,
             max_output_tokens=max_output_tokens,
             temperature=temperature,
@@ -155,6 +262,9 @@ async def create(
             compression=compression,
         )
 
+        if stream:
+            return AsyncStream(await self._client.stream("POST", "/responses", json=body), _parse_event)
+
         response = await self._client.request("POST", "/responses", json=body)
         return Response.model_validate(response.json())
 
@@ -163,6 +273,7 @@ def _build_request_body(
     *,
     model: str,
     input: Union[str, Sequence[ResponseInputItem]],
+    stream: bool,
     instructions: Optional[str],
     max_output_tokens: Optional[int],
     temperature: Optional[float],
@@ -181,6 +292,8 @@ def _build_request_body(
     else:
         body["input"] = list(input)
 
+    if stream:
+        body["stream"] = True
     if instructions is not None:
         body["instructions"] = instructions
     if max_output_tokens is not None:

apertis/types/__init__.py

@@ -69,6 +69,7 @@
     ResponseReasoningItem,
     ResponseFunctionToolCall,
     ResponseUnknownOutputItem,
+    ResponseStreamEvent,
 )
 from apertis.types.messages import (
     Message,
@@ -156,6 +157,7 @@
     "ResponseReasoningItem",
     "ResponseFunctionToolCall",
     "ResponseUnknownOutputItem",
+    "ResponseStreamEvent",
     # Messages API
     "Message",
     "MessageParam",

apertis/types/responses.py

@@ -147,6 +147,29 @@ def output_text(self) -> str:
         )
 
 
+class ResponseStreamEvent(BaseModel):
+    """One event of a streamed Responses API call.
+
+    ``type`` names the event (``response.output_text.delta``, ``response.completed``, ...).
+    Fields not declared here are kept and readable as attributes. Events that the gateway
+    builds itself carry no ``sequence_number`` or ``item_id``.
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    type: str
+    sequence_number: Optional[int] = None
+    item_id: Optional[str] = None
+    output_index: Optional[int] = None
+    content_index: Optional[int] = None
+    # A string for text, argument and audio deltas; Any so an unforeseen shape cannot end the stream.
+    delta: Any = None
+    text: Optional[str] = None
+    item: Optional[Dict[str, Any]] = None
+    part: Optional[Dict[str, Any]] = None
+    response: Optional[Dict[str, Any]] = None
+
+
 # =============================================================================
 # Request Parameter Types
 # =============================================================================