Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 21 additions & 1 deletion a2a/src/main/java/com/google/adk/a2a/executor/AgentExecutor.java
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import com.google.adk.a2a.converters.EventConverter;
import com.google.adk.a2a.converters.PartConverter;
import com.google.adk.agents.BaseAgent;
import com.google.adk.agents.RunConfig;
import com.google.adk.apps.App;
import com.google.adk.artifacts.BaseArtifactService;
import com.google.adk.events.Event;
Expand All @@ -38,6 +39,7 @@
import io.a2a.spec.Artifact;
import io.a2a.spec.InvalidAgentResponseError;
import io.a2a.spec.Message;
import io.a2a.spec.MessageSendParams;
import io.a2a.spec.Part;
import io.a2a.spec.TaskArtifactUpdateEvent;
import io.a2a.spec.TaskState;
Expand All @@ -63,6 +65,7 @@
public class AgentExecutor implements io.a2a.server.agentexecution.AgentExecutor {
private static final Logger logger = LoggerFactory.getLogger(AgentExecutor.class);
private static final String USER_ID_PREFIX = "A2A_USER_";
private static final String A2A_METADATA_KEY = "a2a_metadata";
private final Map<String, Disposable> activeTasks = new ConcurrentHashMap<>();
private final Runner.Builder runnerBuilder;
private final AgentExecutorConfig agentExecutorConfig;
Expand Down Expand Up @@ -217,7 +220,7 @@ public void execute(RequestContext ctx, EventQueue eventQueue) {
getUserId(ctx),
session.id(),
content,
agentExecutorConfig.runConfig());
runConfigWithA2aMetadata(ctx));
});
})
.concatMap(
Expand Down Expand Up @@ -273,6 +276,23 @@ private String getUserId(RequestContext ctx) {
return USER_ID_PREFIX + ctx.getContextId();
}

/**
* Returns the configured run config enriched with the caller's incoming A2A request metadata
* under the {@code a2a_metadata} key, so downstream processing can read it via {@link
* RunConfig#customMetadata()}.
*/
private RunConfig runConfigWithA2aMetadata(RequestContext ctx) {
RunConfig runConfig = agentExecutorConfig.runConfig();
MessageSendParams params = ctx.getParams();
Map<String, Object> requestMetadata = params == null ? null : params.metadata();
if (requestMetadata == null || requestMetadata.isEmpty()) {
return runConfig;
}
Map<String, Object> customMetadata = new HashMap<>(runConfig.customMetadata());
customMetadata.put(A2A_METADATA_KEY, requestMetadata);
return runConfig.toBuilder().customMetadata(customMetadata).build();
}

private Maybe<Session> prepareSession(
RequestContext ctx, String appName, BaseSessionService service) {
return service
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

import com.google.adk.agents.BaseAgent;
import com.google.adk.agents.InvocationContext;
import com.google.adk.agents.RunConfig;
import com.google.adk.apps.App;
import com.google.adk.artifacts.InMemoryArtifactService;
import com.google.adk.events.Event;
Expand All @@ -24,6 +25,7 @@
import io.a2a.server.agentexecution.RequestContext;
import io.a2a.server.events.EventQueue;
import io.a2a.spec.Message;
import io.a2a.spec.MessageSendParams;
import io.a2a.spec.TaskArtifactUpdateEvent;
import io.a2a.spec.TaskState;
import io.a2a.spec.TaskStatus;
Expand Down Expand Up @@ -342,6 +344,62 @@ public void execute_runnerSucceeds_registerCompletedTaskFails_noFailedTaskRegist
assertThat(statusEvents).isEmpty();
}

@Test
public void execute_propagatesRequestMetadataIntoRunConfig() {
testAgent.setEventsToEmit(Flowable.empty());
AgentExecutor executor =
new AgentExecutor.Builder()
.agentExecutorConfig(AgentExecutorConfig.builder().build())
.app(App.builder().name("test_app").rootAgent(testAgent).build())
.sessionService(new InMemorySessionService())
.artifactService(new InMemoryArtifactService())
.build();

Message message =
new Message.Builder()
.messageId("msg-1")
.role(Message.Role.USER)
.parts(ImmutableList.of(new TextPart("trigger")))
.build();
MessageSendParams params =
new MessageSendParams.Builder()
.message(message)
.metadata(ImmutableMap.of("key", "value"))
.build();
RequestContext ctx = mock(RequestContext.class);
when(ctx.getMessage()).thenReturn(message);
when(ctx.getTaskId()).thenReturn("task-1");
when(ctx.getContextId()).thenReturn("ctx-1");
when(ctx.getParams()).thenReturn(params);

executor.execute(ctx, eventQueue);

// The runner passes the enriched run config down to the agent's invocation context.
RunConfig runConfig = testAgent.lastInvocationContext.runConfig();
assertThat(runConfig.customMetadata())
.containsEntry("a2a_metadata", ImmutableMap.of("key", "value"));
}

@Test
public void execute_withoutRequestMetadata_leavesRunConfigCustomMetadataEmpty() {
testAgent.setEventsToEmit(Flowable.empty());
AgentExecutor executor =
new AgentExecutor.Builder()
.agentExecutorConfig(AgentExecutorConfig.builder().build())
.app(App.builder().name("test_app").rootAgent(testAgent).build())
.sessionService(new InMemorySessionService())
.artifactService(new InMemoryArtifactService())
.build();

// createRequestContext() does not stub getParams(), mirroring a request with no metadata.
RequestContext ctx = createRequestContext();

executor.execute(ctx, eventQueue);

RunConfig runConfig = testAgent.lastInvocationContext.runConfig();
assertThat(runConfig.customMetadata()).doesNotContainKey("a2a_metadata");
}

private RequestContext createRequestContext() {
Message message =
new Message.Builder()
Expand Down Expand Up @@ -507,6 +565,7 @@ public void execute_withDefaultArtifactPerRun_emitsMessageAndLastChunk() {

private static final class TestAgent extends BaseAgent {
private Flowable<Event> eventsToEmit;
private volatile InvocationContext lastInvocationContext;

TestAgent() {
this(Flowable.empty());
Expand All @@ -524,6 +583,7 @@ void setEventsToEmit(Flowable<Event> events) {

@Override
protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
this.lastInvocationContext = invocationContext;
return eventsToEmit;
}

Expand Down