feat(messages): add stream, thinking and extra_body

README.md

@@ -404,6 +404,28 @@ if message.stop_reason == "tool_use":
             print(f"Tool: {block.name}, Input: {block.input}")
 ```
 
+Streaming and extended thinking:
+
+```python
+stream = client.messages.create(
+    model="claude-sonnet-4-6",
+    messages=[{"role": "user", "content": "What is 27 * 453?"}],
+    max_tokens=4096,
+    thinking={"type": "enabled", "budget_tokens": 2048},
+    stream=True,
+)
+
+for event in stream:
+    if event.type == "content_block_delta":
+        if event.delta.type == "thinking_delta":
+            print(event.delta.thinking, end="")
+        elif event.delta.type == "text_delta":
+            print(event.delta.text, end="")
+```
+
+Without `stream`, thinking arrives as `thinking` blocks in `message.content`, before
+the text. `extra_body={...}` merges extra fields into the top level of the request.
+
 ### Responses API
 
 Use OpenAI Responses API format for advanced use cases:
@@ -559,6 +581,10 @@ the `publish` confirmation value.
 ## Changelog
 
 ### Unreleased
+- Add `stream`, `thinking` and `extra_body` to `messages.create()`, with typed stream
+  events and `thinking` / `redacted_thinking` content blocks.
+- Streamed requests that fail with an HTTP error now raise `APIError` (previously
+  `httpx.ResponseNotRead`).
 - Parse real Responses API output: `output_text` and `refusal` message parts,
   `reasoning` and `function_call` items, and unmodelled item types (kept as
   `ResponseUnknownOutputItem`). Add `Response.output_text` and the `queued` /

apertis/_base_client.py

@@ -130,7 +130,6 @@ def stream(
         headers: Optional[Mapping[str, str]] = None,
     ) -> httpx.Response:
         """Make a streaming HTTP request."""
-        import json as json_lib
         url = self._build_url(path)
         request_headers = self._build_headers(headers)
 
@@ -145,6 +144,7 @@ def stream(
             response = self._client.send(request, stream=True)
 
             if response.status_code >= 400:
+                response.read()  # the error body is needed for the APIError
                 self._raise_for_status(response)
 
             return response
@@ -269,6 +269,7 @@ async def stream(
             response = await self._client.send(request, stream=True)
 
             if response.status_code >= 400:
+                await response.aread()  # the error body is needed for the APIError
                 await self._raise_for_status(response)
 
             return response

apertis/_streaming.py

@@ -3,25 +3,47 @@
 from __future__ import annotations
 
 import json
-from typing import TYPE_CHECKING, AsyncIterator, Iterator, Optional
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncIterator,
+    Callable,
+    Dict,
+    Generic,
+    Iterator,
+    Optional,
+    TypeVar,
+)
 
 from apertis.types.chat import ChatCompletionChunk
 
 if TYPE_CHECKING:
     import httpx
 
+_T = TypeVar("_T")
 
-class Stream:
+# Turns one decoded `data:` payload into an item; returning None skips it.
+ParseFn = Callable[[Dict[str, Any], "httpx.Response"], Optional[_T]]
+
+
+def _parse_chat_chunk(data: Dict[str, Any], response: "httpx.Response") -> ChatCompletionChunk:
+    return ChatCompletionChunk.model_validate(data)
+
+
+class Stream(Generic[_T]):
     """Synchronous streaming response handler."""
 
-    def __init__(self, response: "httpx.Response") -> None:
+    def __init__(
+        self, response: "httpx.Response", parse: ParseFn[_T] = _parse_chat_chunk  # type: ignore[assignment]
+    ) -> None:
         self._response = response
+        self._parse = parse
         self._iterator: Optional[Iterator[str]] = None
 
-    def __iter__(self) -> "Stream":
+    def __iter__(self) -> "Stream[_T]":
         return self
 
-    def __next__(self) -> ChatCompletionChunk:
+    def __next__(self) -> _T:
         if self._iterator is None:
             self._iterator = self._response.iter_lines()
 
@@ -34,13 +56,19 @@ def __next__(self) -> ChatCompletionChunk:
                     raise StopIteration
                 try:
                     chunk_data = json.loads(data)
-                    return ChatCompletionChunk.model_validate(chunk_data)
                 except json.JSONDecodeError:
                     continue
+                try:
+                    item = self._parse(chunk_data, self._response)
+                except BaseException:
+                    self.close()
+                    raise
+                if item is not None:
+                    return item
 
         raise StopIteration
 
-    def __enter__(self) -> "Stream":
+    def __enter__(self) -> "Stream[_T]":
         return self
 
     def __exit__(self, *args: object) -> None:
@@ -51,17 +79,20 @@ def close(self) -> None:
         self._response.close()
 
 
-class AsyncStream:
+class AsyncStream(Generic[_T]):
     """Asynchronous streaming response handler."""
 
-    def __init__(self, response: "httpx.Response") -> None:
+    def __init__(
+        self, response: "httpx.Response", parse: ParseFn[_T] = _parse_chat_chunk  # type: ignore[assignment]
+    ) -> None:
         self._response = response
+        self._parse = parse
         self._iterator: Optional[AsyncIterator[str]] = None
 
-    def __aiter__(self) -> "AsyncStream":
+    def __aiter__(self) -> "AsyncStream[_T]":
         return self
 
-    async def __anext__(self) -> ChatCompletionChunk:
+    async def __anext__(self) -> _T:
         if self._iterator is None:
             self._iterator = self._response.aiter_lines()
 
@@ -74,13 +105,19 @@ async def __anext__(self) -> ChatCompletionChunk:
                     raise StopAsyncIteration
                 try:
                     chunk_data = json.loads(data)
-                    return ChatCompletionChunk.model_validate(chunk_data)
                 except json.JSONDecodeError:
                     continue
+                try:
+                    item = self._parse(chunk_data, self._response)
+                except BaseException:
+                    await self.close()
+                    raise
+                if item is not None:
+                    return item
 
         raise StopAsyncIteration
 
-    async def __aenter__(self) -> "AsyncStream":
+    async def __aenter__(self) -> "AsyncStream[_T]":
         return self
 
     async def __aexit__(self, *args: object) -> None:

apertis/resources/chat/completions.py

@@ -14,6 +14,7 @@
 from apertis.types.chat import (
     AudioConfig,
     ChatCompletion,
+    ChatCompletionChunk,
     ChatCompletionMessageParam,
     ChatCompletionToolParam,
     CompressionConfig,
@@ -76,7 +77,7 @@ def create(
         stream_options: StreamOptions | None = None,
         # Provider-specific params
         extra_body: dict[str, Any] | None = None,
-    ) -> Stream: ...
+    ) -> Stream[ChatCompletionChunk]: ...
 
     @overload
     def create(
@@ -164,7 +165,7 @@ def create(
         stream_options: StreamOptions | None = None,
         # Provider-specific params
         extra_body: dict[str, Any] | None = None,
-    ) -> ChatCompletion | Stream: ...
+    ) -> ChatCompletion | Stream[ChatCompletionChunk]: ...
 
     def create(
         self,
@@ -207,7 +208,7 @@ def create(
         stream_options: StreamOptions | None = None,
         # Provider-specific params
         extra_body: dict[str, Any] | None = None,
-    ) -> ChatCompletion | Stream:
+    ) -> ChatCompletion | Stream[ChatCompletionChunk]:
         """Create a chat completion.
 
         Args:
@@ -495,7 +496,7 @@ async def create(
         compression: CompressionConfig | None = None,
         stream_options: StreamOptions | None = None,
         extra_body: dict[str, Any] | None = None,
-    ) -> AsyncStream: ...
+    ) -> AsyncStream[ChatCompletionChunk]: ...
 
     @overload
     async def create(
@@ -569,7 +570,7 @@ async def create(
         compression: CompressionConfig | None = None,
         stream_options: StreamOptions | None = None,
         extra_body: dict[str, Any] | None = None,
-    ) -> ChatCompletion | AsyncStream: ...
+    ) -> ChatCompletion | AsyncStream[ChatCompletionChunk]: ...
 
     async def create(
         self,
@@ -605,7 +606,7 @@ async def create(
         compression: CompressionConfig | None = None,
         stream_options: StreamOptions | None = None,
         extra_body: dict[str, Any] | None = None,
-    ) -> ChatCompletion | AsyncStream:
+    ) -> ChatCompletion | AsyncStream[ChatCompletionChunk]:
         """Create a chat completion asynchronously.
 
         See Completions.create() for parameter documentation.