feat(a2a): Introduce /init command for a2a server (#13419)

packages/a2a-server/src/http/app.test.ts

@@ -1061,5 +1061,118 @@ describe('E2E Tests', () => {
       expect(response.status).toBe(200);
       expect(response.body.data).toBe('success');
     });
+
+    it('should include agentExecutor in context', async () => {
+      const mockCommand = {
+        name: 'context-check-command',
+        description: 'checks context',
+        execute: vi.fn(async (context: CommandContext) => {
+          if (!context.agentExecutor) {
+            throw new Error('agentExecutor missing');
+          }
+          return { name: 'context-check-command', data: 'success' };
+        }),
+      };
+      vi.spyOn(commandRegistry, 'get').mockReturnValue(mockCommand);
+
+      const agent = request.agent(app);
+      const res = await agent
+        .post('/executeCommand')
+        .send({ command: 'context-check-command', args: [] })
+        .set('Content-Type', 'application/json')
+        .expect(200);
+
+      expect(res.body.data).toBe('success');
+    });
+
+    describe('/executeCommand streaming', () => {
+      it('should execute a streaming command and stream back events', (done: (
+        err?: unknown,
+      ) => void) => {
+        const executeSpy = vi.fn(async (context: CommandContext) => {
+          context.eventBus?.publish({
+            kind: 'status-update',
+            status: { state: 'working' },
+            taskId: 'test-task',
+            contextId: 'test-context',
+            final: false,
+          });
+          context.eventBus?.publish({
+            kind: 'status-update',
+            status: { state: 'completed' },
+            taskId: 'test-task',
+            contextId: 'test-context',
+            final: true,
+          });
+          return { name: 'stream-test', data: 'done' };
+        });
+
+        const mockStreamCommand = {
+          name: 'stream-test',
+          description: 'A test streaming command',
+          streaming: true,
+          execute: executeSpy,
+        };
+        vi.spyOn(commandRegistry, 'get').mockReturnValue(mockStreamCommand);
+
+        const agent = request.agent(app);
+        agent
+          .post('/executeCommand')
+          .send({ command: 'stream-test', args: [] })
+          .set('Content-Type', 'application/json')
+          .set('Accept', 'text/event-stream')
+          .on('response', (res) => {
+            let data = '';
+            res.on('data', (chunk: Buffer) => {
+              data += chunk.toString();
+            });
+            res.on('end', () => {
+              try {
+                const events = streamToSSEEvents(data);
+                expect(events.length).toBe(2);
+                expect(events[0].result).toEqual({
+                  kind: 'status-update',
+                  status: { state: 'working' },
+                  taskId: 'test-task',
+                  contextId: 'test-context',
+                  final: false,
+                });
+                expect(events[1].result).toEqual({
+                  kind: 'status-update',
+                  status: { state: 'completed' },
+                  taskId: 'test-task',
+                  contextId: 'test-context',
+                  final: true,
+                });
+                expect(executeSpy).toHaveBeenCalled();
+                done();
+              } catch (e) {
+                done(e);
+              }
+            });
+          })
+          .end();
+      });
+
+      it('should handle non-streaming commands gracefully', async () => {
+        const mockNonStreamCommand = {
+          name: 'non-stream-test',
+          description: 'A test non-streaming command',
+          execute: vi
+            .fn()
+            .mockResolvedValue({ name: 'non-stream-test', data: 'done' }),
+        };
+        vi.spyOn(commandRegistry, 'get').mockReturnValue(mockNonStreamCommand);
+
+        const agent = request.agent(app);
+        const res = await agent
+          .post('/executeCommand')
+          .send({ command: 'non-stream-test', args: [] })
+          .set('Content-Type', 'application/json')
+          .expect(200);
+
+        expect(res.body).toEqual({ name: 'non-stream-test', data: 'done' });
+      });
+    });
   });
 });

packages/a2a-server/src/http/app.ts

@@ -6,9 +6,14 @@
 
 import express from 'express';
 
-import type { AgentCard } from '@a2a-js/sdk';
+import type { AgentCard, Message } from '@a2a-js/sdk';
 import type { TaskStore } from '@a2a-js/sdk/server';
-import { DefaultRequestHandler, InMemoryTaskStore } from '@a2a-js/sdk/server';
+import {
+  DefaultRequestHandler,
+  InMemoryTaskStore,
+  DefaultExecutionEventBus,
+  type AgentExecutionEvent,
+} from '@a2a-js/sdk/server';
 import { A2AExpressApp } from '@a2a-js/sdk/server/express'; // Import server components
 import { v4 as uuidv4 } from 'uuid';
 import { logger } from '../utils/logger.js';
@@ -73,6 +78,76 @@ export function updateCoderAgentCardUrl(port: number) {
   coderAgentCard.url = `http://localhost:${port}/`;
 }
 
+async function handleExecuteCommand(
+  req: express.Request,
+  res: express.Response,
+  context: {
+    config: Awaited<ReturnType<typeof loadConfig>>;
+    git: GitService | undefined;
+    agentExecutor: CoderAgentExecutor;
+  },
+) {
+  logger.info('[CoreAgent] Received /executeCommand request: ', req.body);
+  const { command, args } = req.body;
+  try {
+    if (typeof command !== 'string') {
+      return res.status(400).json({ error: 'Invalid "command" field.' });
+    }
+
+    if (args && !Array.isArray(args)) {
+      return res.status(400).json({ error: '"args" field must be an array.' });
+    }
+
+    const commandToExecute = commandRegistry.get(command);
+
+    if (commandToExecute?.requiresWorkspace) {
+      if (!process.env['CODER_AGENT_WORKSPACE_PATH']) {
+        return res.status(400).json({
+          error: `Command "${command}" requires a workspace, but CODER_AGENT_WORKSPACE_PATH is not set.`,
+        });
+      }
+    }
+
+    if (!commandToExecute) {
+      return res.status(404).json({ error: `Command not found: ${command}` });
+    }
+
+    if (commandToExecute.streaming) {
+      const eventBus = new DefaultExecutionEventBus();
+      res.setHeader('Content-Type', 'text/event-stream');
+      const eventHandler = (event: AgentExecutionEvent) => {
+        const jsonRpcResponse = {
+          jsonrpc: '2.0',
+          id: 'taskId' in event ? event.taskId : (event as Message).messageId,
+          result: event,
+        };
+        res.write(`data: ${JSON.stringify(jsonRpcResponse)}\n`);
+      };
+      eventBus.on('event', eventHandler);
+
+      await commandToExecute.execute({ ...context, eventBus }, args ?? []);
+
+      eventBus.off('event', eventHandler);
+      eventBus.finished();
+      return res.end(); // Explicit return for streaming path
+    } else {
+      const result = await commandToExecute.execute(context, args ?? []);
+      logger.info('[CoreAgent] Sending /executeCommand response: ', result);
+      return res.status(200).json(result);
+    }
+  } catch (e) {
+    logger.error(
+      `Error executing /executeCommand: ${command} with args: ${JSON.stringify(
+        args,
+      )}`,
+      e,
+    );
+    const errorMessage =
+      e instanceof Error ? e.message : 'Unknown error executing command';
+    return res.status(500).json({ error: errorMessage });
+  }
+}
+
 export async function createApp() {
   try {
     // Load the server configuration once on startup.
@@ -92,8 +167,6 @@ export async function createApp() {
       await git.initialize();
     }
 
-    const context = { config, git };
-
     // loadEnvironment() is called within getConfig now
     const bucketName = process.env['GCS_BUCKET_NAME'];
     let taskStoreForExecutor: TaskStore;
@@ -113,6 +186,8 @@ export async function createApp() {
 
     const agentExecutor = new CoderAgentExecutor(taskStoreForExecutor);
 
+    const context = { config, git, agentExecutor };
+
     const requestHandler = new DefaultRequestHandler(
       coderAgentCard,
       taskStoreForHandler,
@@ -152,46 +227,8 @@ export async function createApp() {
       }
     });
 
-    expressApp.post('/executeCommand', async (req, res) => {
-      logger.info('[CoreAgent] Received /executeCommand request: ', req.body);
-      try {
-        const { command, args } = req.body;
-
-        if (typeof command !== 'string') {
-          return res.status(400).json({ error: 'Invalid "command" field.' });
-        }
-
-        if (args && !Array.isArray(args)) {
-          return res
-            .status(400)
-            .json({ error: '"args" field must be an array.' });
-        }
-
-        const commandToExecute = commandRegistry.get(command);
-
-        if (commandToExecute?.requiresWorkspace) {
-          if (!process.env['CODER_AGENT_WORKSPACE_PATH']) {
-            return res.status(400).json({
-              error: `Command "${command}" requires a workspace, but CODER_AGENT_WORKSPACE_PATH is not set.`,
-            });
-          }
-        }
-
-        if (!commandToExecute) {
-          return res
-            .status(404)
-            .json({ error: `Command not found: ${command}` });
-        }
-
-        const result = await commandToExecute.execute(context, args ?? []);
-        logger.info('[CoreAgent] Sending /executeCommand response: ', result);
-        return res.status(200).json(result);
-      } catch (e) {
-        logger.error('Error executing /executeCommand:', e);
-        const errorMessage =
-          e instanceof Error ? e.message : 'Unknown error executing command';
-        return res.status(500).json({ error: errorMessage });
-      }
+    expressApp.post('/executeCommand', (req, res) => {
+      void handleExecuteCommand(req, res, context);
     });
 
     expressApp.get('/listCommands', (req, res) => {