diff --git a/README.md b/README.md index e24c20c..193efad 100644 --- a/README.md +++ b/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` / diff --git a/apertis/_base_client.py b/apertis/_base_client.py index 7795f6d..d28b1aa 100644 --- a/apertis/_base_client.py +++ b/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 diff --git a/apertis/_streaming.py b/apertis/_streaming.py index 946d267..7bc4eb9 100644 --- a/apertis/_streaming.py +++ b/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: diff --git a/apertis/resources/chat/completions.py b/apertis/resources/chat/completions.py index 721ce14..3c471e5 100644 --- a/apertis/resources/chat/completions.py +++ b/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. diff --git a/apertis/resources/messages.py b/apertis/resources/messages.py index d70eb75..3169e23 100644 --- a/apertis/resources/messages.py +++ b/apertis/resources/messages.py @@ -2,27 +2,126 @@ 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.messages import Message, MessageParam, ToolDefinition +from apertis.types.messages import Message, MessageParam, MessageStreamEvent, ToolDefinition if TYPE_CHECKING: + import httpx + from apertis._base_client import AsyncClient, SyncClient +# Anthropic error 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, + "request_too_large": 413, + "rate_limit_error": 429, + "api_error": 500, + "overloaded_error": 529, +} + + +def _parse_event(data: Dict[str, Any], response: "httpx.Response") -> Optional[MessageStreamEvent]: + """Map one SSE payload to an event: skip pings, raise on error events.""" + event_type = data.get("type") + if event_type == "ping": + return None + if event_type == "error": + error = data.get("error") + error = error if isinstance(error, dict) else {"message": str(error)} + raise _make_api_error( + error.get("message") or str(data), + # The stream itself is HTTP 200; the error type says what failed. + status_code=_ERROR_STATUS.get(str(error.get("type")), 500), + response=response, + body=data, + ) + return MessageStreamEvent.model_validate(data) + + class Messages: """Synchronous messages resource (Anthropic native format).""" def __init__(self, client: SyncClient) -> None: self._client = client + @overload + def create( + self, + *, + model: str, + messages: Sequence[MessageParam], + max_tokens: int, + stream: Literal[True], + system: Optional[str] = None, + temperature: Optional[float] = None, + top_p: Optional[float] = None, + top_k: Optional[int] = None, + stop_sequences: Optional[List[str]] = None, + tools: Optional[List[ToolDefinition]] = None, + tool_choice: Optional[Dict[str, Any]] = None, + metadata: Optional[Dict[str, str]] = None, + thinking: Optional[Dict[str, Any]] = None, + compression: Optional[CompressionConfig] = None, + extra_body: Optional[Dict[str, Any]] = None, + ) -> Stream[MessageStreamEvent]: ... + + @overload + def create( + self, + *, + model: str, + messages: Sequence[MessageParam], + max_tokens: int, + stream: Literal[False] = False, + system: Optional[str] = None, + temperature: Optional[float] = None, + top_p: Optional[float] = None, + top_k: Optional[int] = None, + stop_sequences: Optional[List[str]] = None, + tools: Optional[List[ToolDefinition]] = None, + tool_choice: Optional[Dict[str, Any]] = None, + metadata: Optional[Dict[str, str]] = None, + thinking: Optional[Dict[str, Any]] = None, + compression: Optional[CompressionConfig] = None, + extra_body: Optional[Dict[str, Any]] = None, + ) -> Message: ... + + @overload + def create( + self, + *, + model: str, + messages: Sequence[MessageParam], + max_tokens: int, + stream: bool = False, + system: Optional[str] = None, + temperature: Optional[float] = None, + top_p: Optional[float] = None, + top_k: Optional[int] = None, + stop_sequences: Optional[List[str]] = None, + tools: Optional[List[ToolDefinition]] = None, + tool_choice: Optional[Dict[str, Any]] = None, + metadata: Optional[Dict[str, str]] = None, + thinking: Optional[Dict[str, Any]] = None, + compression: Optional[CompressionConfig] = None, + extra_body: Optional[Dict[str, Any]] = None, + ) -> Union[Message, Stream[MessageStreamEvent]]: ... + def create( self, *, model: str, messages: Sequence[MessageParam], max_tokens: int, + stream: bool = False, system: Optional[str] = None, temperature: Optional[float] = None, top_p: Optional[float] = None, @@ -31,14 +130,17 @@ def create( tools: Optional[List[ToolDefinition]] = None, tool_choice: Optional[Dict[str, Any]] = None, metadata: Optional[Dict[str, str]] = None, + thinking: Optional[Dict[str, Any]] = None, compression: Optional[CompressionConfig] = None, - ) -> Message: + extra_body: Optional[Dict[str, Any]] = None, + ) -> Union[Message, Stream[MessageStreamEvent]]: """Create a message using Anthropic's native format. Args: model: ID of the model to use (e.g., "claude-sonnet-4-6"). messages: List of messages in the conversation. max_tokens: Maximum tokens to generate. + stream: If True, returns an iterator of MessageStreamEvent. system: System prompt. temperature: Sampling temperature (0-1). top_p: Nucleus sampling parameter. @@ -47,14 +149,18 @@ def create( tools: List of tools the model can use. tool_choice: Controls tool selection. metadata: Metadata to attach to the message. + thinking: Extended thinking, e.g. {"type": "enabled", "budget_tokens": 1024}. + compression: Context compression configuration. + extra_body: Extra fields merged into the top level of the request body. Returns: - Message object with generated content. + Message, or a Stream of MessageStreamEvent when stream is True. """ body = _build_request_body( model=model, messages=messages, max_tokens=max_tokens, + stream=stream, system=system, temperature=temperature, top_p=top_p, @@ -63,9 +169,14 @@ def create( tools=tools, tool_choice=tool_choice, metadata=metadata, + thinking=thinking, compression=compression, + extra_body=extra_body, ) + if stream: + return Stream(self._client.stream("POST", "/messages", json=body), _parse_event) + response = self._client.request("POST", "/messages", json=body) return Message.model_validate(response.json()) @@ -76,12 +187,76 @@ class AsyncMessages: def __init__(self, client: AsyncClient) -> None: self._client = client + @overload + async def create( + self, + *, + model: str, + messages: Sequence[MessageParam], + max_tokens: int, + stream: Literal[True], + system: Optional[str] = None, + temperature: Optional[float] = None, + top_p: Optional[float] = None, + top_k: Optional[int] = None, + stop_sequences: Optional[List[str]] = None, + tools: Optional[List[ToolDefinition]] = None, + tool_choice: Optional[Dict[str, Any]] = None, + metadata: Optional[Dict[str, str]] = None, + thinking: Optional[Dict[str, Any]] = None, + compression: Optional[CompressionConfig] = None, + extra_body: Optional[Dict[str, Any]] = None, + ) -> AsyncStream[MessageStreamEvent]: ... + + @overload + async def create( + self, + *, + model: str, + messages: Sequence[MessageParam], + max_tokens: int, + stream: Literal[False] = False, + system: Optional[str] = None, + temperature: Optional[float] = None, + top_p: Optional[float] = None, + top_k: Optional[int] = None, + stop_sequences: Optional[List[str]] = None, + tools: Optional[List[ToolDefinition]] = None, + tool_choice: Optional[Dict[str, Any]] = None, + metadata: Optional[Dict[str, str]] = None, + thinking: Optional[Dict[str, Any]] = None, + compression: Optional[CompressionConfig] = None, + extra_body: Optional[Dict[str, Any]] = None, + ) -> Message: ... + + @overload + async def create( + self, + *, + model: str, + messages: Sequence[MessageParam], + max_tokens: int, + stream: bool = False, + system: Optional[str] = None, + temperature: Optional[float] = None, + top_p: Optional[float] = None, + top_k: Optional[int] = None, + stop_sequences: Optional[List[str]] = None, + tools: Optional[List[ToolDefinition]] = None, + tool_choice: Optional[Dict[str, Any]] = None, + metadata: Optional[Dict[str, str]] = None, + thinking: Optional[Dict[str, Any]] = None, + compression: Optional[CompressionConfig] = None, + extra_body: Optional[Dict[str, Any]] = None, + ) -> Union[Message, AsyncStream[MessageStreamEvent]]: ... + async def create( self, *, model: str, messages: Sequence[MessageParam], max_tokens: int, + stream: bool = False, system: Optional[str] = None, temperature: Optional[float] = None, top_p: Optional[float] = None, @@ -90,8 +265,10 @@ async def create( tools: Optional[List[ToolDefinition]] = None, tool_choice: Optional[Dict[str, Any]] = None, metadata: Optional[Dict[str, str]] = None, + thinking: Optional[Dict[str, Any]] = None, compression: Optional[CompressionConfig] = None, - ) -> Message: + extra_body: Optional[Dict[str, Any]] = None, + ) -> Union[Message, AsyncStream[MessageStreamEvent]]: """Create a message asynchronously. See Messages.create() for parameter documentation. @@ -100,6 +277,7 @@ async def create( model=model, messages=messages, max_tokens=max_tokens, + stream=stream, system=system, temperature=temperature, top_p=top_p, @@ -108,9 +286,15 @@ async def create( tools=tools, tool_choice=tool_choice, metadata=metadata, + thinking=thinking, compression=compression, + extra_body=extra_body, ) + if stream: + response = await self._client.stream("POST", "/messages", json=body) + return AsyncStream(response, _parse_event) + response = await self._client.request("POST", "/messages", json=body) return Message.model_validate(response.json()) @@ -120,6 +304,7 @@ def _build_request_body( model: str, messages: Sequence[MessageParam], max_tokens: int, + stream: bool, system: Optional[str], temperature: Optional[float], top_p: Optional[float], @@ -128,7 +313,9 @@ def _build_request_body( tools: Optional[List[ToolDefinition]], tool_choice: Optional[Dict[str, Any]], metadata: Optional[Dict[str, str]], + thinking: Optional[Dict[str, Any]], compression: Optional[CompressionConfig], + extra_body: Optional[Dict[str, Any]], ) -> Dict[str, Any]: """Build request body for messages.""" body: Dict[str, Any] = { @@ -137,6 +324,8 @@ def _build_request_body( "max_tokens": max_tokens, } + if stream: + body["stream"] = True if system is not None: body["system"] = system if temperature is not None: @@ -153,7 +342,11 @@ def _build_request_body( body["tool_choice"] = tool_choice if metadata is not None: body["metadata"] = metadata + if thinking is not None: + body["thinking"] = thinking if compression is not None: body["compression"] = compression + if extra_body is not None: + body.update(extra_body) return body diff --git a/apertis/types/__init__.py b/apertis/types/__init__.py index 7c5906f..ca7b915 100644 --- a/apertis/types/__init__.py +++ b/apertis/types/__init__.py @@ -71,6 +71,11 @@ TextBlock, ToolUseBlock, ToolDefinition, + ThinkingBlock, + RedactedThinkingBlock, + MessageStreamEvent, + MessageStreamDelta, + MessageDeltaUsage, ) from apertis.types.rerank import ( RerankResponse, @@ -146,6 +151,11 @@ "TextBlock", "ToolUseBlock", "ToolDefinition", + "ThinkingBlock", + "RedactedThinkingBlock", + "MessageStreamEvent", + "MessageStreamDelta", + "MessageDeltaUsage", # Rerank API "RerankResponse", "RerankResult", diff --git a/apertis/types/messages.py b/apertis/types/messages.py index c61bd02..d5850a8 100644 --- a/apertis/types/messages.py +++ b/apertis/types/messages.py @@ -29,7 +29,34 @@ class ToolUseBlock(BaseModel): input: Dict[str, Any] -ContentBlock = Union[TextBlock, ToolUseBlock] +class ThinkingBlock(BaseModel): + """Extended thinking content block.""" + + type: Literal["thinking"] + thinking: str + # Empty in a stream's content_block_start; the signature arrives as a signature_delta. + signature: str = "" + + +class RedactedThinkingBlock(BaseModel): + """Thinking content the provider returns encrypted.""" + + type: Literal["redacted_thinking"] + data: str + + +ContentBlock = Union[TextBlock, ToolUseBlock, ThinkingBlock, RedactedThinkingBlock] + + +StopReason = Literal[ + "end_turn", + "max_tokens", + "stop_sequence", + "tool_use", + "pause_turn", + "refusal", + "model_context_window_exceeded", +] class MessageUsage(BaseModel): @@ -47,17 +74,33 @@ class Message(BaseModel): role: Literal["assistant"] content: List[ContentBlock] model: str - stop_reason: Optional[Literal["end_turn", "max_tokens", "stop_sequence", "tool_use"]] = None + stop_reason: Optional[StopReason] = None stop_sequence: Optional[str] = None usage: MessageUsage class MessageStreamDelta(BaseModel): - """Delta in a streaming message.""" + """Delta in a stream event. + + content_block_delta deltas carry `type` (text_delta, input_json_delta, + thinking_delta, signature_delta) and the matching field; message_delta + deltas carry `stop_reason` and `stop_sequence` and have no `type`. + """ - type: Literal["text_delta", "input_json_delta"] + type: Optional[str] = None text: Optional[str] = None partial_json: Optional[str] = None + thinking: Optional[str] = None + signature: Optional[str] = None + stop_reason: Optional[StopReason] = None + stop_sequence: Optional[str] = None + + +class MessageDeltaUsage(BaseModel): + """Cumulative usage sent with a message_delta event.""" + + output_tokens: int + input_tokens: Optional[int] = None class MessageStreamEvent(BaseModel): @@ -75,6 +118,7 @@ class MessageStreamEvent(BaseModel): index: Optional[int] = None content_block: Optional[ContentBlock] = None delta: Optional[MessageStreamDelta] = None + usage: Optional[MessageDeltaUsage] = None # ============================================================================= diff --git a/openspec/changes/add-messages-stream-and-thinking/.openspec.yaml b/openspec/changes/add-messages-stream-and-thinking/.openspec.yaml new file mode 100644 index 0000000..a26662f --- /dev/null +++ b/openspec/changes/add-messages-stream-and-thinking/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-10-07 diff --git a/openspec/changes/add-messages-stream-and-thinking/proposal.md b/openspec/changes/add-messages-stream-and-thinking/proposal.md new file mode 100644 index 0000000..b459ba8 --- /dev/null +++ b/openspec/changes/add-messages-stream-and-thinking/proposal.md @@ -0,0 +1,17 @@ +## Why + +`messages.create` cannot stream or request extended thinking, although the gateway's +`/v1/messages` supports both (issue #3). Callers fall back to the Anthropic SDK. + +## What Changes + +- `messages.create` (sync and async) gains `stream`, `thinking` and `extra_body`. +- `stream=True` returns an iterator of Anthropic message stream events. +- `Message.content` accepts `thinking` and `redacted_thinking` blocks. +- The SSE stream classes take a parse function, so chat and messages share one reader. + +## Impact + +- `apertis/_streaming.py`, `apertis/resources/messages.py`, `apertis/types/messages.py`, + tests and README. Non-streaming calls without the new arguments send the same body as + before. diff --git a/openspec/changes/add-messages-stream-and-thinking/specs/messages-stream-and-thinking/spec.md b/openspec/changes/add-messages-stream-and-thinking/specs/messages-stream-and-thinking/spec.md new file mode 100644 index 0000000..b09b5a9 --- /dev/null +++ b/openspec/changes/add-messages-stream-and-thinking/specs/messages-stream-and-thinking/spec.md @@ -0,0 +1,41 @@ +## ADDED Requirements + +### Requirement: Messages streaming +`messages.create` SHALL accept `stream=True` on sync and async clients, send +`"stream": true`, and return an iterator of typed Anthropic stream events +(`message_start`, `content_block_start`, `content_block_delta`, `content_block_stop`, +`message_delta`, `message_stop`) in arrival order. The iterator SHALL end at the end of +the SSE body without a `[DONE]` sentinel. + +#### Scenario: Text stream +- **WHEN** the API streams a text answer +- **THEN** the caller receives every event in order, `text_delta` deltas carry the text, + and the `message_delta` event exposes `stop_reason` and output `usage` + +#### Scenario: Keep-alive and error events +- **WHEN** the stream contains a `ping` event +- **THEN** the SDK skips it +- **WHEN** the stream contains an `error` event +- **THEN** the SDK raises `APIError` carrying the error body + +#### Scenario: HTTP error on a streamed request +- **WHEN** a streamed chat or messages request returns an HTTP error status +- **THEN** the SDK raises the matching `APIError` subclass with the error message + +### Requirement: Extended thinking +`messages.create` SHALL accept a `thinking` object and send it unchanged. Responses and +stream events SHALL expose `thinking` and `redacted_thinking` content blocks and +`thinking_delta` / `signature_delta` deltas. + +#### Scenario: Thinking response +- **WHEN** a non-streaming response contains a `thinking` block before a `text` block +- **THEN** `Message.content` holds a typed thinking block with its text and signature, + followed by the text block + +### Requirement: Extra body fields +`messages.create` SHALL accept `extra_body` and merge its keys into the top level of the +request body, so fields the SDK does not name yet reach the API. + +#### Scenario: Unnamed field +- **WHEN** a caller passes `extra_body={"service_tier": "auto"}` +- **THEN** the request body contains `"service_tier": "auto"` at the top level diff --git a/openspec/changes/add-messages-stream-and-thinking/tasks.md b/openspec/changes/add-messages-stream-and-thinking/tasks.md new file mode 100644 index 0000000..6df63cc --- /dev/null +++ b/openspec/changes/add-messages-stream-and-thinking/tasks.md @@ -0,0 +1,7 @@ +## 1. Messages stream and thinking + +- [x] 1.1 Tests first: text stream, thinking stream, ping/error events, thinking response, extra_body (sync and async) +- [x] 1.2 Parse-function hook in `Stream` / `AsyncStream` +- [x] 1.3 Stream event, delta and thinking block types +- [x] 1.4 `stream`, `thinking`, `extra_body` on `messages.create` with overloads +- [x] 1.5 README and changelog diff --git a/tests/test_messages.py b/tests/test_messages.py index d6f8368..f95a8e3 100644 --- a/tests/test_messages.py +++ b/tests/test_messages.py @@ -2,11 +2,14 @@ from __future__ import annotations +import json + import httpx import pytest import respx -from apertis import Apertis +from apertis import Apertis, APIError, AsyncApertis, InternalServerError, RateLimitError +from apertis.types.messages import RedactedThinkingBlock, TextBlock, ThinkingBlock class TestMessagesCreate: @@ -187,3 +190,205 @@ def test_create_message_with_image(self, client: Apertis) -> None: ) assert "cat" in message.content[0].text.lower() + + +def _sse(*events: dict) -> bytes: + """Encode events the way the Anthropic-native endpoint streams them.""" + return b"".join( + f"event: {e['type']}\ndata: {json.dumps(e)}\n\n".encode() for e in events + ) + + +# Event shapes follow the Anthropic Messages streaming format (anthropic-python 1.11 types). +_THINKING_STREAM = _sse( + { + "type": "message_start", + "message": { + "id": "msg_1", + "type": "message", + "role": "assistant", + "content": [], + "model": "claude-sonnet-4-6", + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 25, "output_tokens": 1}, + }, + }, + {"type": "ping"}, + {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "2+2 is 4."}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": "sig=="}}, + {"type": "content_block_stop", "index": 0}, + {"type": "content_block_start", "index": 1, "content_block": {"type": "text", "text": ""}}, + {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "4"}}, + {"type": "content_block_stop", "index": 1}, + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 15}, + }, + {"type": "message_stop"}, +) + + +class TestMessagesStream: + """stream=True yields typed Anthropic stream events.""" + + @respx.mock + def test_sync_stream_with_thinking(self, client: Apertis) -> None: + route = respx.post("https://api.apertis.ai/v1/messages").mock( + return_value=httpx.Response(200, content=_THINKING_STREAM) + ) + + stream = client.messages.create( + model="claude-sonnet-4-6", + messages=[{"role": "user", "content": "2+2?"}], + max_tokens=2048, + stream=True, + thinking={"type": "enabled", "budget_tokens": 1024}, + ) + events = list(stream) + + body = json.loads(route.calls[0].request.content) + assert body["stream"] is True + assert body["thinking"] == {"type": "enabled", "budget_tokens": 1024} + assert [e.type for e in events] == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_delta", + "content_block_stop", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] # ping skipped + assert isinstance(events[1].content_block, ThinkingBlock) + assert events[2].delta.thinking == "2+2 is 4." + assert events[3].delta.signature == "sig==" + assert events[6].delta.text == "4" + assert events[8].delta.stop_reason == "end_turn" + assert events[8].usage.output_tokens == 15 + + @respx.mock + async def test_async_stream(self, async_client: AsyncApertis) -> None: + respx.post("https://api.apertis.ai/v1/messages").mock( + return_value=httpx.Response(200, content=_THINKING_STREAM) + ) + + stream = await async_client.messages.create( + model="claude-sonnet-4-6", + messages=[{"role": "user", "content": "2+2?"}], + max_tokens=2048, + stream=True, + ) + text = "".join( + [e.delta.text async for e in stream if e.type == "content_block_delta" and e.delta.text] + ) + + assert text == "4" + + @respx.mock + def test_error_event_raises(self, client: Apertis) -> None: + respx.post("https://api.apertis.ai/v1/messages").mock( + return_value=httpx.Response( + 200, + content=_sse( + {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}} + ), + ) + ) + + stream = client.messages.create( + model="claude-sonnet-4-6", + messages=[{"role": "user", "content": "Hi"}], + max_tokens=16, + stream=True, + ) + with pytest.raises(APIError, match="Overloaded") as exc: + list(stream) + assert exc.value.body == {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}} + + +class TestMessagesThinkingAndExtraBody: + @respx.mock + def test_thinking_blocks_and_extra_body(self, client: Apertis) -> None: + route = respx.post("https://api.apertis.ai/v1/messages").mock( + return_value=httpx.Response( + 200, + json={ + "id": "msg_1", + "type": "message", + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "Add them.", "signature": "sig=="}, + {"type": "redacted_thinking", "data": "opaque"}, + {"type": "text", "text": "4"}, + ], + "model": "claude-sonnet-4-6", + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 20}, + }, + ) + ) + + message = client.messages.create( + model="claude-sonnet-4-6", + messages=[{"role": "user", "content": "2+2?"}], + max_tokens=2048, + thinking={"type": "enabled", "budget_tokens": 1024}, + extra_body={"service_tier": "auto"}, + ) + + body = json.loads(route.calls[0].request.content) + assert body["service_tier"] == "auto" + assert "stream" not in body + thinking, redacted, text = message.content + assert isinstance(thinking, ThinkingBlock) + assert (thinking.thinking, thinking.signature) == ("Add them.", "sig==") + assert isinstance(redacted, RedactedThinkingBlock) + assert isinstance(text, TextBlock) + + +class TestMessagesStreamErrors: + @respx.mock + def test_error_event_maps_type_and_closes_response(self, client: Apertis) -> None: + respx.post("https://api.apertis.ai/v1/messages").mock( + return_value=httpx.Response( + 200, + content=_sse( + {"type": "error", "error": {"type": "rate_limit_error", "message": "Slow down"}} + ), + ) + ) + + stream = client.messages.create( + model="claude-sonnet-4-6", + messages=[{"role": "user", "content": "Hi"}], + max_tokens=16, + stream=True, + ) + with pytest.raises(RateLimitError, match="Slow down"): + next(stream) + assert stream._response.is_closed + + @respx.mock + async def test_async_overloaded_is_server_error(self, async_client: AsyncApertis) -> None: + respx.post("https://api.apertis.ai/v1/messages").mock( + return_value=httpx.Response( + 200, + content=_sse({"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}), + ) + ) + + stream = await async_client.messages.create( + model="claude-sonnet-4-6", + messages=[{"role": "user", "content": "Hi"}], + max_tokens=16, + stream=True, + ) + with pytest.raises(InternalServerError, match="Overloaded"): + await stream.__anext__() + assert stream._response.is_closed diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 2daeef3..56182a5 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -3,9 +3,10 @@ from __future__ import annotations import httpx +import pytest import respx -from apertis import Apertis, AsyncApertis +from apertis import Apertis, APIError, AsyncApertis, NotFoundError WEB_SEARCH_STREAM = b"\n".join( @@ -60,3 +61,31 @@ async def test_async_web_search_stream_status_is_not_dropped( "Web searching...\n\n", "Result", ] + + +@respx.mock +def test_sync_stream_http_error_raises_api_error(client: Apertis) -> None: + """An HTTP error on a streamed request surfaces as APIError, not httpx.ResponseNotRead.""" + respx.post("https://api.apertis.ai/v1/chat/completions").mock( + return_value=httpx.Response(400, json={"error": {"message": "bad model"}}) + ) + + with pytest.raises(APIError, match="bad model"): + client.chat.completions.create( + model="nope", messages=[{"role": "user", "content": "Hi"}], stream=True + ) + + +@respx.mock +async def test_async_stream_http_error_raises_api_error(async_client: AsyncApertis) -> None: + respx.post("https://api.apertis.ai/v1/messages").mock( + return_value=httpx.Response(404, json={"error": {"message": "no such model"}}) + ) + + with pytest.raises(NotFoundError, match="no such model"): + await async_client.messages.create( + model="nope", + messages=[{"role": "user", "content": "Hi"}], + max_tokens=16, + stream=True, + )