Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
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
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

package org.springframework.ai.anthropic;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Base64;
import java.util.HashMap;
Expand All @@ -29,6 +30,7 @@
import com.anthropic.client.AnthropicClient;
import com.anthropic.client.AnthropicClientAsync;
import com.anthropic.core.JsonValue;
import com.anthropic.core.RequestOptions;
import com.anthropic.core.http.HttpResponseFor;
import com.anthropic.core.http.StreamResponse;
import com.anthropic.models.messages.Base64ImageSource;
Expand Down Expand Up @@ -330,7 +332,9 @@ public Flux<ChatResponse> internalStream(Prompt prompt, @Nullable ChatResponse p
// stream start) can be captured. The SDK exposes this as a blocking
// StreamResponse, so events are pulled on a boundedElastic worker.
Flux<ChatResponse> chatResponseFlux = Mono
.fromFuture(() -> this.anthropicClientAsync.messages().withRawResponse().createStreaming(request))
.fromFuture(() -> this.anthropicClientAsync.messages()
.withRawResponse()
.createStreaming(request, requestOptionsFor(prompt)))
.flatMapMany(rawResponse -> {
streamingState.setRateLimit(AnthropicRateLimit.from(rawResponse.headers()));
StreamResponse<RawMessageStreamEvent> streamResponse = rawResponse.parse();
Expand Down Expand Up @@ -552,7 +556,7 @@ public ChatResponse internalCall(Prompt prompt, @Nullable ChatResponse previousC

HttpResponseFor<Message> rawResponse = this.anthropicClient.messages()
.withRawResponse()
.create(request);
.create(request, requestOptionsFor(prompt));
Message message = rawResponse.parse();
RateLimit rateLimit = AnthropicRateLimit.from(rawResponse.headers());

Expand Down Expand Up @@ -586,6 +590,40 @@ public ChatResponse internalCall(Prompt prompt, @Nullable ChatResponse previousC
return response;
}

/**
* Resolves the effective {@link AnthropicChatOptions} for a prompt — the merged
* default/per-request options set on {@link Prompt#getOptions()}, or an empty
* {@link AnthropicChatOptions} if the prompt carries a different {@link ChatOptions}
* implementation.
* @param prompt the prompt with message history and options
* @return the effective Anthropic chat options
*/
private static AnthropicChatOptions resolveAnthropicOptions(Prompt prompt) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

In-line these comments instead of javadoc. This level of comment details are not required for private methods.

ChatOptions options = prompt.getOptions();
return options instanceof AnthropicChatOptions anthropicOptions ? anthropicOptions
: AnthropicChatOptions.builder().build();
}

/**
* Resolves the per-call {@link RequestOptions} (currently just {@code timeout} — the
* SDK's {@code RequestOptions} doesn't expose {@code maxRetries}/{@code proxy}/custom
* headers, so those remain construction-time-only settings) from the effective
* {@link AnthropicChatOptions} on the given prompt. Without this, a {@code timeout}
* set via {@link AnthropicChatOptions#getTimeout()} — whether on {@code ChatClient}
* default options or a per-request override — would silently have no effect: the
* Anthropic Java SDK only honors the {@code timeout} supplied on this
* {@link RequestOptions} argument; a call made without one falls back to whatever
* {@link com.anthropic.core.Timeout} the client was built with once, at construction
* time (see {@link AnthropicSetup}).
* @param prompt the prompt with message history and options
* @return request options carrying the resolved timeout, or
* {@link RequestOptions#none()} if none is set
*/
private static RequestOptions requestOptionsFor(Prompt prompt) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Too much javadoc on a private method. Can you cut it down to what is absolutely necessary? We usually prefer this much detail only on the public API's.

Duration timeout = resolveAnthropicOptions(prompt).getTimeout();
return timeout != null ? RequestOptions.builder().timeout(timeout).build() : RequestOptions.none();
}

/**
* Creates a {@link MessageCreateParams} request from a Spring AI {@link Prompt}. Maps
* message types to Anthropic format: TOOL messages become user messages with
Expand All @@ -599,9 +637,7 @@ MessageCreateParams createRequest(Prompt prompt, boolean stream) {

MessageCreateParams.Builder builder = MessageCreateParams.builder();

ChatOptions options = prompt.getOptions();
AnthropicChatOptions requestOptions = options instanceof AnthropicChatOptions anthropicOptions
? anthropicOptions : AnthropicChatOptions.builder().build();
AnthropicChatOptions requestOptions = resolveAnthropicOptions(prompt);

// Set required fields
builder.model(requestOptions.getModel()).maxTokens(requestOptions.getMaxTokens());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,11 @@ public class AnthropicChatOptions implements ToolCallingChatOptions, StructuredO
private final String model;

/**
* Request timeout for the Anthropic client. Defaults to 60 seconds if not specified.
* Request timeout for the Anthropic client, applied per call — set as a
* {@code ChatClient} default option or a per-request override, this bounds only that
* call, not every call made by the underlying {@link AnthropicChatModel}. Defaults to
* 60 seconds if not specified anywhere (including at {@link AnthropicChatModel}
* construction time).
*/
private final @Nullable Duration timeout;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import com.anthropic.client.AnthropicClient;
import com.anthropic.client.AnthropicClientAsync;
import com.anthropic.core.JsonValue;
import com.anthropic.core.RequestOptions;
import com.anthropic.core.http.Headers;
import com.anthropic.core.http.HttpResponseFor;
import com.anthropic.core.http.StreamResponse;
Expand Down Expand Up @@ -125,14 +126,15 @@ class AnthropicChatModelTests {
void setUp() {
given(this.anthropicClient.messages()).willReturn(this.messageService);
given(this.messageService.withRawResponse()).willReturn(this.messageServiceWithRawResponse);
given(this.messageServiceWithRawResponse.create(any(MessageCreateParams.class))).willAnswer(invocation -> {
MessageCreateParams params = invocation.getArgument(0);
Message message = this.messageService.create(params);
HttpResponseFor<Message> rawResponse = mock(HttpResponseFor.class);
given(rawResponse.parse()).willReturn(message);
given(rawResponse.headers()).willReturn(Headers.builder().build());
return rawResponse;
});
given(this.messageServiceWithRawResponse.create(any(MessageCreateParams.class), any(RequestOptions.class)))
.willAnswer(invocation -> {
MessageCreateParams params = invocation.getArgument(0);
Message message = this.messageService.create(params);
HttpResponseFor<Message> rawResponse = mock(HttpResponseFor.class);
given(rawResponse.parse()).willReturn(message);
given(rawResponse.headers()).willReturn(Headers.builder().build());
return rawResponse;
});

this.chatModel = AnthropicChatModel.builder()
.anthropicClient(this.anthropicClient)
Expand Down Expand Up @@ -361,7 +363,8 @@ void streamingThinkingBlockIsReplayedBeforeToolUseBlock() {

given(this.anthropicClientAsync.messages()).willReturn(this.messageServiceAsync);
given(this.messageServiceAsync.withRawResponse()).willReturn(this.messageServiceAsyncWithRawResponse);
given(this.messageServiceAsyncWithRawResponse.createStreaming(any(MessageCreateParams.class)))
given(this.messageServiceAsyncWithRawResponse.createStreaming(any(MessageCreateParams.class),
any(RequestOptions.class)))
.willReturn(CompletableFuture.completedFuture(rawResponse));
Message finalResponse = createMockMessage("Done.", StopReason.END_TURN);
given(this.messageService.create(any(MessageCreateParams.class))).willReturn(finalResponse);
Expand Down Expand Up @@ -849,14 +852,15 @@ void rateLimitHeadersArePopulatedInMetadata() {
.put("anthropic-ratelimit-tokens-reset", resetAt.toString())
.build();

given(this.messageServiceWithRawResponse.create(any(MessageCreateParams.class))).willAnswer(invocation -> {
MessageCreateParams params = invocation.getArgument(0);
Message message = this.messageService.create(params);
HttpResponseFor<Message> rawResponse = mock(HttpResponseFor.class);
given(rawResponse.parse()).willReturn(message);
given(rawResponse.headers()).willReturn(rateLimitHeaders);
return rawResponse;
});
given(this.messageServiceWithRawResponse.create(any(MessageCreateParams.class), any(RequestOptions.class)))
.willAnswer(invocation -> {
MessageCreateParams params = invocation.getArgument(0);
Message message = this.messageService.create(params);
HttpResponseFor<Message> rawResponse = mock(HttpResponseFor.class);
given(rawResponse.parse()).willReturn(message);
given(rawResponse.headers()).willReturn(rateLimitHeaders);
return rawResponse;
});

ChatResponse response = this.chatModel.call(new Prompt("test"));

Expand All @@ -883,7 +887,8 @@ void streamingClosesStreamResponse() {

given(this.anthropicClientAsync.messages()).willReturn(this.messageServiceAsync);
given(this.messageServiceAsync.withRawResponse()).willReturn(this.messageServiceAsyncWithRawResponse);
given(this.messageServiceAsyncWithRawResponse.createStreaming(any(MessageCreateParams.class)))
given(this.messageServiceAsyncWithRawResponse.createStreaming(any(MessageCreateParams.class),
any(RequestOptions.class)))
.willReturn(CompletableFuture.completedFuture(rawResponse));

this.chatModel.stream(new Prompt("test")).collectList().block();
Expand Down Expand Up @@ -936,7 +941,8 @@ void streamingAttachesRateLimitHeadersToResponse() {

given(this.anthropicClientAsync.messages()).willReturn(this.messageServiceAsync);
given(this.messageServiceAsync.withRawResponse()).willReturn(this.messageServiceAsyncWithRawResponse);
given(this.messageServiceAsyncWithRawResponse.createStreaming(any(MessageCreateParams.class)))
given(this.messageServiceAsyncWithRawResponse.createStreaming(any(MessageCreateParams.class),
any(RequestOptions.class)))
.willReturn(CompletableFuture.completedFuture(rawResponse));

List<ChatResponse> responses = this.chatModel.stream(new Prompt("test")).collectList().block();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,209 @@
/*
* Copyright 2023-present the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.springframework.ai.anthropic;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

import com.anthropic.client.AnthropicClientAsync;
import com.anthropic.core.RequestOptions;
import com.anthropic.core.http.Headers;
import com.anthropic.core.http.HttpResponseFor;
import com.anthropic.core.http.StreamResponse;
import com.anthropic.errors.AnthropicIoException;
import com.anthropic.models.messages.MessageCreateParams;
import com.anthropic.models.messages.RawMessageStreamEvent;
import com.anthropic.services.async.MessageServiceAsync;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;

import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.messages.UserMessage;
import org.springframework.ai.chat.model.MessageAggregator;
import org.springframework.ai.chat.prompt.Prompt;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;

/**
* Verifies that {@link AnthropicChatOptions#getTimeout()} actually bounds the HTTP call
* whether it's set at {@link AnthropicChatModel} construction time, as a
* {@link ChatClient} default option, or as a per-request {@code ChatClient} override —
* not just at construction time.
*
* @author seeun0210

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Use your full real name.

*/
class AnthropicChatModelTimeoutTests {

private static final String MESSAGES_RESPONSE = """
{"id":"msg-test","type":"message","role":"assistant","content":[{"type":"text","text":"ok"}],"model":"claude-haiku-4-5-20251001","stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}
""";

@Test
void constructionTimeTimeoutBoundsTheCall() throws Exception {
try (MockWebServer server = new MockWebServer()) {
server.enqueue(new MockResponse().setHeadersDelay(3, TimeUnit.SECONDS)
.setResponseCode(200)
.setHeader("Content-Type", "application/json")
.setBody(MESSAGES_RESPONSE));
server.start();

AnthropicChatModel chatModel = AnthropicChatModel.builder()
.options(AnthropicChatOptions.builder()
.baseUrl(server.url("/").toString())
.apiKey("test-key")
.model("claude-haiku-4-5-20251001")
.maxTokens(10)
.timeout(Duration.ofMillis(500))
.build())
.build();

// Root cause class varies across the SDK's automatic retries
// (SocketTimeoutException
// on a fresh connection vs. SocketException "Socket closed" when a retry
// races a
// connection the timeout already tore down) — assert on the stable wrapper
// type
// instead of pinning an exact root cause class.
assertThatThrownBy(() -> chatModel.call(new Prompt("hi")))
.as("a 500ms construction-time timeout must interrupt a call whose response is delayed 3s")
.isInstanceOf(AnthropicIoException.class);
}
}

@Test
void perRequestChatClientTimeoutOverridesConstructionTimeTimeout() throws Exception {
try (MockWebServer server = new MockWebServer()) {
// Construction-time timeout is intentionally too short (500ms) for the 2s
// delayed response below. A per-request ChatClient override of 5s should
// let the call succeed instead of failing at the construction-time bound.
server.enqueue(new MockResponse().setHeadersDelay(2, TimeUnit.SECONDS)
.setResponseCode(200)
.setHeader("Content-Type", "application/json")
.setBody(MESSAGES_RESPONSE));
server.start();

AnthropicChatModel chatModel = AnthropicChatModel.builder()
.options(AnthropicChatOptions.builder()
.baseUrl(server.url("/").toString())
.apiKey("test-key")
.model("claude-haiku-4-5-20251001")
.maxTokens(10)
.timeout(Duration.ofMillis(500))
.build())
.build();
ChatClient chatClient = ChatClient.builder(chatModel).build();

assertThatCode(() -> {
String content = chatClient.prompt()
.user("hi")
.options(AnthropicChatOptions.builder().timeout(Duration.ofSeconds(5)))
.call()
.content();
assertThat(content).isEqualTo("ok");
}).as("a 5s per-request ChatClient timeout must override the 500ms construction-time default")
.doesNotThrowAnyException();
}
}

@Test
void chatClientDefaultOptionsTimeoutOverridesConstructionTimeTimeout() throws Exception {
try (MockWebServer server = new MockWebServer()) {
server.enqueue(new MockResponse().setHeadersDelay(2, TimeUnit.SECONDS)
.setResponseCode(200)
.setHeader("Content-Type", "application/json")
.setBody(MESSAGES_RESPONSE));
server.start();

AnthropicChatModel chatModel = AnthropicChatModel.builder()
.options(AnthropicChatOptions.builder()
.baseUrl(server.url("/").toString())
.apiKey("test-key")
.model("claude-haiku-4-5-20251001")
.maxTokens(10)
.timeout(Duration.ofMillis(500))
.build())
.build();
ChatClient chatClient = ChatClient.builder(chatModel)
.defaultOptions(AnthropicChatOptions.builder().timeout(Duration.ofSeconds(5)))
.build();

assertThatCode(() -> chatClient.prompt().user("hi").call().content())
.as("a 5s ChatClient default-options timeout must override the 500ms construction-time default")
.doesNotThrowAnyException();
}
}

/**
* Mirrors {@link #perRequestChatClientTimeoutOverridesConstructionTimeTimeout()} for
* the streaming ({@code internalStream()}) code path. A real delayed-response
* MockWebServer setup isn't used here because it would additionally require
* fabricating a valid Anthropic SSE event stream; asserting the
* {@link RequestOptions} actually passed to {@code createStreaming(...)} is a more
* direct and no less faithful check that the fix applies to both call sites.
*/
@Test
@SuppressWarnings("unchecked")
void perRequestChatClientTimeoutIsForwardedOnTheStreamingPath() {
AnthropicClientAsync anthropicClientAsync = mock(AnthropicClientAsync.class);
MessageServiceAsync messageServiceAsync = mock(MessageServiceAsync.class);
MessageServiceAsync.WithRawResponse messageServiceAsyncWithRawResponse = mock(
MessageServiceAsync.WithRawResponse.class);

StreamResponse<RawMessageStreamEvent> streamResponse = mock(StreamResponse.class);
given(streamResponse.stream()).willReturn(Stream.empty());
HttpResponseFor<StreamResponse<RawMessageStreamEvent>> rawResponse = mock(HttpResponseFor.class);
given(rawResponse.parse()).willReturn(streamResponse);
given(rawResponse.headers()).willReturn(Headers.builder().build());

given(anthropicClientAsync.messages()).willReturn(messageServiceAsync);
given(messageServiceAsync.withRawResponse()).willReturn(messageServiceAsyncWithRawResponse);
given(messageServiceAsyncWithRawResponse.createStreaming(any(MessageCreateParams.class),
any(RequestOptions.class)))
.willReturn(CompletableFuture.completedFuture(rawResponse));

AnthropicChatModel chatModel = AnthropicChatModel.builder()
.anthropicClientAsync(anthropicClientAsync)
.options(AnthropicChatOptions.builder()
.model("claude-haiku-4-5-20251001")
.maxTokens(10)
.timeout(Duration.ofMillis(500))
.build())
.build();
ChatClient chatClient = ChatClient.builder(chatModel).build();

Prompt prompt = new Prompt(List.of(new UserMessage("hi")),
AnthropicChatOptions.builder().timeout(Duration.ofSeconds(5)).build());
new MessageAggregator().aggregate(chatClient.prompt(prompt).stream().chatResponse(), response -> {
}).collectList().block();

ArgumentCaptor<RequestOptions> captor = ArgumentCaptor.forClass(RequestOptions.class);
verify(messageServiceAsyncWithRawResponse).createStreaming(any(MessageCreateParams.class), captor.capture());
assertThat(captor.getValue().getTimeout().request()).isEqualTo(Duration.ofSeconds(5));
}

}