A2A Executor Patterns

Purpose: Provide production-ready executor patterns for implementing Agent-to-Agent (A2A) protocol task handlers with proper error handling, retry logic, and execution flows.

Activation Triggers:

• Implementing A2A protocol executors

• Building task handler functions

• Creating agent execution flows

• Managing task lifecycle and state

• Implementing retry and error recovery

• Building executor middleware

• Task validation and sanitization

Key Resources:

• templates/basic-executor.ts

• Simple synchronous executor

• templates/basic-executor.py

• Python synchronous executor

• templates/async-executor.ts

• Asynchronous task executor

• templates/async-executor.py

• Python async executor

• templates/streaming-executor.ts

• Streaming result executor

• templates/streaming-executor.py

• Python streaming executor

• templates/batch-executor.ts

• Batch task processing

• scripts/validate-executor.sh

• Validate executor implementation

• scripts/test-executor.sh

• Test executor against A2A spec

• examples/

• Production executor implementations

Core Executor Patterns

1. Basic Executor (Synchronous)

When to use: Simple, fast tasks with immediate results

Template: templates/basic-executor.ts or templates/basic-executor.py

Pattern:

async function executeTask(task: A2ATask): Promise<A2AResult> { // 1. Validate input validateTask(task)

// 2. Execute task const result = await processTask(task)

// 3. Return result return { status: 'completed', result, taskId: task.id } }

Best for: Quick operations, validation tasks, simple transformations

2. Async Executor (Long-Running)

When to use: Tasks that take time and need status updates

Template: templates/async-executor.ts or templates/async-executor.py

Pattern:

• Accept task and return task ID immediately

• Process task asynchronously

• Provide status endpoint

• Send completion callback

Best for: LLM inference, file processing, data analysis

3. Streaming Executor

When to use: Results should be delivered incrementally

Template: templates/streaming-executor.ts or templates/streaming-executor.py

Pattern:

• Open stream connection

• Send partial results as available

• Close stream on completion

• Handle backpressure

Best for: Text generation, real-time data, progressive results

4. Batch Executor

When to use: Processing multiple related tasks efficiently

Template: templates/batch-executor.ts

Pattern:

• Accept multiple tasks

• Group by similarity

• Process in parallel batches

• Return aggregated results

Best for: Bulk operations, parallel processing, resource optimization

Execution Flow Components

1. Task Validation

function validateTask(task: A2ATask): void { // Validate required fields if (!task.id) throw new ValidationError('Task ID required') if (!task.type) throw new ValidationError('Task type required')

// Validate task parameters validateParameters(task.parameters)

// Check executor capabilities if (!supportsTaskType(task.type)) { throw new UnsupportedTaskError(task.type) } }

Purpose: Catch errors early, provide clear feedback

2. Error Handling

async function executeWithErrorHandling(task: A2ATask) { try { return await executeTask(task) } catch (error) { if (error instanceof ValidationError) { return { status: 'failed', error: error.message } } if (error instanceof RetryableError) { return scheduleRetry(task) } // Log and return generic error logger.error('Task execution failed', { taskId: task.id, error }) return { status: 'failed', error: 'Internal error' } } }

Error Types:

• ValidationError

• Invalid input, don't retry

• RetryableError

• Temporary failure, safe to retry

• FatalError

• Permanent failure, abort

3. Retry Logic

const retryConfig = { maxAttempts: 3, backoff: 'exponential', // or 'linear', 'fixed' initialDelay: 1000, // ms maxDelay: 30000 }

async function executeWithRetry( task: A2ATask, attempt: number = 1 ): Promise<A2AResult> { try { return await executeTask(task) } catch (error) { if (attempt >= retryConfig.maxAttempts) { throw new MaxRetriesExceededError(task.id) }

if (error instanceof RetryableError) {
  const delay = calculateBackoff(attempt)
  await sleep(delay)
  return executeWithRetry(task, attempt + 1)
}

throw error

} }

Retry Strategies:

• Exponential backoff: delay = initialDelay * (2 ^ attempt)

• Linear backoff: delay = initialDelay * attempt

• Fixed delay: delay = initialDelay

4. Task State Management

interface TaskState { id: string status: 'pending' | 'running' | 'completed' | 'failed' result?: any error?: string startTime: Date endTime?: Date attempts: number }

class TaskStore { private tasks = new Map<string, TaskState>()

createTask(id: string): TaskState { const state: TaskState = { id, status: 'pending', startTime: new Date(), attempts: 0 } this.tasks.set(id, state) return state }

updateTask(id: string, update: Partial<TaskState>): void { const state = this.tasks.get(id) if (state) { Object.assign(state, update) } }

getTask(id: string): TaskState | undefined { return this.tasks.get(id) } }

Executor Middleware

1. Logging Middleware

function loggingMiddleware( executor: Executor ): Executor { return async (task) => { logger.info('Task started', { taskId: task.id }) const start = Date.now()

try {
  const result = await executor(task)
  const duration = Date.now() - start
  logger.info('Task completed', { taskId: task.id, duration })
  return result
} catch (error) {
  const duration = Date.now() - start
  logger.error('Task failed', { taskId: task.id, duration, error })
  throw error
}

} }

2. Metrics Middleware

function metricsMiddleware( executor: Executor ): Executor { return async (task) => { metrics.increment('tasks.started', { type: task.type }) const start = Date.now()

try {
  const result = await executor(task)
  const duration = Date.now() - start
  metrics.timing('tasks.duration', duration, { type: task.type })
  metrics.increment('tasks.completed', { type: task.type })
  return result
} catch (error) {
  metrics.increment('tasks.failed', { type: task.type })
  throw error
}

} }

3. Rate Limiting Middleware

function rateLimitMiddleware( executor: Executor, limit: { requests: number, window: number } ): Executor { const limiter = new RateLimiter(limit.requests, limit.window)

return async (task) => { await limiter.acquire() try { return await executor(task) } finally { limiter.release() } } }

Production Best Practices

1. Timeouts

async function executeWithTimeout( task: A2ATask, timeoutMs: number ): Promise<A2AResult> { return Promise.race([ executeTask(task), new Promise((_, reject) => setTimeout(() => reject(new TimeoutError()), timeoutMs) ) ]) }

2. Resource Cleanup

async function executeWithCleanup(task: A2ATask) { const resources = []

try { const resource = await allocateResource() resources.push(resource)

return await executeTask(task, resource)

} finally { // Always cleanup, even on error await Promise.all( resources.map(r => r.cleanup()) ) } }

3. Graceful Shutdown

class GracefulExecutor { private activeTasks = new Set<string>() private shuttingDown = false

async execute(task: A2ATask): Promise<A2AResult> { if (this.shuttingDown) { throw new Error('Executor is shutting down') }

this.activeTasks.add(task.id)

try {
  return await executeTask(task)
} finally {
  this.activeTasks.delete(task.id)
}

}

async shutdown(): Promise<void> { this.shuttingDown = true

// Wait for active tasks to complete
while (this.activeTasks.size > 0) {
  await sleep(100)
}

} }

Common Executor Types

1. LLM Executor

Example: examples/llm-executor.ts

Executes LLM inference tasks with streaming

2. Function Executor

Example: examples/function-executor.ts

Calls functions/tools and returns results

3. Workflow Executor

Example: examples/workflow-executor.ts

Orchestrates multi-step workflows

4. Validation Executor

Example: examples/validation-executor.ts

Validates data and returns compliance results

Validation and Testing

Scripts:

• scripts/validate-executor.sh

• Validate executor structure

• scripts/test-executor.sh

• Test against A2A spec

Run validation:

bash scripts/validate-executor.sh your-executor.ts

Run tests:

bash scripts/test-executor.sh your-executor.ts

Resources

TypeScript Templates:

• basic-executor.ts

• Simple sync executor

• async-executor.ts

• Async with status tracking

• streaming-executor.ts

• Streaming results

• batch-executor.ts

• Batch processing

Python Templates:

• basic-executor.py

• Simple sync executor

• async-executor.py

• Async with status tracking

• streaming-executor.py

• Streaming results

Scripts:

• validate-executor.sh

• Structure validation

• test-executor.sh

• A2A spec compliance

Examples:

• llm-executor.ts

• LLM inference executor

• function-executor.ts

• Function calling executor

• workflow-executor.ts

• Multi-step workflows

• validation-executor.ts

• Data validation

Protocol Version: A2A Protocol v1.0 Runtime: Node.js 18+, Python 3.9+

Best Practice: Start with basic executor, add complexity (async, streaming, batching) only as needed