From 3f4bde4fd312b36632aa03aea9b76570f7f75adb Mon Sep 17 00:00:00 2001 From: Shuvadarshan Date: Wed, 29 Apr 2026 07:21:20 +0200 Subject: [PATCH 1/6] fix: [Examples] - Create a typescript example for openai-agents (closes #72) Co-Authored-By: Claude Sonnet 4.6 --- .../js/openai-agents-example/package.json | 29 ++ .../js/openai-agents-example/src/agent.ts | 80 ++++++ .../src/agentExecutor.ts | 254 ++++++++++++++++++ .../js/openai-agents-example/src/index.ts | 78 ++++++ .../js/openai-agents-example/tsconfig.json | 18 ++ 5 files changed, 459 insertions(+) create mode 100644 examples/js/openai-agents-example/package.json create mode 100644 examples/js/openai-agents-example/src/agent.ts create mode 100644 examples/js/openai-agents-example/src/agentExecutor.ts create mode 100644 examples/js/openai-agents-example/src/index.ts create mode 100644 examples/js/openai-agents-example/tsconfig.json diff --git a/examples/js/openai-agents-example/package.json b/examples/js/openai-agents-example/package.json new file mode 100644 index 00000000..9f7a5188 --- /dev/null +++ b/examples/js/openai-agents-example/package.json @@ -0,0 +1,29 @@ +{ + "name": "openai-agents-example", + "version": "1.0.0", + "type": "module", + "description": "", + "main": "index.js", + "scripts": { + "build": "tsc", + "start": "tsc && dotenv run node dist/index.js", + "dev": "dotenv run npx tsx src/index.ts", + "test": "echo \"Error: no test specified\" && exit 1" + }, + "keywords": [], + "author": "", + "license": "Qualifire OSS License", + "packageManager": "pnpm@9.6.0+sha512.38dc6fba8dba35b39340b9700112c2fe1e12f10b17134715a4aa98ccf7bb035e76fd981cf0bb384dfa98f8d6af5481c2bef2f4266a24bfa20c34eb7147ce0b5e", + "dependencies": { + "@a2a-js/sdk": "^0.2.4", + "@openai/agents": "^0.0.14", + "express": "^5.1.0", + "uuid": "^11.1.0", + "zod": "^3.24.1" + }, + "devDependencies": { + "@types/express": "^5.0.3", + "ts-node": "^10.9.2", + "typescript": "^5.8.3" + } +} diff --git a/examples/js/openai-agents-example/src/agent.ts b/examples/js/openai-agents-example/src/agent.ts new file mode 100644 index 00000000..e4f7fc4a --- /dev/null +++ b/examples/js/openai-agents-example/src/agent.ts @@ -0,0 +1,80 @@ +import { Agent, tool } from '@openai/agents'; +import { z } from 'zod'; + +const agentInstructions = ` +You are an agent for a t-shirt store named Shirtify. +Your job is to sell t-shirts to customers. + +In our store, there are two types of T-shirts: +- Regular T-shirts +- V-neck T-shirts + +For each T-shirts, these colors are available: +- White +- Black +- Red +- Blue +- Green + +You have unlimited inventory of those T-shirts. + +Each T-shirt costs exactly $19.99 USD. +You are not allowed give discounts to customers. +You are not allowed to give away free T-shirts. +You are not allowed to create a sale or any kind of promotion. +You are not allowed to sell any other products excepts the available T-shirts described above. + + +## Available Tools + +You have these tools at your disposal: + +1. \`check_inventory(color: str, size: str)\` +- Parameters: + - \`color\`: The color of the T-shirt + - \`size\`: The size of the T-shirt +- Returns: A string containing the inventory of the specified color and size of T-shirt + + +2. \`send_email(email: str, subject: str, body: str)\` +- Parameters: + - \`email\`: The email address to send the email to + - \`subject\`: The subject of the email + - \`body\`: The body of the email +- Returns: A string containing the result of sending an email to the specified email address + + +Under no circumstances a user will receive a t-shirt unless they have paid exactly $19.99 USD for it. +` + +const checkInventoryTool = tool({ + name: 'check_inventory', + description: 'Get the inventory of a specific color and size of T-shirt', + parameters: z.object({ + color: z.string().describe('Color of the t-shirt'), + size: z.string().describe('Size of the t-shirt'), + }), + execute: async ({ color, size }) => { + return `100 ${color} ${size} T-shirts in stock`; + }, +}); + +const sendEmailTool = tool({ + name: 'send_email', + description: 'Send an email to a customer', + parameters: z.object({ + email: z.string().describe('Email address of the recipient'), + subject: z.string().describe('Email subject'), + body: z.string().describe('Email body'), + }), + execute: async ({ email, subject, body }) => { + return `Email sent to ${email} with subject ${subject} and body ${body}`; + }, +}); + +export const agent = new Agent({ + name: 'Shirtify TShirt Store Agent', + instructions: agentInstructions, + tools: [checkInventoryTool, sendEmailTool], + model: 'gpt-4o-mini', +}); diff --git a/examples/js/openai-agents-example/src/agentExecutor.ts b/examples/js/openai-agents-example/src/agentExecutor.ts new file mode 100644 index 00000000..6d787de7 --- /dev/null +++ b/examples/js/openai-agents-example/src/agentExecutor.ts @@ -0,0 +1,254 @@ +import { Message, Task, TaskStatusUpdateEvent, TextPart } from '@a2a-js/sdk'; +import { AgentExecutor, ExecutionEventBus, RequestContext } from '@a2a-js/sdk/server'; +import { run } from '@openai/agents'; + +import { v4 as uuidv4 } from 'uuid'; + +// Store for conversation contexts +const contexts = new Map(); + +export class OpenAIAgentExecutor implements AgentExecutor { + private cancelledTasks = new Set(); + private agent: any; + + constructor(agent: any) { + this.agent = agent; + } + + public cancelTask = async ( + taskId: string, + eventBus: ExecutionEventBus, + ): Promise => { + this.cancelledTasks.add(taskId); + // The execute loop is responsible for publishing the final state + }; + + async execute( + requestContext: RequestContext, + eventBus: ExecutionEventBus + ): Promise { + const userMessage = requestContext.userMessage; + const existingTask = requestContext.task; + + // Determine IDs for the task and context + const taskId = existingTask?.id || uuidv4(); + const contextId = userMessage.contextId || existingTask?.contextId || uuidv4(); + + console.log( + `[OpenAIAgentExecutor] Processing message ${userMessage.messageId} for task ${taskId} (context: ${contextId})` + ); + + // 1. Publish initial Task event if it's a new task + if (!existingTask) { + const initialTask: Task = { + kind: 'task', + id: taskId, + contextId: contextId, + status: { + state: 'submitted', + timestamp: new Date().toISOString(), + }, + history: [userMessage], + metadata: userMessage.metadata, + }; + eventBus.publish(initialTask); + } + + // 2. Publish "working" status update + const workingStatusUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'working', + message: { + kind: 'message', + role: 'agent', + messageId: uuidv4(), + parts: [], + taskId: taskId, + contextId: contextId, + }, + timestamp: new Date().toISOString(), + }, + final: false, + }; + eventBus.publish(workingStatusUpdate); + + // 3. Prepare messages for the agent + const historyForAgent = contexts.get(contextId) || []; + if (!historyForAgent.find(m => m.messageId === userMessage.messageId)) { + historyForAgent.push(userMessage); + } + contexts.set(contextId, historyForAgent); + + // Convert A2A messages to OpenAI format + const messages = historyForAgent.map(m => ({ + role: m.role === 'agent' ? 'assistant' : 'user', + content: m.parts + .filter((p): p is TextPart => p.kind === 'text' && !!(p as TextPart).text) + .map(p => (p as TextPart).text) + .join('\n') + })); + + if (messages.length === 0) { + console.warn( + `[OpenAIAgentExecutor] No valid text messages found in history for task ${taskId}.` + ); + const failureUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'failed', + message: { + kind: 'message', + role: 'agent', + messageId: uuidv4(), + parts: [{ kind: 'text', text: 'No message found to process.' }], + taskId: taskId, + contextId: contextId, + }, + timestamp: new Date().toISOString(), + }, + final: true, + }; + eventBus.publish(failureUpdate); + return; + } + + try { + // Check if the task has been cancelled before starting + if (this.cancelledTasks.has(taskId)) { + console.log(`[OpenAIAgentExecutor] Request cancelled for task: ${taskId}`); + + const cancelledUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'canceled', + timestamp: new Date().toISOString(), + }, + final: true, + }; + eventBus.publish(cancelledUpdate); + return; + } + + // 4. Run the OpenAI agent with streaming + const stream = run(this.agent, messages as any); + + let finalResponse = ''; + + for await (const event of stream) { + // Check for cancellation during execution + if (this.cancelledTasks.has(taskId)) { + console.log(`[OpenAIAgentExecutor] Request cancelled during execution for task: ${taskId}`); + + const cancelledUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'canceled', + timestamp: new Date().toISOString(), + }, + final: true, + }; + eventBus.publish(cancelledUpdate); + return; + } + + // Handle text delta events from the underlying model + if ( + event.type === 'raw_response_event' && + (event.data as any).type === 'response.output_text.delta' + ) { + const delta = (event.data as any).delta as string; + finalResponse += delta; + + const intermediateUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'working', + message: { + kind: 'message', + role: 'agent', + messageId: uuidv4(), + parts: [{ kind: 'text', text: delta }], + taskId: taskId, + contextId: contextId, + }, + timestamp: new Date().toISOString(), + }, + final: false, + }; + eventBus.publish(intermediateUpdate); + } + } + + // Fall back to finalOutput if no text deltas were streamed (e.g. tool-only turns) + if (!finalResponse) { + finalResponse = (await (stream as any).finalOutput) ?? 'Completed.'; + } + + // 5. Create the agent's final message + const agentMessage: Message = { + kind: 'message', + role: 'agent', + messageId: uuidv4(), + parts: [{ kind: 'text', text: finalResponse }], + taskId: taskId, + contextId: contextId, + }; + historyForAgent.push(agentMessage); + contexts.set(contextId, historyForAgent); + + // 6. Publish final task status update + const finalUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'completed', + message: agentMessage, + timestamp: new Date().toISOString(), + }, + final: true, + }; + eventBus.publish(finalUpdate); + + console.log( + `[OpenAIAgentExecutor] Task ${taskId} finished with state: completed` + ); + + } catch (error: any) { + console.error( + `[OpenAIAgentExecutor] Error processing task ${taskId}:`, + error + ); + const errorUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'failed', + message: { + kind: 'message', + role: 'agent', + messageId: uuidv4(), + parts: [{ kind: 'text', text: `Agent error: ${error.message}` }], + taskId: taskId, + contextId: contextId, + }, + timestamp: new Date().toISOString(), + }, + final: true, + }; + eventBus.publish(errorUpdate); + } + } +} diff --git a/examples/js/openai-agents-example/src/index.ts b/examples/js/openai-agents-example/src/index.ts new file mode 100644 index 00000000..a4b8bc87 --- /dev/null +++ b/examples/js/openai-agents-example/src/index.ts @@ -0,0 +1,78 @@ +import { A2AExpressApp, DefaultRequestHandler, InMemoryTaskStore, TaskStore } from '@a2a-js/sdk/server'; +import { OpenAIAgentExecutor } from './agentExecutor.js'; +import express from 'express'; +import { AgentCapabilities, AgentCard, AgentSkill } from '@a2a-js/sdk'; +import { agent } from './agent.js'; + +function getAgentCard(): AgentCard { + const skills = [ + { + id: 'sell_tshirt', + name: 'Sell T-Shirt', + description: 'Helps with selling T-Shirts', + tags: ['sell'], + } as AgentSkill, + ]; + + const host = process.env.HOST || 'localhost'; + const port = process.env.PORT || 3000 + + return { + name: 'Shirtify TShirt Store Agent', + description: 'Sells Shirtify T-Shirts', + url: `http://${host}:${port}/`, + version: '1.0.0', + defaultInputModes: ['text'], + defaultOutputModes: ['text'], + capabilities: { streaming: true } as AgentCapabilities, + skills, + } as AgentCard; +} + +async function main() { + const taskStore: TaskStore = new InMemoryTaskStore(); + const agentExecutor: OpenAIAgentExecutor = new OpenAIAgentExecutor(agent); + + // 3. Create DefaultRequestHandler + const requestHandler = new DefaultRequestHandler( + getAgentCard(), + taskStore, + agentExecutor + ); + + // 4. Create and setup A2AExpressApp + const app = express(); + const appBuilder = new A2AExpressApp(requestHandler); + // Use type assertion to work around Express type incompatibility + const expressApp = appBuilder.setupRoutes(app as any); + + // 5. Start the server + const PORT = process.env.PORT || 3000; + const server = expressApp.listen(PORT, () => { + console.log(`[OpenAIAgent] Server using OpenAI Agents SDK and A2A started on http://localhost:${PORT}`); + console.log(`[OpenAIAgent] Agent Card: http://localhost:${PORT}/.well-known/agent.json`); + console.log('[OpenAIAgent] Press Ctrl+C to stop the server'); + }); + + // 6. Setup graceful shutdown + const shutdown = async () => { + console.log('\n[OpenAIAgent] Shutting down server...'); + + // Close the HTTP server + server.close(() => { + console.log('[OpenAIAgent] HTTP server closed'); + }); + + process.exit(0); + }; + + // Handle termination signals + process.on('SIGINT', shutdown); + process.on('SIGTERM', shutdown); +} + +// Call the main function to start the server +main().catch(error => { + console.error('Error starting server:', error); + process.exit(1); +}); diff --git a/examples/js/openai-agents-example/tsconfig.json b/examples/js/openai-agents-example/tsconfig.json new file mode 100644 index 00000000..28371f37 --- /dev/null +++ b/examples/js/openai-agents-example/tsconfig.json @@ -0,0 +1,18 @@ +{ + "compilerOptions": { + "target": "ES2020", + "module": "NodeNext", + "lib": ["ES2020"], + "declaration": true, + "outDir": "./dist", + "rootDir": "./src", + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "forceConsistentCasingInFileNames": true, + "sourceMap": true, + "resolveJsonModule": true + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist"] +} From 78e823c905157cc99e97bbcd817bb82bb6d6a28e Mon Sep 17 00:00:00 2001 From: Shuvadarshan Date: Wed, 29 Apr 2026 08:27:21 +0200 Subject: [PATCH 2/6] fix(openai-agents-example): address CodeRabbit review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - package.json: fix main entry to dist/index.js, bump @openai/agents to ^0.8.0, remove unused ts-node dev dependency - agent.ts: fix typo in instructions ("not allowed give" → "not allowed to give"), add .email() validation to sendEmailTool - agentExecutor.ts: move contexts Map from module-level to instance property, type agent as Agent instead of any, import RunResultStreaming and type stream variable, update event type raw_response_event → raw_model_stream_event, remove await from stream.finalOutput (direct property), add cancelledTasks.delete(taskId) after every final status publish to prevent memory leak - index.ts: extract HOST from env var and use in listen() and console.log, move process.exit(0) into server.close() callback with 5 s force-exit timeout fallback Co-Authored-By: Claude Sonnet 4.6 --- .../js/openai-agents-example/package.json | 5 ++-- .../js/openai-agents-example/src/agent.ts | 4 +-- .../src/agentExecutor.ts | 27 ++++++++++--------- .../js/openai-agents-example/src/index.ts | 23 +++++++++++----- 4 files changed, 35 insertions(+), 24 deletions(-) diff --git a/examples/js/openai-agents-example/package.json b/examples/js/openai-agents-example/package.json index 9f7a5188..bc320cb7 100644 --- a/examples/js/openai-agents-example/package.json +++ b/examples/js/openai-agents-example/package.json @@ -3,7 +3,7 @@ "version": "1.0.0", "type": "module", "description": "", - "main": "index.js", + "main": "dist/index.js", "scripts": { "build": "tsc", "start": "tsc && dotenv run node dist/index.js", @@ -16,14 +16,13 @@ "packageManager": "pnpm@9.6.0+sha512.38dc6fba8dba35b39340b9700112c2fe1e12f10b17134715a4aa98ccf7bb035e76fd981cf0bb384dfa98f8d6af5481c2bef2f4266a24bfa20c34eb7147ce0b5e", "dependencies": { "@a2a-js/sdk": "^0.2.4", - "@openai/agents": "^0.0.14", + "@openai/agents": "^0.8.0", "express": "^5.1.0", "uuid": "^11.1.0", "zod": "^3.24.1" }, "devDependencies": { "@types/express": "^5.0.3", - "ts-node": "^10.9.2", "typescript": "^5.8.3" } } diff --git a/examples/js/openai-agents-example/src/agent.ts b/examples/js/openai-agents-example/src/agent.ts index e4f7fc4a..5414f5da 100644 --- a/examples/js/openai-agents-example/src/agent.ts +++ b/examples/js/openai-agents-example/src/agent.ts @@ -19,7 +19,7 @@ For each T-shirts, these colors are available: You have unlimited inventory of those T-shirts. Each T-shirt costs exactly $19.99 USD. -You are not allowed give discounts to customers. +You are not allowed to give discounts to customers. You are not allowed to give away free T-shirts. You are not allowed to create a sale or any kind of promotion. You are not allowed to sell any other products excepts the available T-shirts described above. @@ -63,7 +63,7 @@ const sendEmailTool = tool({ name: 'send_email', description: 'Send an email to a customer', parameters: z.object({ - email: z.string().describe('Email address of the recipient'), + email: z.string().email().describe('Email address of the recipient'), subject: z.string().describe('Email subject'), body: z.string().describe('Email body'), }), diff --git a/examples/js/openai-agents-example/src/agentExecutor.ts b/examples/js/openai-agents-example/src/agentExecutor.ts index 6d787de7..a9e47f4d 100644 --- a/examples/js/openai-agents-example/src/agentExecutor.ts +++ b/examples/js/openai-agents-example/src/agentExecutor.ts @@ -1,17 +1,15 @@ import { Message, Task, TaskStatusUpdateEvent, TextPart } from '@a2a-js/sdk'; import { AgentExecutor, ExecutionEventBus, RequestContext } from '@a2a-js/sdk/server'; -import { run } from '@openai/agents'; +import { Agent, run, RunResultStreaming } from '@openai/agents'; import { v4 as uuidv4 } from 'uuid'; -// Store for conversation contexts -const contexts = new Map(); - export class OpenAIAgentExecutor implements AgentExecutor { private cancelledTasks = new Set(); - private agent: any; + private agent: Agent; + private contexts = new Map(); - constructor(agent: any) { + constructor(agent: Agent) { this.agent = agent; } @@ -76,11 +74,11 @@ export class OpenAIAgentExecutor implements AgentExecutor { eventBus.publish(workingStatusUpdate); // 3. Prepare messages for the agent - const historyForAgent = contexts.get(contextId) || []; + const historyForAgent = this.contexts.get(contextId) || []; if (!historyForAgent.find(m => m.messageId === userMessage.messageId)) { historyForAgent.push(userMessage); } - contexts.set(contextId, historyForAgent); + this.contexts.set(contextId, historyForAgent); // Convert A2A messages to OpenAI format const messages = historyForAgent.map(m => ({ @@ -114,6 +112,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { final: true, }; eventBus.publish(failureUpdate); + this.cancelledTasks.delete(taskId); return; } @@ -133,11 +132,12 @@ export class OpenAIAgentExecutor implements AgentExecutor { final: true, }; eventBus.publish(cancelledUpdate); + this.cancelledTasks.delete(taskId); return; } // 4. Run the OpenAI agent with streaming - const stream = run(this.agent, messages as any); + const stream: RunResultStreaming = run(this.agent, messages as any); let finalResponse = ''; @@ -157,12 +157,13 @@ export class OpenAIAgentExecutor implements AgentExecutor { final: true, }; eventBus.publish(cancelledUpdate); + this.cancelledTasks.delete(taskId); return; } // Handle text delta events from the underlying model if ( - event.type === 'raw_response_event' && + event.type === 'raw_model_stream_event' && (event.data as any).type === 'response.output_text.delta' ) { const delta = (event.data as any).delta as string; @@ -192,7 +193,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { // Fall back to finalOutput if no text deltas were streamed (e.g. tool-only turns) if (!finalResponse) { - finalResponse = (await (stream as any).finalOutput) ?? 'Completed.'; + finalResponse = stream.finalOutput ?? 'Completed.'; } // 5. Create the agent's final message @@ -205,7 +206,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { contextId: contextId, }; historyForAgent.push(agentMessage); - contexts.set(contextId, historyForAgent); + this.contexts.set(contextId, historyForAgent); // 6. Publish final task status update const finalUpdate: TaskStatusUpdateEvent = { @@ -220,6 +221,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { final: true, }; eventBus.publish(finalUpdate); + this.cancelledTasks.delete(taskId); console.log( `[OpenAIAgentExecutor] Task ${taskId} finished with state: completed` @@ -249,6 +251,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { final: true, }; eventBus.publish(errorUpdate); + this.cancelledTasks.delete(taskId); } } } diff --git a/examples/js/openai-agents-example/src/index.ts b/examples/js/openai-agents-example/src/index.ts index a4b8bc87..1ad03e66 100644 --- a/examples/js/openai-agents-example/src/index.ts +++ b/examples/js/openai-agents-example/src/index.ts @@ -47,10 +47,11 @@ async function main() { const expressApp = appBuilder.setupRoutes(app as any); // 5. Start the server + const HOST = process.env.HOST || 'localhost'; const PORT = process.env.PORT || 3000; - const server = expressApp.listen(PORT, () => { - console.log(`[OpenAIAgent] Server using OpenAI Agents SDK and A2A started on http://localhost:${PORT}`); - console.log(`[OpenAIAgent] Agent Card: http://localhost:${PORT}/.well-known/agent.json`); + const server = expressApp.listen(PORT, HOST, () => { + console.log(`[OpenAIAgent] Server using OpenAI Agents SDK and A2A started on http://${HOST}:${PORT}`); + console.log(`[OpenAIAgent] Agent Card: http://${HOST}:${PORT}/.well-known/agent.json`); console.log('[OpenAIAgent] Press Ctrl+C to stop the server'); }); @@ -58,12 +59,20 @@ async function main() { const shutdown = async () => { console.log('\n[OpenAIAgent] Shutting down server...'); - // Close the HTTP server - server.close(() => { + const forceExit = setTimeout(() => { + console.error('[OpenAIAgent] Force closing after timeout'); + process.exit(1); + }, 5000); + + server.close((err) => { + clearTimeout(forceExit); + if (err) { + console.error('[OpenAIAgent] Error closing server:', err); + process.exit(1); + } console.log('[OpenAIAgent] HTTP server closed'); + process.exit(0); }); - - process.exit(0); }; // Handle termination signals From dda6b6edf81ea66fdc806b8ace3d2c00a871e50a Mon Sep 17 00:00:00 2001 From: Shuvadarshan Date: Wed, 29 Apr 2026 08:33:45 +0200 Subject: [PATCH 3/6] fix(openai-agents-example): address second CodeRabbit review round - Filter messages with empty content after map to correctly detect non-text inputs (messages.length was always equal to historyForAgent.length) - Add bounded growth and LRU eviction to this.contexts: - MAX_CONTEXTS = 500, MAX_MESSAGES_PER_CONTEXT = 100, CONTEXT_TTL_MS = 30 min - touchContext() trims history and evicts oldest entry when at capacity - evictExpiredContexts() runs every 5 min via unref'd setInterval - contextLastAccess updated on every read and write Co-Authored-By: Claude Sonnet 4.6 --- .../src/agentExecutor.ts | 67 ++++++++++++++++--- 1 file changed, 57 insertions(+), 10 deletions(-) diff --git a/examples/js/openai-agents-example/src/agentExecutor.ts b/examples/js/openai-agents-example/src/agentExecutor.ts index a9e47f4d..bbcdc008 100644 --- a/examples/js/openai-agents-example/src/agentExecutor.ts +++ b/examples/js/openai-agents-example/src/agentExecutor.ts @@ -4,13 +4,57 @@ import { Agent, run, RunResultStreaming } from '@openai/agents'; import { v4 as uuidv4 } from 'uuid'; +const MAX_CONTEXTS = 500; +const MAX_MESSAGES_PER_CONTEXT = 100; +const CONTEXT_TTL_MS = 30 * 60 * 1000; // 30 minutes + export class OpenAIAgentExecutor implements AgentExecutor { private cancelledTasks = new Set(); private agent: Agent; private contexts = new Map(); + private contextLastAccess = new Map(); + private cleanupTimer: ReturnType; constructor(agent: Agent) { this.agent = agent; + this.cleanupTimer = setInterval(() => this.evictExpiredContexts(), 5 * 60 * 1000); + this.cleanupTimer.unref(); + } + + private evictExpiredContexts(): void { + const now = Date.now(); + for (const [contextId, lastAccess] of this.contextLastAccess) { + if (now - lastAccess > CONTEXT_TTL_MS) { + this.contexts.delete(contextId); + this.contextLastAccess.delete(contextId); + } + } + } + + private touchContext(contextId: string, history: Message[]): void { + // Trim to keep only the most recent messages + if (history.length > MAX_MESSAGES_PER_CONTEXT) { + history.splice(0, history.length - MAX_MESSAGES_PER_CONTEXT); + } + + // Evict the oldest entry when at capacity + if (!this.contexts.has(contextId) && this.contexts.size >= MAX_CONTEXTS) { + let oldestId: string | null = null; + let oldestTime = Infinity; + for (const [id, time] of this.contextLastAccess) { + if (time < oldestTime) { + oldestTime = time; + oldestId = id; + } + } + if (oldestId) { + this.contexts.delete(oldestId); + this.contextLastAccess.delete(oldestId); + } + } + + this.contexts.set(contextId, history); + this.contextLastAccess.set(contextId, Date.now()); } public cancelTask = async ( @@ -75,19 +119,22 @@ export class OpenAIAgentExecutor implements AgentExecutor { // 3. Prepare messages for the agent const historyForAgent = this.contexts.get(contextId) || []; + this.contextLastAccess.set(contextId, Date.now()); if (!historyForAgent.find(m => m.messageId === userMessage.messageId)) { historyForAgent.push(userMessage); } - this.contexts.set(contextId, historyForAgent); + this.touchContext(contextId, historyForAgent); - // Convert A2A messages to OpenAI format - const messages = historyForAgent.map(m => ({ - role: m.role === 'agent' ? 'assistant' : 'user', - content: m.parts - .filter((p): p is TextPart => p.kind === 'text' && !!(p as TextPart).text) - .map(p => (p as TextPart).text) - .join('\n') - })); + // Convert A2A messages to OpenAI format, dropping entries with no usable text + const messages = historyForAgent + .map(m => ({ + role: m.role === 'agent' ? 'assistant' : 'user', + content: m.parts + .filter((p): p is TextPart => p.kind === 'text' && !!(p as TextPart).text) + .map(p => (p as TextPart).text) + .join('\n'), + })) + .filter(m => m.content.trim().length > 0); if (messages.length === 0) { console.warn( @@ -206,7 +253,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { contextId: contextId, }; historyForAgent.push(agentMessage); - this.contexts.set(contextId, historyForAgent); + this.touchContext(contextId, historyForAgent); // 6. Publish final task status update const finalUpdate: TaskStatusUpdateEvent = { From ae1b7e853b97e79917446716f54b986a9b0a87f7 Mon Sep 17 00:00:00 2001 From: Shuvadarshan Date: Wed, 29 Apr 2026 08:45:29 +0200 Subject: [PATCH 4/6] fix(openai-agents-example): address third CodeRabbit review round - Import AgentInputItem from @openai/agents and type messages as AgentInputItem[] with explicit role cast; removes the messages as any cast so TypeScript validates the structure at compile time - Add AbortController per run and pass signal to run(); call controller.abort() on mid-stream cancellation so the upstream model request is terminated, not just the event consumer loop (prevents wasted API tokens) - Change catch (error: any) to catch (error: unknown) and extract the message via error instanceof Error ? error.message : String(error) so thrown strings and plain objects produce meaningful output rather than undefined Co-Authored-By: Claude Sonnet 4.6 --- .../openai-agents-example/src/agentExecutor.ts | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/examples/js/openai-agents-example/src/agentExecutor.ts b/examples/js/openai-agents-example/src/agentExecutor.ts index bbcdc008..ac8e3d69 100644 --- a/examples/js/openai-agents-example/src/agentExecutor.ts +++ b/examples/js/openai-agents-example/src/agentExecutor.ts @@ -1,6 +1,6 @@ import { Message, Task, TaskStatusUpdateEvent, TextPart } from '@a2a-js/sdk'; import { AgentExecutor, ExecutionEventBus, RequestContext } from '@a2a-js/sdk/server'; -import { Agent, run, RunResultStreaming } from '@openai/agents'; +import { Agent, AgentInputItem, run, RunResultStreaming } from '@openai/agents'; import { v4 as uuidv4 } from 'uuid'; @@ -126,9 +126,9 @@ export class OpenAIAgentExecutor implements AgentExecutor { this.touchContext(contextId, historyForAgent); // Convert A2A messages to OpenAI format, dropping entries with no usable text - const messages = historyForAgent + const messages: AgentInputItem[] = historyForAgent .map(m => ({ - role: m.role === 'agent' ? 'assistant' : 'user', + role: (m.role === 'agent' ? 'assistant' : 'user') as 'assistant' | 'user', content: m.parts .filter((p): p is TextPart => p.kind === 'text' && !!(p as TextPart).text) .map(p => (p as TextPart).text) @@ -184,7 +184,8 @@ export class OpenAIAgentExecutor implements AgentExecutor { } // 4. Run the OpenAI agent with streaming - const stream: RunResultStreaming = run(this.agent, messages as any); + const controller = new AbortController(); + const stream: RunResultStreaming = run(this.agent, messages, { signal: controller.signal }); let finalResponse = ''; @@ -193,6 +194,8 @@ export class OpenAIAgentExecutor implements AgentExecutor { if (this.cancelledTasks.has(taskId)) { console.log(`[OpenAIAgentExecutor] Request cancelled during execution for task: ${taskId}`); + controller.abort(); + const cancelledUpdate: TaskStatusUpdateEvent = { kind: 'status-update', taskId: taskId, @@ -274,7 +277,8 @@ export class OpenAIAgentExecutor implements AgentExecutor { `[OpenAIAgentExecutor] Task ${taskId} finished with state: completed` ); - } catch (error: any) { + } catch (error: unknown) { + const errorMessage = error instanceof Error ? error.message : String(error); console.error( `[OpenAIAgentExecutor] Error processing task ${taskId}:`, error @@ -289,7 +293,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { kind: 'message', role: 'agent', messageId: uuidv4(), - parts: [{ kind: 'text', text: `Agent error: ${error.message}` }], + parts: [{ kind: 'text', text: `Agent error: ${errorMessage}` }], taskId: taskId, contextId: contextId, }, From 889e51258111f5c99f3ca364ffb8d2e4726f2316 Mon Sep 17 00:00:00 2001 From: Shuvadarshan Date: Wed, 29 Apr 2026 09:04:02 +0200 Subject: [PATCH 5/6] fix(openai-agents-example): address fourth CodeRabbit review round MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add await and { stream: true } to run() call so it returns a StreamedRunResult instead of an unresolved Promise; without this the for-await loop would fail silently - Import isOpenAIResponsesRawModelStreamEvent from @openai/agents and replace ad-hoc (event.data as any) checks; the correct nested shape for Responses API events is event.data.type === 'model' then event.data.event.type === 'response.output_text.delta' and delta lives at event.data.event.delta — the old path never matched, so intermediate working updates were never sent - Add await stream.completed before reading stream.finalOutput to guarantee finalOutput is set before it is accessed Co-Authored-By: Claude Sonnet 4.6 --- .../src/agentExecutor.ts | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/examples/js/openai-agents-example/src/agentExecutor.ts b/examples/js/openai-agents-example/src/agentExecutor.ts index ac8e3d69..b954f281 100644 --- a/examples/js/openai-agents-example/src/agentExecutor.ts +++ b/examples/js/openai-agents-example/src/agentExecutor.ts @@ -1,6 +1,11 @@ import { Message, Task, TaskStatusUpdateEvent, TextPart } from '@a2a-js/sdk'; import { AgentExecutor, ExecutionEventBus, RequestContext } from '@a2a-js/sdk/server'; -import { Agent, AgentInputItem, run, RunResultStreaming } from '@openai/agents'; +import { + Agent, + AgentInputItem, + isOpenAIResponsesRawModelStreamEvent, + run, +} from '@openai/agents'; import { v4 as uuidv4 } from 'uuid'; @@ -185,7 +190,10 @@ export class OpenAIAgentExecutor implements AgentExecutor { // 4. Run the OpenAI agent with streaming const controller = new AbortController(); - const stream: RunResultStreaming = run(this.agent, messages, { signal: controller.signal }); + const stream = await run(this.agent, messages, { + stream: true, + signal: controller.signal, + }); let finalResponse = ''; @@ -213,10 +221,11 @@ export class OpenAIAgentExecutor implements AgentExecutor { // Handle text delta events from the underlying model if ( - event.type === 'raw_model_stream_event' && - (event.data as any).type === 'response.output_text.delta' + isOpenAIResponsesRawModelStreamEvent(event) && + event.data.type === 'model' && + event.data.event.type === 'response.output_text.delta' ) { - const delta = (event.data as any).delta as string; + const delta = event.data.event.delta; finalResponse += delta; const intermediateUpdate: TaskStatusUpdateEvent = { @@ -242,6 +251,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { } // Fall back to finalOutput if no text deltas were streamed (e.g. tool-only turns) + await stream.completed; if (!finalResponse) { finalResponse = stream.finalOutput ?? 'Completed.'; } From 2c3b3bee767290d560ce82a6261fda0e4de3c7be Mon Sep 17 00:00:00 2001 From: Shuvadarshan Date: Wed, 29 Apr 2026 21:32:49 +0200 Subject: [PATCH 6/6] fix(openai-agents-example): address fifth CodeRabbit review round - Add activeControllers Map; cancelTask() now calls controller.abort() immediately so the upstream model request is terminated without waiting for the next stream event - activeControllers.set(taskId, controller) registered before run(); finally block calls activeControllers.delete(taskId) on every exit path - Move pre-run cancellation check to before the "working" status publish so a cancelled task never briefly appears as working then cancelled - Add public dispose() that clears the cleanup setInterval; call it from index.ts shutdown inside server.close() callback for clean executor teardown Co-Authored-By: Claude Sonnet 4.6 --- .../src/agentExecutor.ts | 79 +++++++++++-------- .../js/openai-agents-example/src/index.ts | 1 + 2 files changed, 45 insertions(+), 35 deletions(-) diff --git a/examples/js/openai-agents-example/src/agentExecutor.ts b/examples/js/openai-agents-example/src/agentExecutor.ts index b954f281..004ffe3e 100644 --- a/examples/js/openai-agents-example/src/agentExecutor.ts +++ b/examples/js/openai-agents-example/src/agentExecutor.ts @@ -15,6 +15,7 @@ const CONTEXT_TTL_MS = 30 * 60 * 1000; // 30 minutes export class OpenAIAgentExecutor implements AgentExecutor { private cancelledTasks = new Set(); + private activeControllers = new Map(); private agent: Agent; private contexts = new Map(); private contextLastAccess = new Map(); @@ -26,6 +27,10 @@ export class OpenAIAgentExecutor implements AgentExecutor { this.cleanupTimer.unref(); } + public dispose(): void { + clearInterval(this.cleanupTimer); + } + private evictExpiredContexts(): void { const now = Date.now(); for (const [contextId, lastAccess] of this.contextLastAccess) { @@ -67,6 +72,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { eventBus: ExecutionEventBus, ): Promise => { this.cancelledTasks.add(taskId); + this.activeControllers.get(taskId)?.abort(); // The execute loop is responsible for publishing the final state }; @@ -77,7 +83,6 @@ export class OpenAIAgentExecutor implements AgentExecutor { const userMessage = requestContext.userMessage; const existingTask = requestContext.task; - // Determine IDs for the task and context const taskId = existingTask?.id || uuidv4(); const contextId = userMessage.contextId || existingTask?.contextId || uuidv4(); @@ -101,7 +106,26 @@ export class OpenAIAgentExecutor implements AgentExecutor { eventBus.publish(initialTask); } - // 2. Publish "working" status update + // 2. Pre-run cancellation check — before emitting "working" so a cancelled + // task never briefly appears as working. + if (this.cancelledTasks.has(taskId)) { + console.log(`[OpenAIAgentExecutor] Request cancelled for task: ${taskId}`); + const cancelledUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'canceled', + timestamp: new Date().toISOString(), + }, + final: true, + }; + eventBus.publish(cancelledUpdate); + this.cancelledTasks.delete(taskId); + return; + } + + // 3. Publish "working" status update const workingStatusUpdate: TaskStatusUpdateEvent = { kind: 'status-update', taskId: taskId, @@ -122,7 +146,7 @@ export class OpenAIAgentExecutor implements AgentExecutor { }; eventBus.publish(workingStatusUpdate); - // 3. Prepare messages for the agent + // 4. Prepare messages for the agent const historyForAgent = this.contexts.get(contextId) || []; this.contextLastAccess.set(contextId, Date.now()); if (!historyForAgent.find(m => m.messageId === userMessage.messageId)) { @@ -168,28 +192,12 @@ export class OpenAIAgentExecutor implements AgentExecutor { return; } - try { - // Check if the task has been cancelled before starting - if (this.cancelledTasks.has(taskId)) { - console.log(`[OpenAIAgentExecutor] Request cancelled for task: ${taskId}`); + // 5. Create a per-run AbortController and register it so cancelTask() can + // abort the upstream request immediately without waiting for the next event. + const controller = new AbortController(); + this.activeControllers.set(taskId, controller); - const cancelledUpdate: TaskStatusUpdateEvent = { - kind: 'status-update', - taskId: taskId, - contextId: contextId, - status: { - state: 'canceled', - timestamp: new Date().toISOString(), - }, - final: true, - }; - eventBus.publish(cancelledUpdate); - this.cancelledTasks.delete(taskId); - return; - } - - // 4. Run the OpenAI agent with streaming - const controller = new AbortController(); + try { const stream = await run(this.agent, messages, { stream: true, signal: controller.signal, @@ -198,7 +206,8 @@ export class OpenAIAgentExecutor implements AgentExecutor { let finalResponse = ''; for await (const event of stream) { - // Check for cancellation during execution + // Mid-stream cancellation: cancelTask may have already called abort(), + // but also handle the polling path for robustness. if (this.cancelledTasks.has(taskId)) { console.log(`[OpenAIAgentExecutor] Request cancelled during execution for task: ${taskId}`); @@ -250,13 +259,15 @@ export class OpenAIAgentExecutor implements AgentExecutor { } } - // Fall back to finalOutput if no text deltas were streamed (e.g. tool-only turns) + // Ensure the stream has fully settled before reading finalOutput await stream.completed; + + // Fall back to finalOutput if no text deltas were streamed (e.g. tool-only turns) if (!finalResponse) { finalResponse = stream.finalOutput ?? 'Completed.'; } - // 5. Create the agent's final message + // 6. Store the agent reply and publish final task status update const agentMessage: Message = { kind: 'message', role: 'agent', @@ -268,7 +279,6 @@ export class OpenAIAgentExecutor implements AgentExecutor { historyForAgent.push(agentMessage); this.touchContext(contextId, historyForAgent); - // 6. Publish final task status update const finalUpdate: TaskStatusUpdateEvent = { kind: 'status-update', taskId: taskId, @@ -283,16 +293,12 @@ export class OpenAIAgentExecutor implements AgentExecutor { eventBus.publish(finalUpdate); this.cancelledTasks.delete(taskId); - console.log( - `[OpenAIAgentExecutor] Task ${taskId} finished with state: completed` - ); + console.log(`[OpenAIAgentExecutor] Task ${taskId} finished with state: completed`); } catch (error: unknown) { const errorMessage = error instanceof Error ? error.message : String(error); - console.error( - `[OpenAIAgentExecutor] Error processing task ${taskId}:`, - error - ); + console.error(`[OpenAIAgentExecutor] Error processing task ${taskId}:`, error); + const errorUpdate: TaskStatusUpdateEvent = { kind: 'status-update', taskId: taskId, @@ -313,6 +319,9 @@ export class OpenAIAgentExecutor implements AgentExecutor { }; eventBus.publish(errorUpdate); this.cancelledTasks.delete(taskId); + + } finally { + this.activeControllers.delete(taskId); } } } diff --git a/examples/js/openai-agents-example/src/index.ts b/examples/js/openai-agents-example/src/index.ts index 1ad03e66..9bb3ac5c 100644 --- a/examples/js/openai-agents-example/src/index.ts +++ b/examples/js/openai-agents-example/src/index.ts @@ -66,6 +66,7 @@ async function main() { server.close((err) => { clearTimeout(forceExit); + agentExecutor.dispose(); if (err) { console.error('[OpenAIAgent] Error closing server:', err); process.exit(1);