Explanations

• 1: Architecture
• 1.1: A2A-Compliant EDA Broker Architecture
• 1.2: Hexagonal Architecture & A2A Protocol Implementation
• 2: Core Concepts
• 2.1: The Agent2Agent Principle
• 2.2: Agent2Agent (A2A) Protocol Migration
• 2.3: Understanding Tasks in Agent2Agent Communication
• 3: Features
• 3.1: Distributed Tracing & OpenTelemetry
• 3.2: Architecture Evolution: From Build Tags to Unified Abstractions
• 3.3: Performance and Scaling Considerations
• 3.4: The Unified Abstraction Library

1 - Architecture

Architecture Explanations

Understand the fundamental architecture and design principles behind AgentHub’s distributed agent system.

Available Documentation

• Broker Architecture - Central broker design and communication patterns

1.1 - A2A-Compliant EDA Broker Architecture

AgentHub A2A-Compliant EDA Broker Architecture

This document explains the internal architecture of the AgentHub Event-Driven Architecture (EDA) broker, how it implements Agent2Agent (A2A) protocol-compliant communication patterns, and the design decisions behind its hybrid approach.

Architectural Overview

The AgentHub broker serves as a centralized Event-Driven Architecture hub that transports Agent2Agent (A2A) protocol-compliant messages between distributed agents. It combines the scalability benefits of EDA with the interoperability guarantees of the A2A protocol.

┌─────────────────────────────────────────────────────────────────┐
│                     AgentHub Broker                             │
├─────────────────────────────────────────────────────────────────┤
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐│
│  │   Task Router   │    │   Subscriber    │    │   Progress      ││
│  │                 │    │   Manager       │    │   Tracker       ││
│  │ • Route tasks   │    │                 │    │                 ││
│  │ • Apply filters │    │ • Manage agent  │    │ • Track task    ││
│  │ • Broadcast     │    │   subscriptions │    │   progress      ││
│  │ • Load balance  │    │ • Handle        │    │ • Update        ││
│  │                 │    │   disconnects   │    │   requesters    ││
│  └─────────────────┘    └─────────────────┘    └─────────────────┘│
├─────────────────────────────────────────────────────────────────┤
│                        gRPC Interface                           │
├─────────────────────────────────────────────────────────────────┤
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐│
│  │ PublishTask     │    │SubscribeToTasks│    │SubscribeToTask  ││
│  │ PublishResult   │    │SubscribeToRes  │    │ Progress        ││
│  │ PublishProgress │    │                 │    │                 ││
│  └─────────────────┘    └─────────────────┘    └─────────────────┘│
└─────────────────────────────────────────────────────────────────┘

Core Components

1. Event Bus Server

The main server implementation at broker/main.go:22 provides the central coordination point:

type eventBusServer struct {
    pb.UnimplementedEventBusServer

    // Subscription management
    taskSubscribers         map[string][]chan *pb.TaskMessage
    taskResultSubscribers   map[string][]chan *pb.TaskResult
    taskProgressSubscribers map[string][]chan *pb.TaskProgress
    taskMu                  sync.RWMutex
}

Key characteristics:

• Thread-safe: Uses sync.RWMutex to protect concurrent access to subscriber maps
• Channel-based: Uses Go channels for efficient message passing
• Non-blocking: Implements timeouts to prevent blocking on slow consumers
• Stateless: No persistent storage - all state is in-memory

2. Task Routing Engine

The routing logic determines how tasks are delivered to agents:

Direct Routing

When a task specifies a ResponderAgentId, it’s routed directly to that agent:

if responderID := req.GetTask().GetResponderAgentId(); responderID != "" {
    if subs, ok := s.taskSubscribers[responderID]; ok {
        targetChannels = subs
    }
}

Broadcast Routing

When no specific responder is set, tasks are broadcast to all subscribed agents:

} else {
    // Broadcast to all task subscribers
    for _, subs := range s.taskSubscribers {
        targetChannels = append(targetChannels, subs...)
    }
}

Routing Features

• Immediate delivery: Tasks are routed immediately upon receipt
• Multiple subscribers: Single agent can have multiple subscription channels
• Timeout protection: 5-second timeout prevents blocking on unresponsive agents
• Error isolation: Failed delivery to one agent doesn’t affect others

3. Subscription Management

The broker manages three types of subscriptions:

Task Subscriptions

Agents subscribe to receive tasks assigned to them:

func (s *eventBusServer) SubscribeToTasks(req *pb.SubscribeToTasksRequest, stream pb.EventBus_SubscribeToTasksServer) error

• Agent-specific: Tasks are delivered based on agent ID
• Type filtering: Optional filtering by task types
• Long-lived streams: Connections persist until agent disconnects
• Automatic cleanup: Subscriptions are removed when connections close

Result Subscriptions

Publishers subscribe to receive results of tasks they requested:

func (s *eventBusServer) SubscribeToTaskResults(req *pb.SubscribeToTaskResultsRequest, stream pb.EventBus_SubscribeToTaskResultsServer) error

Progress Subscriptions

Publishers can track progress of long-running tasks:

func (s *eventBusServer) SubscribeToTaskProgress(req *pb.SubscribeToTaskResultsRequest, stream pb.EventBus_SubscribeToTaskProgressServer) error

4. Message Flow Architecture

Task Publication Flow

1. Validation: Incoming tasks are validated for required fields
2. Routing: Tasks are routed to appropriate subscribers
3. Delivery: Messages are sent via Go channels with timeout protection
4. Response: Publisher receives acknowledgment of successful publication

Result Flow

1. Receipt: Agents publish task results back to the broker
2. Broadcasting: Results are broadcast to all result subscribers
3. Filtering: Subscribers receive results for their requested tasks
4. Delivery: Results are streamed back to requesting agents

Progress Flow

1. Updates: Executing agents send periodic progress updates
2. Distribution: Progress updates are sent to interested subscribers
3. Real-time delivery: Updates are streamed immediately upon receipt

Design Decisions and Trade-offs

In-Memory State Management

Decision: Store all subscription state in memory using Go maps and channels.

Benefits:

• High performance: No database overhead for message routing
• Low latency: Sub-millisecond message routing
• Simplicity: Easier to develop, test, and maintain
• Concurrent efficiency: Go’s garbage collector handles channel cleanup

Trade-offs:

• No persistence: Broker restart loses all subscription state
• Memory usage: Large numbers of agents increase memory requirements
• Single point of failure: No built-in redundancy

When this works well:

• Development and testing environments
• Small to medium-scale deployments
• Scenarios where agents can re-establish subscriptions on broker restart

Asynchronous Message Delivery

Decision: Use Go channels with timeout-based delivery.

Implementation:

go func(ch chan *pb.TaskMessage, task pb.TaskMessage) {
    select {
    case ch <- &task:
        // Message sent successfully
    case <-ctx.Done():
        log.Printf("Context cancelled while sending task %s", task.GetTaskId())
    case <-time.After(5 * time.Second):
        log.Printf("Timeout sending task %s. Dropping message.", task.GetTaskId())
    }
}(subChan, taskToSend)

Benefits:

• Non-blocking: Slow agents don’t block the entire system
• Fault tolerance: Timeouts prevent resource leaks
• Scalability: Concurrent delivery to multiple agents
• Resource protection: Prevents unbounded queue growth

Trade-offs:

• Message loss: Timed-out messages are dropped
• Complexity: Requires careful timeout tuning
• No delivery guarantees: No acknowledgment of successful processing

gRPC Streaming for Subscriptions

Decision: Use bidirectional gRPC streams for agent subscriptions.

Benefits:

• Real-time delivery: Messages are pushed immediately
• Connection awareness: Broker knows when agents disconnect
• Flow control: gRPC handles backpressure automatically
• Type safety: Protocol Buffer messages ensure data consistency

Trade-offs:

• Connection overhead: Each agent maintains persistent connections
• Resource usage: Streams consume memory and file descriptors
• Network sensitivity: Transient network issues can break connections

Concurrent Access Patterns

Decision: Use read-write mutexes with channel-based message passing.

Implementation:

s.taskMu.RLock()
// Read subscriber information
var targetChannels []chan *pb.TaskMessage
for _, subs := range s.taskSubscribers {
    targetChannels = append(targetChannels, subs...)
}
s.taskMu.RUnlock()

// Send messages without holding locks
for _, subChan := range targetChannels {
    go func(ch chan *pb.TaskMessage, task pb.TaskMessage) {
        // Async delivery
    }(subChan, taskToSend)
}

Benefits:

• High concurrency: Multiple readers can access subscriptions simultaneously
• Lock-free delivery: Message sending doesn’t hold locks
• Deadlock prevention: Clear lock ordering and minimal critical sections
• Performance: Read operations are optimized for the common case

Scalability Characteristics

Throughput

• Task routing: ~10,000+ tasks/second on modern hardware
• Concurrent connections: Limited by file descriptor limits (typically ~1,000s)
• Memory usage: ~1KB per active subscription

Latency

• Task routing: <1ms for local network delivery
• End-to-end: <10ms for simple task processing cycles
• Progress updates: Real-time streaming with minimal buffering

Resource Usage

• CPU: Low CPU usage, primarily network I/O bound
• Memory: Linear growth with number of active subscriptions
• Network: Efficient binary Protocol Buffer encoding

Error Handling and Resilience

Connection Failures

• Automatic cleanup: Subscriptions are removed when connections close
• Graceful degradation: Failed agents don’t affect others
• Reconnection support: Agents can re-establish subscriptions

Message Delivery Failures

• Timeout handling: Messages that can’t be delivered are dropped
• Logging: All failures are logged for debugging
• Isolation: Per-agent timeouts prevent cascading failures

Resource Protection

• Channel buffering: Limited buffer sizes prevent memory exhaustion
• Timeout mechanisms: Prevent resource leaks from stuck operations
• Graceful shutdown: Proper cleanup during server shutdown

Monitoring and Observability

Built-in Logging

The broker provides comprehensive logging:

• Task routing decisions
• Subscription lifecycle events
• Error conditions and recovery
• Performance metrics

Integration Points

• Health checks: HTTP endpoints for monitoring
• Metrics export: Prometheus/metrics integration points
• Distributed tracing: Context propagation support

Future Enhancements

Persistence Layer

• Database backend: Store subscription state for broker restarts
• Message queuing: Durable task queues for reliability
• Transaction support: Atomic message delivery guarantees

Clustering Support

• Horizontal scaling: Multiple broker instances
• Load balancing: Distribute agents across brokers
• Consensus protocols: Consistent state across brokers

Advanced Routing

• Capability-based routing: Route tasks based on agent capabilities
• Load-aware routing: Consider agent load in routing decisions
• Geographic routing: Route based on agent location

Security Enhancements

• Authentication: Agent identity verification
• Authorization: Task-level access controls
• Encryption: TLS for all communications

The AgentHub broker architecture provides a solid foundation for Agent2Agent communication while maintaining simplicity and performance. Its design supports the immediate needs of most agent systems while providing clear paths for future enhancement as requirements evolve.

1.2 - Hexagonal Architecture & A2A Protocol Implementation

Hexagonal Architecture & A2A Protocol Implementation

This document explains how AgentHub implements hexagonal architecture principles with the Agent2Agent (A2A) protocol, gRPC communication, and event-driven design patterns.

Overview

AgentHub follows hexagonal architecture (also known as Ports and Adapters) to achieve:

• Domain isolation: Core A2A protocol logic separated from infrastructure
• Testability: Clean interfaces enable comprehensive testing
• Flexibility: Multiple adapters for different communication protocols
• Maintainability: Clear separation of concerns and dependencies

System Architecture

graph TB
    subgraph "AgentHub Ecosystem"
        subgraph "External Agents"
            A["Agent A<br/>(Chat REPL)"]
            B["Agent B<br/>(Chat Responder)"]
            C["Agent C<br/>(Custom Agent)"]
        end

        subgraph "AgentHub Broker"
            subgraph "Adapters (Infrastructure)"
                GRPC["gRPC Server<br/>Adapter"]
                HEALTH["Health Check<br/>Adapter"]
                METRICS["Metrics<br/>Adapter"]
                TRACING["Tracing Adapter<br/>(OTLP/Jaeger)"]
            end

            subgraph "Ports (Interfaces)"
                SP["AgentHub<br/>Service Port"]
                PP["Message<br/>Publisher Port"]
                EP["Event<br/>Subscriber Port"]
                OP["Observability<br/>Port"]
            end

            subgraph "Domain (Core Logic)"
                A2A["A2A Protocol<br/>Engine"]
                ROUTER["Event Router<br/>& Broker"]
                VALIDATOR["Message<br/>Validator"]
                CONTEXT["Context<br/>Manager"]
                TASK["Task<br/>Lifecycle"]
            end
        end

        subgraph "External Systems"
            OTLP["OTLP Collector<br/>& Jaeger"]
            STORE["Event Store<br/>(Memory)"]
        end
    end

    %% External agent connections
    A -->|"gRPC calls<br/>(PublishMessage,<br/>SubscribeToMessages)"| GRPC
    B -->|"gRPC calls"| GRPC
    C -->|"gRPC calls"| GRPC

    %% Adapter to Port connections
    GRPC -->|"implements"| SP
    HEALTH -->|"implements"| OP
    METRICS -->|"implements"| OP
    TRACING -->|"implements"| OP

    %% Port to Domain connections
    SP -->|"delegates to"| A2A
    PP -->|"delegates to"| ROUTER
    EP -->|"delegates to"| ROUTER
    OP -->|"observes"| A2A

    %% Domain internal connections
    A2A -->|"uses"| VALIDATOR
    A2A -->|"uses"| CONTEXT
    A2A -->|"uses"| TASK
    ROUTER -->|"persists events"| STORE
    TRACING -->|"exports traces"| OTLP

    %% Styling
    classDef agents fill:#add8e6
    classDef adapters fill:#ffa500
    classDef ports fill:#e0ffff
    classDef domain fill:#ffb6c1
    classDef external fill:#dda0dd

    class A,B,C agents
    class GRPC,HEALTH,METRICS,TRACING adapters
    class SP,PP,EP,OP ports
    class A2A,ROUTER,VALIDATOR,CONTEXT,TASK domain
    class OTLP,STORE external

Architecture Notes:

• Domain Core: Pure A2A protocol logic with message validation, event routing, context correlation, and task state management
• Ports: Clean, technology-agnostic interfaces providing testable contracts and dependency inversion
• Adapters: Infrastructure concerns including gRPC communication, observability exports, and protocol adaptations

A2A Message Flow

sequenceDiagram
    participant REPL as Chat REPL<br/>Agent
    participant gRPC as gRPC<br/>Adapter
    participant A2A as A2A Protocol<br/>Engine
    participant Router as Event<br/>Router
    participant Responder as Chat Responder<br/>Agent

    rect rgb(240, 248, 255)
        Note over REPL, Router: A2A Message Publishing
        REPL->>+gRPC: PublishMessage(A2AMessage)
        gRPC->>+A2A: validateA2AMessage()
        A2A->>A2A: check MessageId, Role, Content
        A2A-->>-gRPC: validation result
        gRPC->>+Router: routeA2AEvent(messageEvent)
        Router->>Router: identify subscribers<br/>by agent_id/broadcast
        Router->>Router: create tracing span<br/>with A2A attributes
        Router-->>Responder: deliver message event
        Router-->>-gRPC: routing success
        gRPC-->>-REPL: PublishResponse(event_id)
    end

    rect rgb(255, 248, 240)
        Note over Responder, Router: A2A Message Processing
        Responder->>+gRPC: SubscribeToMessages(agent_id)
        gRPC->>Router: register subscriber
        Router-->>gRPC: subscription stream
        gRPC-->>-Responder: message stream
        Note over Responder: Process A2A message<br/>with tracing spans
        Responder->>+gRPC: PublishMessage(A2AResponse)
        gRPC->>A2A: validateA2AMessage()
        A2A->>A2A: check AGENT role,<br/>ContextId correlation
        gRPC->>Router: routeA2AEvent(responseEvent)
        Router-->>REPL: deliver response event
        gRPC-->>-Responder: PublishResponse
    end

    Note over REPL, Responder: A2A Protocol ensures:<br/>• Message structure compliance<br/>• Role semantics (USER/AGENT)<br/>• Context correlation<br/>• Event-driven routing

Core Components

1. A2A Protocol Engine (Domain Core)

The heart of the system implementing A2A protocol specifications:

// Core domain logic - technology agnostic
type A2AProtocolEngine struct {
    messageValidator MessageValidator
    contextManager   ContextManager
    taskLifecycle    TaskLifecycle
}

// A2A message validation
func (e *A2AProtocolEngine) ValidateMessage(msg *Message) error {
    // A2A compliance checks
    if msg.MessageId == "" { return ErrMissingMessageId }
    if msg.Role == ROLE_UNSPECIFIED { return ErrInvalidRole }
    if len(msg.Content) == 0 { return ErrEmptyContent }
    return nil
}

2. Event Router (Domain Core)

Manages event-driven communication between agents:

type EventRouter struct {
    messageSubscribers map[string][]chan *AgentEvent
    taskSubscribers    map[string][]chan *AgentEvent
    eventSubscribers   map[string][]chan *AgentEvent
}

func (r *EventRouter) RouteEvent(event *AgentEvent) error {
    // Route based on A2A metadata
    routing := event.GetRouting()
    subscribers := r.getSubscribers(routing.ToAgentId, event.PayloadType)

    // Deliver with tracing
    for _, sub := range subscribers {
        go r.deliverWithTracing(sub, event)
    }
}

3. gRPC Adapter (Infrastructure)

Translates between gRPC and domain logic:

type GrpcAdapter struct {
    a2aEngine    A2AProtocolEngine
    eventRouter  EventRouter
    tracer       TracingAdapter
}

func (a *GrpcAdapter) PublishMessage(ctx context.Context, req *PublishMessageRequest) (*PublishResponse, error) {
    // Start tracing span
    ctx, span := a.tracer.StartA2AMessageSpan(ctx, "publish_message", req.Message.MessageId, req.Message.Role)
    defer span.End()

    // Validate using domain logic
    if err := a.a2aEngine.ValidateMessage(req.Message); err != nil {
        a.tracer.RecordError(span, err)
        return nil, err
    }

    // Route using domain logic
    event := a.createA2AEvent(req)
    if err := a.eventRouter.RouteEvent(event); err != nil {
        return nil, err
    }

    return &PublishResponse{Success: true, EventId: event.EventId}, nil
}

Hexagonal Architecture Benefits

1. Domain Isolation

• A2A protocol logic is pure, testable business logic
• No infrastructure dependencies in the core domain
• Technology-agnostic implementation

2. Adapter Pattern

• gRPC Adapter: Handles Protocol Buffer serialization/deserialization
• Tracing Adapter: OTLP/Jaeger integration without domain coupling
• Health Adapter: Service health monitoring
• Metrics Adapter: Prometheus metrics collection

3. Port Interfaces

// Clean, testable interfaces
type MessagePublisher interface {
    PublishMessage(ctx context.Context, msg *Message) (*PublishResponse, error)
}

type EventSubscriber interface {
    SubscribeToMessages(ctx context.Context, agentId string) (MessageStream, error)
}

type ObservabilityPort interface {
    StartSpan(ctx context.Context, operation string) (context.Context, Span)
    RecordMetric(name string, value float64, labels map[string]string)
}

4. Dependency Inversion

• Domain depends on abstractions (ports), not concrete implementations
• Adapters depend on domain through well-defined interfaces
• Easy testing with mock implementations

A2A Protocol Integration

Message Structure Compliance

classDiagram
    class A2AMessage {
        +string MessageId
        +string ContextId
        +Role Role
        +Part Content
        +Metadata Metadata
        +string TaskId
    }

    class Part {
        +string Text
        +bytes Data
        +FileData File
    }

    class EventMetadata {
        +string FromAgentId
        +string ToAgentId
        +string EventType
        +Priority Priority
    }

    class Role {
        <<enumeration>>
        USER
        AGENT
    }

    class Metadata {
        +Fields map
    }

    A2AMessage "1" --> "0..*" Part : contains
    A2AMessage "1" --> "1" EventMetadata : routed_with
    A2AMessage "1" --> "1" Role : has
    A2AMessage "1" --> "0..1" Metadata : includes

Event-Driven Architecture

The system implements pure event-driven architecture:

1. Publishers emit A2A-compliant events
2. Broker routes events based on metadata
3. Subscribers receive relevant events
4. Correlation through ContextId maintains conversation flow

Observability Integration

Distributed Tracing

sequenceDiagram
    participant A as Agent A
    participant B as Broker
    participant AB as Agent B
    participant OTLP as OTLP Collector
    participant J as Jaeger

    A->>+B: PublishMessage<br/>[trace_id: 123]
    B->>B: Create A2A spans<br/>with structured attributes
    B->>+AB: RouteEvent<br/>[trace_id: 123]
    AB->>AB: Process with<br/>child spans
    AB->>-B: PublishResponse<br/>[trace_id: 123]
    B->>-A: Success<br/>[trace_id: 123]

    par Observability Export
        B->>OTLP: Export spans<br/>with A2A attributes
        OTLP->>J: Store traces
        J->>J: Build trace timeline<br/>with correlation
    end

    Note over A, J: End-to-end tracing<br/>with A2A protocol visibility

Structured Attributes

Each span includes A2A-specific attributes:

• a2a.message.id
• a2a.message.role
• a2a.context.id
• a2a.event.type
• a2a.routing.from_agent
• a2a.routing.to_agent

Testing Strategy

Unit Testing (Domain Core)

func TestA2AEngine_ValidateMessage(t *testing.T) {
    engine := NewA2AProtocolEngine()

    // Test A2A compliance
    msg := &Message{
        MessageId: "test_msg_123",
        Role: ROLE_USER,
        Content: []*Part{{Text: "hello"}},
    }

    err := engine.ValidateMessage(msg)
    assert.NoError(t, err)
}

Integration Testing (Adapters)

func TestGrpcAdapter_PublishMessage(t *testing.T) {
    // Mock domain dependencies
    mockEngine := &MockA2AEngine{}
    mockRouter := &MockEventRouter{}

    adapter := NewGrpcAdapter(mockEngine, mockRouter)

    // Test adapter behavior
    resp, err := adapter.PublishMessage(ctx, validRequest)
    assert.NoError(t, err)
    assert.True(t, resp.Success)
}

Conclusion

AgentHub’s hexagonal architecture with A2A protocol provides:

1. Clean Architecture: Separation of concerns with domain-driven design
2. A2A Compliance: Full protocol implementation with validation
3. Event-Driven Design: Scalable, loosely-coupled communication
4. Rich Observability: Comprehensive tracing and metrics
5. Testability: Clean interfaces enable thorough testing
6. Flexibility: Easy to extend with new adapters and protocols

This architecture ensures maintainable, scalable, and observable agent communication while maintaining strict A2A protocol compliance.

2 - Core Concepts

Core Concepts

Explore the fundamental concepts, principles, and mental models that underpin AgentHub’s agent-to-agent communication system.

Available Documentation

• The Agent-to-Agent Principle - Core philosophy of direct agent communication
• The Tasks - Understanding task messages and data structures

2.1 - The Agent2Agent Principle

The Agent2Agent Protocol and AgentHub Implementation

This document explores the core principles of Google’s Agent2Agent protocol and how AgentHub implements a communication broker based on these concepts. We distinguish between the Agent2Agent protocol specification (task structures and communication patterns) and our custom AgentHub broker implementation.

Agent2Agent vs AgentHub: What’s What

Agent2Agent Protocol (Google)

The Agent2Agent protocol defines:

• A2A Message Structures: Message, Task, Artifact with structured content parts
• Task State Management: TaskState enums (SUBMITTED, WORKING, COMPLETED, FAILED, CANCELLED)
• Communication Patterns: Asynchronous task delegation with context-aware message handling

AgentHub Implementation (This Project)

AgentHub provides:

• Hybrid EDA+A2A Broker: Centralized gRPC service implementing A2A protocol within Event-Driven Architecture
• A2A-Compliant Pub/Sub: Publisher-subscriber pattern using native A2A message structures
• A2A Subscription Mechanisms: SubscribeToTasks, SubscribeToMessages, SubscribeToAgentEvents methods
• A2A Agent Implementations: Sample agents using A2ATaskPublisher and A2ATaskSubscriber abstractions

Philosophy and Core Concepts

Beyond Simple Request-Response

Traditional software architectures rely heavily on synchronous request-response patterns where a client requests a service and waits for an immediate response. While effective for simple operations, this pattern has limitations when dealing with:

• Complex, multi-step processes that require coordination between multiple specialized services
• Long-running operations that may take minutes or hours to complete
• Dynamic workload distribution where the best processor for a task may vary over time
• Autonomous decision-making where agents need to collaborate without central coordination

The Agent2Agent protocol addresses these limitations by defining task structures and communication patterns for autonomous agents. AgentHub implements a broker-based system that enables agents to communicate using Agent2Agent-inspired task structures:

1. Delegating work to other agents based on their capabilities
2. Accepting and processing tasks according to their specializations
3. Reporting progress during long-running operations
4. Making collaborative decisions about task distribution and execution

Autonomous Collaboration

In an Agent2Agent system, each agent operates with a degree of autonomy, making decisions about:

• Which tasks to accept based on current capacity and capabilities
• How to prioritize work when multiple tasks are pending
• When to delegate subtasks to other specialized agents
• How to report progress and handle failures

This autonomy enables the system to be more resilient, scalable, and adaptive compared to centrally-controlled architectures.

Key Design Principles

1. Asynchronous Communication

Agent2Agent communication is fundamentally asynchronous. When Agent A requests work from Agent B:

• Agent A doesn’t block waiting for completion
• Agent B can process the task when resources are available
• Progress updates provide visibility into long-running operations
• Results are delivered when the work is complete

This asynchronicity enables:

• Better resource utilization as agents aren’t blocked waiting
• Improved scalability as systems can handle more concurrent operations
• Enhanced resilience as temporary agent unavailability doesn’t block the entire system

2. Rich A2A Task Semantics

The Agent2Agent protocol defines rich task structures with flexible message content that AgentHub implements:

message Task {
  string id = 1;                         // Unique task identifier
  string context_id = 2;                 // Conversation/workflow context
  TaskStatus status = 3;                 // Current status with latest message
  repeated Message history = 4;          // Complete message history
  repeated Artifact artifacts = 5;       // Task output artifacts
  google.protobuf.Struct metadata = 6;   // Additional context
}

message Message {
  string message_id = 1;                 // Unique message identifier
  string context_id = 2;                 // Conversation context
  string task_id = 3;                    // Associated task
  Role role = 4;                         // USER or AGENT
  repeated Part content = 5;             // Structured content parts
  google.protobuf.Struct metadata = 6;   // Message metadata
}

message TaskStatus {
  TaskState state = 1;                   // SUBMITTED, WORKING, COMPLETED, etc.
  Message update = 2;                    // Latest status message
  google.protobuf.Timestamp timestamp = 3; // Status timestamp
}

This rich A2A structure enables:

• Context-aware routing based on conversation context and message content
• Flexible content handling through structured Part types (text, data, files)
• Workflow coordination via shared context IDs across related tasks
• Complete communication history for debugging and audit trails
• Structured artifact delivery for rich result types

3. A2A Status Updates and Progress Tracking

Long-running tasks benefit from A2A status updates through the message history:

// Progress updates are A2A messages within the task
message TaskStatus {
  TaskState state = 1;                   // Current execution state
  Message update = 2;                    // Latest status message from agent
  google.protobuf.Timestamp timestamp = 3; // When this status was set
}

// Progress information is conveyed through message content
message Message {
  // ... other fields
  repeated Part content = 5;             // Can include progress details
}

// Example progress message content
Part progressPart = {
  part: {
    data: {
      data: {
        "progress_percentage": 65,
        "phase": "data_analysis",
        "estimated_remaining": "2m30s"
      },
      description: "Processing progress update"
    }
  }
}

This A2A approach enables:

• Rich progress communication through structured message content
• Complete audit trails via message history preservation
• Context-aware status updates linking progress to specific workflows
• Flexible progress formats supporting text, data, and file-based updates
• Multi-agent coordination through shared context and message threading

4. A2A EDA Routing Flexibility

AgentHub’s A2A implementation supports multiple routing patterns through EDA metadata:

message AgentEventMetadata {
  string from_agent_id = 1;              // Source agent
  string to_agent_id = 2;                // Target agent (empty = broadcast)
  string event_type = 3;                 // Event classification
  repeated string subscriptions = 4;      // Topic-based routing
  Priority priority = 5;                 // Delivery priority
}

• Direct A2A addressing: Tasks sent to specific agents via to_agent_id
• Broadcast A2A addressing: Tasks sent to all subscribed agents (empty to_agent_id)
• Topic-based A2A routing: Tasks routed via subscription filters and event types
• Context-aware routing: Tasks routed based on A2A context and conversation state

This hybrid EDA+A2A approach enables sophisticated routing patterns while maintaining A2A protocol compliance.

Architectural Patterns

Microservices Enhancement

In a microservices architecture, Agent2Agent can enhance service communication by:

• Replacing synchronous HTTP calls with asynchronous task delegation
• Adding progress visibility to long-running service operations
• Enabling service composition through task chaining
• Improving resilience through task retry and timeout mechanisms

Event-Driven Architecture with A2A Protocol

AgentHub integrates A2A protocol within Event-Driven Architecture by:

• Wrapping A2A messages in EDA event envelopes for routing and delivery
• Preserving A2A semantics while leveraging EDA scalability and reliability
• Enabling A2A conversation contexts within event-driven message flows
• Supporting A2A task coordination alongside traditional event broadcasting
• Providing A2A-compliant APIs that internally use EDA for transport

// A2A message wrapped in EDA event
type AgentEvent struct {
    EventId   string
    Timestamp timestamppb.Timestamp

    // A2A-compliant payload
    Payload oneof {
        a2a.Message message = 10
        a2a.Task task = 11
        TaskStatusUpdateEvent status_update = 12
        TaskArtifactUpdateEvent artifact_update = 13
    }

    // EDA routing metadata
    Routing AgentEventMetadata
}

Workflow Orchestration

Complex business processes can be modeled as Agent2Agent workflows:

1. Process Initiation: A workflow agent receives a high-level business request
2. Task Decomposition: The request is broken down into specific tasks
3. Agent Coordination: Tasks are distributed to specialized agents
4. Progress Aggregation: Individual task progress is combined into overall workflow status
5. Result Assembly: Task results are combined into a final business outcome

Benefits and Trade-offs

Benefits

Scalability: Asynchronous operation and agent autonomy enable horizontal scaling without central bottlenecks.

Resilience: Agent failures don’t cascade as easily since tasks can be retried or redistributed.

Flexibility: New agent types can be added without modifying existing agents.

Observability: Rich task semantics and progress reporting provide excellent visibility into system operations.

Modularity: Agents can be developed, deployed, and scaled independently.

Trade-offs

Complexity: The system requires more sophisticated error handling and state management compared to simple request-response patterns.

Latency: For simple operations, the overhead of task creation and routing may add latency compared to direct calls.

Debugging: Distributed, asynchronous operations can be more challenging to debug than synchronous call chains.

Consistency: Managing data consistency across asynchronous agent operations requires careful design.

When to Use Agent2Agent

Agent2Agent is particularly well-suited for:

Complex Processing Pipelines

When work involves multiple steps that can be performed by different specialized agents:

• Data ingestion → validation → transformation → analysis → reporting
• Image upload → virus scan → thumbnail generation → metadata extraction
• Order processing → inventory check → payment processing → fulfillment

Long-Running Operations

When operations take significant time and users need progress feedback:

• Large file processing
• Machine learning model training
• Complex data analysis
• Batch job processing

Dynamic Load Distribution

When workload characteristics vary and different agents may be better suited for different tasks:

• Multi-tenant systems with varying customer requirements
• Resource-intensive operations that need specialized hardware
• Geographic distribution where local processing is preferred

System Integration

When connecting heterogeneous systems that need to coordinate:

• Third-party service coordination
• Cross-platform workflows

A2A Protocol Comparison with Other Patterns

vs. Message Queues

Traditional message queues provide asynchronous communication but lack:

• A2A structured message parts (text, data, files)
• A2A conversation context and task threading
• A2A bidirectional artifact delivery
• A2A complete message history preservation
• A2A flexible content types and metadata

vs. RPC/HTTP APIs

RPC and HTTP APIs provide structured communication but are typically:

• Synchronous (blocking) vs A2A asynchronous task delegation
• Lacking A2A-style progress tracking through message history
• Point-to-point rather than A2A context-aware routing
• Without A2A structured content parts and artifact handling
• Missing A2A conversation threading and workflow coordination

vs. Event Sourcing

Event sourcing provides audit trails and state reconstruction but:

• Focuses on state changes rather than A2A work coordination
• Lacks A2A structured task status and message threading
• Doesn’t provide A2A artifact-based result delivery
• Requires more complex patterns vs A2A’s built-in conversation context
• Missing A2A’s multi-modal content handling (text, data, files)

A2A Protocol Future Evolution

The A2A protocol and AgentHub implementation opens possibilities for:

Intelligent A2A Agent Networks

Agents that learn from A2A conversation contexts and message patterns to make better delegation decisions based on historical performance and capability matching.

Self-Organizing A2A Systems

Agent networks that automatically reconfigure based on A2A workflow patterns, context relationships, and agent availability, using A2A metadata for intelligent routing decisions.

Cross-Organization A2A Collaboration

Extending A2A protocols across organizational boundaries for B2B workflow automation, leveraging A2A’s structured content parts and artifact handling for secure inter-org communication.

AI Agent A2A Integration

Natural integration points for AI agents that can:

• Parse A2A message content parts for semantic understanding
• Generate appropriate A2A responses with structured artifacts
• Maintain A2A conversation context across complex multi-turn interactions
• Make autonomous decisions about A2A task acceptance based on content analysis

Enhanced A2A Features

• A2A Protocol Extensions: Custom Part types for domain-specific content
• Advanced A2A Routing: ML-based routing decisions using conversation context
• A2A Federation: Cross-cluster A2A communication with context preservation
• A2A Analytics: Deep insights from conversation patterns and artifact flows

The A2A protocol represents a foundational shift toward more intelligent, context-aware, and collaborative software systems that can handle complex distributed workflows while maintaining strong semantics, complete audit trails, and rich inter-agent communication patterns.

2.2 - Agent2Agent (A2A) Protocol Migration

Agent2Agent (A2A) Protocol Migration

This document explains the migration of AgentHub to full Agent2Agent (A2A) protocol compliance while maintaining the essential Event-Driven Architecture (EDA) patterns that make the system scalable and resilient.

What is the Agent2Agent Protocol?

The Agent2Agent (A2A) protocol is a standardized specification for communication between AI agents. It defines:

• Standardized Message Formats: Using Message, Part, Task, and Artifact structures
• Task Lifecycle Management: Clear states (SUBMITTED, WORKING, COMPLETED, FAILED, CANCELLED)
• Agent Discovery: Using AgentCard for capability advertisement
• Interoperability: Ensuring agents can communicate across different platforms

Why Migrate to A2A?

Benefits of A2A Compliance

1. Interoperability: AgentHub can now communicate with any A2A-compliant agent or system
2. Standardization: Clear, well-defined message formats reduce integration complexity
3. Ecosystem Compatibility: Join the growing ecosystem of A2A-compatible tools
4. Future-Proofing: Built on industry standards rather than custom protocols

Maintained EDA Benefits

• Scalability: Event-driven routing scales to thousands of agents
• Resilience: Asynchronous communication handles network partitions gracefully
• Flexibility: Topic-based routing and priority queues enable sophisticated workflows
• Observability: Built-in tracing and metrics for production deployments

Hybrid Architecture

AgentHub implements a hybrid approach that combines the best of both worlds:

┌─────────────────────────────────────────────────────────────────┐
│                   A2A Protocol Layer                           │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────┐│
│  │ A2A Message │  │  A2A Task   │  │ A2A Artifact│  │A2A Agent││
│  │  (standard) │  │ (standard)  │  │ (standard)  │  │  Card   ││
│  └─────────────┘  └─────────────┘  └─────────────┘  └─────────┘│
├─────────────────────────────────────────────────────────────────┤
│                    EDA Transport Layer                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────┐│
│  │ AgentEvent  │  │Event Router │  │ Subscribers │  │Priority ││
│  │  Wrapper    │  │             │  │  Manager    │  │ Queues  ││
│  └─────────────┘  └─────────────┘  └─────────────┘  └─────────┘│
├─────────────────────────────────────────────────────────────────┤
│                      gRPC Infrastructure                       │
└─────────────────────────────────────────────────────────────────┘

How It Works

1. A2A Messages are created using standard A2A structures (Message, Task, etc.)
2. EDA Wrapper wraps A2A messages in AgentEvent for transport
3. Event Routing uses EDA patterns (pub/sub, priority, topics) for delivery
4. A2A Compliance ensures messages follow A2A protocol semantics

API Changes

Before (Legacy API)

// Legacy TaskMessage (deprecated)
taskPublisher.PublishTask(ctx, &agenthub.PublishTaskRequest{
    TaskType: "greeting",
    Parameters: map[string]interface{}{
        "name": "Claude",
    },
    RequesterAgentID: "my_agent",
    ResponderAgentID: "target_agent",
})

After (A2A-Compliant API)

// A2A-compliant task publishing
content := []*pb.Part{
    {
        Part: &pb.Part_Text{
            Text: "Hello! Please provide a greeting for Claude.",
        },
    },
}

task, err := taskPublisher.PublishTask(ctx, &agenthub.A2APublishTaskRequest{
    TaskType:         "greeting",
    Content:          content,
    RequesterAgentID: "my_agent",
    ResponderAgentID: "target_agent",
    Priority:         pb.Priority_PRIORITY_MEDIUM,
    ContextID:        "conversation_123",
})

Message Structure Changes

A2A Message Format

message Message {
  string message_id = 1;       // Unique message identifier
  string context_id = 2;       // Conversation context
  string task_id = 3;          // Associated task (optional)
  Role role = 4;               // USER or AGENT
  repeated Part content = 5;   // Message content parts
  google.protobuf.Struct metadata = 6; // Additional metadata
}

message Part {
  oneof part {
    string text = 1;           // Text content
    DataPart data = 2;         // Structured data
    FilePart file = 3;         // File reference
  }
}

A2A Task Format

message Task {
  string id = 1;                    // Task identifier
  string context_id = 2;            // Conversation context
  TaskStatus status = 3;            // Current status
  repeated Message history = 4;     // Message history
  repeated Artifact artifacts = 5;  // Task outputs
  google.protobuf.Struct metadata = 6; // Task metadata
}

enum TaskState {
  TASK_STATE_SUBMITTED = 0;    // Task created
  TASK_STATE_WORKING = 1;      // Task in progress
  TASK_STATE_COMPLETED = 2;    // Task completed successfully
  TASK_STATE_FAILED = 3;       // Task failed
  TASK_STATE_CANCELLED = 4;    // Task cancelled
}

Migration Guide

For Publishers

1. Replace TaskPublisher with A2ATaskPublisher
2. Use A2APublishTaskRequest with A2A Part structures
3. Handle returned A2A Task objects

For Subscribers

1. Replace TaskSubscriber with A2ATaskSubscriber
2. Update handlers to process A2A Task and Message objects
3. Return A2A Artifact objects instead of custom results

For Custom Integrations

1. Update protobuf imports to use events/a2a package
2. Replace custom message structures with A2A equivalents
3. Use AgentHub service instead of EventBus

Backward Compatibility

The migration maintains wire-level compatibility through:

• Deprecated Types: Legacy message types marked as deprecated but still supported
• Automatic Conversion: EDA broker converts between legacy and A2A formats when needed
• Graceful Migration: Existing agents can migrate incrementally

Testing A2A Compliance

Run the demo to verify A2A compliance:

# Terminal 1: Start A2A broker
make run-server

# Terminal 2: Start A2A subscriber
make run-subscriber

# Terminal 3: Start A2A publisher
make run-publisher

Expected output shows successful A2A task processing:

• Publisher: “Published A2A task”
• Subscriber: “Task processing completed”
• Artifacts generated in A2A format

Best Practices

1. Use A2A Types: Always use A2A message structures for new code
2. Context Management: Use context_id to group related messages
3. Proper Parts: Structure content using appropriate Part types
4. Artifact Returns: Return structured Artifact objects from tasks
5. Status Updates: Properly manage task lifecycle states

The A2A migration ensures AgentHub remains both standards-compliant and highly scalable through its hybrid EDA+A2A architecture.

2.3 - Understanding Tasks in Agent2Agent Communication

Understanding Tasks in Agent2Agent Communication

Tasks are the fundamental unit of work exchange in the Agent2Agent protocol. This document provides a deep dive into task semantics, lifecycle, and design patterns.

Task Anatomy

Core Components

Every task in the Agent2Agent system consists of several key components that define its identity, purpose, and execution context:

A2A Task Identity

string id = 1;                         // Unique task identifier
string context_id = 2;                 // Optional conversation context

The id serves as a unique identifier that allows all participants to track the task throughout its lifecycle. It should be globally unique and meaningful for debugging purposes.

The context_id groups related tasks in a conversation or workflow context, enabling sophisticated multi-task coordination patterns.

Task classification in A2A is handled through the initial Message content rather than a separate task_type field, providing more flexibility for complex task descriptions.

A2A Task Status and History

TaskStatus status = 3;                 // Current task status
repeated Message history = 4;          // Message history for this task
repeated Artifact artifacts = 5;       // Task output artifacts
google.protobuf.Struct metadata = 6;   // Task metadata

In A2A, task data is contained within Message content using the structured Part format:

// A2A task request message
message Message {
  string message_id = 1;
  string context_id = 2;
  string task_id = 3;
  Role role = 4;                    // USER (requester) or AGENT (responder)
  repeated Part content = 5;        // Structured task content
}

message Part {
  oneof part {
    string text = 1;               // Text description
    DataPart data = 2;             // Structured data
    FilePart file = 3;             // File references
  }
}

// Example: A2A data analysis task
taskMessage := &a2a.Message{
    MessageId: "msg_" + uuid.New().String(),
    ContextId: "analysis_workflow_123",
    TaskId:    "task_analysis_456",
    Role:      a2a.Role_USER,
    Content: []*a2a.Part{
        {
            Part: &a2a.Part_Text{
                Text: "Please perform trend analysis on Q4 sales data",
            },
        },
        {
            Part: &a2a.Part_Data{
                Data: &a2a.DataPart{
                    Data: analysisParams, // Structured parameters
                    Description: "Analysis configuration",
                },
            },
        },
    },
}

Metadata in A2A tasks provides additional context for execution, auditing, or debugging:

// A2A task metadata
taskMetadata, _ := structpb.NewStruct(map[string]interface{}{
    "workflow_id":     "workflow_abc123",
    "user_id":         "user_456",
    "request_source":  "web_ui",
    "correlation_id":  "trace_789",
    "priority":        "high",
    "expected_duration": "5m",
})

task := &a2a.Task{
    Id:        "task_analysis_456",
    ContextId: "analysis_workflow_123",
    Metadata:  taskMetadata,
}

A2A Agent Coordination

In A2A, agent coordination is handled through the EDA routing metadata:

message AgentEventMetadata {
  string from_agent_id = 1;           // Source agent identifier
  string to_agent_id = 2;             // Target agent ID (empty = broadcast)
  string event_type = 3;              // Event classification
  repeated string subscriptions = 4;   // Topic-based routing tags
  Priority priority = 5;              // Delivery priority
}

This enables flexible routing patterns:

• from_agent_id identifies the requesting agent
• to_agent_id can specify a target agent or be empty for broadcast
• subscriptions enable topic-based routing for specialized agents
• priority ensures urgent tasks get precedence

A2A Execution Context

A2A handles execution context through the TaskStatus structure:

message TaskStatus {
  TaskState state = 1;                   // SUBMITTED, WORKING, COMPLETED, FAILED, CANCELLED
  Message update = 2;                    // Latest status message
  google.protobuf.Timestamp timestamp = 3; // Status timestamp
}

enum TaskState {
  TASK_STATE_SUBMITTED = 0;
  TASK_STATE_WORKING = 1;
  TASK_STATE_COMPLETED = 2;
  TASK_STATE_FAILED = 3;
  TASK_STATE_CANCELLED = 4;
}

This context helps agents make intelligent scheduling decisions:

• deadline enables time-sensitive prioritization
• priority provides explicit urgency ranking
• created_at enables age-based scheduling policies

Task Lifecycle

1. A2A Task Creation and Publishing

A2A tasks begin their lifecycle when a requesting agent creates a task with an initial message:

// Create A2A task with initial request message
task := &a2a.Task{
    Id:        "task_analysis_" + uuid.New().String(),
    ContextId: "workflow_orchestration_123",
    Status: &a2a.TaskStatus{
        State: a2a.TaskState_TASK_STATE_SUBMITTED,
        Update: &a2a.Message{
            MessageId: "msg_" + uuid.New().String(),
            TaskId:    "task_analysis_" + uuid.New().String(),
            Role:      a2a.Role_USER,
            Content: []*a2a.Part{
                {
                    Part: &a2a.Part_Text{
                        Text: "Please analyze the quarterly sales data for trends",
                    },
                },
                {
                    Part: &a2a.Part_Data{
                        Data: &a2a.DataPart{
                            Data: analysisParams,
                            Description: "Analysis configuration",
                        },
                    },
                },
            },
        },
        Timestamp: timestamppb.Now(),
    },
}

// Publish to AgentHub broker
client.PublishTaskUpdate(ctx, &pb.PublishTaskUpdateRequest{
    Task: task,
    Routing: &pb.AgentEventMetadata{
        FromAgentId: "data_orchestrator",
        ToAgentId:   "data_processor_01", // Optional: specific agent
        EventType:   "task.submitted",
        Priority:    pb.Priority_PRIORITY_HIGH,
    },
})

2. A2A Task Discovery and Acceptance

Agents subscribe to A2A task events and evaluate whether to accept them:

// Agent receives A2A task event
func (a *Agent) evaluateA2ATask(event *pb.AgentEvent) bool {
    task := event.GetTask()
    if task == nil || task.Status.State != a2a.TaskState_TASK_STATE_SUBMITTED {
        return false
    }

    // Analyze task content to understand requirements
    requestMessage := task.Status.Update
    taskDescription := a.extractTaskDescription(requestMessage)

    // Check if agent can handle this task type
    if !a.canHandleTaskType(taskDescription) {
        return false
    }

    // Check capacity constraints
    if a.getCurrentLoad() > a.maxCapacity {
        return false
    }

    // Estimate duration from task content and metadata
    estimatedDuration := a.estimateA2ATaskDuration(task)
    if estimatedDuration > a.maxTaskDuration {
        return false
    }

    return true
}

func (a *Agent) extractTaskDescription(msg *a2a.Message) string {
    for _, part := range msg.Content {
        if textPart := part.GetText(); textPart != "" {
            return textPart
        }
    }
    return ""
}

3. A2A Task Execution with Progress Reporting

Accepted A2A tasks enter the execution phase with regular status updates:

func (a *Agent) executeA2ATask(task *a2a.Task) {
    // Update task to WORKING state
    a.updateTaskStatus(task, a2a.TaskState_TASK_STATE_WORKING, "Task started")

    // Phase 1: Preparation
    a.updateTaskStatus(task, a2a.TaskState_TASK_STATE_WORKING, "Preparing data analysis")
    prepareResult := a.prepareA2AExecution(task)

    // Phase 2: Main processing
    a.updateTaskStatus(task, a2a.TaskState_TASK_STATE_WORKING, "Processing data - 50% complete")
    processResult := a.processA2AData(prepareResult)

    // Phase 3: Finalization
    a.updateTaskStatus(task, a2a.TaskState_TASK_STATE_WORKING, "Finalizing results - 75% complete")
    finalResult := a.finalizeA2AResults(processResult)

    // Completion with artifacts
    a.completeTaskWithArtifacts(task, finalResult)
}

func (a *Agent) updateTaskStatus(task *a2a.Task, state a2a.TaskState, message string) {
    statusUpdate := &a2a.Message{
        MessageId: "msg_" + uuid.New().String(),
        TaskId:    task.Id,
        Role:      a2a.Role_AGENT,
        Content: []*a2a.Part{
            {
                Part: &a2a.Part_Text{
                    Text: message,
                },
            },
        },
    }

    task.Status = &a2a.TaskStatus{
        State:     state,
        Update:    statusUpdate,
        Timestamp: timestamppb.Now(),
    }

    // Publish task update
    a.client.PublishTaskUpdate(context.Background(), &pb.PublishTaskUpdateRequest{
        Task: task,
        Routing: &pb.AgentEventMetadata{
            FromAgentId: a.agentId,
            EventType:   "task.status_update",
        },
    })
}

4. A2A Result Delivery

A2A task completion delivers results through structured artifacts:

func (a *Agent) completeTaskWithArtifacts(task *a2a.Task, resultData interface{}) {
    // Create completion message
    completionMessage := &a2a.Message{
        MessageId: "msg_" + uuid.New().String(),
        TaskId:    task.Id,
        Role:      a2a.Role_AGENT,
        Content: []*a2a.Part{
            {
                Part: &a2a.Part_Text{
                    Text: "Analysis completed successfully",
                },
            },
        },
    }

    // Create result artifact
    resultArtifact := &a2a.Artifact{
        ArtifactId:  "artifact_" + uuid.New().String(),
        Name:        "Analysis Results",
        Description: "Quarterly sales trend analysis",
        Parts: []*a2a.Part{
            {
                Part: &a2a.Part_Data{
                    Data: &a2a.DataPart{
                        Data:        resultData.(structpb.Struct),
                        Description: "Analysis results and metrics",
                    },
                },
            },
        },
    }

    // Update task to completed
    task.Status = &a2a.TaskStatus{
        State:     a2a.TaskState_TASK_STATE_COMPLETED,
        Update:    completionMessage,
        Timestamp: timestamppb.Now(),
    }
    task.Artifacts = append(task.Artifacts, resultArtifact)

    // Publish final task update
    a.client.PublishTaskUpdate(context.Background(), &pb.PublishTaskUpdateRequest{
        Task: task,
        Routing: &pb.AgentEventMetadata{
            FromAgentId: a.agentId,
            EventType:   "task.completed",
        },
    })

    // Publish artifact separately
    a.client.PublishTaskArtifact(context.Background(), &pb.PublishTaskArtifactRequest{
        TaskId:   task.Id,
        Artifact: resultArtifact,
        Routing: &pb.AgentEventMetadata{
            FromAgentId: a.agentId,
            EventType:   "task.artifact",
        },
    })
}

A2A Task Design Patterns

1. Simple A2A Request-Response

The most basic pattern where one agent requests work from another using A2A messages:

Agent A ──[A2A Task]──> AgentHub ──[TaskEvent]──> Agent B
Agent A <─[Artifact]─── AgentHub <─[TaskUpdate]── Agent B

A2A Implementation:

// Agent A creates task
task := &a2a.Task{
    Id: "simple_task_123",
    Status: &a2a.TaskStatus{
        State: a2a.TaskState_TASK_STATE_SUBMITTED,
        Update: &a2a.Message{
            Role: a2a.Role_USER,
            Content: []*a2a.Part{{Part: &a2a.Part_Text{Text: "Convert CSV to JSON"}}},
        },
    },
}

// Agent B responds with artifact
artifact := &a2a.Artifact{
    Name: "Converted Data",
    Parts: []*a2a.Part{{Part: &a2a.Part_File{File: &a2a.FilePart{FileId: "converted.json"}}}},
}

Use cases:

• File format conversion
• Simple calculations
• Data validation
• Content generation

2. A2A Broadcast Processing

One agent broadcasts a task to multiple potential processors using A2A context-aware routing:

Agent A ──[A2A Task]──> AgentHub ──[TaskEvent]──> Agent B₁
                                ├─[TaskEvent]──> Agent B₂
                                └─[TaskEvent]──> Agent B₃

A2A Implementation:

// Broadcast task with shared context
task := &a2a.Task{
    Id:        "broadcast_task_456",
    ContextId: "parallel_processing_context",
    Status: &a2a.TaskStatus{
        State: a2a.TaskState_TASK_STATE_SUBMITTED,
        Update: &a2a.Message{
            Role: a2a.Role_USER,
            Content: []*a2a.Part{
                {Part: &a2a.Part_Text{Text: "Process data chunk"}},
                {Part: &a2a.Part_Data{Data: &a2a.DataPart{Data: chunkData}}},
            },
        },
    },
}

// Publish without specific target (broadcast)
client.PublishTaskUpdate(ctx, &pb.PublishTaskUpdateRequest{
    Task: task,
    Routing: &pb.AgentEventMetadata{
        FromAgentId: "orchestrator",
        // No ToAgentId = broadcast
        EventType: "task.broadcast",
    },
})

Use cases:

• Distributed computation
• Load testing
• Content distribution
• Parallel processing

3. A2A Pipeline Processing

Tasks flow through a series of specialized agents using shared A2A context:

Agent A ──[A2A Task₁]──> Agent B ──[A2A Task₂]──> Agent C ──[A2A Task₃]──> Agent D
       <──[Final Artifact]───────────────────────────────────────────────────┘

A2A Implementation:

// Shared context for pipeline
pipelineContext := "data_pipeline_" + uuid.New().String()

// Stage 1: Data extraction
task1 := &a2a.Task{
    Id:        "extract_" + uuid.New().String(),
    ContextId: pipelineContext,
    Status: &a2a.TaskStatus{
        State: a2a.TaskState_TASK_STATE_SUBMITTED,
        Update: &a2a.Message{
            Role: a2a.Role_USER,
            Content: []*a2a.Part{{Part: &a2a.Part_Text{Text: "Extract data from source"}}},
        },
    },
}

// Stage 2: Data transformation (triggered by Stage 1 completion)
task2 := &a2a.Task{
    Id:        "transform_" + uuid.New().String(),
    ContextId: pipelineContext, // Same context
    Status: &a2a.TaskStatus{
        State: a2a.TaskState_TASK_STATE_SUBMITTED,
        Update: &a2a.Message{
            Role: a2a.Role_USER,
            Content: []*a2a.Part{{Part: &a2a.Part_Text{Text: "Transform extracted data"}}},
        },
    },
}

// Context linking enables pipeline coordination

Use cases:

• Data processing pipelines
• Image processing workflows
• Document processing chains
• ETL operations

4. A2A Hierarchical Decomposition

Complex tasks are broken down into subtasks using A2A context hierarchy:

Agent A ──[A2A ComplexTask]──> Coordinator
                                  ├──[A2A SubTask₁]──> Specialist₁
                                  ├──[A2A SubTask₂]──> Specialist₂
                                  └──[A2A SubTask₃]──> Specialist₃

A2A Implementation:

// Parent task
parentTask := &a2a.Task{
    Id:        "complex_analysis_789",
    ContextId: "business_workflow_123",
    Status: &a2a.TaskStatus{
        State: a2a.TaskState_TASK_STATE_SUBMITTED,
        Update: &a2a.Message{
            Role: a2a.Role_USER,
            Content: []*a2a.Part{{Part: &a2a.Part_Text{Text: "Perform comprehensive business analysis"}}},
        },
    },
}

// Coordinator creates subtasks with hierarchical context
subtask1 := &a2a.Task{
    Id:        "financial_analysis_790",
    ContextId: "business_workflow_123", // Same parent context
    Metadata: map[string]interface{}{
        "parent_task_id": "complex_analysis_789",
        "subtask_type":   "financial",
    },
}

subtask2 := &a2a.Task{
    Id:        "market_analysis_791",
    ContextId: "business_workflow_123", // Same parent context
    Metadata: map[string]interface{}{
        "parent_task_id": "complex_analysis_789",
        "subtask_type":   "market",
    },
}

// Context enables coordination and result aggregation

Use cases:

• Complex business workflows
• Multi-step analysis
• Orchestrated services
• Batch job coordination

5. Competitive Processing

Multiple agents compete to handle the same task (first-come-first-served):

Agent A ──[Task]──> Broker ──[Task]──> Agent B₁ (accepts)
                           ├─[Task]──> Agent B₂ (rejects)
                           └─[Task]──> Agent B₃ (rejects)

Use cases:

• Resource-constrained environments
• Load balancing
• Fault tolerance
• Performance optimization

A2A Task Content and Semantics

A2A Message-Based Classification

In A2A, task classification is handled through message content rather than rigid type fields, providing more flexibility:

Content-Based Classification

// Data processing task
message := &a2a.Message{
    Content: []*a2a.Part{
        {Part: &a2a.Part_Text{Text: "Analyze quarterly sales data for trends"}},
        {Part: &a2a.Part_Data{Data: &a2a.DataPart{Description: "Analysis parameters"}}},
    },
}

// Image processing task
message := &a2a.Message{
    Content: []*a2a.Part{
        {Part: &a2a.Part_Text{Text: "Generate product image with specifications"}},
        {Part: &a2a.Part_Data{Data: &a2a.DataPart{Description: "Image requirements"}}},
    },
}

// Notification task
message := &a2a.Message{
    Content: []*a2a.Part{
        {Part: &a2a.Part_Text{Text: "Send completion notification to user"}},
        {Part: &a2a.Part_Data{Data: &a2a.DataPart{Description: "Notification details"}}},
    },
}

Operation-Based Classification

create.*        - Creation operations
update.*        - Modification operations
delete.*        - Removal operations
analyze.*       - Analysis operations
transform.*     - Transformation operations

Complexity-Based Classification

simple.*        - Quick, low-resource tasks
standard.*      - Normal processing tasks
complex.*       - Resource-intensive tasks
background.*    - Long-running batch tasks

A2A Content Design Guidelines

Be Explicit: Include all information needed for execution in structured Parts

// Good: Explicit A2A content
content := []*a2a.Part{
    {
        Part: &a2a.Part_Text{
            Text: "Convert CSV file to JSON format with specific options",
        },
    },
    {
        Part: &a2a.Part_Data{
            Data: &a2a.DataPart{
                Data: structpb.NewStruct(map[string]interface{}{
                    "source_format":   "csv",
                    "target_format":   "json",
                    "include_headers": true,
                    "delimiter":       ",",
                    "encoding":        "utf-8",
                }),
                Description: "Conversion parameters",
            },
        },
    },
    {
        Part: &a2a.Part_File{
            File: &a2a.FilePart{
                FileId:   "source_data.csv",
                Filename: "data.csv",
                MimeType: "text/csv",
            },
        },
    },
}

// Poor: Ambiguous A2A content
content := []*a2a.Part{
    {
        Part: &a2a.Part_Text{
            Text: "Convert file", // Too vague
        },
    },
}

Use Standard Data Types: Leverage common formats for interoperability

// Good: Standard formats
{
  "timestamp": "2024-01-15T10:30:00Z",      // ISO 8601
  "amount": "123.45",                        // String for precision
  "coordinates": {"lat": 40.7128, "lng": -74.0060}
}

Include Validation Information: Help agents validate inputs

{
  "email": "user@example.com",
  "email_format": "rfc5322",
  "max_length": 254,
  "required": true
}

A2A Error Handling and Edge Cases

A2A Task Rejection

Agents should provide meaningful rejection reasons using A2A message format:

func (a *Agent) rejectA2ATask(task *a2a.Task, reason string) {
    // Create rejection message
    rejectionMessage := &a2a.Message{
        MessageId: "msg_" + uuid.New().String(),
        TaskId:    task.Id,
        Role:      a2a.Role_AGENT,
        Content: []*a2a.Part{
            {
                Part: &a2a.Part_Text{
                    Text: "Task rejected: " + reason,
                },
            },
            {
                Part: &a2a.Part_Data{
                    Data: &a2a.DataPart{
                        Data: structpb.NewStruct(map[string]interface{}{
                            "rejection_reason": reason,
                            "agent_id":         a.agentId,
                            "timestamp":        time.Now().Unix(),
                        }),
                        Description: "Rejection details",
                    },
                },
            },
        },
    }

    // Update task status to failed
    task.Status = &a2a.TaskStatus{
        State:     a2a.TaskState_TASK_STATE_FAILED,
        Update:    rejectionMessage,
        Timestamp: timestamppb.Now(),
    }

    a.publishTaskUpdate(task)
}

Common rejection reasons:

• UNSUPPORTED_TASK_TYPE: Agent doesn’t handle this task type
• CAPACITY_EXCEEDED: Agent is at maximum capacity
• DEADLINE_IMPOSSIBLE: Cannot complete within deadline
• INVALID_PARAMETERS: Task parameters are malformed
• RESOURCE_UNAVAILABLE: Required external resources unavailable

Timeout Handling

Both requesters and processors should handle timeouts gracefully:

// Requester timeout
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()

select {
case result := <-resultChannel:
    // Process result
case <-ctx.Done():
    // Handle timeout - possibly retry or fail
}

// Processor timeout
func (a *Agent) executeWithTimeout(task *pb.TaskMessage) {
    deadline := task.GetDeadline().AsTime()
    ctx, cancel := context.WithDeadline(context.Background(), deadline)
    defer cancel()

    select {
    case result := <-a.processTask(ctx, task):
        a.publishResult(task, result, pb.TaskStatus_TASK_STATUS_COMPLETED)
    case <-ctx.Done():
        a.publishResult(task, nil, pb.TaskStatus_TASK_STATUS_FAILED, "Deadline exceeded")
    }
}

Partial Results

For long-running tasks, consider supporting partial results:

type PartialResult struct {
    TaskId          string
    CompletedPortion float64    // 0.0 to 1.0
    IntermediateData interface{}
    CanResume       bool
    ResumeToken     string
}

Best Practices

Task Design

1. Make task types granular but not too fine-grained
2. Design for idempotency when possible
3. Include retry information in metadata
4. Use consistent parameter naming across similar task types
5. Version your task schemas to enable evolution

Performance Considerations

1. Batch related tasks when appropriate
2. Use appropriate priority levels to avoid starvation
3. Set realistic deadlines based on historical performance
4. Include resource hints to help with scheduling
5. Monitor task completion rates to identify bottlenecks

Security Considerations

1. Validate all task parameters before processing
2. Sanitize user-provided data in task parameters
3. Include authorization context in metadata
4. Log task execution for audit trails
5. Encrypt sensitive parameters when necessary

A2A tasks form the foundation of Agent2Agent communication, enabling sophisticated distributed processing patterns through structured messages, artifacts, and context-aware coordination. The A2A protocol’s flexible message format and EDA integration provide robust, scalable agent networks with clear semantics and strong observability. Proper A2A task design leverages the protocol’s strengths for building maintainable, interoperable agent systems.

3 - Features

Feature Explanations

Detailed explanations of AgentHub’s advanced features, their design rationale, and implementation details.

Available Documentation

• Distributed Tracing - OpenTelemetry integration and tracing architecture
• Architecture Evolution - Evolution from build tags to unified abstractions
• Performance and Scaling - Performance characteristics and scaling strategies
• Unified Abstraction Library - High-level API design and abstractions

3.1 - Distributed Tracing & OpenTelemetry

🔍 Distributed Tracing & OpenTelemetry

Understanding-oriented: Deep dive into distributed tracing concepts, OpenTelemetry architecture, and how AgentHub implements comprehensive observability for event-driven systems.

The Problem: Observing Distributed Systems

Traditional monolithic applications are relatively easy to debug—everything happens in one process, on one machine, with one log file. But modern event-driven architectures like AgentHub present unique challenges:

The Complexity of Event-Driven Systems

Request Flow in AgentHub:
User → Publisher Agent → AgentHub Broker → Subscriber Agent → Result → Publisher Agent

Each step involves:

• Different processes (potentially on different machines)
• Asynchronous communication (events, not direct calls)
• Multiple protocol layers (gRPC, HTTP, network)
• Independent failure modes (network partitions, service crashes)
• Varying performance characteristics (CPU, memory, I/O)

Traditional Debugging Challenges

Without distributed tracing:

Publisher logs:   "Published task task_123 at 10:00:01"
Broker logs:     "Received task from agent_pub at 10:00:01"
                 "Routed task to agent_sub at 10:00:01"
Subscriber logs: "Processing task task_456 at 10:00:02"
                 "Completed task task_789 at 10:00:03"

Questions you can’t answer:

• Which subscriber processed task_123?
• How long did task_123 take end-to-end?
• Where did task_123 fail?
• What was the complete flow for a specific request?

The Solution: Distributed Tracing

Distributed tracing solves these problems by creating a unified view of requests as they flow through multiple services.

Core Concepts

Trace

A trace represents a complete request journey through the system. In AgentHub, a trace might represent:

• Publishing a task
• Processing the task
• Publishing the result
• Receiving the result

Trace ID: a1b2c3d4e5f67890
Duration: 150ms
Services: 3 (publisher, broker, subscriber)
Spans: 5
Status: Success

Span

A span represents a single operation within a trace. Each span has:

• Name: What operation it represents
• Start/End time: When it happened
• Tags: Metadata about the operation
• Logs: Events that occurred during the operation
• Status: Success, error, or timeout

Span: "publish_event"
  Service: agenthub-publisher
  Duration: 25ms
  Tags:
    event.type: "greeting"
    event.id: "task_123"
    responder.agent: "agent_demo_subscriber"
  Status: OK

Span Context

The glue that connects spans across service boundaries. Contains:

• Trace ID: Unique identifier for the entire request
• Span ID: Unique identifier for the current operation
• Trace Flags: Sampling decisions, debug mode, etc.

How Tracing Works in AgentHub

1. Trace Initiation

When a publisher creates a task, it starts a new trace:

// Publisher starts a trace
ctx, span := tracer.Start(ctx, "publish_event")
defer span.End()

// Add metadata
span.SetAttributes(
    attribute.String("event.type", "greeting"),
    attribute.String("event.id", taskID),
)

2. Context Propagation

The trace context is injected into the task metadata:

// Inject trace context into task headers
headers := make(map[string]string)
otel.GetTextMapPropagator().Inject(ctx, propagation.MapCarrier(headers))

// Embed headers in task metadata
task.Metadata = &structpb.Struct{
    Fields: map[string]*structpb.Value{
        "trace_headers": structpb.NewStructValue(&structpb.Struct{
            Fields: stringMapToStructFields(headers),
        }),
    },
}

3. Context Extraction

The broker and subscriber extract the trace context:

// Extract trace context from task metadata
if metadata := task.GetMetadata(); metadata != nil {
    if traceHeaders, ok := metadata.Fields["trace_headers"]; ok {
        headers := structFieldsToStringMap(traceHeaders.GetStructValue().Fields)
        ctx = otel.GetTextMapPropagator().Extract(ctx, propagation.MapCarrier(headers))
    }
}

// Continue the trace
ctx, span := tracer.Start(ctx, "process_event")
defer span.End()

4. Complete Request Flow

The result is a complete trace showing the entire request journey:

Trace: a1b2c3d4e5f67890
├── publish_event (agenthub-publisher) [25ms]
│   ├── event.type: greeting
│   └── event.id: task_123
├── route_task (agenthub-broker) [2ms]
│   ├── source.agent: agent_demo_publisher
│   └── target.agent: agent_demo_subscriber
├── consume_event (agenthub-subscriber) [5ms]
│   └── messaging.operation: receive
├── process_task (agenthub-subscriber) [98ms]
│   ├── task.type: greeting
│   ├── task.parameter.name: Claude
│   └── processing.status: completed
└── publish_result (agenthub-subscriber) [20ms]
    └── result.status: success

OpenTelemetry Architecture

OpenTelemetry is the observability framework that powers AgentHub’s tracing implementation.

The OpenTelemetry Stack

┌─────────────────────────────────────────────────────────┐
│                    Applications                         │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐      │
│  │  Publisher  │ │   Broker    │ │ Subscriber  │      │
│  └─────────────┘ └─────────────┘ └─────────────┘      │
└─────────────────┬───────────────┬───────────────┬─────┘
                  │               │               │
┌─────────────────▼───────────────▼───────────────▼─────┐
│              OpenTelemetry SDK                        │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐      │
│  │   Tracer    │ │    Meter    │ │   Logger    │      │
│  └─────────────┘ └─────────────┘ └─────────────┘      │
└─────────────────────────────────┬───────────────────────┘
                                  │
┌─────────────────────────────────▼───────────────────────┐
│            OpenTelemetry Collector                     │
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐      │
│  │  Receivers  │ │ Processors  │ │  Exporters  │      │
│  └─────────────┘ └─────────────┘ └─────────────┘      │
└─────────────────┬───────────────┬───────────────┬─────┘
                  │               │               │
┌─────────────────▼─────┐ ┌───────▼───────┐ ┌─────▼─────┐
│      Jaeger           │ │  Prometheus   │ │   Logs    │
│   (Tracing)           │ │  (Metrics)    │ │(Logging)  │
└───────────────────────┘ └───────────────┘ └───────────┘

Core Components

Tracer

Creates and manages spans:

tracer := otel.Tracer("agenthub-publisher")
ctx, span := tracer.Start(ctx, "publish_event")
defer span.End()

Meter

Creates and manages metrics:

meter := otel.Meter("agenthub-publisher")
counter, _ := meter.Int64Counter("events_published_total")
counter.Add(ctx, 1)

Propagators

Handle context propagation across service boundaries:

// Inject context
otel.GetTextMapPropagator().Inject(ctx, carrier)

// Extract context
ctx = otel.GetTextMapPropagator().Extract(ctx, carrier)

Exporters

Send telemetry data to backend systems:

• OTLP Exporter: Sends to OpenTelemetry Collector
• Jaeger Exporter: Sends directly to Jaeger
• Prometheus Exporter: Exposes metrics for Prometheus

AgentHub’s OpenTelemetry Implementation

Configuration

func NewObservability(config Config) (*Observability, error) {
    // Create resource (service identification)
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName(config.ServiceName),
            semconv.ServiceVersion(config.ServiceVersion),
        ),
    )

    // Setup tracing
    traceExporter, err := otlptracegrpc.New(ctx,
        otlptracegrpc.WithEndpoint(config.JaegerEndpoint),
        otlptracegrpc.WithInsecure(),
    )

    tracerProvider := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(traceExporter),
        sdktrace.WithResource(res),
        sdktrace.WithSampler(sdktrace.AlwaysSample()),
    )

    otel.SetTracerProvider(tracerProvider)

    // Setup metrics
    meterProvider := sdkmetric.NewMeterProvider(
        sdkmetric.WithResource(res),
        sdkmetric.WithReader(promExporter),
    )

    otel.SetMeterProvider(meterProvider)
}

Custom slog Handler Integration

AgentHub’s custom logging handler automatically correlates logs with traces:

func (h *ObservabilityHandler) Handle(ctx context.Context, r slog.Record) error {
    // Extract trace context
    if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() {
        spanCtx := span.SpanContext()
        attrs = append(attrs,
            slog.String("trace_id", spanCtx.TraceID().String()),
            slog.String("span_id", spanCtx.SpanID().String()),
        )
    }

    // Structured log output with trace correlation
    logData := map[string]interface{}{
        "time":     r.Time.Format(time.RFC3339),
        "level":    r.Level.String(),
        "msg":      r.Message,
        "trace_id": spanCtx.TraceID().String(),
        "span_id":  spanCtx.SpanID().String(),
        "service":  h.serviceName,
    }
}

Observability Patterns in Event-Driven Systems

Pattern 1: Event Correlation

Challenge: Correlating events across async boundaries Solution: Inject trace context into event metadata

// Publisher injects context
headers := make(map[string]string)
otel.GetTextMapPropagator().Inject(ctx, propagation.MapCarrier(headers))
event.Metadata["trace_headers"] = headers

// Consumer extracts context
ctx = otel.GetTextMapPropagator().Extract(ctx, propagation.MapCarrier(event.Metadata["trace_headers"]))

Pattern 2: Async Operation Tracking

Challenge: Tracking operations that complete asynchronously Solution: Create child spans that can outlive their parents

// Start async operation
ctx, span := tracer.Start(ctx, "async_operation")

go func() {
    defer span.End()
    // Long-running async work
    processTask()
    span.SetStatus(2, "") // Success
}()

// Parent can continue/return immediately

Pattern 3: Error Propagation

Challenge: Understanding how errors flow through the system Solution: Record errors at each span and propagate error status

if err != nil {
    span.RecordError(err)
    span.SetStatus(1, err.Error()) // Error status

    // Optionally add error details
    span.SetAttributes(
        attribute.String("error.type", "validation_error"),
        attribute.String("error.message", err.Error()),
    )
}

Pattern 4: Performance Attribution

Challenge: Understanding where time is spent in complex flows Solution: Detailed span hierarchy with timing

// High-level operation
ctx, span := tracer.Start(ctx, "process_task")
defer span.End()

// Sub-operations
ctx, validateSpan := tracer.Start(ctx, "validate_input")
// ... validation logic
validateSpan.End()

ctx, computeSpan := tracer.Start(ctx, "compute_result")
// ... computation logic
computeSpan.End()

ctx, persistSpan := tracer.Start(ctx, "persist_result")
// ... persistence logic
persistSpan.End()