This is the multi-page printable view of this section. Click here to print.
Detailed explanations of AgentHub’s advanced features, their design rationale, and implementation details.
Understanding-oriented: Deep dive into distributed tracing concepts, OpenTelemetry architecture, and how AgentHub implements comprehensive observability for event-driven 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:
Request Flow in AgentHub:
User → Publisher Agent → AgentHub Broker → Subscriber Agent → Result → Publisher Agent
Each step involves:
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:
Distributed tracing solves these problems by creating a unified view of requests as they flow through multiple services.
A trace represents a complete request journey through the system. In AgentHub, a trace might represent:
Trace ID: a1b2c3d4e5f67890
Duration: 150ms
Services: 3 (publisher, broker, subscriber)
Spans: 5
Status: Success
A span represents a single operation within a trace. Each span has:
Span: "publish_event"
Service: agenthub-publisher
Duration: 25ms
Tags:
event.type: "greeting"
event.id: "task_123"
responder.agent: "agent_demo_subscriber"
Status: OK
The glue that connects spans across service boundaries. Contains:
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),
)
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),
}),
},
}
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()
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 is the observability framework that powers AgentHub’s tracing implementation.
┌─────────────────────────────────────────────────────────┐
│ Applications │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Publisher │ │ Broker │ │ Subscriber │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────┬───────────────┬───────────────┬─────┘
│ │ │
┌─────────────────▼───────────────▼───────────────▼─────┐
│ OpenTelemetry SDK │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Tracer │ │ Meter │ │ Logger │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────┬───────────────────────┘
│
┌─────────────────────────────────▼───────────────────────┐
│ OpenTelemetry Collector │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Receivers │ │ Processors │ │ Exporters │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────┬───────────────┬───────────────┬─────┘
│ │ │
┌─────────────────▼─────┐ ┌───────▼───────┐ ┌─────▼─────┐
│ Jaeger │ │ Prometheus │ │ Logs │
│ (Tracing) │ │ (Metrics) │ │(Logging) │
└───────────────────────┘ └───────────────┘ └───────────┘
Creates and manages spans:
tracer := otel.Tracer("agenthub-publisher")
ctx, span := tracer.Start(ctx, "publish_event")
defer span.End()
Creates and manages metrics:
meter := otel.Meter("agenthub-publisher")
counter, _ := meter.Int64Counter("events_published_total")
counter.Add(ctx, 1)
Handle context propagation across service boundaries:
// Inject context
otel.GetTextMapPropagator().Inject(ctx, carrier)
// Extract context
ctx = otel.GetTextMapPropagator().Extract(ctx, carrier)
Send telemetry data to backend systems:
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)
}
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,
}
}
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"]))
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
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()),
)
}
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()
Not every request needs to be traced. Sampling reduces overhead:
// Probability sampling (trace 10% of requests)
sdktrace.WithSampler(sdktrace.ParentBased(
sdktrace.TraceIDRatioBased(0.1),
))
// Rate limiting sampling (max 100 traces/second)
sdktrace.WithSampler(sdktrace.ParentBased(
sdktrace.RateLimited(100),
))
Add business context to spans:
span.SetAttributes(
attribute.String("user.id", userID),
attribute.String("tenant.id", tenantID),
attribute.Int("batch.size", len(items)),
attribute.String("workflow.type", workflowType),
)
Add timestamped events within spans:
span.AddEvent("validation.started")
// ... validation logic
span.AddEvent("validation.completed", trace.WithAttributes(
attribute.Int("validation.rules.evaluated", ruleCount),
))
Propagate key-value pairs across the entire trace:
// Set baggage
ctx = baggage.ContextWithValues(ctx,
baggage.String("user.tier", "premium"),
baggage.String("feature.flag", "new_algorithm"),
)
// Read baggage in any service
if member := baggage.FromContext(ctx).Member("user.tier"); member.Value() == "premium" {
// Use premium algorithm
}
AgentHub’s observability adds:
Symptoms: No traces appear in Jaeger Causes:
Debugging:
# Check if spans are being created
curl http://localhost:8080/metrics | grep trace
# Check collector logs
docker-compose logs otel-collector
# Verify Jaeger connectivity
curl http://localhost:16686/api/traces
Symptoms: Spans appear disconnected Causes:
Debugging:
// Always check if context contains active span
if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() {
fmt.Printf("Active trace: %s\n", span.SpanContext().TraceID())
} else {
fmt.Println("No active trace context")
}
Symptoms: Observability causing OOM errors Causes:
Solutions:
// Configure memory limits
config := sdktrace.NewTracerProvider(
sdktrace.WithSpanLimits(sdktrace.SpanLimits{
AttributeCountLimit: 128,
EventCountLimit: 128,
LinkCountLimit: 128,
}),
)
Distributed tracing transforms debugging from guesswork into precise investigation. AgentHub’s implementation with OpenTelemetry provides:
The investment in observability pays dividends in:
🎯 Ready to Implement?
Hands-on: Observability Demo Tutorial
Production: Add Observability to Your Agent
Deep Dive: Observability Architecture
Understanding-oriented: Learn how AgentHub evolved from build tag-based conditional compilation to a unified abstraction approach that dramatically simplifies development while providing comprehensive observability.
AgentHub originally used Go build tags to handle different deployment scenarios:
go build)go build -tags observability)Problems with Build Tags:
AgentHub now uses a unified abstraction layer (internal/agenthub/) that provides:
The unified abstraction provides these key components:
// Single server implementation with built-in observability
server, err := agenthub.NewAgentHubServer(config)
if err != nil {
return err
}
// Automatic OpenTelemetry, metrics, health checks
err = server.Start(ctx)
// Single client implementation with built-in observability
client, err := agenthub.NewAgentHubClient(config)
if err != nil {
return err
}
// Automatic tracing, metrics, structured logging
err = client.Start(ctx)
// High-level abstractions with automatic correlation
publisher := &agenthub.TaskPublisher{
Client: client.Client,
TraceManager: client.TraceManager,
// Built-in observability
}
subscriber := agenthub.NewTaskSubscriber(client, agentID)
// Automatic task processing with tracing
File Structure (Legacy):
agents/publisher/
├── main.go # Basic version (~200 lines)
├── main_observability.go # Observable version (~380 lines)
├── shared.go # Common code
└── config.go # Configuration
broker/
├── main.go # Basic broker (~150 lines)
├── main_observability.go # Observable broker (~300 lines)
└── server.go # Core logic
Build Commands (Legacy):
# Basic build
go build -o bin/publisher agents/publisher/
# Observable build
go build -tags observability -o bin/publisher-obs agents/publisher/
# Testing observable features
go test -tags observability ./...
File Structure (Current):
agents/publisher/
└── main.go # Single implementation (~50 lines)
agents/subscriber/
└── main.go # Single implementation (~60 lines)
broker/
└── main.go # Single implementation (~30 lines)
internal/agenthub/ # Unified abstraction layer
├── grpc.go # Client/server with observability
├── subscriber.go # Task processing abstractions
├── broker.go # Event bus implementation
└── metadata.go # Correlation and metadata
Build Commands (Current):
# Single build for all use cases
go build -o bin/publisher agents/publisher/
go build -o bin/subscriber agents/subscriber/
go build -o bin/broker broker/
# Testing (no special tags needed)
go test ./...
Instead of build tags, features are now controlled via environment variables:
# Observability configuration
export JAEGER_ENDPOINT="http://localhost:14268/api/traces"
export OTEL_SERVICE_NAME="agenthub"
export OTEL_SERVICE_VERSION="1.0.0"
# Health and metrics ports
export BROKER_HEALTH_PORT="8080"
# Broker connection
export AGENTHUB_BROKER_ADDR="localhost"
export AGENTHUB_BROKER_PORT="50051"
The unified abstractions automatically configure features based on environment:
// Observability is automatically configured
config := agenthub.NewGRPCConfig("publisher")
client, err := agenthub.NewAgentHubClient(config)
// If JAEGER_ENDPOINT is set → tracing enabled
// If BROKER_HEALTH_PORT is set → health server enabled
// Always includes structured logging and basic metrics
For users migrating from the old build tag approach:
# OLD: Basic builds
go build -o bin/publisher agents/publisher/
# NEW: Same command (unchanged)
go build -o bin/publisher agents/publisher/
# OLD: Observable builds
go build -tags observability -o bin/publisher-obs agents/publisher/
# NEW: Same binary, configure via environment
export JAEGER_ENDPOINT="http://localhost:14268/api/traces"
go build -o bin/publisher agents/publisher/
# OLD: Testing with tags
go test -tags observability ./...
# NEW: Standard testing
go test ./...
# OLD: Feature controlled by build tags
go build -tags observability
# NEW: Feature controlled by environment
export JAEGER_ENDPOINT="http://localhost:14268/api/traces"
export OTEL_SERVICE_NAME="my-service"
The move from build tags to unified abstractions represents a fundamental shift:
Build Tags Philosophy (Old):
Unified Abstractions Philosophy (New):
The unified approach has minimal overhead:
Binary Size:
- Old basic: ~8MB
- Old observable: ~15MB
- New unified: ~12MB
Memory Usage:
- Baseline: ~10MB
- With observability: ~15MB (when enabled)
- Without observability: ~10MB (minimal overhead)
Startup Time:
- With observability enabled: ~150ms
- With observability disabled: ~50ms
The abstractions use lazy initialization:
// Observability components only initialize if configured
if config.JaegerEndpoint != "" {
// Initialize tracing
}
if config.HealthPort != "" {
// Start health server
}
// Always minimal logging and basic metrics
The unified abstractions maintain backward compatibility:
This architectural evolution demonstrates how AgentHub prioritizes developer experience and operational simplicity while maintaining full observability capabilities. The move from build tags to unified abstractions represents a maturation of the platform toward cloud-native best practices.
This document explores the performance characteristics of AgentHub, scaling patterns, and optimization strategies for different deployment scenarios.
Test Environment:
Measured Performance:
Task routing is the core performance bottleneck in AgentHub:
// Fast path: Direct agent routing
if responderID := req.GetTask().GetResponderAgentId(); responderID != "" {
if subs, ok := s.taskSubscribers[responderID]; ok {
targetChannels = subs // O(1) lookup
}
}
Optimization factors:
Protocol Buffers provide efficient serialization:
// Memory usage breakdown per agent:
type agentMemoryFootprint struct {
SubscriptionState int // ~200 bytes (map entry + channel)
ChannelBuffer int // ~800 bytes (10 message buffer * 80 bytes avg)
ConnectionOverhead int // ~2KB (gRPC stream state)
// Total: ~3KB per active agent
}
Memory optimization strategies:
Increasing resources on a single broker instance:
// Configure for CPU optimization
export GOMAXPROCS=8 // Match available CPU cores
// Memory-optimized configuration
subChan := make(chan *pb.TaskMessage, 5) // Smaller buffers for memory-constrained environments
// vs
subChan := make(chan *pb.TaskMessage, 50) // Larger buffers for high-throughput environments
Distributing load across multiple broker instances:
Static Partitioning:
Agent Groups:
├── Broker 1: agents_1-1000
├── Broker 2: agents_1001-2000
└── Broker 3: agents_2001-3000
Hash-based Partitioning:
func selectBroker(agentID string) string {
hash := fnv.New32a()
hash.Write([]byte(agentID))
brokerIndex := hash.Sum32() % uint32(len(brokers))
return brokers[brokerIndex]
}
Specialized Brokers:
Task Routing:
├── Broker 1: data_processing, analytics
├── Broker 2: image_processing, ml_inference
└── Broker 3: notifications, logging
Regional Distribution:
Geographic Deployment:
├── US-East: Broker cluster for East Coast agents
├── US-West: Broker cluster for West Coast agents
└── EU: Broker cluster for European agents
type LoadBalancer struct {
brokers []string
current int
mu sync.Mutex
}
func (lb *LoadBalancer) NextBroker() string {
lb.mu.Lock()
defer lb.mu.Unlock()
broker := lb.brokers[lb.current]
lb.current = (lb.current + 1) % len(lb.brokers)
return broker
}
type BrokerMetrics struct {
ActiveAgents int
TasksPerSec float64
CPUUsage float64
MemoryUsage float64
}
func selectBestBroker(brokers []BrokerMetrics) int {
// Select broker with lowest load score
bestIndex := 0
bestScore := calculateLoadScore(brokers[0])
for i, broker := range brokers[1:] {
score := calculateLoadScore(broker)
if score < bestScore {
bestScore = score
bestIndex = i + 1
}
}
return bestIndex
}
For high-throughput scenarios, implement message batching:
type BatchProcessor struct {
tasks []*pb.TaskMessage
batchSize int
timeout time.Duration
ticker *time.Ticker
}
func (bp *BatchProcessor) processBatch() {
batch := make([]*pb.TaskMessage, len(bp.tasks))
copy(batch, bp.tasks)
bp.tasks = bp.tasks[:0] // Clear slice
// Process entire batch
go bp.routeBatch(batch)
}
Optimize gRPC connections for better resource utilization:
type ConnectionPool struct {
connections map[string]*grpc.ClientConn
maxConns int
mu sync.RWMutex
}
func (cp *ConnectionPool) GetConnection(addr string) (*grpc.ClientConn, error) {
cp.mu.RLock()
if conn, exists := cp.connections[addr]; exists {
cp.mu.RUnlock()
return conn, nil
}
cp.mu.RUnlock()
// Create new connection
return cp.createConnection(addr)
}
Dynamically adjust channel buffer sizes based on load:
func calculateOptimalBufferSize(avgTaskRate float64, processingTime time.Duration) int {
// Buffer size = rate * processing time + safety margin
bufferSize := int(avgTaskRate * processingTime.Seconds()) + 10
// Clamp to reasonable bounds
if bufferSize < 5 {
return 5
}
if bufferSize > 100 {
return 100
}
return bufferSize
}
Reduce memory allocations in hot paths:
// Use sync.Pool for frequent allocations
var taskPool = sync.Pool{
New: func() interface{} {
return &pb.TaskMessage{}
},
}
func processTaskOptimized(task *pb.TaskMessage) {
// Reuse task objects
pooledTask := taskPool.Get().(*pb.TaskMessage)
defer taskPool.Put(pooledTask)
// Copy and process
*pooledTask = *task
// ... processing logic
}
type ThroughputMetrics struct {
TasksPerSecond float64
ResultsPerSecond float64
ProgressPerSecond float64
MessagesPerSecond float64
}
type LatencyMetrics struct {
RoutingLatency time.Duration // Broker routing time
ProcessingLatency time.Duration // Agent processing time
EndToEndLatency time.Duration // Total task completion time
P50, P95, P99 time.Duration // Percentile latencies
}
type ResourceMetrics struct {
ActiveAgents int
ActiveTasks int
MemoryUsage int64
CPUUsage float64
GoroutineCount int
OpenConnections int
}
import "github.com/prometheus/client_golang/prometheus"
var (
taskCounter = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "agenthub_tasks_total",
Help: "Total number of tasks processed",
},
[]string{"task_type", "status"},
)
latencyHistogram = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Name: "agenthub_task_duration_seconds",
Help: "Task processing duration",
Buckets: prometheus.DefBuckets,
},
[]string{"task_type"},
)
)
Symptoms: Slow task processing times Causes: Network latency, overloaded agents, inefficient routing Solutions: Optimize routing, add caching, scale horizontally
Symptoms: Increasing memory usage over time Causes: Unclosed channels, goroutine leaks, connection leaks Solutions: Proper cleanup, monitoring, garbage collection tuning
Symptoms: New agents can’t connect Causes: OS file descriptor limits, broker resource limits Solutions: Increase limits, implement connection pooling
Symptoms: Tasks not reaching agents or results not returned Causes: Timeout issues, network problems, buffer overflows Solutions: Increase timeouts, improve error handling, adjust buffer sizes
func loadTest() {
// Create multiple publishers
publishers := make([]Publisher, 10)
for i := range publishers {
publishers[i] = NewPublisher(fmt.Sprintf("publisher_%d", i))
}
// Send tasks concurrently
taskRate := 1000 // tasks per second
duration := 60 * time.Second
ticker := time.NewTicker(time.Duration(1e9 / taskRate))
timeout := time.After(duration)
for {
select {
case <-ticker.C:
publisher := publishers[rand.Intn(len(publishers))]
go publisher.PublishTask(generateRandomTask())
case <-timeout:
return
}
}
}
The AgentHub architecture provides solid performance for most use cases and clear scaling paths for growing deployments. Regular monitoring and optimization ensure continued performance as your agent ecosystem evolves.
The AgentHub Unified Abstraction Library (internal/agenthub/) is a comprehensive set of A2A protocol-compliant abstractions that dramatically simplifies the development of A2A agents and brokers while providing built-in observability, environment-based configuration, and automatic correlation tracking.
Before (Legacy approach):
broker/main_observability.go: 380+ lines of boilerplateAfter (Unified abstractions):
broker/main.go: 29 lines using abstractionsgrpc.go)Provides a complete gRPC server abstraction with:
// Create and start a broker in one line
func StartBroker(ctx context.Context) error {
config := NewGRPCConfig("broker")
server, err := NewAgentHubServer(config)
if err != nil {
return err
}
return server.Start(ctx)
}
Provides a complete gRPC client abstraction with:
// Create a client with built-in observability
config := agenthub.NewGRPCConfig("publisher")
client, err := agenthub.NewAgentHubClient(config)
a2a.go)Simplifies A2A task publishing with:
a2aPublisher := &agenthub.A2ATaskPublisher{
Client: client.Client,
TraceManager: client.TraceManager,
MetricsManager: client.MetricsManager,
Logger: client.Logger,
ComponentName: "a2a_publisher",
}
// Create A2A task with structured message content
task := &a2a.Task{
Id: "task_greeting_" + uuid.New().String(),
ContextId: "conversation_123",
Status: &a2a.TaskStatus{
State: a2a.TaskState_TASK_STATE_SUBMITTED,
Update: &a2a.Message{
MessageId: "msg_" + uuid.New().String(),
Role: a2a.Role_USER,
Content: []*a2a.Part{
{
Part: &a2a.Part_Text{
Text: "Please process greeting task",
},
},
{
Part: &a2a.Part_Data{
Data: &a2a.DataPart{
Data: greetingParams,
Description: "Greeting parameters",
},
},
},
},
},
Timestamp: timestamppb.Now(),
},
}
err := a2aPublisher.PublishA2ATask(ctx, task, &pb.AgentEventMetadata{
FromAgentId: "publisher_id",
ToAgentId: "subscriber_id",
EventType: "task.submitted",
Priority: pb.Priority_PRIORITY_MEDIUM,
})
Provides full observability for A2A task processing:
a2a_subscriber.go)Complete A2A subscriber implementation with:
a2aSubscriber := agenthub.NewA2ATaskSubscriber(client, agentID)
a2aSubscriber.RegisterDefaultA2AHandlers()
// Custom A2A task handlers
a2aSubscriber.RegisterA2ATaskHandler("greeting", func(ctx context.Context, event *pb.AgentEvent) error {
task := event.GetTask()
if task == nil {
return fmt.Errorf("no task in event")
}
// Process A2A task content
requestMessage := task.Status.Update
response := a2aSubscriber.ProcessA2AMessage(ctx, requestMessage)
// Create completion artifact
artifact := &a2a.Artifact{
ArtifactId: "artifact_" + uuid.New().String(),
Name: "Greeting Response",
Description: "Processed greeting task result",
Parts: []*a2a.Part{
{
Part: &a2a.Part_Text{
Text: response,
},
},
},
}
// Complete task with artifact
return a2aSubscriber.CompleteA2ATaskWithArtifact(ctx, task, artifact)
})
go a2aSubscriber.SubscribeToA2ATasks(ctx)
go a2aSubscriber.SubscribeToA2AMessages(ctx)
a2a_broker.go)Complete A2A-compliant AgentHub service implementation that handles:
// A2A broker service with unified abstractions
type A2ABrokerService struct {
// A2A-specific components
MessageRouter *A2AMessageRouter
TaskManager *A2ATaskManager
ContextManager *A2AContextManager
ArtifactManager *A2AArtifactManager
// EDA integration
EventBus *EDAEventBus
SubscriptionMgr *A2ASubscriptionManager
// Observability
TraceManager *TraceManager
MetricsManager *A2AMetricsManager
}
The library uses environment variables for zero-configuration A2A setup:
# Core AgentHub A2A Settings
export AGENTHUB_BROKER_ADDR=localhost
export AGENTHUB_BROKER_PORT=50051
# A2A Protocol Configuration
export AGENTHUB_A2A_PROTOCOL_VERSION=1.0
export AGENTHUB_MESSAGE_BUFFER_SIZE=100
export AGENTHUB_CONTEXT_TIMEOUT=30s
export AGENTHUB_ARTIFACT_MAX_SIZE=10MB
# Observability Endpoints
export JAEGER_ENDPOINT=127.0.0.1:4317
export OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:4317
# A2A Health Check Ports
export AGENTHUB_HEALTH_PORT=8080
export A2A_PUBLISHER_HEALTH_PORT=8081
export A2A_SUBSCRIBER_HEALTH_PORT=8082
/health, /ready, /metrics// A2A task ID generation
taskID := fmt.Sprintf("task_%s_%s", taskDescription, uuid.New().String())
// A2A message ID generation
messageID := fmt.Sprintf("msg_%d_%s", time.Now().Unix(), uuid.New().String())
// A2A context ID for conversation threading
contextID := fmt.Sprintf("ctx_%s_%s", workflowType, uuid.New().String())
Before (Legacy EventBus):
// 50+ lines of observability setup
obs, err := observability.New(ctx, observability.Config{...})
server := grpc.NewServer(grpc.UnaryInterceptor(...))
pb.RegisterEventBusServer(server, &eventBusService{...})
// Manual task message creation
task := &pb.TaskMessage{
TaskId: "task_123",
TaskType: "greeting",
// ... manual field population
}
After (A2A Abstractions):
// One line A2A broker startup
err := agenthub.StartA2ABroker(ctx)
// A2A task creation with abstractions
task := a2aPublisher.CreateA2ATask("greeting", greetingContent, "conversation_123")
err := a2aPublisher.PublishA2ATask(ctx, task, routingMetadata)
Let the library handle configuration automatically:
source .envrc # Load all environment variables
go run broker/main.go
Extend functionality with custom A2A task handlers:
a2aSubscriber.RegisterA2ATaskHandler("my_task", myCustomA2AHandler)
// A2A handler signature with event and context
func myCustomA2AHandler(ctx context.Context, event *pb.AgentEvent) error {
task := event.GetTask()
// Process A2A message content
return a2aSubscriber.CompleteA2ATaskWithArtifact(ctx, task, resultArtifact)
}
The library provides comprehensive observability by default - no additional setup required.
The library provides structured loggers with A2A trace correlation:
// A2A-aware logging with context
client.Logger.InfoContext(ctx, "Processing A2A task",
"task_id", task.GetId(),
"context_id", task.GetContextId(),
"message_count", len(task.GetHistory()),
"current_state", task.GetStatus().GetState().String(),
)
The A2A abstraction library is designed for A2A protocol extension:
// Custom A2A Part type
type CustomPart struct {
CustomData interface{} `json:"custom_data"`
Format string `json:"format"`
}
// Custom A2A artifact processor
type CustomArtifactProcessor struct {
SupportedTypes []string
ProcessFunc func(ctx context.Context, artifact *a2a.Artifact) error
}
// Custom A2A context manager
type CustomContextManager struct {
ContextRules map[string]ContextRule
RouteFunc func(contextId string, message *a2a.Message) []string
}
This A2A-compliant unified approach provides a solid foundation for building complex multi-agent systems with full Agent2Agent protocol support while maintaining simplicity, comprehensive observability, and rich conversation capabilities.