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 dda10f62 [api][runtime][python] Support running a ReActAgent as a
general-purpose sub-agent (#1206)
dda10f62 is described below
commit dda10f6259caefe60b5a8beea076b855eb116e48
Author: Yi Zhang <[email protected]>
AuthorDate: Wed Oct 14 11:27:06 2026 +0800
[api][runtime][python] Support running a ReActAgent as a general-purpose
sub-agent (#1206)
Generated-by: Qoder
---
.../apache/flink/agents/api/agents/ReActAgent.java | 127 ++++--
.../flink/agents/api/agents/ReActAgentTest.java | 218 ++++++++++
.../docs/development/multi-agent/subagents.md | 203 +++++++++
docs/content/docs/development/react_agent.md | 4 +
python/flink_agents/api/agents/react_agent.py | 77 +++-
.../api/agents/tests/test_react_agent.py | 120 ++++++
.../react_agent_subagent_test.py | 344 ++++++++++++++++
.../flink_agents/runtime/flink_runner_context.py | 18 +-
.../runtime/subagent/ReActAgentSubagentTest.java | 458 +++++++++++++++++++++
9 files changed, 1515 insertions(+), 54 deletions(-)
diff --git
a/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java
b/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java
index 70a858bb..372a80d7 100644
--- a/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java
+++ b/api/src/main/java/org/apache/flink/agents/api/agents/ReActAgent.java
@@ -81,6 +81,15 @@ public class ReActAgent extends Agent {
private static final String SKILLS_SOURCES_ARGUMENT = "skills_sources";
private static final String SKILLS_RESOURCE_NAME = "_react_agent_skills";
+ /**
+ * Action config flag recording that this agent registered {@link
#DEFAULT_USER_PROMPT}, as
+ * {@code output_schema} does for {@link #DEFAULT_SCHEMA_PROMPT}. A
sub-agent's resource lookup
+ * falls back to the root agent's resources, so {@link #startAction} looks
up a built-in prompt
+ * only when the running scope's own action config declares it; otherwise
an unconfigured child
+ * would pick up a parent ReActAgent's prompt under the same name.
+ */
+ private static final String HAS_USER_PROMPT = "has_user_prompt";
+
private static final ObjectMapper mapper = new ObjectMapper();
public ReActAgent(
@@ -171,6 +180,7 @@ public class ReActAgent extends Agent {
if (prompt != null) {
this.addResource(DEFAULT_USER_PROMPT, ResourceType.PROMPT, prompt);
+ actionConfig.put(HAS_USER_PROMPT, true);
}
try {
@@ -260,21 +270,33 @@ public class ReActAgent extends Agent {
descriptor.getLanguage(), descriptor.getModule(),
descriptor.getClazz(), arguments);
}
+ /**
+ * Whether {@code input} is a map holding a single "input" entry - the one
structured shape
+ * {@link #startAction} converts without a user-supplied prompt, so a
sub-agent declared with a
+ * single "input" field needs no prompt. Keyed by equals, so a non-String
key never matches.
+ */
+ private static boolean isSingleInputMap(Object input) {
+ return input instanceof Map
+ && ((Map<?, ?>) input).size() == 1
+ && ((Map<?, ?>) input).containsKey("input");
+ }
+
public static void startAction(Event event, RunnerContext ctx) {
// Built-in events reach an action as their concrete subclass
(restored at the JSON
// boundary), so the dispatched InputEvent is cast directly.
InputEvent inputEvent = (InputEvent) event;
Object input = inputEvent.getInput();
- Prompt userPrompt;
- try {
- userPrompt = (Prompt) ctx.getResource(DEFAULT_USER_PROMPT,
ResourceType.PROMPT);
- } catch (Exception e) {
- userPrompt = null;
- }
+ Prompt userPrompt =
+ Boolean.TRUE.equals(ctx.getActionConfigValue(HAS_USER_PROMPT))
+ ? declaredPrompt(ctx, DEFAULT_USER_PROMPT)
+ : null;
List<ChatMessage> inputMessages = new ArrayList<>();
- if (ClassUtils.isPrimitiveOrWrapper(input.getClass())) {
+ // A String is not a primitive wrapper, but the documented input
contract covers it like
+ // one: without a prompt it becomes the user message, with a prompt it
fills the {input}
+ // placeholder.
+ if (input instanceof String ||
ClassUtils.isPrimitiveOrWrapper(input.getClass())) {
if (userPrompt != null) {
inputMessages =
userPrompt.formatMessages(
@@ -284,38 +306,65 @@ public class ReActAgent extends Agent {
}
} else {
if (userPrompt == null) {
- throw new RuntimeException(
- String.format(
- "The input type is %s, which is not primitive
types,"
- + " user should provide prompt to help
convert it to ChatMessage",
- input.getClass()));
- }
-
- Map<String, String> fields = new HashMap<>();
- if (input instanceof Row) {
- Row userInput = (Row) input;
- for (String name :
Objects.requireNonNull(userInput.getFieldNames(true))) {
- fields.put(name, String.valueOf(userInput.getField(name)));
- }
- } else { // regard as pojo
- ObjectMapper objectMapper = new ObjectMapper();
- try {
- fields =
mapper.readValue(objectMapper.writeValueAsString(input), Map.class);
- } catch (JsonProcessingException e) {
+ if (isSingleInputMap(input)) {
+ // The one structured shape that works without a
user-supplied prompt: a
+ // single "input" entry, the arguments of a model-driven
call against an
+ // input schema declaring only that field. The value
becomes the user message
+ // directly, mirroring the prompt-less primitive branch.
+ inputMessages.add(
+ new ChatMessage(
+ MessageRole.USER,
+ String.valueOf(((Map<?, ?>)
input).get("input"))));
+ } else {
throw new RuntimeException(
String.format(
- "Input must be primitive type, Row or
Pojo, but is %s",
+ "The input type is %s, which is not
primitive types,"
+ + " user should provide prompt to
help convert it to"
+ + " ChatMessage",
input.getClass()));
}
- }
+ } else {
+ Map<String, String> fields = new HashMap<>();
+ if (input instanceof Row) {
+ Row userInput = (Row) input;
+ for (String name :
Objects.requireNonNull(userInput.getFieldNames(true))) {
+ fields.put(name,
String.valueOf(userInput.getField(name)));
+ }
+ } else if (input instanceof Map) {
+ for (Map.Entry<?, ?> entry : ((Map<?, ?>)
input).entrySet()) {
+ fields.put(
+ String.valueOf(entry.getKey()),
String.valueOf(entry.getValue()));
+ }
+ } else { // regard as pojo
+ ObjectMapper objectMapper = new ObjectMapper();
+ try {
+ // A read into the raw Map.class ignores the declared
String values, so JSON
+ // numbers and booleans come back as Integer/Boolean.
Stringify every value
+ // like the Map branch above, otherwise Prompt#format
fails with a bare
+ // ClassCastException when it casts a placeholder
value to String.
+ Map<?, ?> pojoFields =
+
mapper.readValue(objectMapper.writeValueAsString(input), Map.class);
+ for (Map.Entry<?, ?> entry : pojoFields.entrySet()) {
+ fields.put(
+ String.valueOf(entry.getKey()),
+ String.valueOf(entry.getValue()));
+ }
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(
+ String.format(
+ "Input must be primitive type, Row,
Map or Pojo, but is %s",
+ input.getClass()));
+ }
+ }
- inputMessages = userPrompt.formatMessages(MessageRole.USER,
fields);
+ inputMessages = userPrompt.formatMessages(MessageRole.USER,
fields);
+ }
}
Object outputSchema = ctx.getActionConfigValue("output_schema");
- Prompt schemaPrompt = promptOrNull(ctx, DEFAULT_SCHEMA_PROMPT);
- if (schemaPrompt != null) {
+ if (outputSchema != null) {
+ Prompt schemaPrompt = declaredPrompt(ctx, DEFAULT_SCHEMA_PROMPT);
// A plan built before the native prompt existed has none, and
then adds no
// instruction on the native path.
Prompt instruction =
@@ -367,6 +416,24 @@ public class ReActAgent extends Agent {
}
}
+ /**
+ * Resolves a built-in prompt the running scope's action config declares.
The constructor
+ * registers the prompt together with its declaration, so a failed lookup
is a broken plan and
+ * is reported rather than silently dropping the prompt.
+ */
+ private static Prompt declaredPrompt(RunnerContext ctx, String name) {
+ try {
+ return (Prompt) ctx.getResource(name, ResourceType.PROMPT);
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ String.format(
+ "Prompt %s is declared by this agent's action
config but cannot be"
+ + " resolved.",
+ name),
+ e);
+ }
+ }
+
@Action(EventType.ChatResponseEvent)
public static void stopAction(Event event, RunnerContext ctx) {
ChatResponseEvent chatResponse = (ChatResponseEvent) event;
diff --git
a/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java
b/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java
index d4474a5c..b4561398 100644
--- a/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java
+++ b/api/src/test/java/org/apache/flink/agents/api/agents/ReActAgentTest.java
@@ -43,6 +43,7 @@ import org.apache.flink.agents.api.tools.Tool;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.types.Row;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
@@ -62,6 +63,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.tuple;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -312,6 +314,7 @@ public class ReActAgentTest {
when(ctx.getResource("_default_user_prompt", ResourceType.PROMPT))
.thenReturn((Resource) prompts.get("_default_user_prompt"));
when(ctx.getActionConfigValue("output_schema")).thenReturn(config.get("output_schema"));
+
when(ctx.getActionConfigValue("has_user_prompt")).thenReturn(config.get("has_user_prompt"));
return ctx;
}
@@ -474,6 +477,221 @@ public class ReActAgentTest {
assertThat(actual.<List<String>>getArgument("skills")).containsExactly("a",
"b");
}
+ // --- startAction input dispatch ---
+
+ @Test
+ @DisplayName("A primitive input without a prompt becomes the user message")
+ public void testPrimitiveInputWithoutPromptBecomesUserMessage() {
+ RunnerContext ctx = mock(RunnerContext.class);
+
+ ReActAgent.startAction(new InputEvent(42), ctx);
+
+ assertThat(soleUserMessage(ctx).getText()).isEqualTo("42");
+ }
+
+ @Test
+ @DisplayName("A string input without a prompt becomes the user message")
+ public void testStringInputWithoutPromptBecomesUserMessage() {
+ RunnerContext ctx = mock(RunnerContext.class);
+
+ ReActAgent.startAction(new InputEvent("analyze this review"), ctx);
+
+ assertThat(soleUserMessage(ctx).getText()).isEqualTo("analyze this
review");
+ }
+
+ @Test
+ @DisplayName("A string input with a prompt fills the {input} placeholder")
+ public void testStringInputWithPromptFillsInputPlaceholder() throws
Exception {
+ RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task:
{input}"));
+
+ ReActAgent.startAction(new InputEvent("analyze this review"), ctx);
+
+ assertThat(soleUserMessage(ctx).getText()).isEqualTo("Task: analyze
this review");
+ }
+
+ @Test
+ @DisplayName("A map input without a prompt reports that a prompt is
required")
+ public void testMapInputWithoutPromptReportsPromptRequirement() {
+ RunnerContext ctx = mock(RunnerContext.class);
+
+ assertThatThrownBy(
+ () ->
+ ReActAgent.startAction(
+ new InputEvent(Map.of("task", "analyze
this review")), ctx))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("not primitive types");
+ }
+
+ @Test
+ @DisplayName("A map input with a prompt uses the keys as placeholders")
+ public void testMapInputWithPromptUsesKeysAsPlaceholders() throws
Exception {
+ RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task: {task}"));
+
+ ReActAgent.startAction(new InputEvent(Map.of("task", "analyze this
review")), ctx);
+
+ assertThat(soleUserMessage(ctx).getText()).isEqualTo("Task: analyze
this review");
+ }
+
+ @Test
+ @DisplayName("A Row input with a prompt uses the field names as
placeholders")
+ public void testRowInputWithPromptUsesFieldNamesAsPlaceholders() throws
Exception {
+ RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task: {task}"));
+ Row row = Row.withNames();
+ row.setField("task", "analyze this review");
+
+ ReActAgent.startAction(new InputEvent(row), ctx);
+
+ assertThat(soleUserMessage(ctx).getText()).isEqualTo("Task: analyze
this review");
+ }
+
+ @Test
+ @DisplayName("A Pojo with a non-String field fills placeholders with its
stringified value")
+ public void
testPojoWithNonStringFieldFillsPlaceholdersWithStringifiedValue() throws
Exception {
+ RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task:
{count}"));
+ WithCount input = new WithCount();
+ input.count = 5;
+
+ ReActAgent.startAction(new InputEvent(input), ctx);
+
+ assertThat(soleUserMessage(ctx).getText()).isEqualTo("Task: 5");
+ }
+
+ @Test
+ @DisplayName("An input of an unsupported kind reports the accepted kinds")
+ public void testInputOfUnsupportedKindReportsAcceptedKinds() throws
Exception {
+ RunnerContext ctx = contextWithPrompt(Prompt.fromText("Task:
{input}"));
+
+ assertThatThrownBy(() -> ReActAgent.startAction(new InputEvent(new
FieldLess()), ctx))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Row, Map or Pojo")
+ .hasMessageContaining("FieldLess");
+ }
+
+ // --- built-in prompt scoping ---
+
+ @Test
+ @DisplayName("A configured agent declares its built-in prompts in the
start action config")
+ public void testConfiguredAgentDeclaresBuiltInPromptsInActionConfig() {
+ ReActAgent agent =
+ new ReActAgent(
+ chatModelDescriptor(), Prompt.fromText("Task:
{input}"), WithCount.class);
+
+ Map<String, Object> config = agent.getActions().get("startAction").f2;
+ assertThat(config).containsEntry("has_user_prompt", true);
+ assertThat(config).containsEntry("output_schema", WithCount.class);
+ }
+
+ @Test
+ @DisplayName("A bare agent declares no built-in prompt in the start action
config")
+ public void testBareAgentDeclaresNoBuiltInPrompt() {
+ Map<String, Object> config =
bareAgent().getActions().get("startAction").f2;
+
+ assertThat(config).doesNotContainKeys("has_user_prompt",
"output_schema");
+ }
+
+ @Test
+ @DisplayName("A prompt reachable but not declared by this agent is not
looked up")
+ public void testUndeclaredPromptsAreNotLookedUp() throws Exception {
+ // As a sub-agent, a lookup missing locally falls back to the root
agent's resources, so
+ // a parent ReActAgent's prompts are reachable under the same names.
+ RunnerContext ctx = mock(RunnerContext.class);
+ when(ctx.getResource("_default_user_prompt", ResourceType.PROMPT))
+ .thenReturn(Prompt.fromText("PARENT ONLY: {input}"));
+ when(ctx.getResource("_default_schema_prompt", ResourceType.PROMPT))
+ .thenReturn(Prompt.fromText("PARENT SCHEMA"));
+
+ ReActAgent.startAction(new InputEvent("review the diff"), ctx);
+
+ assertThat(soleUserMessage(ctx).getText()).isEqualTo("review the
diff");
+ verify(ctx, never()).getResource(anyString(), any());
+ }
+
+ @Test
+ @DisplayName("A declared output schema inserts the schema instruction
before the user message")
+ public void testDeclaredOutputSchemaInsertsSchemaInstruction() throws
Exception {
+ RunnerContext ctx = mock(RunnerContext.class);
+
when(ctx.getActionConfigValue("output_schema")).thenReturn(WithCount.class);
+ when(ctx.getResource("_default_schema_prompt", ResourceType.PROMPT))
+ .thenReturn(Prompt.fromText("match the schema"));
+
+ ReActAgent.startAction(new InputEvent("review the diff"), ctx);
+
+ ChatRequestEvent request = soleRequest(ctx);
+ assertThat(request.getMessages()).hasSize(2);
+
assertThat(request.getMessages().get(0).getRole()).isEqualTo(MessageRole.SYSTEM);
+ assertThat(request.getMessages().get(0).getText()).isEqualTo("match
the schema");
+ assertThat(request.getMessages().get(1).getText()).isEqualTo("review
the diff");
+ assertThat(request.getOutputSchema()).isEqualTo(WithCount.class);
+ }
+
+ @Test
+ @DisplayName("A declared prompt that cannot be resolved is reported, not
dropped")
+ public void testDeclaredPromptThatCannotBeResolvedIsReported() throws
Exception {
+ RunnerContext ctx = mock(RunnerContext.class);
+ when(ctx.getActionConfigValue("has_user_prompt")).thenReturn(true);
+ when(ctx.getResource("_default_user_prompt", ResourceType.PROMPT))
+ .thenThrow(new IllegalArgumentException("Resource not found"));
+
+ assertThatThrownBy(() -> ReActAgent.startAction(new InputEvent("review
the diff"), ctx))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("_default_user_prompt")
+ .hasCauseInstanceOf(IllegalArgumentException.class);
+ }
+
+ // --- single-input map ---
+
+ @Test
+ @DisplayName("A single-input map without a prompt passes the value
through")
+ public void testSingleInputMapWithoutPromptBecomesUserMessage() {
+ RunnerContext ctx = mock(RunnerContext.class);
+
+ ReActAgent.startAction(new InputEvent(Map.of("input", "analyze this
review")), ctx);
+
+ assertThat(soleUserMessage(ctx).getText()).isEqualTo("analyze this
review");
+ }
+
+ @Test
+ @DisplayName("A multi-key map without a prompt still reports that a prompt
is required")
+ public void testMultiKeyMapWithoutPromptStillReportsPromptRequirement() {
+ RunnerContext ctx = mock(RunnerContext.class);
+
+ assertThatThrownBy(
+ () ->
+ ReActAgent.startAction(
+ new InputEvent(Map.of("input", "x",
"other", "y")), ctx))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("not primitive types");
+ }
+
+ private static ReActAgent bareAgent() {
+ return new ReActAgent(chatModelDescriptor(), null, null);
+ }
+
+ private static ResourceDescriptor chatModelDescriptor() {
+ return
ResourceDescriptor.Builder.newBuilder("com.example.ChatModel").build();
+ }
+
+ private static RunnerContext contextWithPrompt(Prompt prompt) throws
Exception {
+ RunnerContext ctx = mock(RunnerContext.class);
+ when(ctx.getActionConfigValue("has_user_prompt")).thenReturn(true);
+ when(ctx.getResource("_default_user_prompt",
ResourceType.PROMPT)).thenReturn(prompt);
+ return ctx;
+ }
+
+ private static ChatRequestEvent soleRequest(RunnerContext ctx) {
+ ArgumentCaptor<Event> captor = ArgumentCaptor.forClass(Event.class);
+ verify(ctx).sendEvent(captor.capture());
+ return (ChatRequestEvent) captor.getValue();
+ }
+
+ private static ChatMessage soleUserMessage(RunnerContext ctx) {
+ ChatRequestEvent request = soleRequest(ctx);
+ assertThat(request.getMessages()).hasSize(1);
+ ChatMessage message = request.getMessages().get(0);
+ assertThat(message.getRole()).isEqualTo(MessageRole.USER);
+ return message;
+ }
+
private static ReActAgent agentWithSchema(Object outputSchema) {
return agentWithSchema(outputSchema, null);
}
diff --git a/docs/content/docs/development/multi-agent/subagents.md
b/docs/content/docs/development/multi-agent/subagents.md
index 86e61089..0871c679 100644
--- a/docs/content/docs/development/multi-agent/subagents.md
+++ b/docs/content/docs/development/multi-agent/subagents.md
@@ -94,6 +94,10 @@ This shorthand does not declare an input schema. Use the
builder when exposing
the child to a model. Recursive registration, such as a child registering its
parent, is rejected when the plan is built.
+Register the resources a child uses, such as its chat model connection, on the
+child itself. In Python, a child resolves resources against its own plan only,
+so a resource registered solely on the parent is not visible to the child.
+
## Declare an external sub-agent
Implement `SubagentSetup` (or the appropriate runtime base class) and register
a
@@ -131,3 +135,202 @@ The Python setup declares these configuration fields and
implements the
invocation contract. Both languages construct the setup at runtime from the
descriptor. Existing code that passes a setup instance to `addResource` or
`add_resource` must instead pass a descriptor with the same configuration.
+
+## Delegate from a chat model
+
+List the child's registered name under `subagents` of the parent's chat model:
+
+```java
+ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
+ .addInitialArgument("connection", "ollamaConnection")
+ .addInitialArgument("model", "qwen3:8b")
+ .addInitialArgument("subagents", List.of("reviewer"))
+ .build();
+```
+
+```python
+ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP,
+ connection="ollama_connection",
+ model="qwen3:8b",
+ subagents=["reviewer"],
+)
+```
+
+The model sees the child as a tool named `_subagent_reviewer`, described by the
+registration's description and input schema. The arguments the model sends
+become the child's input, and the child's outputs return to the model as the
+tool observation, so it can continue reasoning over them. For an internal
+sub-agent, the observation is the JSON encoding of the full output list
+described in [Call from an action](#call-from-an-action), even when the child
+emits a single output, so its shape does not depend on how many outputs the
+child emits.
+
+## Call from an action
+
+An action can submit input to a child directly and await its outcome. This path
+does not need an input schema.
+
+```java
+public class DelegatingAgent extends Agent {
+
+ @Action(EventType.InputEvent)
+ public static void delegate(Event event, RunnerContext ctx) throws
Exception {
+ SubagentSetup child =
+ (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+ SubagentResult result = child.submit(ctx, "review the diff").await();
+ if (result.isSuccess()) {
+ String answer = String.valueOf(((List<?>)
result.getResult()).get(0));
+ ctx.sendEvent(new OutputEvent(answer));
+ } else {
+ ctx.sendEvent(new OutputEvent("failed: " +
result.getErrorMessage()));
+ }
+ }
+}
+```
+
+```python
+class DelegatingAgent(Agent):
+ @action(InputEvent.EVENT_TYPE)
+ @staticmethod
+ async def delegate(event: Event, ctx: RunnerContext) -> None:
+ child = ctx.get_resource("reviewer", ResourceType.AGENT)
+ future = await child.submit(ctx, "review the diff")
+ result = await future
+ if result.success:
+ ctx.send_event(OutputEvent(output=result.result[0]))
+ else:
+ ctx.send_event(OutputEvent(output=f"failed:
{result.error_message}"))
+```
+
+The awaited value is a `SubagentResult`. On success, its `result` field
+(`getResult()` in Java) carries the child's result. For an internal sub-agent,
+such as `reviewer` here, that is the list of outputs the child emitted, which
is
+why the example takes the first element. An external sub-agent returns whatever
+its `SubagentSetup` implementation produces, such as a string or an object. On
+failure, `success` is `False` (`isSuccess()` is `false` in Java), and
+`error_message` (`getErrorMessage()`) carries the reason.
+
+## Use a ReActAgent as the child
+
+A ReActAgent runs as a child like any other agent, keeping its own event loop,
+resources and conversation. It uses only its own prompt and output schema, even
+when the parent is also a ReActAgent.
+
+```java
+ReActAgent reviewer =
+ new ReActAgent(
+
ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
+ .addInitialArgument("connection", "reviewerConnection")
+ .addInitialArgument("model", "qwen3:8b")
+ .build(),
+ null,
+ null);
+reviewer.addResource(
+ "reviewerConnection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+
ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_CONNECTION)
+ .build());
+
+parent.addResource(
+ "reviewer",
+ ResourceType.AGENT,
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(reviewer)
+ .description("Reviews a code diff and reports problems.")
+ .inputSchema(
+
"{\"type\":\"object\",\"properties\":{\"input\":{\"type\":\"string\"}},"
+ + "\"required\":[\"input\"]}")
+ .build());
+```
+
+```python
+reviewer = ReActAgent(
+ chat_model=ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP,
+ connection="reviewer_connection",
+ model="qwen3:8b",
+ ),
+)
+reviewer.add_resource(
+ "reviewer_connection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+ ResourceDescriptor(clazz=ResourceName.ChatModel.OLLAMA_CONNECTION),
+)
+
+parent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder()
+ .agent(reviewer)
+ .description("Reviews a code diff and reports problems.")
+ .input_schema(
+ '{"type":"object","properties":{"input":{"type":"string"}},'
+ '"required":["input"]}'
+ )
+ .build(),
+)
+```
+
+Called from an action, the child takes a string input as its user message, or
+uses it to fill the `{input}` placeholder of its prompt. Delegated to by a
+model, the child receives the arguments as a map keyed by the input schema's
+field names and turns that map into its user message, so the schema and the
+child's prompt form one contract:
+
+- **With a prompt**, the schema's field names must match the prompt's
+ placeholders: each argument fills the placeholder of the same name, as
+ described in [Prompt]({{< ref "docs/development/react_agent#prompt" >}}).
+- **Without a prompt**, declare a single `input` field, as in the example
above.
+ A map holding only `input` passes its value through as the user message; any
+ other map is rejected because there is no prompt to format it.
+
+A child with a prompt declares the fields its placeholders use:
+
+```java
+ReActAgent reviewer =
+ new ReActAgent(
+ chatModelDescriptor,
+ Prompt.fromText(
+ "Review the following code diff and report any
problems:\n\n{diff}"),
+ null);
+
+parent.addResource(
+ "reviewer",
+ ResourceType.AGENT,
+ new InternalSubagentResourceDescriptorBuilder()
+ .agent(reviewer)
+ .description("Reviews a code diff and reports problems.")
+ // "diff" matches the {diff} placeholder of the child's prompt.
+ .inputSchema(
+
"{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}},"
+ + "\"required\":[\"diff\"]}")
+ .build());
+```
+
+```python
+reviewer = ReActAgent(
+ chat_model=chat_model_descriptor,
+ prompt=Prompt.from_text(
+ "Review the following code diff and report any problems:\n\n{diff}"
+ ),
+)
+
+parent.add_resource(
+ "reviewer",
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder()
+ .agent(reviewer)
+ .description("Reviews a code diff and reports problems.")
+ # "diff" matches the {diff} placeholder of the child's prompt.
+ .input_schema(
+ '{"type":"object","properties":{"diff":{"type":"string"}},'
+ '"required":["diff"]}'
+ )
+ .build(),
+)
+```
+
+The framework does not check the schema against the prompt. A field the prompt
+does not use is ignored, and a placeholder no field fills is left in the prompt
+as literal text, so keep the two in step when changing either.
diff --git a/docs/content/docs/development/react_agent.md
b/docs/content/docs/development/react_agent.md
index 32ae5518..ba7b68ca 100644
--- a/docs/content/docs/development/react_agent.md
+++ b/docs/content/docs/development/react_agent.md
@@ -399,3 +399,7 @@ The constructor registers those sources on the agent, so
they do not also need `
**Which declaration wins.** Sources reach the agent from three levels.
`@Skills`/`@skills` methods are merged first, then the execution environment's
`SKILLS` resources, then the agent's own. A declaration from a later level wins
when two of them resolve to the same skill name.
+## As a Sub-Agent
+
+A ReActAgent can run as a sub-agent of another agent, delegated to by the
parent's chat model or called from the parent's actions. See [Use a ReActAgent
as the child]({{< ref
"docs/development/multi-agent/subagents#use-a-reactagent-as-the-child" >}}).
+
diff --git a/python/flink_agents/api/agents/react_agent.py
b/python/flink_agents/api/agents/react_agent.py
index d1661a87..c7052a94 100644
--- a/python/flink_agents/api/agents/react_agent.py
+++ b/python/flink_agents/api/agents/react_agent.py
@@ -52,6 +52,12 @@ _DEFAULT_CHAT_MODEL = "_default_chat_model"
_DEFAULT_SCHEMA_PROMPT = "_default_schema_prompt"
_DEFAULT_NATIVE_SCHEMA_PROMPT = "_default_native_schema_prompt"
_DEFAULT_USER_PROMPT = "_default_user_prompt"
+# Action config flag recording that the agent registered _DEFAULT_USER_PROMPT,
+# as output_schema does for _DEFAULT_SCHEMA_PROMPT. start_action looks up a
+# built-in prompt only when the running scope's own action config declares it,
+# keeping a sub-agent's prompts its own on a runtime whose child scope falls
+# back to the root agent's resources.
+_HAS_USER_PROMPT = "has_user_prompt"
_OUTPUT_SCHEMA = "_output_schema"
_SKILLS_ARGUMENT = "skills"
_SKILLS_SOURCES_ARGUMENT = "skills_sources"
@@ -252,6 +258,7 @@ class ReActAgent(Agent):
output_schema=OutputSchema(output_schema=output_schema)
if output_schema
else None,
+ has_user_prompt=bool(prompt),
)
@staticmethod
@@ -261,12 +268,11 @@ class ReActAgent(Agent):
# at the JSON boundary), so the dispatched InputEvent is used directly.
usr_input = cast("InputEvent", event).input
- try:
- prompt = cast(
- "Prompt", ctx.get_resource(_DEFAULT_USER_PROMPT,
ResourceType.PROMPT)
- )
- except KeyError:
- prompt = None
+ prompt = (
+ _declared_prompt(ctx, _DEFAULT_USER_PROMPT)
+ if ctx.get_action_config_value(key=_HAS_USER_PROMPT)
+ else None
+ )
if isinstance(usr_input, bool | str | int | float | type(None)):
usr_input = str(usr_input)
@@ -278,26 +284,36 @@ class ReActAgent(Agent):
usr_msgs = [ChatMessage.user(usr_input)]
else:
if not prompt:
- err_msg = (
- f"Input type is {usr_input.__class__}, which is not
primitive types. "
- f"User should provide prompt to help convert it to
ChatMessage."
+ if isinstance(usr_input, dict) and set(usr_input.keys()) ==
{"input"}:
+ # The one structured shape that works without a
user-supplied
+ # prompt: a single "input" entry, the arguments of a
model-driven
+ # call against an input schema declaring only that field.
The
+ # value becomes the user message directly, mirroring the
+ # prompt-less primitive branch.
+ usr_msgs = [ChatMessage.user(str(usr_input["input"]))]
+ else:
+ err_msg = (
+ f"Input type is {usr_input.__class__}, which is not
primitive types. "
+ f"User should provide prompt to help convert it to
ChatMessage."
+ )
+ raise RuntimeError(err_msg)
+ else:
+ if isinstance(usr_input, Row):
+ usr_input = usr_input.as_dict(recursive=True)
+ elif isinstance(usr_input, dict):
+ pass
+ else: # regard as pojo
+ usr_input = usr_input.__dict__
+ # Convert Any values to str to match format_messages signature
+ str_usr_input = {k: str(v) for k, v in usr_input.items()}
+ usr_msgs = prompt.format_messages(
+ role=MessageRole.USER, **str_usr_input
)
- raise RuntimeError(err_msg)
- if isinstance(usr_input, Row):
- usr_input = usr_input.as_dict(recursive=True)
- elif isinstance(usr_input, dict):
- pass
- else: # regard as pojo
- usr_input = usr_input.__dict__
- # Convert Any values to str to match format_messages signature
- str_usr_input = {k: str(v) for k, v in usr_input.items()}
- usr_msgs = prompt.format_messages(role=MessageRole.USER,
**str_usr_input)
-
- schema_prompt = ReActAgent._prompt_or_none(ctx, _DEFAULT_SCHEMA_PROMPT)
output_schema = ctx.get_action_config_value(key="output_schema")
- if schema_prompt is not None:
+ if output_schema is not None:
+ schema_prompt = _declared_prompt(ctx, _DEFAULT_SCHEMA_PROMPT)
# A plan built before the native prompt existed has none, and then
adds
# no instruction on the native path.
instruction = (
@@ -364,3 +380,20 @@ class ReActAgent(Agent):
output = response.text
ctx.send_event(OutputEvent(output=output))
+
+
+def _declared_prompt(ctx: RunnerContext, name: str) -> Prompt:
+ """Resolve a built-in prompt the running scope's action config declares.
+
+ The constructor registers the prompt together with its declaration, so a
+ failed lookup is a broken plan and is reported rather than silently
dropping
+ the prompt.
+ """
+ try:
+ return cast("Prompt", ctx.get_resource(name, ResourceType.PROMPT))
+ except KeyError as e:
+ err_msg = (
+ f"Prompt {name} is declared by this agent's action config but
cannot "
+ f"be resolved."
+ )
+ raise RuntimeError(err_msg) from e
diff --git a/python/flink_agents/api/agents/tests/test_react_agent.py
b/python/flink_agents/api/agents/tests/test_react_agent.py
index 3435a47b..0fc71a44 100644
--- a/python/flink_agents/api/agents/tests/test_react_agent.py
+++ b/python/flink_agents/api/agents/tests/test_react_agent.py
@@ -16,6 +16,7 @@
# limitations under the License.
#################################################################################
from typing import Any, Callable
+from unittest.mock import MagicMock
import pytest
from pydantic import BaseModel, Field
@@ -27,6 +28,9 @@ from flink_agents.api.agents.react_agent import (
_DEFAULT_SCHEMA_PROMPT,
ReActAgent,
)
+from flink_agents.api.chat_message import MessageRole
+from flink_agents.api.events.event import InputEvent
+from flink_agents.api.prompts.prompt import Prompt
from flink_agents.api.resource import (
JavaResourceDescriptor,
ResourceDescriptor,
@@ -94,6 +98,122 @@ def _expected_prompt(rendered: Any) -> str:
return f"The final response should be json format, and match the schema
{rendered}."
+def _bare_agent() -> ReActAgent:
+ """A ReActAgent with neither a prompt nor an output schema."""
+ return ReActAgent(
+ chat_model=ResourceDescriptor(clazz=_CHAT_MODEL_CLASS,
model="qwen3:8b")
+ )
+
+
+def _context_without_prompt() -> tuple[MagicMock, list]:
+ """A mock context owning no prompt resource, collecting sent events."""
+ ctx = MagicMock()
+ ctx.get_resource = MagicMock(side_effect=KeyError("no such resource"))
+ ctx.get_action_config_value = MagicMock(return_value=None)
+ sent_events: list = []
+ ctx.send_event = MagicMock(side_effect=lambda e: sent_events.append(e))
+ return ctx, sent_events
+
+
+def _sole_user_text(sent_events: list) -> str:
+ """The text of the single user message of the emitted chat request."""
+ assert len(sent_events) == 1
+ messages = sent_events[0].messages
+ assert len(messages) == 1
+ assert messages[0].role == MessageRole.USER
+ return messages[0].text
+
+
+def _start_action_config(agent: ReActAgent) -> dict:
+ """The action config the agent registered for its start action."""
+ return agent._actions["start_action"][2]
+
+
+# --- built-in prompt scoping ---
+
+
+def test_configured_agent_declares_built_in_prompts_in_action_config() -> None:
+ """A configured agent declares its built-in prompts in the start action
config."""
+ agent = ReActAgent(
+ chat_model=ResourceDescriptor(clazz=_CHAT_MODEL_CLASS,
model="qwen3:8b"),
+ prompt=Prompt.from_text("Task: {input}"),
+ output_schema=Person,
+ )
+
+ config = _start_action_config(agent)
+ assert config["has_user_prompt"] is True
+ assert config["output_schema"] is not None
+
+
+def test_bare_agent_declares_no_built_in_prompt() -> None:
+ """A bare agent declares no built-in prompt in the start action config."""
+ config = _start_action_config(_bare_agent())
+
+ assert config["has_user_prompt"] is False
+ assert config["output_schema"] is None
+
+
+def test_undeclared_prompts_are_not_looked_up() -> None:
+ """A prompt reachable but not declared by this agent is not looked up."""
+ ctx, sent_events = _context_without_prompt()
+ # A runtime whose child scope falls back to the root agent's resources
makes a
+ # parent ReActAgent's prompts reachable under the same names.
+ ctx.get_resource = MagicMock(return_value=Prompt.from_text("PARENT ONLY:
{input}"))
+
+ ReActAgent.start_action(InputEvent(input="review the diff"), ctx)
+
+ assert _sole_user_text(sent_events) == "review the diff"
+ ctx.get_resource.assert_not_called()
+
+
+def test_declared_output_schema_inserts_schema_instruction() -> None:
+ """A declared output schema inserts the schema instruction before the user
message."""
+ ctx, sent_events = _context_without_prompt()
+ output_schema = _start_action_config(_agent(Person))["output_schema"]
+ ctx.get_action_config_value = MagicMock(
+ side_effect=lambda key: output_schema if key == "output_schema" else
None
+ )
+ ctx.get_resource = MagicMock(return_value=Prompt.from_text("match the
schema"))
+
+ ReActAgent.start_action(InputEvent(input="review the diff"), ctx)
+
+ messages = sent_events[0].messages
+ assert [m.text for m in messages] == ["match the schema", "review the
diff"]
+ assert messages[1].role == MessageRole.USER
+ ctx.get_resource.assert_any_call(_DEFAULT_SCHEMA_PROMPT,
ResourceType.PROMPT)
+
+
+def test_declared_prompt_that_cannot_be_resolved_is_reported() -> None:
+ """A declared prompt that cannot be resolved is reported, not dropped."""
+ ctx, _ = _context_without_prompt()
+ ctx.get_action_config_value = MagicMock(
+ side_effect=lambda key: key == "has_user_prompt" or None
+ )
+
+ with pytest.raises(RuntimeError, match="_default_user_prompt"):
+ ReActAgent.start_action(InputEvent(input="review the diff"), ctx)
+
+
+# --- start_action input dispatch ---
+
+
+def test_single_input_map_without_prompt_becomes_user_message() -> None:
+ """A single-input map without a prompt passes the value through."""
+ ctx, sent_events = _context_without_prompt()
+
+ ReActAgent.start_action(InputEvent(input={"input": "analyze this
review"}), ctx)
+
+ assert _sole_user_text(sent_events) == "analyze this review"
+
+
+def test_multi_key_map_without_prompt_still_reports_prompt_requirement() ->
None:
+ """A multi-key map without a prompt still reports that a prompt is
required."""
+ ctx, _ = _context_without_prompt()
+
+ with pytest.raises(RuntimeError, match="not primitive types"):
+ ReActAgent.start_action(InputEvent(input={"input": "x", "other":
"y"}), ctx)
+
+
def test_unrenderable_output_schema_raises_naming_the_model() -> None:
"""A schema that cannot be rendered fails at construction, not at the
provider."""
with pytest.raises(TypeError, match="Unrenderable cannot be rendered"):
diff --git
a/python/flink_agents/e2e_tests/e2e_tests_integration/react_agent_subagent_test.py
b/python/flink_agents/e2e_tests/e2e_tests_integration/react_agent_subagent_test.py
new file mode 100644
index 00000000..c7e8681e
--- /dev/null
+++
b/python/flink_agents/e2e_tests/e2e_tests_integration/react_agent_subagent_test.py
@@ -0,0 +1,344 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# 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.
+################################################################################
+"""E2E tests for a real ReActAgent running as an internal sub-agent.
+
+The children here are the built-in agent itself, not hand-written echo agents:
+its start action, the framework's chat model action and its stop action must
+all run inside the child scope, driven by scripted connections instead of a
+live model. The model-driven test additionally runs the whole delegation loop:
+the parent model issues a ``_subagent_child`` tool call, the framework routes
+it to the child, and the child's answer flows back as the tool observation
+that closes the loop.
+
+The delegated child is registered with a single-field ``input`` schema, so
+that test also proves the declared schema survives compilation and the
+child's prompt-free ``{input}`` map path accepts the model's arguments as-is.
+
+The third test lets the child declare a tool of its own and scripts two model
+rounds for it: the first reply asks for the tool, the tool executes inside
+the child scope, and the second round sees the tool result, closing the
+reasoning/tool loop in isolation.
+"""
+
+import os
+import sysconfig
+from pathlib import Path
+from typing import Any
+
+from pyflink.common import Encoder
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import KeySelector, StreamExecutionEnvironment
+from pyflink.datastream.connectors.file_system import StreamingFileSink
+
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.agents.react_agent import ReActAgent
+from flink_agents.api.chat_message import (
+ ChatMessage,
+ MessageRole,
+ TextBlock,
+ ToolCallBlock,
+ ToolResultBlock,
+)
+from flink_agents.api.chat_models.chat_model import (
+ BaseChatModelConnection,
+ BaseChatModelSetup,
+)
+from flink_agents.api.chat_result import ChatResult
+from flink_agents.api.decorators import action
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+from flink_agents.api.resource import ResourceDescriptor, ResourceType
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import InternalSubagentResourceDescriptorBuilder
+from flink_agents.api.tools.tool import Tool
+
+os.environ["PYTHONPATH"] = sysconfig.get_paths()["purelib"]
+
+CHILD_SCOPE = "child"
+
+
+class InputKeySelector(KeySelector):
+ """Keys every element by itself."""
+
+ def get_key(self, value: Any) -> Any:
+ """Return the element itself as key."""
+ return value
+
+
+class ScriptedModelSetup(BaseChatModelSetup):
+ """A setup whose behavior lives entirely in its scripted connection."""
+
+ @property
+ def model_kwargs(self) -> dict[str, Any]:
+ """Return chat model settings."""
+ return {}
+
+
+class ParentScriptedConnection(BaseChatModelConnection):
+ """Delegates once, then answers with the observed child output."""
+
+ def chat(
+ self,
+ messages: Any,
+ tools: list | None = None,
+ output_schema: Any = None,
+ **kwargs: Any,
+ ) -> ChatResult:
+ """Issue the delegation tool call, then echo the observation back."""
+ self._reject_unsupported_output_schema(output_schema)
+ tool_messages = [m for m in messages if m.role == MessageRole.TOOL]
+ if not tool_messages:
+ # The child is declared with a single "input" field, so the model
+ # builds the one argument the agent's prompt-free path accepts.
+ tool_call = ToolCallBlock(
+ call_id="call-1",
+ name="_subagent_child",
+ input={"input": "review the diff"},
+ )
+ return ChatResult(
+ message=ChatMessage.assistant(
+ [TextBlock(text="delegating"), tool_call]
+ )
+ )
+ result = tool_messages[0].blocks[0]
+ assert isinstance(result, ToolResultBlock)
+ return ChatResult(
+ message=ChatMessage.assistant(f"observed: {result.text}")
+ )
+
+
+class ChildScriptedConnection(BaseChatModelConnection):
+ """Echoes the last user message, revealing what the child agent
received."""
+
+ def chat(
+ self,
+ messages: Any,
+ tools: list | None = None,
+ output_schema: Any = None,
+ **kwargs: Any,
+ ) -> ChatResult:
+ """Answer with the child's own user message."""
+ self._reject_unsupported_output_schema(output_schema)
+ return ChatResult(
+ message=ChatMessage.assistant(f"child saw: {messages[-1].text}")
+ )
+
+
+def shout(text: str) -> str:
+ """Return the text in upper case, marked as tool-produced."""
+ return f"{text.upper()}!"
+
+
+class ToolCallingChildConnection(BaseChatModelConnection):
+ """Asks for the child's tool first, then answers from the tool result."""
+
+ def chat(
+ self,
+ messages: Any,
+ tools: list | None = None,
+ output_schema: Any = None,
+ **kwargs: Any,
+ ) -> ChatResult:
+ """Emit the tool call on the first round, consume the result on the
second."""
+ self._reject_unsupported_output_schema(output_schema)
+ tool_messages = [m for m in messages if m.role == MessageRole.TOOL]
+ if not tool_messages:
+ tool_call = ToolCallBlock(
+ call_id="call-1", name="shout", input={"text": "hello world"}
+ )
+ return ChatResult(
+ message=ChatMessage.assistant(
+ [TextBlock(text="let me check"), tool_call]
+ )
+ )
+ result = tool_messages[0].blocks[0]
+ assert isinstance(result, ToolResultBlock)
+ return ChatResult(
+ message=ChatMessage.assistant(f"tool says: {result.text}")
+ )
+
+
+def _child_react_agent() -> ReActAgent:
+ """A ReActAgent with neither a prompt nor an output schema.
+
+ The connection is registered on the child itself, keeping the child plan
+ self-contained: a child scope resolves resources against the child plan
+ only, so the connection must live there.
+ """
+ child = ReActAgent(
+ chat_model=ResourceDescriptor(
+
clazz=f"{ScriptedModelSetup.__module__}.{ScriptedModelSetup.__name__}",
+ connection="child_connection",
+ model="mock-model",
+ ),
+ )
+ child.add_resource(
+ "child_connection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+ ResourceDescriptor(
+
clazz=f"{ChildScriptedConnection.__module__}.{ChildScriptedConnection.__name__}",
+ ),
+ )
+ return child
+
+
+def _tool_calling_child_react_agent() -> ReActAgent:
+ """A ReActAgent whose model asks for the child's own tool across two
rounds."""
+ child = ReActAgent(
+ chat_model=ResourceDescriptor(
+
clazz=f"{ScriptedModelSetup.__module__}.{ScriptedModelSetup.__name__}",
+ connection="tool_child_connection",
+ model="mock-model",
+ tools=["shout"],
+ ),
+ )
+ child.add_resource(
+ "tool_child_connection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+ ResourceDescriptor(
+
clazz=f"{ToolCallingChildConnection.__module__}.{ToolCallingChildConnection.__name__}",
+ ),
+ )
+ child.add_resource("shout", ResourceType.TOOL, Tool.from_callable(shout))
+ return child
+
+
+class CallerRootAgent(Agent):
+ """Caller agent awaiting the ReAct child from an async action."""
+
+ @action(InputEvent.EVENT_TYPE)
+ @staticmethod
+ async def orchestrate(event: Event, ctx: RunnerContext) -> None:
+ """Await the child and emit its outcome."""
+ child = ctx.get_resource(CHILD_SCOPE, ResourceType.AGENT)
+ future = await child.submit(ctx, "review the diff")
+ result = await future
+ if result.success:
+ ctx.send_event(OutputEvent(output=result.result[0]))
+ else:
+ msg = f"failed:{result.error_message}"
+ ctx.send_event(OutputEvent(output=msg))
+
+
+def _delegating_parent() -> ReActAgent:
+ """A bare ReActAgent parent that declares the child to its model.
+
+ Sub-classing ReActAgent would trip the plan compiler, which rejects
+ inherited ``@action`` methods on a concrete agent, so the parent is a
+ plain instance carrying instance-level resources instead.
+ """
+ parent = ReActAgent(
+ chat_model=ResourceDescriptor(
+
clazz=f"{ScriptedModelSetup.__module__}.{ScriptedModelSetup.__name__}",
+ connection="parent_connection",
+ model="mock-model",
+ subagents=[CHILD_SCOPE],
+ ),
+ )
+ parent.add_resource(
+ "parent_connection",
+ ResourceType.CHAT_MODEL_CONNECTION,
+ ResourceDescriptor(
+
clazz=f"{ParentScriptedConnection.__module__}.{ParentScriptedConnection.__name__}",
+ ),
+ )
+ # Only a declared input schema lets the parent model call the child.
+ parent.add_resource(
+ CHILD_SCOPE,
+ ResourceType.AGENT,
+ InternalSubagentResourceDescriptorBuilder()
+ .agent(_child_react_agent())
+ .description("Reviews a code diff.")
+ .input_schema(
+ '{"type":"object","properties":{"input":{"type":"string"}},'
+ '"required":["input"]}'
+ )
+ .build(),
+ )
+ return parent
+
+
+def _run_and_collect(root: Agent, tmp_path: Path) -> str:
+ """Run the root agent over one element and collect what the job emitted."""
+ env = StreamExecutionEnvironment.get_execution_environment()
+ env.set_parallelism(1)
+ input_stream = env.from_collection(["hello"])
+
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ output_datastream = (
+ agents_env.from_datastream(input=input_stream,
key_selector=InputKeySelector())
+ .apply(root)
+ .to_datastream()
+ )
+
+ result_dir = tmp_path / "results"
+ result_dir.mkdir(parents=True, exist_ok=True)
+ output_datastream.map(str, Types.STRING()).add_sink(
+ StreamingFileSink.for_row_format(
+ base_path=str(result_dir.absolute()),
+ encoder=Encoder.simple_string_encoder(),
+ ).build()
+ )
+ agents_env.execute()
+ return "".join(p.read_text() for p in result_dir.rglob("*") if p.is_file())
+
+
+def test_caller_driven_react_subagent(tmp_path: Path) -> None:
+ """A submitted string runs the child's ReAct loop and returns its
answer."""
+ root = CallerRootAgent()
+ root.add_resource(CHILD_SCOPE, ResourceType.AGENT, _child_react_agent())
+
+ contents = _run_and_collect(root, tmp_path)
+
+ # The child connection saw the submitted string as its user message, which
is
+ # the prompt-free primitive path, and the child's reply is the call result.
+ assert "child saw: review the diff" in contents, (
+ f"missing child output; collected: {contents!r}"
+ )
+
+
+def test_model_driven_react_subagent_delegation(tmp_path: Path) -> None:
+ """A model-issued delegation reaches the declared child and closes the
loop."""
+ contents = _run_and_collect(_delegating_parent(), tmp_path)
+
+ # The final answer echoes the tool observation, proving the whole loop: the
+ # model's {"input": ...} arguments became the child's user message through
+ # the prompt-free {input} map path, and the child's reply came back as the
+ # observation. A child's own output never reaches the sink directly -- only
+ # this parent answer does -- so both fragments here mean the round-trip
+ # completed; the assertion stays split because the observation carries the
+ # result in the framework's string form, not verbatim.
+ assert "observed:" in contents, f"no observation echoed; collected:
{contents!r}"
+ assert "child saw: review the diff" in contents, (
+ f"child output missing from the loop; collected: {contents!r}"
+ )
+
+
+def test_child_tool_call_loop_runs_inside_the_child_scope(tmp_path: Path) ->
None:
+ """The child's own tool executes inside the child scope across two
rounds."""
+ root = CallerRootAgent()
+ root.add_resource(
+ CHILD_SCOPE, ResourceType.AGENT, _tool_calling_child_react_agent()
+ )
+
+ contents = _run_and_collect(root, tmp_path)
+
+ # The tool result flowed back into the child's second model round, which
+ # answered from it; the child's tool never touches the parent's resources.
+ assert "tool says: HELLO WORLD!" in contents, (
+ f"tool loop output missing; collected: {contents!r}"
+ )
diff --git a/python/flink_agents/runtime/flink_runner_context.py
b/python/flink_agents/runtime/flink_runner_context.py
index 23e0654c..21485641 100644
--- a/python/flink_agents/runtime/flink_runner_context.py
+++ b/python/flink_agents/runtime/flink_runner_context.py
@@ -767,6 +767,20 @@ class FlinkRunnerContext(RunnerContext, ExecutionReporter):
self.__scoped_resource_caches[plan_json] = entry
return entry
+ def __active_plan(self) -> Any:
+ """The plan in effect: the child plan's while inside a sub-agent call,
else
+ the root plan's.
+
+ Mirrors the Java context's ``currentPlan()``: action-config lookups
must
+ resolve against the plan whose action is executing, so a child agent's
+ action reads its own config instead of failing on the root plan, which
+ does not carry that action.
+ """
+ plan_json = self._j_runner_context.getActiveScopePlanJson()
+ if plan_json is None:
+ return self.__agent_plan
+ return self.__scoped_plan_and_cache(plan_json)[0]
+
def __observe_subagent_setup(self, resource: Any) -> None:
"""Wire a lazily materialized sub-agent handle into the task lifecycle.
@@ -902,14 +916,14 @@ class FlinkRunnerContext(RunnerContext,
ExecutionReporter):
@override
def action_config(self) -> Dict[str, Any]:
"""Get config of the action."""
- return self.__agent_plan.get_action_config(
+ return self.__active_plan().get_action_config(
self._j_runner_context.getActionName()
)
@override
def get_action_config_value(self, key: str) -> Any:
"""Get config of the action."""
- return self.__agent_plan.get_action_config_value(
+ return self.__active_plan().get_action_config_value(
action_name=self._j_runner_context.getActionName(), key=key
)
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/ReActAgentSubagentTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/ReActAgentSubagentTest.java
new file mode 100644
index 00000000..f27e0680
--- /dev/null
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/subagent/ReActAgentSubagentTest.java
@@ -0,0 +1,458 @@
+/*
+ * 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.runtime.subagent;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.InputEvent;
+import org.apache.flink.agents.api.OutputEvent;
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.agents.ReActAgent;
+import org.apache.flink.agents.api.annotation.ToolParam;
+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.MessageRole;
+import org.apache.flink.agents.api.chat.messages.TextBlock;
+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.model.BaseChatModelConnection;
+import org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.prompt.Prompt;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import
org.apache.flink.agents.api.subagent.InternalSubagentResourceDescriptorBuilder;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.api.subagent.SubagentSetup;
+import org.apache.flink.agents.api.tools.Tool;
+import org.apache.flink.agents.plan.AgentPlan;
+import org.apache.flink.agents.runtime.operator.ActionExecutionOperator;
+import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.condition.EnabledForJreRange;
+import org.junit.jupiter.api.condition.JRE;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end coverage for a real {@link ReActAgent} running as an internal
sub-agent, over the
+ * operator harness. Unlike {@link InternalSubagentCallTest}, whose children
are hand-written echo
+ * agents, the child here is the built-in agent itself: its start action, the
framework's chat model
+ * action and its stop action must all run inside the child scope, driven by a
scripted connection
+ * instead of a live model.
+ *
+ * <p>The second test drives the model path: the parent model issues a {@code
_subagent_child} tool
+ * call, the framework routes it to the child, and the child's answer flows
back as the tool
+ * observation that closes the loop. The child is registered with a
single-field {@code input}
+ * schema, so this also proves the declared schema survives compilation and
the child's prompt-free
+ * {@code {input}} map path accepts the model's arguments as-is.
+ *
+ * <p>The third test lets the child declare a tool of its own and scripts two
model rounds for it:
+ * the first reply asks for the tool, the tool executes inside the child
scope, and the second round
+ * sees the tool result, closing the reasoning/tool loop in isolation.
+ *
+ * <p>The last two tests configure the parent ReActAgent with a prompt and an
output schema and
+ * leave the child bare. A child scope's resource lookup falls back to the
root agent's resources,
+ * so they pin that the child's built-in prompts stay its own rather than
resolving to the parent's
+ * under the same names.
+ *
+ * <p>Requires JDK 21+ for the same mailbox-suspension reason as {@link
InternalSubagentCallTest}.
+ */
+@EnabledForJreRange(min = JRE.JAVA_21)
+public class ReActAgentSubagentTest {
+
+ @BeforeEach
+ void resetScripts() {
+ ParentScriptedConnection.reset();
+ ChildScriptedConnection.reset();
+ }
+
+ // --- scripted chat model ---
+
+ /** Materialized from a descriptor; the connection carries the script. */
+ public static class ScriptedModelSetup extends BaseChatModelSetup {
+
+ public ScriptedModelSetup(ResourceDescriptor descriptor,
ResourceContext resourceContext) {
+ super(descriptor, resourceContext);
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return new HashMap<>();
+ }
+ }
+
+ /** Serves the parent agent; replies come from a static script. */
+ public static class ParentScriptedConnection extends
BaseChatModelConnection {
+
+ static final Deque<ChatMessage> REPLIES = new ArrayDeque<>();
+ static final List<List<ChatMessage>> CONVERSATIONS = new ArrayList<>();
+
+ public ParentScriptedConnection(
+ ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ }
+
+ static void reset() {
+ REPLIES.clear();
+ CONVERSATIONS.clear();
+ }
+
+ @Override
+ public ChatResult chat(
+ List<ChatMessage> messages, List<Tool> tools, Map<String,
Object> modelParams) {
+ CONVERSATIONS.add(new ArrayList<>(messages));
+ return new ChatResult(
+ REPLIES.isEmpty() ? ChatMessage.assistant("no script
left") : REPLIES.poll());
+ }
+ }
+
+ /** Serves the child agent; replies come from a static script. */
+ public static class ChildScriptedConnection extends
BaseChatModelConnection {
+
+ static final Deque<ChatMessage> REPLIES = new ArrayDeque<>();
+ static final List<List<ChatMessage>> CONVERSATIONS = new ArrayList<>();
+ static final List<List<String>> OFFERED_TOOLS = new ArrayList<>();
+
+ public ChildScriptedConnection(
+ ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ }
+
+ static void reset() {
+ REPLIES.clear();
+ CONVERSATIONS.clear();
+ OFFERED_TOOLS.clear();
+ }
+
+ @Override
+ public ChatResult chat(
+ List<ChatMessage> messages, List<Tool> tools, Map<String,
Object> modelParams) {
+ CONVERSATIONS.add(new ArrayList<>(messages));
+ OFFERED_TOOLS.add(
+ tools == null
+ ? List.of()
+ :
tools.stream().map(Tool::getName).collect(Collectors.toList()));
+ return new ChatResult(
+ REPLIES.isEmpty() ? ChatMessage.assistant("no script
left") : REPLIES.poll());
+ }
+ }
+
+ // --- caller actions ---
+
+ @SuppressWarnings("unused")
+ public static void callReActChild(Event event, RunnerContext ctx) throws
Exception {
+ SubagentSetup child = (SubagentSetup) ctx.getResource("child",
ResourceType.AGENT);
+ SubagentResult result = child.submit(ctx, "review the diff").await();
+ ctx.sendEvent(
+ new OutputEvent(
+ result.isSuccess()
+ ? firstOutput(result)
+ : "unexpected: " + result.getErrorMessage()));
+ }
+
+ // --- tests ---
+
+ @Test
+ @Timeout(60)
+ void callerDrivenCallRunsTheReActLoopInsideTheChildScope() throws
Exception {
+ ChildScriptedConnection.REPLIES.add(new
ChatMessage(MessageRole.ASSISTANT, "reviewed: ok"));
+
+ Agent parent = new Agent();
+ registerConnections(parent);
+ parent.addResource("child", ResourceType.AGENT,
bareReactAgent("child_conn"));
+ parent.addAction(
+ new String[] {InputEvent.EVENT_TYPE},
+ ReActAgentSubagentTest.class.getMethod(
+ "callReActChild", Event.class, RunnerContext.class));
+
+ List<Object> output = run(parent, 7L);
+
+ // The child's stop action output is the caller-visible result, and
the submitted string
+ // became the child's user message through the prompt-free primitive
path.
+ assertThat(output).containsExactly("reviewed: ok");
+ assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(1);
+
assertThat(latestUserText(ChildScriptedConnection.CONVERSATIONS.get(0)))
+ .isEqualTo("review the diff");
+ }
+
+ @Test
+ @Timeout(60)
+ void modelDrivenDelegationRoutesToTheDeclaredChildAndClosesTheLoop()
throws Exception {
+ ToolCallBlock delegation = delegation();
+
ParentScriptedConnection.REPLIES.add(assistantWithToolCall("delegating",
delegation));
+ ParentScriptedConnection.REPLIES.add(
+ new ChatMessage(MessageRole.ASSISTANT, "delegated and
reviewed"));
+ ChildScriptedConnection.REPLIES.add(new
ChatMessage(MessageRole.ASSISTANT, "reviewed: ok"));
+
+ // The registration declares the child's schema, which carries it
through open().
+ ReActAgent parent =
+ new ReActAgent(
+
ResourceDescriptor.Builder.newBuilder(ScriptedModelSetup.class.getName())
+ .addInitialArgument("connection",
"parent_conn")
+ .addInitialArgument("subagents",
List.of("child"))
+ .build(),
+ null,
+ null);
+ registerConnections(parent);
+ parent.addResource(
+ "child", ResourceType.AGENT,
declaredChild(bareReactAgent("child_conn")));
+
+ List<Object> output = run(parent, 7L);
+
+ assertThat(output).containsExactly("delegated and reviewed");
+ // The model's arguments reached the child as its user message through
the prompt-free
+ // {input} map path, and the child's answer came back as the closing
observation.
+ assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(1);
+
assertThat(latestUserText(ChildScriptedConnection.CONVERSATIONS.get(0)))
+ .isEqualTo("review the diff");
+ assertThat(ParentScriptedConnection.CONVERSATIONS).hasSize(2);
+ assertThat(allTexts(ParentScriptedConnection.CONVERSATIONS.get(1)))
+ .anySatisfy(text -> assertThat(text).contains("reviewed: ok"));
+ }
+
+ @Test
+ @Timeout(60)
+ void toolCallLoopRunsInsideTheChildScope() throws Exception {
+ ToolCallBlock lookup = new ToolCallBlock("call-1", "shout",
Map.of("text", "hello world"));
+ ChildScriptedConnection.REPLIES.add(assistantWithToolCall("let me
check", lookup));
+ ChildScriptedConnection.REPLIES.add(
+ new ChatMessage(MessageRole.ASSISTANT, "tool says: HELLO
WORLD!"));
+
+ Agent parent = new Agent();
+ registerConnections(parent);
+ parent.addResource("child", ResourceType.AGENT,
toolCallingReactAgent());
+ parent.addAction(
+ new String[] {InputEvent.EVENT_TYPE},
+ ReActAgentSubagentTest.class.getMethod(
+ "callReActChild", Event.class, RunnerContext.class));
+
+ List<Object> output = run(parent, 7L);
+
+ // The tool schema was offered on the first round, the tool executed
inside the child
+ // scope, and the second round saw the tool result before answering.
+ assertThat(output).containsExactly("tool says: HELLO WORLD!");
+
assertThat(ChildScriptedConnection.OFFERED_TOOLS.get(0)).containsExactly("shout");
+ assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(2);
+ assertThat(ChildScriptedConnection.CONVERSATIONS.get(1))
+ .anySatisfy(
+ message -> {
+
assertThat(message.getRole()).isEqualTo(MessageRole.TOOL);
+ ToolResultBlock result = (ToolResultBlock)
message.getBlocks().get(0);
+ assertThat(result.getText()).isEqualTo("HELLO
WORLD!");
+ });
+ }
+
+ @Test
+ @Timeout(60)
+ void bareChildDoesNotInheritTheParentPrompt() throws Exception {
+
ParentScriptedConnection.REPLIES.add(assistantWithToolCall("delegating",
delegation()));
+ ParentScriptedConnection.REPLIES.add(
+ new ChatMessage(MessageRole.ASSISTANT, "delegated and
reviewed"));
+ ChildScriptedConnection.REPLIES.add(new
ChatMessage(MessageRole.ASSISTANT, "reviewed: ok"));
+
+ ReActAgent parent =
+ delegatingParent(Prompt.fromText("PARENT ONLY: {input}"), /*
outputSchema= */ null);
+
+ List<Object> output = run(parent, 7L);
+
+ assertThat(output).containsExactly("delegated and reviewed");
+ // The parent prompt applies to the parent's own input only; the bare
child takes the
+ // delegated value through its prompt-free {input} map path.
+
assertThat(latestUserText(ParentScriptedConnection.CONVERSATIONS.get(0)))
+ .isEqualTo("PARENT ONLY: 7");
+ assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(1);
+
assertThat(latestUserText(ChildScriptedConnection.CONVERSATIONS.get(0)))
+ .isEqualTo("review the diff");
+ }
+
+ @Test
+ @Timeout(60)
+ void bareChildDoesNotInheritTheParentOutputSchema() throws Exception {
+
ParentScriptedConnection.REPLIES.add(assistantWithToolCall("delegating",
delegation()));
+ ParentScriptedConnection.REPLIES.add(
+ new ChatMessage(MessageRole.ASSISTANT,
"{\"verdict\":\"ok\"}"));
+ ChildScriptedConnection.REPLIES.add(new
ChatMessage(MessageRole.ASSISTANT, "reviewed: ok"));
+
+ ReActAgent parent = delegatingParent(/* prompt= */ null,
Verdict.class);
+
+ run(parent, 7L);
+
+ // The parent's own request carries its schema instruction, which
proves the instruction
+ // is reachable by name; the bare child's request must not.
+ assertThat(allTexts(ParentScriptedConnection.CONVERSATIONS.get(0)))
+ .anySatisfy(text ->
assertThat(text).contains(SCHEMA_INSTRUCTION));
+ assertThat(ChildScriptedConnection.CONVERSATIONS).hasSize(1);
+ List<ChatMessage> childConversation =
ChildScriptedConnection.CONVERSATIONS.get(0);
+ assertThat(allTexts(childConversation))
+ .noneSatisfy(text ->
assertThat(text).contains(SCHEMA_INSTRUCTION));
+ assertThat(childConversation)
+ .extracting(ChatMessage::getRole)
+ .containsExactly(MessageRole.USER);
+ }
+
+ // --- helpers ---
+
+ private static final String SCHEMA_INSTRUCTION = "The final response
should be json format";
+
+ /** The parent's structured answer, giving it an output schema the bare
child must not see. */
+ public static class Verdict {
+ public String verdict;
+ }
+
+ private static ToolCallBlock delegation() {
+ return new ToolCallBlock("call-1", "_subagent_child", Map.of("input",
"review the diff"));
+ }
+
+ private static ChatMessage assistantWithToolCall(String text,
ToolCallBlock toolCall) {
+ return ChatMessage.assistant(List.of(new TextBlock(text), toolCall));
+ }
+
+ /** A ReActAgent parent configured as given, declaring a bare ReActAgent
child to its model. */
+ private static ReActAgent delegatingParent(
+ @Nullable Prompt prompt, @Nullable Class<?> outputSchema) {
+ ReActAgent parent =
+ new ReActAgent(
+
ResourceDescriptor.Builder.newBuilder(ScriptedModelSetup.class.getName())
+ .addInitialArgument("connection",
"parent_conn")
+ .addInitialArgument("subagents",
List.of("child"))
+ .build(),
+ prompt,
+ outputSchema);
+ registerConnections(parent);
+ parent.addResource(
+ "child", ResourceType.AGENT,
declaredChild(bareReactAgent("child_conn")));
+ return parent;
+ }
+
+ /**
+ * Declares {@code child} with a single {@code input} field, the arguments
shape the prompt-free
+ * map path of a ReActAgent passes through as the user message.
+ */
+ private static ResourceDescriptor declaredChild(Agent child) {
+ return new InternalSubagentResourceDescriptorBuilder()
+ .agent(child)
+ .description("Reviews a code diff.")
+ .inputSchema(
+
"{\"type\":\"object\",\"properties\":{\"input\":{\"type\":\"string\"}},"
+ + "\"required\":[\"input\"]}")
+ .build();
+ }
+
+ private static ReActAgent toolCallingReactAgent() throws Exception {
+ ReActAgent child =
+ new ReActAgent(
+
ResourceDescriptor.Builder.newBuilder(ScriptedModelSetup.class.getName())
+ .addInitialArgument("connection", "child_conn")
+ .addInitialArgument("tools", List.of("shout"))
+ .build(),
+ null,
+ null);
+ child.addResource(
+ "shout",
+ ResourceType.TOOL,
+
Tool.fromMethod(ReActAgentSubagentTest.class.getMethod("shout", String.class)));
+ return child;
+ }
+
+ @SuppressWarnings("unused")
+ public static String shout(@ToolParam(name = "text") String text) {
+ return text.toUpperCase(Locale.ROOT) + "!";
+ }
+
+ private static ReActAgent bareReactAgent(String connectionName) {
+ return new ReActAgent(
+
ResourceDescriptor.Builder.newBuilder(ScriptedModelSetup.class.getName())
+ .addInitialArgument("connection", connectionName)
+ .build(),
+ null,
+ null);
+ }
+
+ private static String firstOutput(SubagentResult result) {
+ return String.valueOf(((List<?>) result.getResult()).get(0));
+ }
+
+ private static String latestUserText(List<ChatMessage> messages) {
+ for (int i = messages.size() - 1; i >= 0; i--) {
+ ChatMessage message = messages.get(i);
+ if (message.getRole() == MessageRole.USER) {
+ return message.getText();
+ }
+ }
+ throw new AssertionError("no user message in " + messages);
+ }
+
+ private static List<String> allTexts(List<ChatMessage> messages) {
+ return messages.stream()
+ .flatMap(message -> message.getBlocks().stream())
+ .filter(block -> block instanceof TextBlock || block
instanceof ToolResultBlock)
+ .map(
+ block ->
+ block instanceof TextBlock
+ ? ((TextBlock) block).getText()
+ : ((ToolResultBlock) block).getText())
+ .collect(Collectors.toList());
+ }
+
+ private static void registerConnections(Agent agent) {
+ agent.addResource(
+ "parent_conn",
+ ResourceType.CHAT_MODEL_CONNECTION,
+
ResourceDescriptor.Builder.newBuilder(ParentScriptedConnection.class.getName())
+ .build());
+ agent.addResource(
+ "child_conn",
+ ResourceType.CHAT_MODEL_CONNECTION,
+
ResourceDescriptor.Builder.newBuilder(ChildScriptedConnection.class.getName())
+ .build());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<Object> run(Agent root, long input) throws Exception {
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object>
harness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory<>(new
AgentPlan(root), true),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ harness.open();
+ harness.processElement(new StreamRecord<>(input));
+ ((ActionExecutionOperator<Long, Object>) harness.getOperator())
+ .waitInFlightEventsFinished();
+ return ((List<StreamRecord<Object>>) harness.getRecordOutput())
+
.stream().map(StreamRecord::getValue).collect(Collectors.toList());
+ }
+ }
+}
