[api] Introduce multimodal content blocks

api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java

@@ -178,7 +178,7 @@ public static void stopAction(Event event, RunnerContext ctx) {
         if (response.getExtraArgs().containsKey(STRUCTURED_OUTPUT)) {
             output = response.getExtraArgs().get(STRUCTURED_OUTPUT);
         } else {
-            output = String.valueOf(response.getContent());
+            output = response.getText();
         }
 
         ctx.sendEvent(new OutputEvent(output));

api/src/main/java/org/apache/flink/agents/api/chat/messages/AudioBlock.java

@@ -0,0 +1,54 @@
+/*
+ * 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.api.chat.messages;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** The audio content of a {@link ChatMessage} — see {@link MediaBlock} for the media shape. */
+public final class AudioBlock extends MediaBlock {
+
+    @JsonCreator
+    public AudioBlock(
+            @JsonProperty("media_type") String mediaType,
+            @JsonProperty("data") @Nullable String data,
+            @JsonProperty("url") @Nullable String url,
+            @JsonProperty("name") @Nullable String name,
+            @JsonProperty("size_bytes") @Nullable Long sizeBytes,
+            @JsonProperty("sha256") @Nullable String sha256) {
+        super(mediaType, data, url, name, sizeBytes, sha256);
+    }
+
+    /** Creates an audio block carrying an inline base64 payload. */
+    public static AudioBlock fromBase64(String mediaType, String data) {
+        return new AudioBlock(mediaType, data, null, null, null, null);
+    }
+
+    /** Creates an audio block referencing an externally managed URL or provider file URI. */
+    public static AudioBlock fromUrl(String mediaType, String url) {
+        return new AudioBlock(mediaType, null, url, null, null, null);
+    }
+
+    @Override
+    public String getType() {
+        return "audio";
+    }
+}

api/src/main/java/org/apache/flink/agents/api/chat/messages/ChatMessage.java

@@ -20,21 +20,36 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 /**
  * Chat message class that represents all message types (user, system, assistant, tool) with
- * different roles
+ * different roles.
+ *
+ * <p>Message content is an ordered list of typed {@link ContentBlock}s ({@link TextBlock} plus the
+ * media blocks); a text-only message simply carries one {@link TextBlock}. The string convenience
+ * constructors and factories preserve the text-message experience, and {@link #getText()} is the
+ * ordered concatenation of the text blocks.
+ *
+ * <p>Blocks are immutable and the message snapshots every block list it is handed, so {@link
+ * #getBlocks()} is an unmodifiable view: the content changes only by replacing it through {@link
+ * #setBlocks(List)} or {@link #setText(String)}.
  */
 public class ChatMessage {
 
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
     private MessageRole role;
-    private String content;
+    private List<ContentBlock> blocks;
 
     @JsonProperty("tool_calls")
     private List<Map<String, Object>> toolCalls;
@@ -44,34 +59,61 @@ public class ChatMessage {
 
     /** Default constructor with SYSTEM role */
     public ChatMessage() {
-        this(MessageRole.SYSTEM, null, null, null);
+        this(MessageRole.SYSTEM, (List<ContentBlock>) null, null, null);
+    }
+
+    /** Constructor with role and text content */
+    public ChatMessage(MessageRole role, String text) {
+        this(role, blocksOf(text), null, null);
+    }
+
+    /** Constructor with role and content blocks */
+    public ChatMessage(MessageRole role, List<ContentBlock> blocks) {
+        this(role, blocks, null, null);
     }
 
-    /** Constructor with role and content */
-    public ChatMessage(MessageRole role, String content) {
-        this(role, content, null, null);
+    public ChatMessage(MessageRole role, String text, Map<String, Object> extraArgs) {
+        this(role, blocksOf(text), null, extraArgs);
     }
 
-    public ChatMessage(MessageRole role, String content, Map<String, Object> extraArgs) {
-        this(role, content, null, extraArgs);
+    public ChatMessage(MessageRole role, String text, List<Map<String, Object>> toolCalls) {
+        this(role, blocksOf(text), toolCalls, null);
     }
 
-    public ChatMessage(MessageRole role, String content, List<Map<String, Object>> toolCalls) {
-        this(role, content, toolCalls, null);
+    public ChatMessage(
+            MessageRole role,
+            String text,
+            List<Map<String, Object>> toolCalls,
+            Map<String, Object> extraArgs) {
+        this(role, blocksOf(text), toolCalls, extraArgs);
     }
 
     /** Full constructor */
     public ChatMessage(
             MessageRole role,
-            String content,
+            List<ContentBlock> blocks,
             List<Map<String, Object>> toolCalls,
             Map<String, Object> extraArgs) {
         this.role = role != null ? role : MessageRole.SYSTEM;
-        this.content = content != null ? content : "";
+        this.blocks = snapshotOf(blocks);
         this.toolCalls = toolCalls != null ? toolCalls : new ArrayList<>();
         this.extraArgs = extraArgs != null ? new HashMap<>(extraArgs) : new HashMap<>();
     }
 
+    /** An empty or null text becomes an empty block list rather than an empty text block. */
+    private static List<ContentBlock> blocksOf(String text) {
+        return text == null || text.isEmpty()
+                ? Collections.emptyList()
+                : Collections.singletonList(new TextBlock(text));
+    }
+
+    /** An unmodifiable copy — since blocks are immutable, this freezes the content. */
+    private static List<ContentBlock> snapshotOf(List<ContentBlock> blocks) {
+        return blocks == null || blocks.isEmpty()
+                ? Collections.emptyList()
+                : Collections.unmodifiableList(new ArrayList<>(blocks));
+    }
+
     public MessageRole getRole() {
         return role;
     }
@@ -80,12 +122,19 @@ public void setRole(MessageRole role) {
         this.role = role;
     }
 
-    public String getContent() {
-        return content;
+    /** The content as an unmodifiable list of immutable blocks. */
+    public List<ContentBlock> getBlocks() {
+        return blocks;
+    }
+
+    public void setBlocks(List<ContentBlock> blocks) {
+        this.blocks = snapshotOf(blocks);
     }
 
-    public void setContent(String content) {
-        this.content = content;
+    /** Replaces the content with a single text block (empty text clears the content). */
+    @JsonIgnore
+    public void setText(String text) {
+        this.blocks = blocksOf(text);
     }
 
     @JsonProperty("tool_calls")
@@ -108,9 +157,40 @@ public void setExtraArgs(Map<String, Object> extraArgs) {
         this.extraArgs = extraArgs != null ? extraArgs : new HashMap<>();
     }
 
+    /**
+     * The content blocks as plain maps in the serialized (snake_case, discriminated) shape — the
+     * same representation {@code tool_calls} uses. This is how blocks cross the Python bridge,
+     * which exchanges JSON-friendly lists and maps rather than typed Java objects.
+     */
+    @JsonIgnore
+    public List<Map<String, Object>> getBlocksAsMaps() {
+        return blocks.stream()
+                .map(
+                        block ->
+                                MAPPER.<Map<String, Object>>convertValue(
+                                        block, new TypeReference<Map<String, Object>>() {}))
+                .collect(Collectors.toList());
+    }
+
+    /** Replaces the content with blocks given as plain maps — see {@link #getBlocksAsMaps()}. */
+    @JsonIgnore
+    public void setBlocksFromMaps(List<Map<String, Object>> blockMaps) {
+        this.blocks =
+                blockMaps == null
+                        ? Collections.emptyList()
+                        : Collections.unmodifiableList(
+                                blockMaps.stream()
+                                        .map(map -> MAPPER.convertValue(map, ContentBlock.class))
+                                        .collect(Collectors.toList()));
+    }
+
+    /** The text projection: the ordered concatenation of this message's {@link TextBlock}s. */
     @JsonIgnore
     public String getText() {
-        return this.content;
+        return blocks.stream()
+                .filter(block -> block instanceof TextBlock)
+                .map(block -> ((TextBlock) block).getText())
+                .collect(Collectors.joining());
     }
 
     @JsonIgnore
@@ -124,24 +204,32 @@ public MessageRole getMessageType() {
     }
 
     // Static factory methods for convenience
-    public static ChatMessage user(String content) {
-        return new ChatMessage(MessageRole.USER, content);
+    public static ChatMessage user(String text) {
+        return new ChatMessage(MessageRole.USER, text);
+    }
+
+    public static ChatMessage user(List<ContentBlock> blocks) {
+        return new ChatMessage(MessageRole.USER, blocks);
+    }
+
+    public static ChatMessage system(String text) {
+        return new ChatMessage(MessageRole.SYSTEM, text);
     }
 
-    public static ChatMessage system(String content) {
-        return new ChatMessage(MessageRole.SYSTEM, content);
+    public static ChatMessage assistant(String text) {
+        return new ChatMessage(MessageRole.ASSISTANT, text);
     }
 
-    public static ChatMessage assistant(String content) {
-        return new ChatMessage(MessageRole.ASSISTANT, content);
+    public static ChatMessage assistant(String text, List<Map<String, Object>> toolCalls) {
+        return new ChatMessage(MessageRole.ASSISTANT, text, toolCalls, new HashMap<>());
     }
 
-    public static ChatMessage assistant(String content, List<Map<String, Object>> toolCalls) {
-        return new ChatMessage(MessageRole.ASSISTANT, content, toolCalls, new HashMap<>());
+    public static ChatMessage tool(String text) {
+        return new ChatMessage(MessageRole.TOOL, text);
     }
 
-    public static ChatMessage tool(String content) {
-        return new ChatMessage(MessageRole.TOOL, content);
+    public static ChatMessage tool(List<ContentBlock> blocks) {
+        return new ChatMessage(MessageRole.TOOL, blocks);
     }
 
     @Override
@@ -150,19 +238,19 @@ public boolean equals(Object o) {
         if (!(o instanceof ChatMessage)) return false;
         ChatMessage that = (ChatMessage) o;
         return Objects.equals(role, that.role)
-                && Objects.equals(content, that.content)
+                && Objects.equals(blocks, that.blocks)
                 && Objects.equals(toolCalls, that.toolCalls)
                 && Objects.equals(extraArgs, that.extraArgs);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(role, content, toolCalls, extraArgs);
+        return Objects.hash(role, blocks, toolCalls, extraArgs);
     }
 
     @Override
     public String toString() {
-        return role.getValue() + ": " + content;
+        return role.getValue() + ": " + getText();
     }
 
     /** Return the index of the first system message in the list, or -1 if none. */

api/src/main/java/org/apache/flink/agents/api/chat/messages/ContentBlock.java

@@ -0,0 +1,66 @@
+/*
+ * 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.api.chat.messages;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+
+import java.util.Map;
+
+/**
+ * A single, typed part of a {@link ChatMessage}'s content.
+ *
+ * <p>Blocks are ordered within a message and are immutable value objects: every construction path,
+ * including Jackson deserialization, runs the same validation, so sharing a block instance never
+ * shares mutable state. The concrete type answers how providers route the content ({@link
+ * TextBlock}, {@link ImageBlock}, {@link AudioBlock}, {@link VideoBlock}, {@link DocumentBlock}),
+ * while media encoding is carried by the media type on {@link MediaBlock}.
+ *
+ * <p>The serialized form carries a {@code type} discriminator with fixed values ({@code text},
+ * {@code image}, {@code audio}, {@code video}, {@code document}) shared with the Python API, so
+ * blocks cross the Java/Python boundary as plain JSON.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
+@JsonSubTypes({
+    @JsonSubTypes.Type(value = TextBlock.class, name = "text"),
+    @JsonSubTypes.Type(value = ImageBlock.class, name = "image"),
+    @JsonSubTypes.Type(value = AudioBlock.class, name = "audio"),
+    @JsonSubTypes.Type(value = VideoBlock.class, name = "video"),
+    @JsonSubTypes.Type(value = DocumentBlock.class, name = "document")
+})
+public abstract class ContentBlock {
+
+    /** The wire discriminator of this block: {@code text}, {@code image}, {@code audio}, ... */
+    @JsonIgnore
+    public abstract String getType();
+
+    /**
+     * The log-safe projection of this block, as a plain map in the wire's snake_case shape. Each
+     * block type defines its own logging policy: text passes through unchanged (the Event Log's
+     * level-dependent truncation still applies downstream), while media blocks whitelist their
+     * metadata, omit inline payload bytes, and sanitize URLs. Normal Jackson serialization — the
+     * Java/Python bridge, event serialization, state recovery — is unaffected and preserves the
+     * complete payload.
+     *
+     * <p>The result is intentionally not a valid wire block (media payloads are gone for good), so
+     * it must never be deserialized back into a {@link ContentBlock}.
+     */
+    public abstract Map<String, Object> sanitize();
+}