This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new ddd2c020 [integration][java][python] Send user images and documents to
Anthropic (#1204)
ddd2c020 is described below
commit ddd2c020e59889f3acbd766361ed586a8e5a470a
Author: Edson <[email protected]>
AuthorDate: Fri Oct 16 02:44:32 2026 -0400
[integration][java][python] Send user images and documents to Anthropic
(#1204)
Generated-by: Claude Code 2.1.259 (Claude Opus 5.5)
---
.../calling-llms/chat-models/providers.md | 13 +-
docs/content/docs/development/calling-llms/chat.md | 10 +-
.../anthropic/AnthropicChatModelConnection.java | 5 +-
.../chatmodels/anthropic/AnthropicChatUtils.java | 124 ++++++++++++-
.../AnthropicChatModelConnectionTest.java | 20 --
.../anthropic/AnthropicMultimodalTest.java | 205 +++++++++++++++++++++
.../chat_models/anthropic/anthropic_chat_model.py | 89 ++++++++-
.../anthropic/tests/test_anthropic_multimodal.py | 193 +++++++++++++++++++
.../tests/test_text_only_providers_reject_media.py | 4 -
9 files changed, 628 insertions(+), 35 deletions(-)
diff --git
a/docs/content/docs/development/calling-llms/chat-models/providers.md
b/docs/content/docs/development/calling-llms/chat-models/providers.md
index 488b42a0..f405a468 100644
--- a/docs/content/docs/development/calling-llms/chat-models/providers.md
+++ b/docs/content/docs/development/calling-llms/chat-models/providers.md
@@ -258,6 +258,17 @@ public class MyAgent extends Agent {
{{< /tabs >}}
+**Multimodal input.** The media blocks of a user message are sent as Anthropic
content blocks, in block order, next to its text blocks:
+
+| Block | Sent as |
+|---|---|
+| `TextBlock` | a `text` block |
+| `ImageBlock` | an `image` block, by URL or Base64 data, for `image/jpeg`,
`image/png`, `image/gif` or `image/webp` |
+| `DocumentBlock` | a `document` block titled by the block's `name`:
`application/pdf` by URL or Base64 data, or `text/plain` Base64 data sent as
decoded UTF-8 text |
+| `AudioBlock`, `VideoBlock` | not supported |
+
+Other image or document types, plain text given by URL, audio, video, and
media in assistant messages or tool results raise
`UnsupportedContentBlockException` (Java) or `UnsupportedContentBlockError`
(Python) before the request is sent.
+
#### Available Models
Visit the [Anthropic Models
documentation](https://docs.anthropic.com/en/docs/about-claude/models) for the
complete and up-to-date list of available chat models.
@@ -857,7 +868,7 @@ The Chat Completions connections (OpenAI, Azure OpenAI and
vLLM) send the media
A user message with only text blocks is still sent as a plain string. Media in
system, assistant or tool messages, audio or documents given by URL, other
audio types, and video raise `UnsupportedContentBlockException` (Java) or
`UnsupportedContentBlockError` (Python) before the request is sent. Whether a
model accepts a given part depends on the model and the server.
-The other built-in providers (Anthropic, Gemini, Amazon Bedrock, IBM
watsonx.ai, DashScope and the OpenAI Responses API) do not send media yet: a
message with a media block raises the same error instead of being sent as its
text.
+The other built-in providers (Gemini, Amazon Bedrock, IBM watsonx.ai,
DashScope and the OpenAI Responses API) do not send media yet: a message with a
media block raises the same error instead of being sent as its text.
#### Responses API
diff --git a/docs/content/docs/development/calling-llms/chat.md
b/docs/content/docs/development/calling-llms/chat.md
index b9216033..1cd13e17 100644
--- a/docs/content/docs/development/calling-llms/chat.md
+++ b/docs/content/docs/development/calling-llms/chat.md
@@ -219,10 +219,12 @@ For tools that return text or media, see
**Provider support:** The OpenAI Chat Completions integration, and the Azure
OpenAI
and vLLM integrations built on it, send media blocks to the model; see
[Multimodal Input]({{< ref
"docs/development/calling-llms/chat-models/providers#multimodal-input" >}}) for
what each block becomes. Ollama sends
-Base64 images; see the [Ollama provider section]({{< ref
"docs/development/calling-llms/chat-models/providers#ollama" >}}). The
Anthropic, Gemini, Amazon Bedrock, IBM
-watsonx.ai, DashScope and OpenAI Responses integrations do not send media yet:
a
-message with a media block raises `UnsupportedContentBlockException` (Java) or
-`UnsupportedContentBlockError` (Python).
+Base64 images, and Anthropic sends images and PDF or plain-text documents; see
the
+[Ollama]({{< ref "docs/development/calling-llms/chat-models/providers#ollama"
>}}) and
+[Anthropic]({{< ref
"docs/development/calling-llms/chat-models/providers#anthropic" >}}) provider
+sections. The Gemini, Amazon Bedrock, IBM watsonx.ai, DashScope and OpenAI
Responses
+integrations do not send media yet: a message with a media block raises
+`UnsupportedContentBlockException` (Java) or `UnsupportedContentBlockError`
(Python).
For example, a message can contain a question followed by an image:
diff --git
a/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java
b/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java
index 7331e82a..a9498d65 100644
---
a/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java
+++
b/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java
@@ -458,14 +458,15 @@ public class AnthropicChatModelConnection extends
BaseChatModelConnection {
List<org.apache.flink.agents.api.tools.Tool> tools,
Map<String, Object> modelParams,
Object outputSchema) {
- // Media blocks are not sent yet; fail rather than drop them (#1059).
- UnsupportedContentBlockException.rejectMedia("Anthropic", messages);
try {
BuiltRequest built = buildRequest(messages, tools, modelParams,
outputSchema);
Message response = client.messages().create(built.params);
ChatResult result = convertResponse(built, response);
return result;
+ } catch (UnsupportedContentBlockException e) {
+ // Unchanged, so callers can catch it by type; other build
failures stay wrapped.
+ throw e;
} catch (Exception e) {
throw new RuntimeException("Failed to call Anthropic messages
API.", e);
}
diff --git
a/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatUtils.java
b/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatUtils.java
index f1d6d8f9..db1f431c 100644
---
a/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatUtils.java
+++
b/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatUtils.java
@@ -18,10 +18,15 @@
package org.apache.flink.agents.integrations.chatmodels.anthropic;
import com.anthropic.core.JsonValue;
+import com.anthropic.models.messages.Base64ImageSource;
+import com.anthropic.models.messages.Base64PdfSource;
import com.anthropic.models.messages.ContentBlock;
import com.anthropic.models.messages.ContentBlockParam;
+import com.anthropic.models.messages.DocumentBlockParam;
+import com.anthropic.models.messages.ImageBlockParam;
import com.anthropic.models.messages.Message;
import com.anthropic.models.messages.MessageParam;
+import com.anthropic.models.messages.PlainTextSource;
import com.anthropic.models.messages.RedactedThinkingBlockParam;
import com.anthropic.models.messages.StopReason;
import com.anthropic.models.messages.TextBlockParam;
@@ -29,23 +34,35 @@ import com.anthropic.models.messages.ThinkingBlockParam;
import com.anthropic.models.messages.Tool;
import com.anthropic.models.messages.ToolResultBlockParam;
import com.anthropic.models.messages.ToolUseBlockParam;
+import com.anthropic.models.messages.UrlImageSource;
+import com.anthropic.models.messages.UrlPdfSource;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.chat.messages.Base64Source;
import org.apache.flink.agents.api.chat.messages.ChatMessage;
import org.apache.flink.agents.api.chat.messages.ChatResult;
+import org.apache.flink.agents.api.chat.messages.DocumentBlock;
+import org.apache.flink.agents.api.chat.messages.ImageBlock;
+import org.apache.flink.agents.api.chat.messages.MediaBlock;
import org.apache.flink.agents.api.chat.messages.MessageRole;
import org.apache.flink.agents.api.chat.messages.ReasoningBlock;
import org.apache.flink.agents.api.chat.messages.TextBlock;
import org.apache.flink.agents.api.chat.messages.TokenUsage;
import org.apache.flink.agents.api.chat.messages.ToolCallBlock;
import org.apache.flink.agents.api.chat.messages.ToolResultBlock;
+import
org.apache.flink.agents.api.chat.messages.UnsupportedContentBlockException;
+import org.apache.flink.agents.api.chat.messages.UrlSource;
import org.apache.flink.agents.api.tools.ToolMetadata;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Base64;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
/** Converts Flink Agents messages and tool definitions to and from Anthropic
SDK types. */
@@ -60,6 +77,10 @@ final class AnthropicChatUtils {
private static final String REDACTED_THINKING = "redacted_thinking";
private static final String REDACTED_DATA = "data";
+ /** The image media types the Anthropic Messages API accepts. */
+ private static final Set<String> IMAGE_MEDIA_TYPES =
+ Set.of("image/jpeg", "image/png", "image/gif", "image/webp");
+
private AnthropicChatUtils() {}
static List<TextBlockParam> extractSystemMessages(List<ChatMessage>
messages) {
@@ -69,9 +90,20 @@ final class AnthropicChatUtils {
.collect(Collectors.toList());
}
+ /**
+ * Converts a message to an Anthropic message. Only user messages can
carry media: images (JPEG,
+ * PNG, GIF or WebP) and PDF documents by Base64 data or URL, and
plain-text documents by Base64
+ * data. Any other media, or media in an assistant message or a tool
result, throws {@link
+ * UnsupportedContentBlockException}.
+ */
static MessageParam convertToAnthropicMessage(ChatMessage message) {
if (message.getRole() == MessageRole.TOOL) {
- return convertToolResult((ToolResultBlock)
message.getBlocks().get(0));
+ ToolResultBlock result = (ToolResultBlock)
message.getBlocks().get(0);
+ requireTextOnly(message, result.getBlocks());
+ return convertToolResult(result);
+ }
+ if (message.getRole() != MessageRole.USER) {
+ requireTextOnly(message, message.getBlocks());
}
List<ContentBlockParam> blocks = new ArrayList<>();
for (org.apache.flink.agents.api.chat.messages.ContentBlock block :
message.getBlocks()) {
@@ -90,6 +122,8 @@ final class AnthropicChatUtils {
.name(call.getName())
.input(toJsonValue(call.getInput()))
.build()));
+ } else if (block instanceof MediaBlock) {
+ blocks.add(toMediaBlockParam((MediaBlock) block));
} else if (block instanceof ReasoningBlock) {
ContentBlockParam reasoning =
convertReasoning((ReasoningBlock) block);
if (reasoning != null) {
@@ -109,6 +143,94 @@ final class AnthropicChatUtils {
.build();
}
+ private static ContentBlockParam toMediaBlockParam(MediaBlock block) {
+ if (block instanceof ImageBlock) {
+ String mediaType = essence(block.getMediaType());
+ if (!IMAGE_MEDIA_TYPES.contains(mediaType)) {
+ throw unsupported(block, "images must be JPEG, PNG, GIF or
WebP");
+ }
+ ImageBlockParam.Builder param = ImageBlockParam.builder();
+ if (block.getSource() instanceof UrlSource) {
+ param.source(
+ UrlImageSource.builder()
+ .url(((UrlSource) block.getSource()).getUrl())
+ .build());
+ } else {
+ param.source(
+ Base64ImageSource.builder()
+ .data(((Base64Source)
block.getSource()).getData())
+
.mediaType(Base64ImageSource.MediaType.of(mediaType))
+ .build());
+ }
+ return ContentBlockParam.ofImage(param.build());
+ }
+ if (block instanceof DocumentBlock) {
+ return
ContentBlockParam.ofDocument(toDocumentBlockParam((DocumentBlock) block));
+ }
+ throw unsupported(block, "Anthropic accepts images and PDF or
plain-text documents only");
+ }
+
+ private static DocumentBlockParam toDocumentBlockParam(DocumentBlock
document) {
+ String mediaType = essence(document.getMediaType());
+ DocumentBlockParam.Builder param = DocumentBlockParam.builder();
+ if (document.getName() != null) {
+ param.title(document.getName());
+ }
+ if ("application/pdf".equals(mediaType)) {
+ if (document.getSource() instanceof UrlSource) {
+ param.source(
+ UrlPdfSource.builder()
+ .url(((UrlSource)
document.getSource()).getUrl())
+ .build());
+ } else {
+ param.source(
+ Base64PdfSource.builder()
+ .data(((Base64Source)
document.getSource()).getData())
+ .build());
+ }
+ return param.build();
+ }
+ if ("text/plain".equals(mediaType) && document.getSource() instanceof
Base64Source) {
+ byte[] text;
+ try {
+ text = Base64.getDecoder().decode(((Base64Source)
document.getSource()).getData());
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "A document block's base64 data could not be
decoded.", e);
+ }
+ return param.source(
+ PlainTextSource.builder()
+ .data(new String(text,
StandardCharsets.UTF_8))
+ .build())
+ .build();
+ }
+ throw unsupported(document, "documents must be PDF, or Base64 plain
text, for Anthropic");
+ }
+
+ private static void requireTextOnly(
+ ChatMessage message,
+ List<? extends
org.apache.flink.agents.api.chat.messages.ContentBlock> blocks) {
+ for (org.apache.flink.agents.api.chat.messages.ContentBlock block :
blocks) {
+ if (block instanceof MediaBlock) {
+ throw unsupported(
+ block,
+ "only user messages can carry media, not "
+ + message.getRole().getValue()
+ + " messages");
+ }
+ }
+ }
+
+ /** The media type without parameters, lower-cased. */
+ private static String essence(String mediaType) {
+ return mediaType.split(";", 2)[0].trim().toLowerCase(Locale.ROOT);
+ }
+
+ private static UnsupportedContentBlockException unsupported(
+ org.apache.flink.agents.api.chat.messages.ContentBlock block,
String reason) {
+ return UnsupportedContentBlockException.forBlock("Anthropic", block,
reason);
+ }
+
private static MessageParam convertToolResult(ToolResultBlock result) {
StringBuilder text = new StringBuilder();
for (org.apache.flink.agents.api.chat.messages.ContentBlock block :
result.getBlocks()) {
diff --git
a/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java
b/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java
index 54d9077d..6ca83790 100644
---
a/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java
+++
b/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java
@@ -37,10 +37,8 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.agents.api.chat.messages.ChatMessage;
import org.apache.flink.agents.api.chat.messages.ChatResult;
-import org.apache.flink.agents.api.chat.messages.ImageBlock;
import org.apache.flink.agents.api.chat.messages.MessageRole;
import org.apache.flink.agents.api.chat.messages.ReasoningBlock;
-import
org.apache.flink.agents.api.chat.messages.UnsupportedContentBlockException;
import org.apache.flink.agents.api.chat.model.NativeStructuredOutputSupport;
import org.apache.flink.agents.api.resource.ResourceContext;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
@@ -1277,24 +1275,6 @@ class AnthropicChatModelConnectionTest {
}
}
- @Test
- @DisplayName("Media blocks fail explicitly instead of being dropped")
- void testMediaBlocksFailExplicitly() {
- List<ChatMessage> messages =
- List.of(
- ChatMessage.user(
- List.of(
-
org.apache.flink.agents.api.chat.messages.TextBlock.of(
- "Describe this"),
- ImageBlock.fromBase64("image/png",
"aGVsbG8="))));
-
- assertThatThrownBy(() -> connection().chat(messages, List.of(), new
HashMap<>(), null))
- .isInstanceOf(UnsupportedContentBlockException.class)
- .hasMessage(
- "Anthropic cannot send an image block (image/png,
base64 source): this"
- + " integration sends text only.");
- }
-
@Test
void testNativeReasoningReplayPreservesSignedAndRedactedBlocks() {
ChatMessage message =
diff --git
a/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicMultimodalTest.java
b/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicMultimodalTest.java
new file mode 100644
index 00000000..409a820a
--- /dev/null
+++
b/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicMultimodalTest.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * http://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.apache.flink.agents.integrations.chatmodels.anthropic;
+
+import com.anthropic.core.ObjectMappers;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.agents.api.chat.messages.AudioBlock;
+import org.apache.flink.agents.api.chat.messages.Base64Source;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.ContentBlock;
+import org.apache.flink.agents.api.chat.messages.DocumentBlock;
+import org.apache.flink.agents.api.chat.messages.ImageBlock;
+import org.apache.flink.agents.api.chat.messages.TextBlock;
+import org.apache.flink.agents.api.chat.messages.ToolResultBlock;
+import
org.apache.flink.agents.api.chat.messages.UnsupportedContentBlockException;
+import org.apache.flink.agents.api.chat.messages.VideoBlock;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests how content blocks become Anthropic Messages API content blocks. */
+class AnthropicMultimodalTest {
+
+ private static final String URL = "https://example.com/cat.png?sig=secret";;
+ private static final String BASE64 = "aGVsbG8="; // "hello"
+
+ private static AnthropicChatModelConnection connection() {
+ ResourceDescriptor descriptor =
+
ResourceDescriptor.Builder.newBuilder(AnthropicChatModelConnection.class.getName())
+ .addInitialArgument("api_key", "test-key")
+ .addInitialArgument("model",
"claude-sonnet-4-20250514")
+ .build();
+ return new AnthropicChatModelConnection(
+ descriptor, ResourceContext.fromGetResource((a, b) -> null));
+ }
+
+ private static Map<String, Object> params() {
+ Map<String, Object> params = new HashMap<>();
+ params.put("max_tokens", 256);
+ return params;
+ }
+
+ /** The first request message, as the SDK serializes it. */
+ private static JsonNode wire(ChatMessage message) {
+ return ObjectMappers.jsonMapper()
+ .valueToTree(
+ connection()
+ .buildRequest(List.of(message), List.of(),
params(), null)
+ .params
+ .messages()
+ .get(0));
+ }
+
+ @Test
+ @DisplayName("A text-only user message is sent as before, as one text
block")
+ void testTextOnlyUserMessageIsUnchanged() {
+ JsonNode content =
wire(ChatMessage.user(List.of(TextBlock.of("hi")))).get("content");
+
+ assertThat(content).hasSize(1);
+ assertThat(content.at("/0/type").asText()).isEqualTo("text");
+ assertThat(content.at("/0/text").asText()).isEqualTo("hi");
+ }
+
+ @Test
+ @DisplayName("Images become image blocks in block order, by URL or as
Base64 data")
+ void testImagesBecomeImageBlocks() {
+ JsonNode content =
+ wire(ChatMessage.user(
+ List.of(
+ TextBlock.of("Compare"),
+ ImageBlock.fromUrl("image/png", URL),
+ ImageBlock.fromBase64("image/png",
BASE64))))
+ .get("content");
+
+ assertThat(content).hasSize(3);
+ assertThat(content.at("/0/type").asText()).isEqualTo("text");
+ assertThat(content.at("/0/text").asText()).isEqualTo("Compare");
+ assertThat(content.at("/1/type").asText()).isEqualTo("image");
+ assertThat(content.at("/1/source/type").asText()).isEqualTo("url");
+ assertThat(content.at("/1/source/url").asText()).isEqualTo(URL);
+ assertThat(content.at("/2/source/type").asText()).isEqualTo("base64");
+
assertThat(content.at("/2/source/media_type").asText()).isEqualTo("image/png");
+ assertThat(content.at("/2/source/data").asText()).isEqualTo(BASE64);
+ }
+
+ @Test
+ @DisplayName("PDFs and Base64 plain text become document blocks, titled by
the block name")
+ void testDocumentsBecomeDocumentBlocks() {
+ JsonNode content =
+ wire(ChatMessage.user(
+ List.of(
+ new DocumentBlock(
+ "application/pdf",
+ new Base64Source(BASE64),
+ "report.pdf",
+ null,
+ null),
+ DocumentBlock.fromUrl(
+ "application/pdf",
"https://example.com/a.pdf";),
+ DocumentBlock.fromBase64("text/plain",
BASE64))))
+ .get("content");
+
+ assertThat(content.at("/0/type").asText()).isEqualTo("document");
+ assertThat(content.at("/0/source/type").asText()).isEqualTo("base64");
+
assertThat(content.at("/0/source/media_type").asText()).isEqualTo("application/pdf");
+ assertThat(content.at("/0/source/data").asText()).isEqualTo(BASE64);
+ assertThat(content.at("/0/title").asText()).isEqualTo("report.pdf");
+ assertThat(content.at("/1/source/type").asText()).isEqualTo("url");
+
assertThat(content.at("/1/source/url").asText()).isEqualTo("https://example.com/a.pdf";);
+ assertThat(content.at("/2/source/type").asText()).isEqualTo("text");
+
assertThat(content.at("/2/source/media_type").asText()).isEqualTo("text/plain");
+ assertThat(content.at("/2/source/data").asText()).isEqualTo("hello");
+ }
+
+ @Test
+ @DisplayName("Blocks Anthropic has no block for fail without leaking the
source")
+ void testUnsupportedBlocksFailExplicitly() {
+ List<ContentBlock> unsupported =
+ List.of(
+ AudioBlock.fromBase64("audio/wav", BASE64),
+ VideoBlock.fromUrl("video/mp4", URL),
+ ImageBlock.fromBase64("image/bmp", BASE64),
+ DocumentBlock.fromUrl("text/plain", URL),
+ DocumentBlock.fromBase64("application/msword",
BASE64));
+ for (ContentBlock block : unsupported) {
+ assertThatThrownBy(() ->
wire(ChatMessage.user(List.of(TextBlock.of("hi"), block))))
+ .isInstanceOf(UnsupportedContentBlockException.class)
+ .hasMessageStartingWith("Anthropic cannot send a")
+ .hasMessageContaining(" " + block.getType() + " block")
+ .hasMessageNotContaining("secret")
+ .hasMessageNotContaining(BASE64);
+ }
+ }
+
+ @Test
+ @DisplayName("Media in assistant messages and tool results fails
explicitly")
+ void testMediaOutsideUserMessagesFails() {
+ List<ContentBlock> blocks =
+ List.of(TextBlock.of("see"), ImageBlock.fromUrl("image/png",
URL));
+ ChatMessage assistant = ChatMessage.assistant(blocks);
+ ChatMessage tool =
+ ChatMessage.tool(
+ new ToolResultBlock(
+ "call-1",
+ List.of(TextBlock.of("see"),
ImageBlock.fromUrl("image/png", URL)),
+ false));
+
+ for (ChatMessage message : List.of(assistant, tool)) {
+ assertThatThrownBy(
+ () ->
+ connection()
+ .buildRequest(
+
List.of(ChatMessage.user("hi"), message),
+ List.of(),
+ params(),
+ null))
+ .isInstanceOf(UnsupportedContentBlockException.class)
+ .hasMessageContaining("only user messages can carry
media");
+ }
+ }
+
+ @Test
+ @DisplayName("Undecodable plain-text data fails as invalid, not as
unsupported")
+ void testInvalidBase64TextDocumentFails() {
+ ChatMessage message =
+
ChatMessage.user(List.of(DocumentBlock.fromBase64("text/plain", "not
base64!")));
+
+ assertThatThrownBy(() -> wire(message))
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessage("A document block's base64 data could not be
decoded.");
+ }
+
+ @Test
+ @DisplayName("chat() surfaces media errors with their own types")
+ void testChatKeepsMediaErrorTypes() {
+ ChatMessage message =
ChatMessage.user(List.of(AudioBlock.fromBase64("audio/wav", BASE64)));
+
+ assertThatThrownBy(() -> connection().chat(List.of(message),
List.of(), params(), null))
+ .isExactlyInstanceOf(UnsupportedContentBlockException.class);
+ }
+}
diff --git
a/python/flink_agents/integrations/chat_models/anthropic/anthropic_chat_model.py
b/python/flink_agents/integrations/chat_models/anthropic/anthropic_chat_model.py
index 539a23be..bef145db 100644
---
a/python/flink_agents/integrations/chat_models/anthropic/anthropic_chat_model.py
+++
b/python/flink_agents/integrations/chat_models/anthropic/anthropic_chat_model.py
@@ -15,6 +15,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import base64
+import binascii
import logging
import uuid
from typing import Any, Dict, List, Mapping, Sequence
@@ -27,12 +29,18 @@ from typing_extensions import override
from flink_agents.api.agents.types import OutputSchema, render_output_schema
from flink_agents.api.chat_message import (
+ Base64Source,
ChatMessage,
+ ContentBlock,
+ DocumentBlock,
+ ImageBlock,
+ MediaBlock,
MessageRole,
ReasoningBlock,
TextBlock,
ToolCallBlock,
UnsupportedContentBlockError,
+ UrlSource,
)
from flink_agents.api.chat_models.chat_model import (
BaseChatModelConnection,
@@ -47,6 +55,7 @@ logger = logging.getLogger(__name__)
_ANTHROPIC_METADATA = "anthropic"
_THINKING = "thinking"
_REDACTED_THINKING = "redacted_thinking"
+_IMAGE_MEDIA_TYPES = frozenset({"image/jpeg", "image/png", "image/gif",
"image/webp"})
def to_anthropic_tool(
@@ -66,8 +75,76 @@ def to_anthropic_tool(
}
+def _unsupported(block: ContentBlock, reason: str) ->
UnsupportedContentBlockError:
+ return UnsupportedContentBlockError.for_block("Anthropic", block, reason)
+
+
+def _essence(media_type: str) -> str:
+ """The media type without parameters, lower-cased."""
+ return media_type.split(";", 1)[0].strip().lower()
+
+
+def _to_document_block(block: DocumentBlock) -> Dict[str, Any]:
+ media_type = _essence(block.media_type)
+ if media_type == "application/pdf":
+ if isinstance(block.source, UrlSource):
+ source = {"type": "url", "url": block.source.url}
+ else:
+ source = {
+ "type": "base64",
+ "media_type": "application/pdf",
+ "data": block.source.data,
+ }
+ elif media_type == "text/plain" and isinstance(block.source, Base64Source):
+ try:
+ data = base64.b64decode(block.source.data, validate=True)
+ except binascii.Error as e:
+ msg = "A document block's base64 data could not be decoded."
+ raise ValueError(msg) from e
+ source = {
+ "type": "text",
+ "media_type": "text/plain",
+ "data": data.decode("utf-8", errors="replace"),
+ }
+ else:
+ raise _unsupported(
+ block, "documents must be PDF, or Base64 plain text, for Anthropic"
+ )
+ document: Dict[str, Any] = {"type": "document", "source": source}
+ if block.name is not None:
+ document["title"] = block.name
+ return document
+
+
+def _to_media_block(block: MediaBlock) -> Dict[str, Any]:
+ if isinstance(block, ImageBlock):
+ media_type = _essence(block.media_type)
+ if media_type not in _IMAGE_MEDIA_TYPES:
+ raise _unsupported(block, "images must be JPEG, PNG, GIF or WebP")
+ if isinstance(block.source, UrlSource):
+ source = {"type": "url", "url": block.source.url}
+ else:
+ source = {
+ "type": "base64",
+ "media_type": media_type,
+ "data": block.source.data,
+ }
+ return {"type": "image", "source": source}
+ if isinstance(block, DocumentBlock):
+ return _to_document_block(block)
+ raise _unsupported(
+ block, "Anthropic accepts images and PDF or plain-text documents only"
+ )
+
+
def convert_to_anthropic_message(message: ChatMessage) -> MessageParam:
- """Convert ChatMessage to Anthropic MessageParam format."""
+ """Convert ChatMessage to Anthropic MessageParam format.
+
+ Only user messages can carry media: images (JPEG, PNG, GIF or WebP) and PDF
+ documents by Base64 data or URL, and plain-text documents by Base64 data.
+ Any other media, or media in an assistant message or a tool result, raises
+ UnsupportedContentBlockError.
+ """
blocks = (
message.blocks[0].blocks if message.role == MessageRole.TOOL else
message.blocks
)
@@ -75,6 +152,14 @@ def convert_to_anthropic_message(message: ChatMessage) ->
MessageParam:
for block in blocks:
if isinstance(block, TextBlock):
content.append({"type": "text", "text": block.text})
+ elif isinstance(block, MediaBlock):
+ if message.role != MessageRole.USER:
+ reason = (
+ "only user messages can carry media, "
+ f"not {message.role.value} messages"
+ )
+ raise _unsupported(block, reason)
+ content.append(_to_media_block(block))
elif isinstance(block, ToolCallBlock):
content.append(
{
@@ -503,8 +588,6 @@ class AnthropicChatModelConnection(BaseChatModelConnection):
ChatResult
Model response message
"""
- # Media blocks are not sent yet; fail rather than drop them (#1059).
- UnsupportedContentBlockError.reject_media("Anthropic", messages)
anthropic_tools = None
if tools is not None:
anthropic_tools = [
diff --git
a/python/flink_agents/integrations/chat_models/anthropic/tests/test_anthropic_multimodal.py
b/python/flink_agents/integrations/chat_models/anthropic/tests/test_anthropic_multimodal.py
new file mode 100644
index 00000000..9956de1f
--- /dev/null
+++
b/python/flink_agents/integrations/chat_models/anthropic/tests/test_anthropic_multimodal.py
@@ -0,0 +1,193 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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
+#
+# http://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.
+#################################################################################
+"""How content blocks reach an Anthropic messages request.
+
+Mirrors the Java AnthropicMultimodalTest.
+"""
+
+from typing import Any, Dict
+from unittest.mock import MagicMock
+
+import pytest
+from anthropic.types import Message, Usage
+from anthropic.types import TextBlock as AnthropicTextBlock
+
+from flink_agents.api.chat_message import (
+ AudioBlock,
+ Base64Source,
+ ChatMessage,
+ ContentBlock,
+ DocumentBlock,
+ ImageBlock,
+ TextBlock,
+ ToolResultBlock,
+ UnsupportedContentBlockError,
+ VideoBlock,
+)
+from flink_agents.integrations.chat_models.anthropic.anthropic_chat_model
import (
+ AnthropicChatModelConnection,
+)
+
+URL = "https://example.com/cat.png?sig=secret";
+BASE64 = "aGVsbG8=" # "hello"
+
+
+def _connection() -> tuple[AnthropicChatModelConnection, MagicMock]:
+ connection = AnthropicChatModelConnection(api_key="fake-key")
+ client = MagicMock()
+ client.messages.create.return_value = Message(
+ id="m",
+ model="claude",
+ role="assistant",
+ type="message",
+ stop_reason="end_turn",
+ content=[AnthropicTextBlock(type="text", text="ok")],
+ usage=Usage(input_tokens=1, output_tokens=1),
+ )
+ connection._client = client
+ return connection, client
+
+
+def _chat(*messages: ChatMessage) -> MagicMock:
+ connection, client = _connection()
+ connection.chat(list(messages), model="claude-sonnet-4-5")
+ return client
+
+
+def _sent(message: ChatMessage) -> Dict[str, Any]:
+ """Chat through a mocked client and return the first message as sent."""
+ return _chat(message).messages.create.call_args.kwargs["messages"][0]
+
+
+def test_text_only_user_message_is_unchanged() -> None:
+ """A user message without media is sent as before, as one text block."""
+ assert _sent(ChatMessage.user([TextBlock(text="hi")]))["content"] == [
+ {"type": "text", "text": "hi"}
+ ]
+
+
+def test_images_become_image_blocks() -> None:
+ """Images become image blocks in block order, by URL or as Base64 data."""
+ content = _sent(
+ ChatMessage.user(
+ [
+ TextBlock(text="Compare"),
+ ImageBlock.from_url("image/png", URL),
+ ImageBlock.from_base64("image/png", BASE64),
+ ]
+ )
+ )["content"]
+
+ assert content == [
+ {"type": "text", "text": "Compare"},
+ {"type": "image", "source": {"type": "url", "url": URL}},
+ {
+ "type": "image",
+ "source": {"type": "base64", "media_type": "image/png", "data":
BASE64},
+ },
+ ]
+
+
+def test_documents_become_document_blocks() -> None:
+ """PDFs and Base64 plain text become document blocks, titled by the
name."""
+ content = _sent(
+ ChatMessage.user(
+ [
+ DocumentBlock(
+ media_type="application/pdf",
+ source=Base64Source(data=BASE64),
+ name="report.pdf",
+ ),
+ DocumentBlock.from_url("application/pdf",
"https://example.com/a.pdf";),
+ DocumentBlock.from_base64("text/plain", BASE64),
+ ]
+ )
+ )["content"]
+
+ assert content == [
+ {
+ "type": "document",
+ "source": {
+ "type": "base64",
+ "media_type": "application/pdf",
+ "data": BASE64,
+ },
+ "title": "report.pdf",
+ },
+ {
+ "type": "document",
+ "source": {"type": "url", "url": "https://example.com/a.pdf"},
+ },
+ {
+ "type": "document",
+ "source": {"type": "text", "media_type": "text/plain", "data":
"hello"},
+ },
+ ]
+
+
[email protected](
+ "block",
+ [
+ AudioBlock.from_base64("audio/wav", BASE64),
+ VideoBlock.from_url("video/mp4", URL),
+ ImageBlock.from_base64("image/bmp", BASE64),
+ DocumentBlock.from_url("text/plain", URL),
+ DocumentBlock.from_base64("application/msword", BASE64),
+ ],
+ ids=str,
+)
+def test_unsupported_blocks_fail_explicitly(block: ContentBlock) -> None:
+ """Blocks Anthropic has no block for fail without leaking the source."""
+ connection, client = _connection()
+ message = ChatMessage.user([TextBlock(text="hi"), block])
+
+ with pytest.raises(UnsupportedContentBlockError) as error:
+ connection.chat([message], model="claude-sonnet-4-5")
+
+ text = str(error.value)
+ assert text.startswith("Anthropic cannot send a")
+ assert f" {block.type} block" in text
+ assert "secret" not in text
+ assert BASE64 not in text
+ client.messages.create.assert_not_called()
+
+
[email protected]("kind", ["assistant", "tool"])
+def test_media_outside_user_messages_fails(kind: str) -> None:
+ """Media in assistant messages and tool results fails explicitly."""
+ blocks = [TextBlock(text="see"), ImageBlock.from_url("image/png", URL)]
+ message = (
+ ChatMessage.assistant(blocks)
+ if kind == "assistant"
+ else ChatMessage.tool(ToolResultBlock(call_id="call-1",
blocks=tuple(blocks)))
+ )
+
+ with pytest.raises(
+ UnsupportedContentBlockError, match="only user messages can carry
media"
+ ):
+ _chat(ChatMessage.user("hi"), message)
+
+
+def test_invalid_base64_text_document_fails() -> None:
+ """Undecodable plain-text data fails with ValueError, not as
unsupported."""
+ message = ChatMessage.user([DocumentBlock.from_base64("text/plain", "not
base64!")])
+
+ with pytest.raises(ValueError, match="could not be decoded") as error:
+ _chat(message)
+
+ assert not isinstance(error.value, UnsupportedContentBlockError)
diff --git
a/python/flink_agents/integrations/chat_models/tests/test_text_only_providers_reject_media.py
b/python/flink_agents/integrations/chat_models/tests/test_text_only_providers_reject_media.py
index a11aea10..0cef1641 100644
---
a/python/flink_agents/integrations/chat_models/tests/test_text_only_providers_reject_media.py
+++
b/python/flink_agents/integrations/chat_models/tests/test_text_only_providers_reject_media.py
@@ -33,9 +33,6 @@ from flink_agents.api.chat_message import (
UnsupportedContentBlockError,
)
from flink_agents.api.chat_models.chat_model import BaseChatModelConnection
-from flink_agents.integrations.chat_models.anthropic.anthropic_chat_model
import (
- AnthropicChatModelConnection,
-)
from flink_agents.integrations.chat_models.dashscope_chat_model import (
DashScopeChatModelConnection,
)
@@ -44,7 +41,6 @@ from
flink_agents.integrations.chat_models.watsonx.watsonx_chat_model import (
)
CONNECTIONS: Dict[str, Callable[[], BaseChatModelConnection]] = {
- "Anthropic": lambda: AnthropicChatModelConnection(api_key="fake-key"),
"DashScope": lambda: DashScopeChatModelConnection(api_key="fake-key"),
"IBM watsonx.ai": lambda: WatsonxChatModelConnection(
url="https://us-south.ml.cloud.ibm.com";,
![https://example.com/cat.png?sig=secret"](https://example.com/cat.png?sig=secret"){kind=link}