yunfengzhou-hub commented on code in PR #1268:
URL: https://github.com/apache/flink-agents/pull/1268#discussion_r4300865107
##########
docs/content/docs/development/multi-agent/subagents.md:
##########
@@ -307,30 +443,854 @@ parent.addResource(
+ "\"required\":[\"diff\"]}")
.build());
```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+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 you change either.
+
+### External: write a setup class
+
+An external sub-agent has no plan of its own, and it is not compiled into the
+parent's. The work happens in your own code or in a remote agent service. The
parent
+only submits to it and waits for the outcome. The framework cannot perform
that work
+for you, because it does not know your protocol. So you write the part that
performs
+one invocation, and the framework supplies everything around it. It runs your
code
+under durable execution, and after a failover it replays the recorded result
instead
+of calling your code again.
+
+What you write is a setup class. Its only job is to carry out one invocation,
so it
+declares no actions and no events of its own. It extends a framework base
class and
+implements the methods that class declares.
+
+Registration names the setup class in a descriptor, and both languages
construct the
+setup from the descriptor at runtime. Registering a setup instance instead is
+rejected.
+
+The invocation itself is carried out in one of two modes, and which one you
write
+follows from your remote interface. When one request returns the result, write
a
+*deferred* setup. You own the whole interaction, and the framework only holds
it back
+until the caller asks for the result. When the remote instead hands back a run
that
+you later query for status and result, write an *async (pub/sub)* setup. The
base
+class owns that composition of publishing, polling, and fetching, and you
supply the
+transport calls it is built from.
+
+#### Deferred execution
+
+A deferred setup sends nothing when the caller submits. It only registers the
+invocation. The request goes out when the caller first awaits the result. So a
call
+that is cancelled instead never issues its request.
+
+What you write is the interaction itself, in the `prepare` method of your
setup class.
+`prepare` runs once per invocation and returns the durable call that carries
it. The
+example is one blocking request, and whatever else your call does is yours to
decide,
+because the framework does not know your protocol.
+The framework assigns `session_id` and `call_id` before it calls `prepare`,
and they
+are the only values that a replay reproduces. The durable id you build from
them is
+the key that recovery matches a recorded result against. When the key matches,
your
+call is not run again. So build the key from those two values alone. Anything
else in
+it, a timestamp for instance, makes a replay look up a different key and run
the work
+a second time. Passing the same values to your own remote lets it recognize a
+re-issued request as the same one.
+
+The reconciler is optional. Omitting it leaves one crash window uncovered, and
+[Durability and failover](#durability-and-failover) describes that window.
Supply one
+when the work cannot be repeated.
+
+{{< tabs "Supply a Reconciler" >}}
+
+{{< tab "Python" >}}
```python
-reviewer = ReActAgent(
- chat_model=chat_model_descriptor,
- prompt=Prompt.from_text(
- "Review the following code diff and report any problems:\n\n{diff}"
+def prepare(
+ self, ctx: RunnerContext, prompt: Any, session_id: str, call_id: str
+) -> PreparedTriple:
+ key = f"{session_id}#{call_id}"
+
+ def call() -> SubagentResult:
+ return SubagentResult.ok(post_review(self.endpoint, prompt, key))
+
+ # Consulted only during recovery, when this call is revisited and no
result was
+ # recorded for it. Return the recovered outcome, or raise the recovered
failure.
+ # It takes no arguments.
+ def reconcile() -> SubagentResult:
+ # fetch_review_by_key is your own lookup; not part of Flink Agents.
+ found = fetch_review_by_key(self.endpoint, key)
+ if found is None:
+ raise RuntimeError("the review service has no record of this call")
+ return SubagentResult.ok(found)
+
+ return (key, call, reconcile)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+// Inside the DurableCallable that prepare returns, alongside getId() and
call().
+
+// Consulted only during recovery, when this call is revisited and no result
was
+// recorded for it. Return the recovered outcome, or throw the recovered
failure.
+@Override
+public Callable<SubagentResult> reconciler() {
+ return () -> {
+ // fetchReviewByKey is your own lookup; not part of Flink Agents.
+ String found = fetchReviewByKey(endpoint, key);
+ if (found == null) {
+ throw new IllegalStateException("the review service has no record
of this call");
+ }
+ return SubagentResult.ok(found);
+ };
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+{{< tabs "Deferred Sub-agent Setup" >}}
+
+{{< tab "Python" >}}
+```python
+from typing import Any
+
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import SubagentResult
+from flink_agents.runtime.deferred_subagent import (
+ DeferredSubagentSetup,
+ PreparedTriple,
+)
+
+
+def post_review(endpoint: str, prompt: Any, key: str) -> str:
+ """Your own call to the review service; not part of Flink Agents.
+
+ Keying the work lets the service treat a re-issued request as the same one.
+ """
+ ...
+
+
+class RemoteReviewSetup(DeferredSubagentSetup):
+ endpoint: str = ""
+
+ def prepare(
+ self, ctx: RunnerContext, prompt: Any, session_id: str, call_id: str
+ ) -> PreparedTriple:
+ key = f"{session_id}#{call_id}"
+
+ def call() -> SubagentResult:
+ try:
+ return SubagentResult.ok(post_review(self.endpoint, prompt,
key))
+ except Exception as e:
+ # An expected failure is a value; only a system fault should
escape.
+ return SubagentResult.error(e)
+
+ # The first value is the key recovery matches a recorded result
against, so
+ # it is built from the two ids alone. The third is the reconciler.
+ return (key, call, None)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.runtime.subagent.BaseDeferredSubagentSetup;
+
+/**
+ * Your own call to the review service; not part of Flink Agents. Keying the
work lets
+ * the service treat a re-issued request as the same one.
+ */
+static String postReview(String endpoint, Object prompt, String key) {
+ ...
+}
+
+public class RemoteReviewSetup extends BaseDeferredSubagentSetup {
+
+ private final String endpoint;
+
+ public RemoteReviewSetup(ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ this.endpoint = descriptor.getArgument("endpoint");
+ }
+
+ @Override
+ protected DurableCallable<SubagentResult> prepare(
+ RunnerContext ctx, Object prompt, String sessionId, String callId)
{
+ // Built from the two ids alone, so a replay reproduces it.
+ String key = sessionId + "#" + callId;
+ return new DurableCallable<SubagentResult>() {
+ @Override
+ public String getId() {
+ // The key recovery matches a recorded result against.
+ return key;
+ }
+
+ @Override
+ public Class<SubagentResult> getResultClass() {
+ return SubagentResult.class;
+ }
+
+ @Override
+ public SubagentResult call() {
+ try {
+ return SubagentResult.ok(postReview(endpoint, prompt,
key));
+ } catch (Exception e) {
+ // An expected failure is a value; only a system fault
should escape.
+ return SubagentResult.error(e);
+ }
+ }
+ };
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+Register the setup under the name invocations will use. Every descriptor
argument
+beyond the built-in ones is configuration for the setup to read back. In
Python it
+reads them as model fields, and in Java through the public
+`(ResourceDescriptor, ResourceContext)` constructor. The `endpoint` above is
such
+configuration, not a built-in transport.
+
+{{< tabs "Register an External Sub-agent" >}}
+
+{{< tab "Python" >}}
+```python
+parent.add_resource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz="my_app.agents.RemoteReviewSetup",
+ endpoint="https://example.com/review";,
+ description="Review code using the remote service",
+ input_schema=(
+ '{"type":"object","properties":{"diff":{"type":"string"}},'
+ '"required":["diff"]}'
+ ),
),
)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+parent.addResource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(RemoteReviewSetup.class.getName())
+ .addInitialArgument("endpoint", "https://example.com/review";)
+ .addInitialArgument("description", "Review code using the remote
service")
+ .addInitialArgument(
+ "input_schema",
+
"{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}},"
+ + "\"required\":[\"diff\"]}")
+ .build());
+```
+{{< /tab >}}
+
+{{< tab "YAML" >}}
+```yaml
+agents:
+ - name: orchestrator
+ subagents:
+ - name: remote_reviewer
+ clazz: my_app.agents.RemoteReviewSetup
+ endpoint: https://example.com/review
+ description: Review code using the remote service
+ input_schema:
'{"type":"object","properties":{"diff":{"type":"string"}},"required":["diff"]}'
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The YAML form of that registration is the `subagents` list of an agent entry,
as in
+the YAML tab above. That list can also sit at the shared document level,
outside
+`agents:`, so that several agents in one file reuse it. A chat model setup
declared in
+the same file also takes a `subagents` argument, but it is a list of
registered names
+rather than a list of entries. See
+[delegate from a chat model](#delegate-from-a-chat-model).
+
+Two constraints apply to the entry:
+
+- No class alias is defined for `AGENT` resources, so `clazz` is always a
+ fully-qualified class path.
+- The entry's language must match the loader. A Python YAML file names a Python
+ class and may not set `type: java`. A Java YAML file must set `type: java`.
+
+#### Async (pub/sub) execution
+
+An async sub-agent talks to a service that runs the task for you. The service
accepts
+the work and answers at once, and you query it later for the run's status and
+result.
+
+An async setup publishes the run when the caller submits, through one durable
POST.
+Waiting for the result then polls the run's status until it reaches a terminal
state,
+and fetches it.
+
+That is the difference from the deferred mode, which sends nothing when the
caller
+submits. The mode's name is the framework's own. The POST is the publish, and
the
+polling is the subscribe.
+
+The submit-poll-fetch composition is the same for every such remote, so the
base
+class owns it, along with the durable execution and the crash-window recovery.
You
+supply the three transport calls it is built from:
+
+- `callSubmitRequest` / `call_submit_request` starts the run.
+- `callQueryStatus` / `call_query_status` returns a `RunStatus`, one of
+ `notStarted()`, `running()`, `completed()`, or `failed(error)`. It must be a
+ read-only probe with no side effects. Report an expected failure as a
`FAILED`
+ status.
+- `callFetchResult` / `call_fetch_result` returns the `SubagentResult`. It
must be
+ an idempotent read, because a replay may call it again. The status snapshot
never
+ carries the payload, and a failed run is reported without fetching at all.
+
+{{< tabs "Async Sub-agent Setup" >}}
+
+{{< tab "Python" >}}
+```python
+from typing import Any
+
+from flink_agents.api.subagent import SubagentResult
+from flink_agents.runtime.async_subagent import BaseAsyncSubagentSetup,
RunStatus
+
+# client(...) and task_id stand for your own transport client and run-id
+# scheme; neither is part of Flink Agents.
+
+
+class RemoteRunSetup(BaseAsyncSubagentSetup):
+ endpoint: str = ""
+
+ def call_submit_request(self, session_id: str, call_id: str, prompt: Any)
-> None:
+ client(self.endpoint).submit(prompt, task_id=f"{session_id}#{call_id}")
+
+ def call_query_status(self, session_id: str, call_id: str) -> RunStatus:
+ probe = client(self.endpoint).status(task_id=f"{session_id}#{call_id}")
+ if probe is None:
+ return RunStatus.not_started()
+ if probe.succeeded:
+ return RunStatus.completed()
+ if probe.failed:
+ return RunStatus.failed(probe.error)
+ return RunStatus.running()
+
+ def call_fetch_result(self, session_id: str, call_id: str) ->
SubagentResult:
+ return SubagentResult.ok(
+ client(self.endpoint).fetch(task_id=f"{session_id}#{call_id}")
+ )
+
+ # Optional, and a no-op by default, so a cancelled call leaves the remote
run
+ # going. A replay may propagate the same cancellation again, so your
remote has
+ # to accept it twice.
+ def call_cancel_request(self, session_id: str, call_id: str) -> None:
+ client(self.endpoint).cancel(task_id=f"{session_id}#{call_id}")
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.runtime.subagent.BaseAsyncSubagentSetup;
+
+// client() and taskId(...) stand for your own transport client and run-id
+// scheme; neither is part of Flink Agents.
+public class RemoteRunSetup extends BaseAsyncSubagentSetup {
+
+ private final String endpoint;
+
+ public RemoteRunSetup(ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ this.endpoint = descriptor.getArgument("endpoint");
+ }
+ @Override
+ protected void callSubmitRequest(String sessionId, String callId, Object
prompt)
+ throws Exception {
+ client().submit(String.valueOf(prompt), taskId(sessionId, callId));
+ }
+
+ @Override
+ protected RunStatus callQueryStatus(String sessionId, String callId) {
+ // TaskStatus is your own transport type, not a framework type.
+ TaskStatus probe = client().status(taskId(sessionId, callId));
+ if (probe == null) {
+ return RunStatus.notStarted();
+ }
+ if (probe.succeeded()) {
+ return RunStatus.completed();
+ }
+ if (probe.failed()) {
+ return RunStatus.failed(probe.error());
+ }
+ return RunStatus.running();
+ }
+
+ @Override
+ protected SubagentResult callFetchResult(String sessionId, String callId)
throws Exception {
+ return client().fetchResult(taskId(sessionId, callId));
+ }
+
+ // Optional, and a no-op by default, so a cancelled call leaves the remote
run
+ // going. A replay may propagate the same cancellation again, so your
remote has
+ // to accept it twice.
+ @Override
+ protected void callCancelRequest(String sessionId, String callId) {
+ client().cancel(taskId(sessionId, callId));
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+Register the setup like any external sub-agent. The async form adds one
descriptor
+argument, `status_poll_interval_millis`, and it defaults to 500.
+
+{{< tabs "Register an Async Sub-agent" >}}
+
+{{< tab "Python" >}}
+```python
parent.add_resource(
- "reviewer",
+ "remote_runner",
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(),
+ ResourceDescriptor(
+ clazz="my_app.agents.RemoteRunSetup",
+ endpoint="https://example.com/runs";,
+ description="Run a review task on the remote service",
+ input_schema=(
+ '{"type":"object","properties":{"diff":{"type":"string"}},'
+ '"required":["diff"]}'
+ ),
+ status_poll_interval_millis=500,
+ ),
)
```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+parent.addResource(
+ "remote_runner",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(RemoteRunSetup.class.getName())
+ .addInitialArgument("endpoint", "https://example.com/runs";)
+ .addInitialArgument("description", "Run a review task on the remote
service")
+ .addInitialArgument(
+ "input_schema",
+
"{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}},"
+ + "\"required\":[\"diff\"]}")
+ .addInitialArgument("status_poll_interval_millis", 500L)
+ .build());
+```
+{{< /tab >}}
+
+{{< tab "YAML" >}}
+```yaml
+agents:
+ - name: orchestrator
+ subagents:
+ - name: remote_runner
+ clazz: my_app.agents.RemoteRunSetup
+ endpoint: https://example.com/runs
+ description: Run a review task on the remote service
+ input_schema:
'{"type":"object","properties":{"diff":{"type":"string"}},"required":["diff"]}'
+ status_poll_interval_millis: 500
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The base class also recovers the crash window between "POST sent" and "POST
durably
+recorded". On recovery it probes the run's status first, and it re-posts only
when the
+probe reports that the run never started. Override `reconcileSubmitRequest` /
+`reconcile_submit_request` only if your service needs a different repair.
+
+## Invoke a sub-agent
+
+A sub-agent is invoked in one of two ways, model-driven or caller-driven, and
the two
+reuse different existing abstractions.
+
+Model-driven delegation rides the tool abstraction. The sub-agent is offered
to the
+model as a tool, so your action sends a `ChatRequestEvent` and the model
answers with
+a call to the sub-agent. The result then returns to the model as the tool
+observation, until a final `ChatResponseEvent` reaches your action. This is
the same
+cycle any tool call makes.
+
+Caller-driven delegation cannot ride that cycle, because a tool's result goes
back
+into the chat loop rather than to the action that asked for it. An action that
needs
+the delegated outcome therefore calls the sub-agent itself, with no model in
the loop.
+
+Both ways end in the same call. Your action, or the framework on the model's
behalf,
+runs `submit(ctx, prompt)` on the registered name. That call returns a
+`SubagentFuture`, which is the handle for that one call. Awaiting the handle
yields
+the call's `SubagentResult`.
+
+### Delegate from a chat model
+
+List the registered name under `subagents` of the parent's chat model:
+
+{{< tabs "Declare Sub-agents on a Chat Model" >}}
+
+{{< tab "Python" >}}
+```python
+ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP,
+ connection="ollama_connection",
+ model="qwen3:8b",
+ subagents=["reviewer"],
+)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
+ .addInitialArgument("connection", "ollamaConnection")
+ .addInitialArgument("model", "qwen3:8b")
+ .addInitialArgument("subagents", List.of("reviewer"))
+ .build();
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The model sees `reviewer` as a tool, described by the registration's
description and
+input schema. A schema is required for model-selected calls. A missing
description
+falls back to a generic delegation description.
+
+The arguments the model sends become the sub-agent's input, and its outputs
return to
+the model as the tool observation. The model can then keep reasoning over
them. For
+an internal sub-agent, the observation is the JSON encoding of the full output
list
+described in [the result](#the-result). That holds even when the child emits a
single
+output, so the shape does not depend on how many outputs the child emits.
+
+An external setup class can state the shape of that observation. It declares
+`result_type()` in Python or `getResultType()` in Java, and the default
reports the
+result as it arrived. Declare one to narrow a wider payload, or to convert a
payload
+that is not already JSON-compatible. An internal sub-agent has no such
declaration, so
+its observation is always the JSON encoding of its output list. This shapes the
+observation only. A caller that awaits the handle gets the result the sub-agent
+produced, as [the result](#the-result) describes.
+
+The external sub-agents `remote_reviewer` and `remote_runner` are offered the
same
+way. Only their registration differs.
+
+### Call from an action
+
+An action can submit input to a registered sub-agent directly and await its
+outcome. This path needs no description and no input declaration, so it also
works
+for a child registered as-is.
+
+{{< tabs "Call a Sub-agent From an Action" >}}
+
+{{< tab "Python" >}}
+```python
+class DelegatingAgent(Agent):
+ @action(EventType.InputEvent)
+ @staticmethod
+ async def delegate(event: Event, ctx: RunnerContext) -> None:
+ reviewer = ctx.get_resource("reviewer", ResourceType.AGENT)
+ future = await reviewer.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}"))
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+public class DelegatingAgent extends Agent {
+
+ @Action(EventType.InputEvent)
+ public static void delegate(Event event, RunnerContext ctx) throws
Exception {
+ SubagentSetup reviewer =
+ (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+ SubagentResult result = reviewer.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()));
+ }
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+What you pass to `submit` is what the sub-agent receives. An internal child
gets it
+as its `InputEvent` payload, and a model-driven call passes the arguments map
+instead. What comes back is a `SubagentResult`. For the internal sub-agent
used here,
+its `result` is the list of `OutputEvent` payloads the child emitted. That is
why the
+example reads `result.result[0]`. An external sub-agent returns whatever its
setup
+produces.
+
+In Python `submit` is a coroutine, so the call is awaited in two phases and the
+action must be `async`. In Java `submit` returns the handle directly and
`await()`
+yields the result. The same two calls drive `remote_reviewer` and
`remote_runner`
+unchanged. Only the registration differs.
+
+## Work with open handles
+
+A `SubagentFuture` is live from `submit` until it is awaited or cancelled.
+
+It works like the `AsyncFuture` that
+[async execution]({{< ref
"docs/development/create-agent/actions#async-execution" >}})
+returns. You get it from one call, and you await it to get the outcome. It is
its own
+type rather than an `AsyncFuture`, because it always resolves to a
`SubagentResult`,
+it carries the call's `(session_id, call_id)`, and it owns its durable
execution
+inside the setup.
+
+### Fan out to several sub-agents
+
+`combine` batches handles into a `SubagentFutures`, and it works the way
`gather`
+works for ordinary async futures. `gather` does not accept a `SubagentFuture`,
so
+handles batch this way instead. Every handle in the batch is prepared before
any is
+executed, so deferred requests all go out up front and remote runs overlap.
Results
+come back in the order the handles were added.
+
+{{< tabs "Fan Out to Several Sub-agents" >}}
+
+{{< tab "Python" >}}
+```python
+reviewer = ctx.get_resource("reviewer", ResourceType.AGENT)
+
+first = await reviewer.submit(ctx, "review the diff")
+second = await reviewer.submit(ctx, "review the tests")
+
+outcomes = await first.combine(second)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+SubagentSetup reviewer = (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+
+SubagentFuture first = reviewer.submit(ctx, "review the diff");
+SubagentFuture second = reviewer.submit(ctx, "review the tests");
+
+List<SubagentResult> outcomes = first.combine(second).awaitAll();
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+### Continue a session
+
+Pass the `session_id` of an earlier handle to continue that conversation. The
new
+call gets a fresh `call_id` in the same session. Whether a conversation can be
+continued across actions is up to the implementation.
+
+{{< tabs "Continue a Session" >}}
+
+{{< tab "Python" >}}
+```python
+reviewer = ctx.get_resource("reviewer", ResourceType.AGENT)
+
+first = await reviewer.submit(ctx, "review the diff")
+first_result = await first
+
+second = await reviewer.submit(ctx, "now the tests", first.session_id)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+SubagentSetup reviewer = (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+
+SubagentFuture first = reviewer.submit(ctx, "review the diff");
+SubagentResult firstResult = first.await();
+
+SubagentFuture second = reviewer.submit(ctx, "now the tests",
first.getSessionId());
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+### Cancel and resolve
+
+Resolve every handle you submit. An action that finishes with a handle it
neither
+awaited nor cancelled fails with `Action <name> finished without resolving the
+sub-agent calls it submitted: <ids>`. Cancelling counts as resolving, and
awaiting a
+cancelled handle throws `CancellationException` in Java and raises
`CancelledError` in
+Python.
+
+{{< tabs "Cancel a Sub-agent Call" >}}
+
+{{< tab "Python" >}}
+```python
+from concurrent.futures import CancelledError
+
+reviewer = ctx.get_resource("reviewer", ResourceType.AGENT)
+
+future = await reviewer.submit(ctx, "review the diff")
+future.cancel()
+
+try:
+ await future
+except CancelledError:
+ ctx.send_event(OutputEvent(output="cancelled"))
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+SubagentSetup reviewer = (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+
+SubagentFuture future = reviewer.submit(ctx, "review the diff");
+future.cancel();
+
+try {
+ future.await();
+} catch (CancellationException e) {
+ ctx.sendEvent(new OutputEvent("cancelled"));
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+`cancel()` on a deferred handle takes effect only before the request is issued.
+
+## The outcome of a call
+
+### The result
+
+The awaited value of every call is a `SubagentResult`, which carries the
outcome in
+three fields:
+
+| Field | Java | Carries |
+|---|---|---|
+| `success` | `isSuccess()` | Whether the call succeeded |
+| `result` | `getResult()` | The answer on success. For an internal sub-agent,
the list of `OutputEvent` payloads in emission order. For an external one,
whatever the setup produces, such as a string or an object |
+| `error_message` | `getErrorMessage()` | The reason on failure |
+
+A result is persisted and re-bound after a failover, so it must be
JSON-compatible.
Review Comment:
Fixed. Replay is split per language, and the JSON rule is scoped to
model-facing results and tied to normalization rather than to persistence.
##########
docs/content/docs/development/multi-agent/subagents.md:
##########
@@ -307,30 +443,854 @@ parent.addResource(
+ "\"required\":[\"diff\"]}")
.build());
```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+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 you change either.
+
+### External: write a setup class
+
+An external sub-agent has no plan of its own, and it is not compiled into the
+parent's. The work happens in your own code or in a remote agent service. The
parent
+only submits to it and waits for the outcome. The framework cannot perform
that work
+for you, because it does not know your protocol. So you write the part that
performs
+one invocation, and the framework supplies everything around it. It runs your
code
+under durable execution, and after a failover it replays the recorded result
instead
+of calling your code again.
+
+What you write is a setup class. Its only job is to carry out one invocation,
so it
+declares no actions and no events of its own. It extends a framework base
class and
+implements the methods that class declares.
+
+Registration names the setup class in a descriptor, and both languages
construct the
+setup from the descriptor at runtime. Registering a setup instance instead is
+rejected.
+
+The invocation itself is carried out in one of two modes, and which one you
write
+follows from your remote interface. When one request returns the result, write
a
+*deferred* setup. You own the whole interaction, and the framework only holds
it back
+until the caller asks for the result. When the remote instead hands back a run
that
+you later query for status and result, write an *async (pub/sub)* setup. The
base
+class owns that composition of publishing, polling, and fetching, and you
supply the
+transport calls it is built from.
+
+#### Deferred execution
+
+A deferred setup sends nothing when the caller submits. It only registers the
+invocation. The request goes out when the caller first awaits the result. So a
call
+that is cancelled instead never issues its request.
+
+What you write is the interaction itself, in the `prepare` method of your
setup class.
+`prepare` runs once per invocation and returns the durable call that carries
it. The
+example is one blocking request, and whatever else your call does is yours to
decide,
+because the framework does not know your protocol.
+The framework assigns `session_id` and `call_id` before it calls `prepare`,
and they
+are the only values that a replay reproduces. The durable id you build from
them is
+the key that recovery matches a recorded result against. When the key matches,
your
+call is not run again. So build the key from those two values alone. Anything
else in
+it, a timestamp for instance, makes a replay look up a different key and run
the work
+a second time. Passing the same values to your own remote lets it recognize a
+re-issued request as the same one.
+
+The reconciler is optional. Omitting it leaves one crash window uncovered, and
+[Durability and failover](#durability-and-failover) describes that window.
Supply one
+when the work cannot be repeated.
+
+{{< tabs "Supply a Reconciler" >}}
+
+{{< tab "Python" >}}
```python
-reviewer = ReActAgent(
- chat_model=chat_model_descriptor,
- prompt=Prompt.from_text(
- "Review the following code diff and report any problems:\n\n{diff}"
+def prepare(
+ self, ctx: RunnerContext, prompt: Any, session_id: str, call_id: str
+) -> PreparedTriple:
+ key = f"{session_id}#{call_id}"
+
+ def call() -> SubagentResult:
+ return SubagentResult.ok(post_review(self.endpoint, prompt, key))
+
+ # Consulted only during recovery, when this call is revisited and no
result was
+ # recorded for it. Return the recovered outcome, or raise the recovered
failure.
+ # It takes no arguments.
+ def reconcile() -> SubagentResult:
+ # fetch_review_by_key is your own lookup; not part of Flink Agents.
+ found = fetch_review_by_key(self.endpoint, key)
+ if found is None:
+ raise RuntimeError("the review service has no record of this call")
+ return SubagentResult.ok(found)
+
+ return (key, call, reconcile)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+// Inside the DurableCallable that prepare returns, alongside getId() and
call().
+
+// Consulted only during recovery, when this call is revisited and no result
was
+// recorded for it. Return the recovered outcome, or throw the recovered
failure.
+@Override
+public Callable<SubagentResult> reconciler() {
+ return () -> {
+ // fetchReviewByKey is your own lookup; not part of Flink Agents.
+ String found = fetchReviewByKey(endpoint, key);
+ if (found == null) {
+ throw new IllegalStateException("the review service has no record
of this call");
+ }
+ return SubagentResult.ok(found);
+ };
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+{{< tabs "Deferred Sub-agent Setup" >}}
+
+{{< tab "Python" >}}
+```python
+from typing import Any
+
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.subagent import SubagentResult
+from flink_agents.runtime.deferred_subagent import (
+ DeferredSubagentSetup,
+ PreparedTriple,
+)
+
+
+def post_review(endpoint: str, prompt: Any, key: str) -> str:
+ """Your own call to the review service; not part of Flink Agents.
+
+ Keying the work lets the service treat a re-issued request as the same one.
+ """
+ ...
+
+
+class RemoteReviewSetup(DeferredSubagentSetup):
+ endpoint: str = ""
+
+ def prepare(
+ self, ctx: RunnerContext, prompt: Any, session_id: str, call_id: str
+ ) -> PreparedTriple:
+ key = f"{session_id}#{call_id}"
+
+ def call() -> SubagentResult:
+ try:
+ return SubagentResult.ok(post_review(self.endpoint, prompt,
key))
+ except Exception as e:
+ # An expected failure is a value; only a system fault should
escape.
+ return SubagentResult.error(e)
+
+ # The first value is the key recovery matches a recorded result
against, so
+ # it is built from the two ids alone. The third is the reconciler.
+ return (key, call, None)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.runtime.subagent.BaseDeferredSubagentSetup;
+
+/**
+ * Your own call to the review service; not part of Flink Agents. Keying the
work lets
+ * the service treat a re-issued request as the same one.
+ */
+static String postReview(String endpoint, Object prompt, String key) {
+ ...
+}
+
+public class RemoteReviewSetup extends BaseDeferredSubagentSetup {
+
+ private final String endpoint;
+
+ public RemoteReviewSetup(ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ this.endpoint = descriptor.getArgument("endpoint");
+ }
+
+ @Override
+ protected DurableCallable<SubagentResult> prepare(
+ RunnerContext ctx, Object prompt, String sessionId, String callId)
{
+ // Built from the two ids alone, so a replay reproduces it.
+ String key = sessionId + "#" + callId;
+ return new DurableCallable<SubagentResult>() {
+ @Override
+ public String getId() {
+ // The key recovery matches a recorded result against.
+ return key;
+ }
+
+ @Override
+ public Class<SubagentResult> getResultClass() {
+ return SubagentResult.class;
+ }
+
+ @Override
+ public SubagentResult call() {
+ try {
+ return SubagentResult.ok(postReview(endpoint, prompt,
key));
+ } catch (Exception e) {
+ // An expected failure is a value; only a system fault
should escape.
+ return SubagentResult.error(e);
+ }
+ }
+ };
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+Register the setup under the name invocations will use. Every descriptor
argument
+beyond the built-in ones is configuration for the setup to read back. In
Python it
+reads them as model fields, and in Java through the public
+`(ResourceDescriptor, ResourceContext)` constructor. The `endpoint` above is
such
+configuration, not a built-in transport.
+
+{{< tabs "Register an External Sub-agent" >}}
+
+{{< tab "Python" >}}
+```python
+parent.add_resource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor(
+ clazz="my_app.agents.RemoteReviewSetup",
+ endpoint="https://example.com/review";,
+ description="Review code using the remote service",
+ input_schema=(
+ '{"type":"object","properties":{"diff":{"type":"string"}},'
+ '"required":["diff"]}'
+ ),
),
)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+parent.addResource(
+ "remote_reviewer",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(RemoteReviewSetup.class.getName())
+ .addInitialArgument("endpoint", "https://example.com/review";)
+ .addInitialArgument("description", "Review code using the remote
service")
+ .addInitialArgument(
+ "input_schema",
+
"{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}},"
+ + "\"required\":[\"diff\"]}")
+ .build());
+```
+{{< /tab >}}
+
+{{< tab "YAML" >}}
+```yaml
+agents:
+ - name: orchestrator
+ subagents:
+ - name: remote_reviewer
+ clazz: my_app.agents.RemoteReviewSetup
+ endpoint: https://example.com/review
+ description: Review code using the remote service
+ input_schema:
'{"type":"object","properties":{"diff":{"type":"string"}},"required":["diff"]}'
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The YAML form of that registration is the `subagents` list of an agent entry,
as in
+the YAML tab above. That list can also sit at the shared document level,
outside
+`agents:`, so that several agents in one file reuse it. A chat model setup
declared in
+the same file also takes a `subagents` argument, but it is a list of
registered names
+rather than a list of entries. See
+[delegate from a chat model](#delegate-from-a-chat-model).
+
+Two constraints apply to the entry:
+
+- No class alias is defined for `AGENT` resources, so `clazz` is always a
+ fully-qualified class path.
+- The entry's language must match the loader. A Python YAML file names a Python
+ class and may not set `type: java`. A Java YAML file must set `type: java`.
+
+#### Async (pub/sub) execution
+
+An async sub-agent talks to a service that runs the task for you. The service
accepts
+the work and answers at once, and you query it later for the run's status and
+result.
+
+An async setup publishes the run when the caller submits, through one durable
POST.
+Waiting for the result then polls the run's status until it reaches a terminal
state,
+and fetches it.
+
+That is the difference from the deferred mode, which sends nothing when the
caller
+submits. The mode's name is the framework's own. The POST is the publish, and
the
+polling is the subscribe.
+
+The submit-poll-fetch composition is the same for every such remote, so the
base
+class owns it, along with the durable execution and the crash-window recovery.
You
+supply the three transport calls it is built from:
+
+- `callSubmitRequest` / `call_submit_request` starts the run.
+- `callQueryStatus` / `call_query_status` returns a `RunStatus`, one of
+ `notStarted()`, `running()`, `completed()`, or `failed(error)`. It must be a
+ read-only probe with no side effects. Report an expected failure as a
`FAILED`
+ status.
+- `callFetchResult` / `call_fetch_result` returns the `SubagentResult`. It
must be
+ an idempotent read, because a replay may call it again. The status snapshot
never
+ carries the payload, and a failed run is reported without fetching at all.
+
+{{< tabs "Async Sub-agent Setup" >}}
+
+{{< tab "Python" >}}
+```python
+from typing import Any
+
+from flink_agents.api.subagent import SubagentResult
+from flink_agents.runtime.async_subagent import BaseAsyncSubagentSetup,
RunStatus
+
+# client(...) and task_id stand for your own transport client and run-id
+# scheme; neither is part of Flink Agents.
+
+
+class RemoteRunSetup(BaseAsyncSubagentSetup):
+ endpoint: str = ""
+
+ def call_submit_request(self, session_id: str, call_id: str, prompt: Any)
-> None:
+ client(self.endpoint).submit(prompt, task_id=f"{session_id}#{call_id}")
+
+ def call_query_status(self, session_id: str, call_id: str) -> RunStatus:
+ probe = client(self.endpoint).status(task_id=f"{session_id}#{call_id}")
+ if probe is None:
+ return RunStatus.not_started()
+ if probe.succeeded:
+ return RunStatus.completed()
+ if probe.failed:
+ return RunStatus.failed(probe.error)
+ return RunStatus.running()
+
+ def call_fetch_result(self, session_id: str, call_id: str) ->
SubagentResult:
+ return SubagentResult.ok(
+ client(self.endpoint).fetch(task_id=f"{session_id}#{call_id}")
+ )
+
+ # Optional, and a no-op by default, so a cancelled call leaves the remote
run
+ # going. A replay may propagate the same cancellation again, so your
remote has
+ # to accept it twice.
+ def call_cancel_request(self, session_id: str, call_id: str) -> None:
+ client(self.endpoint).cancel(task_id=f"{session_id}#{call_id}")
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.subagent.SubagentResult;
+import org.apache.flink.agents.runtime.subagent.BaseAsyncSubagentSetup;
+
+// client() and taskId(...) stand for your own transport client and run-id
+// scheme; neither is part of Flink Agents.
+public class RemoteRunSetup extends BaseAsyncSubagentSetup {
+
+ private final String endpoint;
+
+ public RemoteRunSetup(ResourceDescriptor descriptor, ResourceContext
resourceContext) {
+ super(descriptor, resourceContext);
+ this.endpoint = descriptor.getArgument("endpoint");
+ }
+ @Override
+ protected void callSubmitRequest(String sessionId, String callId, Object
prompt)
+ throws Exception {
+ client().submit(String.valueOf(prompt), taskId(sessionId, callId));
+ }
+
+ @Override
+ protected RunStatus callQueryStatus(String sessionId, String callId) {
+ // TaskStatus is your own transport type, not a framework type.
+ TaskStatus probe = client().status(taskId(sessionId, callId));
+ if (probe == null) {
+ return RunStatus.notStarted();
+ }
+ if (probe.succeeded()) {
+ return RunStatus.completed();
+ }
+ if (probe.failed()) {
+ return RunStatus.failed(probe.error());
+ }
+ return RunStatus.running();
+ }
+
+ @Override
+ protected SubagentResult callFetchResult(String sessionId, String callId)
throws Exception {
+ return client().fetchResult(taskId(sessionId, callId));
+ }
+
+ // Optional, and a no-op by default, so a cancelled call leaves the remote
run
+ // going. A replay may propagate the same cancellation again, so your
remote has
+ // to accept it twice.
+ @Override
+ protected void callCancelRequest(String sessionId, String callId) {
+ client().cancel(taskId(sessionId, callId));
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+Register the setup like any external sub-agent. The async form adds one
descriptor
+argument, `status_poll_interval_millis`, and it defaults to 500.
+
+{{< tabs "Register an Async Sub-agent" >}}
+
+{{< tab "Python" >}}
+```python
parent.add_resource(
- "reviewer",
+ "remote_runner",
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(),
+ ResourceDescriptor(
+ clazz="my_app.agents.RemoteRunSetup",
+ endpoint="https://example.com/runs";,
+ description="Run a review task on the remote service",
+ input_schema=(
+ '{"type":"object","properties":{"diff":{"type":"string"}},'
+ '"required":["diff"]}'
+ ),
+ status_poll_interval_millis=500,
+ ),
)
```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+parent.addResource(
+ "remote_runner",
+ ResourceType.AGENT,
+ ResourceDescriptor.Builder.newBuilder(RemoteRunSetup.class.getName())
+ .addInitialArgument("endpoint", "https://example.com/runs";)
+ .addInitialArgument("description", "Run a review task on the remote
service")
+ .addInitialArgument(
+ "input_schema",
+
"{\"type\":\"object\",\"properties\":{\"diff\":{\"type\":\"string\"}},"
+ + "\"required\":[\"diff\"]}")
+ .addInitialArgument("status_poll_interval_millis", 500L)
+ .build());
+```
+{{< /tab >}}
+
+{{< tab "YAML" >}}
+```yaml
+agents:
+ - name: orchestrator
+ subagents:
+ - name: remote_runner
+ clazz: my_app.agents.RemoteRunSetup
+ endpoint: https://example.com/runs
+ description: Run a review task on the remote service
+ input_schema:
'{"type":"object","properties":{"diff":{"type":"string"}},"required":["diff"]}'
+ status_poll_interval_millis: 500
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The base class also recovers the crash window between "POST sent" and "POST
durably
+recorded". On recovery it probes the run's status first, and it re-posts only
when the
+probe reports that the run never started. Override `reconcileSubmitRequest` /
+`reconcile_submit_request` only if your service needs a different repair.
+
+## Invoke a sub-agent
+
+A sub-agent is invoked in one of two ways, model-driven or caller-driven, and
the two
+reuse different existing abstractions.
+
+Model-driven delegation rides the tool abstraction. The sub-agent is offered
to the
+model as a tool, so your action sends a `ChatRequestEvent` and the model
answers with
+a call to the sub-agent. The result then returns to the model as the tool
+observation, until a final `ChatResponseEvent` reaches your action. This is
the same
+cycle any tool call makes.
+
+Caller-driven delegation cannot ride that cycle, because a tool's result goes
back
+into the chat loop rather than to the action that asked for it. An action that
needs
+the delegated outcome therefore calls the sub-agent itself, with no model in
the loop.
+
+Both ways end in the same call. Your action, or the framework on the model's
behalf,
+runs `submit(ctx, prompt)` on the registered name. That call returns a
+`SubagentFuture`, which is the handle for that one call. Awaiting the handle
yields
+the call's `SubagentResult`.
+
+### Delegate from a chat model
+
+List the registered name under `subagents` of the parent's chat model:
+
+{{< tabs "Declare Sub-agents on a Chat Model" >}}
+
+{{< tab "Python" >}}
+```python
+ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP,
+ connection="ollama_connection",
+ model="qwen3:8b",
+ subagents=["reviewer"],
+)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
+ .addInitialArgument("connection", "ollamaConnection")
+ .addInitialArgument("model", "qwen3:8b")
+ .addInitialArgument("subagents", List.of("reviewer"))
+ .build();
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+The model sees `reviewer` as a tool, described by the registration's
description and
+input schema. A schema is required for model-selected calls. A missing
description
+falls back to a generic delegation description.
+
+The arguments the model sends become the sub-agent's input, and its outputs
return to
+the model as the tool observation. The model can then keep reasoning over
them. For
+an internal sub-agent, the observation is the JSON encoding of the full output
list
+described in [the result](#the-result). That holds even when the child emits a
single
+output, so the shape does not depend on how many outputs the child emits.
+
+An external setup class can state the shape of that observation. It declares
+`result_type()` in Python or `getResultType()` in Java, and the default
reports the
+result as it arrived. Declare one to narrow a wider payload, or to convert a
payload
+that is not already JSON-compatible. An internal sub-agent has no such
declaration, so
+its observation is always the JSON encoding of its output list. This shapes the
+observation only. A caller that awaits the handle gets the result the sub-agent
+produced, as [the result](#the-result) describes.
+
+The external sub-agents `remote_reviewer` and `remote_runner` are offered the
same
+way. Only their registration differs.
+
+### Call from an action
+
+An action can submit input to a registered sub-agent directly and await its
+outcome. This path needs no description and no input declaration, so it also
works
+for a child registered as-is.
+
+{{< tabs "Call a Sub-agent From an Action" >}}
+
+{{< tab "Python" >}}
+```python
+class DelegatingAgent(Agent):
+ @action(EventType.InputEvent)
+ @staticmethod
+ async def delegate(event: Event, ctx: RunnerContext) -> None:
+ reviewer = ctx.get_resource("reviewer", ResourceType.AGENT)
+ future = await reviewer.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}"))
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+public class DelegatingAgent extends Agent {
+
+ @Action(EventType.InputEvent)
+ public static void delegate(Event event, RunnerContext ctx) throws
Exception {
+ SubagentSetup reviewer =
+ (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+ SubagentResult result = reviewer.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()));
+ }
+ }
+}
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+What you pass to `submit` is what the sub-agent receives. An internal child
gets it
+as its `InputEvent` payload, and a model-driven call passes the arguments map
+instead. What comes back is a `SubagentResult`. For the internal sub-agent
used here,
+its `result` is the list of `OutputEvent` payloads the child emitted. That is
why the
+example reads `result.result[0]`. An external sub-agent returns whatever its
setup
+produces.
+
+In Python `submit` is a coroutine, so the call is awaited in two phases and the
+action must be `async`. In Java `submit` returns the handle directly and
`await()`
+yields the result. The same two calls drive `remote_reviewer` and
`remote_runner`
+unchanged. Only the registration differs.
+
+## Work with open handles
+
+A `SubagentFuture` is live from `submit` until it is awaited or cancelled.
+
+It works like the `AsyncFuture` that
+[async execution]({{< ref
"docs/development/create-agent/actions#async-execution" >}})
+returns. You get it from one call, and you await it to get the outcome. It is
its own
+type rather than an `AsyncFuture`, because it always resolves to a
`SubagentResult`,
+it carries the call's `(session_id, call_id)`, and it owns its durable
execution
+inside the setup.
+
+### Fan out to several sub-agents
+
+`combine` batches handles into a `SubagentFutures`, and it works the way
`gather`
+works for ordinary async futures. `gather` does not accept a `SubagentFuture`,
so
+handles batch this way instead. Every handle in the batch is prepared before
any is
+executed, so deferred requests all go out up front and remote runs overlap.
Results
+come back in the order the handles were added.
+
+{{< tabs "Fan Out to Several Sub-agents" >}}
+
+{{< tab "Python" >}}
+```python
+reviewer = ctx.get_resource("reviewer", ResourceType.AGENT)
+
+first = await reviewer.submit(ctx, "review the diff")
+second = await reviewer.submit(ctx, "review the tests")
+
+outcomes = await first.combine(second)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+SubagentSetup reviewer = (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+
+SubagentFuture first = reviewer.submit(ctx, "review the diff");
+SubagentFuture second = reviewer.submit(ctx, "review the tests");
+
+List<SubagentResult> outcomes = first.combine(second).awaitAll();
+```
+{{< /tab >}}
+
+{{< /tabs >}}
+
+### Continue a session
+
+Pass the `session_id` of an earlier handle to continue that conversation. The
new
+call gets a fresh `call_id` in the same session. Whether a conversation can be
+continued across actions is up to the implementation.
+
+{{< tabs "Continue a Session" >}}
+
+{{< tab "Python" >}}
+```python
+reviewer = ctx.get_resource("reviewer", ResourceType.AGENT)
+
+first = await reviewer.submit(ctx, "review the diff")
+first_result = await first
+
+second = await reviewer.submit(ctx, "now the tests", first.session_id)
+```
+{{< /tab >}}
+
+{{< tab "Java" >}}
+```java
+SubagentSetup reviewer = (SubagentSetup) ctx.getResource("reviewer",
ResourceType.AGENT);
+
+SubagentFuture first = reviewer.submit(ctx, "review the diff");
+SubagentResult firstResult = first.await();
+
+SubagentFuture second = reviewer.submit(ctx, "now the tests",
first.getSessionId());
Review Comment:
Fixed in both tabs. Swept the other examples: fan-out resolves through
combine, cancel through cancel().
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
