diff --git a/examples/js/openai-agents-example/package.json b/examples/js/openai-agents-example/package.json new file mode 100644 index 00000000..bc320cb7 --- /dev/null +++ b/examples/js/openai-agents-example/package.json @@ -0,0 +1,28 @@ +{ + "name": "openai-agents-example", + "version": "1.0.0", + "type": "module", + "description": "", + "main": "dist/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.8.0", + "express": "^5.1.0", + "uuid": "^11.1.0", + "zod": "^3.24.1" + }, + "devDependencies": { + "@types/express": "^5.0.3", + "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..5414f5da --- /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 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. + + +## 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().email().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..004ffe3e --- /dev/null +++ b/examples/js/openai-agents-example/src/agentExecutor.ts @@ -0,0 +1,327 @@ +import { Message, Task, TaskStatusUpdateEvent, TextPart } from '@a2a-js/sdk'; +import { AgentExecutor, ExecutionEventBus, RequestContext } from '@a2a-js/sdk/server'; +import { + Agent, + AgentInputItem, + isOpenAIResponsesRawModelStreamEvent, + run, +} 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 activeControllers = new Map(); + 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(); + } + + public dispose(): void { + clearInterval(this.cleanupTimer); + } + + 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 ( + taskId: string, + eventBus: ExecutionEventBus, + ): Promise => { + this.cancelledTasks.add(taskId); + this.activeControllers.get(taskId)?.abort(); + // 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; + + 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. 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, + 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); + + // 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)) { + historyForAgent.push(userMessage); + } + this.touchContext(contextId, historyForAgent); + + // Convert A2A messages to OpenAI format, dropping entries with no usable text + const messages: AgentInputItem[] = historyForAgent + .map(m => ({ + 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) + .join('\n'), + })) + .filter(m => m.content.trim().length > 0); + + 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); + this.cancelledTasks.delete(taskId); + return; + } + + // 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); + + try { + const stream = await run(this.agent, messages, { + stream: true, + signal: controller.signal, + }); + + let finalResponse = ''; + + for await (const event of stream) { + // 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}`); + + controller.abort(); + + 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; + } + + // Handle text delta events from the underlying model + if ( + isOpenAIResponsesRawModelStreamEvent(event) && + event.data.type === 'model' && + event.data.event.type === 'response.output_text.delta' + ) { + const delta = event.data.event.delta; + 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); + } + } + + // 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.'; + } + + // 6. Store the agent reply and publish final task status update + const agentMessage: Message = { + kind: 'message', + role: 'agent', + messageId: uuidv4(), + parts: [{ kind: 'text', text: finalResponse }], + taskId: taskId, + contextId: contextId, + }; + historyForAgent.push(agentMessage); + this.touchContext(contextId, historyForAgent); + + const finalUpdate: TaskStatusUpdateEvent = { + kind: 'status-update', + taskId: taskId, + contextId: contextId, + status: { + state: 'completed', + message: agentMessage, + timestamp: new Date().toISOString(), + }, + final: true, + }; + eventBus.publish(finalUpdate); + this.cancelledTasks.delete(taskId); + + 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); + + 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: ${errorMessage}` }], + taskId: taskId, + contextId: contextId, + }, + timestamp: new Date().toISOString(), + }, + final: true, + }; + 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 new file mode 100644 index 00000000..9bb3ac5c --- /dev/null +++ b/examples/js/openai-agents-example/src/index.ts @@ -0,0 +1,88 @@ +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 HOST = process.env.HOST || 'localhost'; + const PORT = process.env.PORT || 3000; + 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'); + }); + + // 6. Setup graceful shutdown + const shutdown = async () => { + console.log('\n[OpenAIAgent] Shutting down server...'); + + const forceExit = setTimeout(() => { + console.error('[OpenAIAgent] Force closing after timeout'); + process.exit(1); + }, 5000); + + server.close((err) => { + clearTimeout(forceExit); + agentExecutor.dispose(); + if (err) { + console.error('[OpenAIAgent] Error closing server:', err); + process.exit(1); + } + 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"] +}