diff --git a/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatModel.java b/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatModel.java index 94841cfc02..f6e3932635 100644 --- a/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatModel.java +++ b/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatModel.java @@ -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; @@ -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; @@ -144,6 +146,7 @@ * @author Sebastien Deleuze * @author Ilayaperumal Gopinathan * @author Jewoo Shin + * @author Seeun Kim * @since 1.0.0 * @see AnthropicChatOptions * @see Anthropic Messages API @@ -330,7 +333,9 @@ public Flux 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 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 streamResponse = rawResponse.parse(); @@ -552,7 +557,7 @@ public ChatResponse internalCall(Prompt prompt, @Nullable ChatResponse previousC HttpResponseFor rawResponse = this.anthropicClient.messages() .withRawResponse() - .create(request); + .create(request, requestOptionsFor(prompt)); Message message = rawResponse.parse(); RateLimit rateLimit = AnthropicRateLimit.from(rawResponse.headers()); @@ -586,6 +591,21 @@ public ChatResponse internalCall(Prompt prompt, @Nullable ChatResponse previousC return response; } + private static AnthropicChatOptions resolveAnthropicOptions(Prompt prompt) { + // Use the prompt's merged AnthropicChatOptions, or empty options if it carries a + // different ChatOptions implementation. + ChatOptions options = prompt.getOptions(); + return options instanceof AnthropicChatOptions anthropicOptions ? anthropicOptions + : AnthropicChatOptions.builder().build(); + } + + private static RequestOptions requestOptionsFor(Prompt prompt) { + // Carry the resolved timeout as per-call RequestOptions; the SDK only honors a + // timeout supplied here, so otherwise AnthropicChatOptions#getTimeout() is ignored. + 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 @@ -599,9 +619,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()); diff --git a/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatOptions.java b/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatOptions.java index 2168e1c66f..2716d1d647 100644 --- a/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatOptions.java +++ b/models/spring-ai-anthropic/src/main/java/org/springframework/ai/anthropic/AnthropicChatOptions.java @@ -63,6 +63,7 @@ * @author Soby Chacko * @author Austin Dase * @author Sebastien Deleuze + * @author Seeun Kim * @since 1.0.0 * @see AnthropicChatModel * @see Anthropic Messages API @@ -99,7 +100,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; diff --git a/models/spring-ai-anthropic/src/test/java/org/springframework/ai/anthropic/AnthropicChatModelTests.java b/models/spring-ai-anthropic/src/test/java/org/springframework/ai/anthropic/AnthropicChatModelTests.java index 4c6058be00..30bc7e3e52 100644 --- a/models/spring-ai-anthropic/src/test/java/org/springframework/ai/anthropic/AnthropicChatModelTests.java +++ b/models/spring-ai-anthropic/src/test/java/org/springframework/ai/anthropic/AnthropicChatModelTests.java @@ -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; @@ -95,6 +96,7 @@ * @author Soby Chacko * @author Sebastien Deleuze * @author Jewoo Shin + * @author Seeun Kim */ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) @@ -125,14 +127,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 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 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) @@ -361,7 +364,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); @@ -849,14 +853,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 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 rawResponse = mock(HttpResponseFor.class); + given(rawResponse.parse()).willReturn(message); + given(rawResponse.headers()).willReturn(rateLimitHeaders); + return rawResponse; + }); ChatResponse response = this.chatModel.call(new Prompt("test")); @@ -883,7 +888,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(); @@ -936,7 +942,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 responses = this.chatModel.stream(new Prompt("test")).collectList().block(); diff --git a/models/spring-ai-anthropic/src/test/java/org/springframework/ai/anthropic/AnthropicChatModelTimeoutTests.java b/models/spring-ai-anthropic/src/test/java/org/springframework/ai/anthropic/AnthropicChatModelTimeoutTests.java new file mode 100644 index 0000000000..a597b65b41 --- /dev/null +++ b/models/spring-ai-anthropic/src/test/java/org/springframework/ai/anthropic/AnthropicChatModelTimeoutTests.java @@ -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 Seeun Kim + */ +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 streamResponse = mock(StreamResponse.class); + given(streamResponse.stream()).willReturn(Stream.empty()); + HttpResponseFor> 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 captor = ArgumentCaptor.forClass(RequestOptions.class); + verify(messageServiceAsyncWithRawResponse).createStreaming(any(MessageCreateParams.class), captor.capture()); + assertThat(captor.getValue().getTimeout().request()).isEqualTo(Duration.ofSeconds(5)); + } + +}