feat(cli): A2A Server Primary Agent Support

- Added a new `/agents use <name>` command to allow users to set a remote agent as the primary agent for the current session.
- Implemented `A2AStreamingAdapter` to seamlessly integrate A2A server task polling with the existing CLI streaming UX.
- Mapped A2A server task status updates to Gemini `Thought` events, providing native thinking visualization in the terminal.
- Updated `nonInteractiveCli` and `useGeminiStream` to dynamically route requests through the `A2AStreamingAdapter` when a session primary agent is configured.
- Ensured context history continuity by maintaining `contextId` and `taskId` across requests within the adapter.

packages/core/src/agents/a2aStreamingAdapter.ts

@@ -0,0 +1,199 @@
+/**
+ * @license
+ * Copyright 2025 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { FinishReason, type PartListUnion } from '@google/genai';
+import { type ServerGeminiStreamEvent, GeminiEventType } from '../core/turn.js';
+import { A2AClientManager } from './a2a-client-manager.js';
+import { extractTaskText, extractIdsFromResponse } from './a2aUtils.js';
+import { ADCHandler } from './remote-invocation.js';
+import type { Config } from '../config/config.js';
+import type { Task } from '@a2a-js/sdk';
+import type { ThoughtSummary } from '../utils/thoughtUtils.js';
+
+export class A2AStreamingAdapter {
+  private static sessionState = new Map<
+    string,
+    { contextId?: string; taskId?: string }
+  >();
+
+  constructor(private readonly config: Config) {}
+
+  async *sendMessageStream(
+    agentName: string,
+    request: PartListUnion,
+    signal: AbortSignal,
+    prompt_id: string,
+  ): AsyncGenerator<ServerGeminiStreamEvent> {
+    const clientManager = A2AClientManager.getInstance();
+    const registry = this.config.getAgentRegistry();
+    if (!registry) {
+      yield {
+        type: GeminiEventType.Error,
+        value: { error: new Error('Agent registry not found.') },
+      };
+      return;
+    }
+
+    const definition = registry.getDiscoveredDefinition(agentName);
+    if (!definition || definition.kind !== 'remote') {
+      yield {
+        type: GeminiEventType.Error,
+        value: { error: new Error(`Remote agent '${agentName}' not found.`) },
+      };
+      return;
+    }
+
+    // Determine query
+    const { partToString } = await import('../utils/partUtils.js');
+    const queryText = partToString(request);
+
+    // Load agent if needed
+    if (!clientManager.getClient(agentName)) {
+      try {
+        await clientManager.loadAgent(
+          agentName,
+          definition.agentCardUrl,
+          new ADCHandler(),
+        );
+      } catch (err) {
+        yield {
+          type: GeminiEventType.Error,
+          value: {
+            error: new Error(`Failed to load agent ${agentName}: ${err}`),
+          },
+        };
+        return;
+      }
+    }
+
+    const priorState = A2AStreamingAdapter.sessionState.get(agentName);
+    const contextId = priorState?.contextId;
+    const taskId = priorState?.taskId;
+
+    const client = clientManager.getClient(agentName);
+    if (!client) {
+      yield {
+        type: GeminiEventType.Error,
+        value: { error: new Error(`Client for ${agentName} not initialized.`) },
+      };
+      return;
+    }
+
+    try {
+      let currentTask: Task | undefined;
+      const response = await client.sendMessage({
+        message: {
+          kind: 'message',
+          role: 'user',
+          messageId: prompt_id,
+          parts: [{ kind: 'text', text: queryText }],
+          contextId,
+          taskId,
+        },
+        configuration: { blocking: false }, // NON-BLOCKING!
+      });
+
+      let currentTaskId = taskId;
+
+      if (response.kind === 'message') {
+        A2AStreamingAdapter.sessionState.set(
+          agentName,
+          extractIdsFromResponse(response),
+        );
+        if (response.parts && response.parts.length > 0) {
+          const finalMessageText = response.parts
+            .map((p) => (p.kind === 'text' ? p.text : ''))
+            .join('\\n');
+          if (finalMessageText) {
+            yield {
+              type: GeminiEventType.Content,
+              value: finalMessageText,
+              traceId: response.messageId,
+            };
+          }
+        }
+      } else if (response.kind === 'task') {
+        currentTask = response;
+        currentTaskId = response.id;
+        A2AStreamingAdapter.sessionState.set(agentName, {
+          contextId: response.contextId,
+          taskId: response.id,
+        });
+      }
+
+      if (currentTask && currentTaskId) {
+        let lastStatusMessage = '';
+        while (true) {
+          if (signal.aborted) {
+            await client.cancelTask({ id: currentTaskId });
+            yield { type: GeminiEventType.UserCancelled };
+            return;
+          }
+
+          currentTask = await client.getTask({ id: currentTaskId });
+
+          const state = currentTask.status?.state;
+          const statusMessage =
+            currentTask.status?.message?.parts
+              ?.map((p) => (p.kind === 'text' ? p.text : ''))
+              .join('\\n') || '';
+
+          if (statusMessage && statusMessage !== lastStatusMessage) {
+            const thought: ThoughtSummary = {
+              subject: 'Remote Action',
+              description: statusMessage,
+            };
+            yield {
+              type: GeminiEventType.Thought,
+              value: thought,
+            };
+            lastStatusMessage = statusMessage;
+          }
+
+          if (
+            state === 'completed' ||
+            state === 'failed' ||
+            state === 'canceled' ||
+            state === 'input-required'
+          ) {
+            A2AStreamingAdapter.sessionState.set(
+              agentName,
+              extractIdsFromResponse(currentTask),
+            );
+            break;
+          }
+
+          await new Promise((resolve) => setTimeout(resolve, 1000));
+        }
+
+        const finalOutput = extractTaskText(currentTask);
+        if (finalOutput) {
+          yield {
+            type: GeminiEventType.Content,
+            value: finalOutput,
+            traceId: currentTask.id,
+          };
+        }
+
+        // Also fire finished event when complete.
+        if (currentTask.status?.state === 'completed') {
+          yield {
+            type: GeminiEventType.Finished,
+            value: {
+              reason: FinishReason.STOP,
+              usageMetadata: undefined,
+            },
+          };
+        }
+      }
+    } catch (err) {
+      yield {
+        type: GeminiEventType.Error,
+        value: { error: new Error(`A2A Execution Error: ${err}`) },
+      };
+    }
+  }
+}

packages/core/src/agents/a2aUtils.ts

@@ -122,7 +122,10 @@ export function extractIdsFromResponse(result: Message | Task): {
   let taskId: string | undefined;
 
   if (result.kind === 'message') {
-    taskId = result.taskId;
+    // We explicitly DO NOT return the taskId for a 'message' response.
+    // In the A2A SDK, when a server returns a final Message instead of a Task,
+    // the task is implicitly complete and should not be passed to subsequent requests,
+    // otherwise the server will throw a TaskNotFoundError.
     contextId = result.contextId;
   } else if (result.kind === 'task') {
     taskId = result.id;

packages/core/src/config/config.ts

@@ -701,6 +701,7 @@ export class Config {
   private lastModeSwitchTime: number = performance.now();
   readonly userHintService: UserHintService;
   private approvedPlanPath: string | undefined;
+  private sessionPrimaryAgent: string | null = null;
 
   constructor(params: ConfigParameters) {
     this.sessionId = params.sessionId;
@@ -1936,6 +1937,14 @@ export class Config {
     return this.telemetrySettings.useCliAuth ?? false;
   }
 
+  getSessionPrimaryAgent(): string | null {
+    return this.sessionPrimaryAgent;
+  }
+
+  setSessionPrimaryAgent(agentName: string | null): void {
+    this.sessionPrimaryAgent = agentName;
+  }
+
   getGeminiClient(): GeminiClient {
     return this.geminiClient;
   }

packages/core/src/index.ts

@@ -195,6 +195,7 @@ export * from './hooks/types.js';
 
 // Export agent types
 export * from './agents/types.js';
+export * from './agents/a2aStreamingAdapter.js';
 
 // Export stdio utils
 export * from './utils/stdio.js';