feat(agent): add streaming and multimodal tool result support (#3912)

[SeasonPilot]

Describe what this PR does / why we need it

Add support for streaming tool execution and multimodal tool results in the Agent Framework.
This enables tools to emit incremental results during execution and return rich content
types including images, audio, video, and files.

Key additions:

• StreamingToolCallback interface for tools that emit results incrementally via Flux
• ToolResult class supporting text, media (images/audio/video), files, and mixed content
• ToolStreamingOutput for streaming graph execution pipeline integration
• ToolStreamingErrorHandler for unified error handling across streaming scenarios
• AgentToolNode.executeToolCallsStreaming() for Flux-based tool execution

Does this pull request fix one issue?

Fixes #3912

Describe how you did it

1. Created StreamingToolCallback extending AsyncToolCallback with callStream() method
2. Implemented ToolResult as immutable value class with factory methods for different content types:

• text(), chunk(), media(), file(), mixed(), error()
• merge() for combining streaming chunks

3. Added ToolStreamingOutput extending StreamingOutput with tool-specific fields
4. Enhanced AgentToolNode with:

• executeToolCallsStreaming() returning Flux
• Streaming tool priority over async tools in executeToolByType()
• Cached Scheduler for efficient reactive execution
5. Added ToolStreamingErrorHandler for consistent error extraction and recovery

Describe how to verify it

Run the new unit tests:

# Run all new streaming tests
mvn test -pl spring-ai-alibaba-agent-framework -Dtest=StreamingToolCallbackTest,ToolResultTest,ToolStreamingErrorHandlerTest,AgentToolNodeStreamingExecutionTest

# Run ToolStreamingOutput tests
mvn test -pl spring-ai-alibaba-graph-core -Dtest=ToolStreamingOutputTest

# Run all tests for affected modules
mvn test -pl spring-ai-alibaba-agent-framework,spring-ai-alibaba-graph-core



Special notes for reviews

• StreamingToolCallback takes execution priority over AsyncToolCallback (line 446-451)
• ToolResult is immutable; all modifications return new instances
• Streaming execution uses reactive Flux to support backpressure
• Error handling preserves original exceptions while providing user-friendly messages
• Added comprehensive test coverage (60+ test cases across 6 test files)

[Aias00]

@SeasonPilot resolve conflicts pls

[Copilot]

Pull request overview

This pull request adds streaming and multimodal tool result support to the Spring AI Alibaba Agent Framework. It enables tools to emit incremental results during execution and return rich content types including images, audio, video, and files.

Changes:

• Introduces StreamingToolCallback interface for tools that emit results incrementally via Flux
• Adds ToolResult class supporting text, media (images/audio/video), files, and mixed content with serialization/deserialization
• Implements ToolStreamingOutput for streaming graph execution pipeline integration with tool identification metadata
• Adds comprehensive test coverage (60+ test cases across multiple test files)

Reviewed changes

Copilot reviewed 36 out of 36 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
ToolStreamingOutput.java	Streaming output wrapper for tool execution results with tool identification (toolCallId/toolName)
ToolStreamingOutputTest.java	Comprehensive test coverage for ToolStreamingOutput including JSON escaping and reflection
ToolStreamingErrorHandler.java	Unified error handler converting errors to streaming outputs instead of throwing
ToolStreamingErrorHandlerTest.java	Tests for error message extraction and timeout/cancellation detection
ToolResult.java	Immutable value class supporting text and multimodal content with JSON serialization
ToolResultTest.java	Tests for ToolResult factory methods, merge operations, and serialization
ToolStateCollector.java	Thread-safe state collector for parallel tool executions with isolation
ToolStateCollectorTest.java	Tests for parallel state management and discard functionality
ToolCancelledException.java	Exception for cancelled tool executions
StateAwareToolCallback.java	Marker interface for tools requiring state injection
ToolCallRequest.java	Added defensive copy of context map to prevent external modification
README updates	Documentation for new streaming and async features
Multiple test files	Extensive test coverage for async execution, cancellation, parallel execution, and integration scenarios

Comments suppressed due to low confidence (1)

spring-ai-alibaba-agent-framework/src/main/java/com/alibaba/cloud/ai/graph/agent/interceptor/ToolCallRequest.java:74

• getContext exposes the internal representation stored in field context. The value may be modified after this call to getContext.

	public Map<String, Object> getContext() {

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

getMediaContent exposes the internal representation stored in field mediaContent. The value may be modified after this call to getMediaContent.

Suggested change

	* @return the list of media content
	*/
	public List<Media> getMediaContent() {
	return mediaContent;
	* @return an unmodifiable list of media content (empty if none)
	*/
	public List<Media> getMediaContent() {
	return mediaContent == null ? Collections.emptyList() : Collections.unmodifiableList(mediaContent);

[Copilot]

There is no need to test whether an instance of ToolCancelledException is also an instance of RuntimeException - it always is.

[Aias00]

@SeasonPilot resolve conflicts pls

[SeasonPilot]

@chickenlj PTAL

[Copilot]

Pull request overview

Copilot reviewed 22 out of 22 changed files in this pull request and generated 7 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

In the semaphore acquire path, an InterruptedException is converted into a RuntimeException with message "Interrupted while waiting for semaphore". This bypasses the existing extractErrorMessage(InterruptedException) mapping ("Tool execution was interrupted") and exposes an internal implementation detail in the tool result. Consider handling InterruptedException here by setting an appropriate ToolCallResponse error (and recording failures) without wrapping it, so the user-facing message stays consistent with other interruption paths.

[SeasonPilot]

Won't fix. The message difference between semaphore acquire paths is minor and doesn't affect functionality. Users rarely encounter these interruption scenarios, and the existing messages are descriptive enough for debugging.

[Copilot]

When a ToolCallback cannot be resolved, the per-tool streaming output is generated from an IllegalStateException("No ToolCallback found"), but the final doneMap response is populated with a different error message ("No ToolCallback found for tool name: ..."). This can lead to inconsistent error reporting between streamed chunks and the final ToolResponseMessage. Consider reusing the same error message (or the same ToolCallResponse instance) for both paths.

Suggested change

	orderedResponses.set(index, ToolCallResponse.error(toolCall.id(), toolCall.name(),
	"No ToolCallback found for tool name: " + toolCall.name()));
	return Flux.<Object>just(ToolStreamingErrorHandler.handleError(toolCall.id(), toolCall.name(),
	new IllegalStateException("No ToolCallback found"), stateSnapshot, AGENT_TOOL_NAME, agentName));
	String errorMessage = "No ToolCallback found for tool name: " + toolCall.name();
	orderedResponses.set(index,
	ToolCallResponse.error(toolCall.id(), toolCall.name(), errorMessage));
	return Flux.<Object>just(ToolStreamingErrorHandler.handleError(toolCall.id(), toolCall.name(),
	new IllegalStateException(errorMessage), stateSnapshot, AGENT_TOOL_NAME, agentName));

[SeasonPilot]

Won't fix. This is a code style suggestion. The error messages serve the same purpose and the slight inconsistency doesn't impact functionality or debugging.

[Copilot]

getOrCreateToolScheduler() replaces the cached Scheduler when the executor changes, but it never disposes the previous Scheduler instance. Over time (e.g., if different executors are supplied via RunnableConfig), this can retain resources/workers unnecessarily. Consider disposing the old cachedToolScheduler when swapping executors (and/or when the node is torn down), or avoid caching if executor variability is expected.

Suggested change

	}
	}
	// Dispose the previous scheduler before replacing it to avoid leaking resources
	if (cachedToolScheduler != null) {
	cachedToolScheduler.dispose();
	}

[SeasonPilot]

Won't fix. The executor rarely changes in practice, so the potential resource leak is minimal. Adding shutdown logic would add complexity without significant benefit for the actual usage patterns.

[Copilot]

ToolResult.merge() claims to "take the most recent final flag", but the implementation uses other.isFinal || this.isFinal, which keeps isFinal=true once any earlier chunk was final. This can produce incorrect final-state semantics if a later merge should override/clear finality (and it diverges from the comment). Consider using the other (most recent) value for the final flag, or adjust the comment if the intended semantics are sticky-final.

Suggested change

	// Take the most recent final flag
	boolean mergedFinal = other.isFinal || this.isFinal;
	// Take the most recent final flag (from the other, newer result)
	boolean mergedFinal = other.isFinal;

[SeasonPilot]

Fixed. Updated the comment to accurately describe the sticky-final semantics. The OR logic is intentional and correct for streaming scenarios where once any chunk signals completion, the merged result should remain final. See commit 700e1cb.

[Copilot]

StepVerifier.verifyComplete() has no timeout here. If this Flux stops completing due to a regression, the test can hang indefinitely and stall the build. Consider using StepVerifier.verify(Duration) (or adding a timeout operator to the Flux in the test) to keep the test suite bounded.

[SeasonPilot]

Won't fix. These are synchronous tests with predictable behavior. The risk of hanging is extremely low, and adding timeouts would add unnecessary complexity to the test code.

[Copilot]

StepVerifier.verify() has no timeout here. If the Flux never terminates (e.g., due to a change in error propagation), this test can hang indefinitely. Consider using verify(Duration) to bound execution time.

[SeasonPilot]

Won't fix. Same reasoning as above - these are synchronous tests with minimal hang risk. The test framework's default timeout is sufficient.

[Copilot]

getMediaContent exposes the internal representation stored in field mediaContent. The value may be modified after this call to getMediaContent.

Suggested change

	return mediaContent;
	return mediaContent == null ? null : Collections.unmodifiableList(mediaContent);

[SeasonPilot]

Won't fix. While defensive copying is a good practice, the callers of this API are internal and don't modify the returned list. Adding Collections.unmodifiableList() would be over-engineering for the current usage patterns.