Consuming: Consume External Agent

Consuming external Agents is a core capability of multi-Agent systems. Unlike calling a plain HTTP API, consuming an A2A Agent means dealing with async tasks, streaming responses, retries, timeout control, and other complex semantics. A robust A2A Client must not only send requests—it must also manage the task lifecycle, handle network partitions, and implement circuit breaking and graceful degradation.

This article will walk through how to build a production-grade A2A consumer, from basic calls to advanced patterns.

Creating the Client: The Foundation of Connection Management

Basic Client Configuration

package main

import (
    "context"
    "crypto/tls"
    "fmt"
    "log"
    "net"
    "net/http"
    "time"

    "google.golang.org/adk/a2a/client"
)

func createClient() (*a2aclient.Client, error) {
    ctx := context.Background()

    // Custom HTTP Transport (production-grade configuration)
    transport := &http.Transport{
        // Connection pool settings
        MaxIdleConns:        100,              // max idle connections
        MaxIdleConnsPerHost: 10,               // max idle connections per host
        MaxConnsPerHost:     50,               // max connections per host
        IdleConnTimeout:     90 * time.Second, // idle connection timeout

        // TLS settings
        TLSClientConfig: &tls.Config{
            MinVersion: tls.VersionTLS12,
            // Production: verify the server certificate
            InsecureSkipVerify: false,
        },

        // Connection timeout
        DialContext: (&net.Dialer{
            Timeout:   5 * time.Second,  // dial timeout
            KeepAlive: 30 * time.Second, // TCP keepalive
        }).DialContext,

        // Response header timeout
        ResponseHeaderTimeout: 10 * time.Second,

        // Expect 100-continue timeout
        ExpectContinueTimeout: 1 * time.Second,

        // Force HTTP/2
        ForceAttemptHTTP2: true,
    }

    client, err := a2aclient.New(ctx,
        a2aclient.WithURL("https://python-agent.example.com/a2a"),
        a2aclient.WithHTTPClient(&http.Client{
            Transport: transport,
            Timeout:   30 * time.Second, // total request timeout
        }),
        a2aclient.WithAPIKey("production-api-key"),
        a2aclient.WithRetry(3, time.Second), // retry 3 times on failure, 1s apart
        a2aclient.WithRequestTimeout(30*time.Second),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to create client: %w", err)
    }

    return client, nil
}

Connection Pool Optimization

The A2A Client’s connection pool configuration directly affects performance:

// Transport configuration for high-concurrency scenarios
transport := &http.Transport{
    // Increase pool size under heavy concurrency
    MaxIdleConns:        500,
    MaxIdleConnsPerHost: 100,
    MaxConnsPerHost:     200,

    // Keep-alive duration (A2A benefits from connection reuse)
    IdleConnTimeout:     5 * time.Minute,

    // Disable compression (if the Agent returns lots of text, compression adds CPU cost)
    DisableCompression: false,

    // TCP connection reuse
    DialContext: (&net.Dialer{
        Timeout:   5 * time.Second,
        KeepAlive: 30 * time.Second,
        // Use dual stack (IPv4 + IPv6)
        DualStack: true,
    }).DialContext,
}

Calling External Agents: Task Lifecycle Management

Synchronous Calls (Simple Scenarios)

func callAgentSync(ctx context.Context, client *a2aclient.Client, input string) (string, error) {
    // Create the task
    task, err := client.SendTask(ctx, &a2a.Task{
        Input: map[string]interface{}{
            "skill": "weather-query",
            "query": input,
        },
    })
    if err != nil {
        return "", fmt.Errorf("failed to send task: %w", err)
    }

    log.Printf("task created: %s (status: %s)", task.ID, task.Status)

    // Poll until completion
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    timeout := time.After(30 * time.Second)

    for {
        select {
        case <-ticker.C:
            task, err = client.GetTask(ctx, task.ID)
            if err != nil {
                return "", fmt.Errorf("failed to get task: %w", err)
            }

            log.Printf("task status: %s", task.Status)

            switch task.Status {
            case a2a.TaskStatusCompleted:
                output, _ := task.Output["result"].(string)
                return output, nil
            case a2a.TaskStatusFailed:
                errMsg, _ := task.Output["error"].(string)
                return "", fmt.Errorf("task failed: %s", errMsg)
            case a2a.TaskStatusCancelled:
                return "", fmt.Errorf("task was cancelled")
            }

        case <-timeout:
            // Cancel the task on timeout
            if err := client.CancelTask(ctx, task.ID); err != nil {
                log.Printf("failed to cancel task: %v", err)
            }
            return "", fmt.Errorf("task timeout")

        case <-ctx.Done():
            return "", ctx.Err()
        }
    }
}

Asynchronous Calls (Recommended for Production)

type AsyncResult struct {
    TaskID string
    Result string
    Error  error
}

func callAgentAsync(ctx context.Context, client *a2aclient.Client, input string) (<-chan AsyncResult, error) {
    // Create the task
    task, err := client.SendTask(ctx, &a2a.Task{
        Input: map[string]interface{}{
            "skill": "data-analysis",
            "query": input,
        },
    })
    if err != nil {
        return nil, err
    }

    resultCh := make(chan AsyncResult, 1)

    // Poll in the background
    go func() {
        defer close(resultCh)

        ticker := time.NewTicker(2 * time.Second)
        defer ticker.Stop()

        timeout := time.After(5 * time.Minute)

        for {
            select {
            case <-ticker.C:
                currentTask, err := client.GetTask(ctx, task.ID)
                if err != nil {
                    resultCh <- AsyncResult{TaskID: task.ID, Error: err}
                    return
                }

                switch currentTask.Status {
                case a2a.TaskStatusCompleted:
                    output, _ := currentTask.Output["result"].(string)
                    resultCh <- AsyncResult{TaskID: task.ID, Result: output}
                    return
                case a2a.TaskStatusFailed:
                    errMsg, _ := currentTask.Output["error"].(string)
                    resultCh <- AsyncResult{
                        TaskID: task.ID,
                        Error:  fmt.Errorf("task failed: %s", errMsg),
                    }
                    return
                case a2a.TaskStatusCancelled:
                    resultCh <- AsyncResult{
                        TaskID: task.ID,
                        Error:  fmt.Errorf("task cancelled"),
                    }
                    return
                }

            case <-timeout:
                client.CancelTask(ctx, task.ID)
                resultCh <- AsyncResult{
                    TaskID: task.ID,
                    Error:  fmt.Errorf("task timeout after 5m"),
                }
                return

            case <-ctx.Done():
                client.CancelTask(ctx, task.ID)
                resultCh <- AsyncResult{TaskID: task.ID, Error: ctx.Err()}
                return
            }
        }
    }()

    return resultCh, nil
}

// Usage example
func main() {
    client, _ := createClient()

    resultCh, err := callAgentAsync(ctx, client, "Analyze Q3 sales data")
    if err != nil {
        log.Fatal(err)
    }

    // Continue doing other things...

    // Wait for the result
    result := <-resultCh
    if result.Error != nil {
        log.Printf("task %s failed: %v", result.TaskID, result.Error)
        return
    }

    log.Printf("task %s completed: %s", result.TaskID, result.Result)
}

Streaming Calls (Real-Time Responses)

func callAgentStreaming(ctx context.Context, client *a2aclient.Client, input string) error {
    // Create a streaming task
    stream, err := client.SendTaskStreaming(ctx, &a2a.Task{
        Input: map[string]interface{}{
            "skill": "code-generation",
            "query": input,
        },
    })
    if err != nil {
        return fmt.Errorf("failed to start streaming: %w", err)
    }
    defer stream.Close()

    // Receive streaming updates in real time
    for {
        select {
        case update, ok := <-stream.Updates():
            if !ok {
                log.Println("stream closed")
                return nil
            }

            switch update.Type {
            case a2a.StreamUpdateTypeStatus:
                log.Printf("status: %s", update.Status)

            case a2a.StreamUpdateTypeOutput:
                // Real-time output (e.g., code snippets)
                chunk, _ := update.Data["chunk"].(string)
                fmt.Print(chunk) // print in real time

            case a2a.StreamUpdateTypeArtifact:
                // Complete artifact (e.g., a generated file)
                artifact := update.Artifact
                log.Printf("artifact received: %s (%d bytes)", artifact.Name, len(artifact.Data))

            case a2a.StreamUpdateTypeError:
                log.Printf("stream error: %s", update.Error)
                return fmt.Errorf("stream error: %s", update.Error)
            }

        case <-ctx.Done():
            return ctx.Err()
        }
    }
}

Wrapping as a Tool: Seamless Integration into Local Agents

Wrapping an external A2A Agent as a local Tool lets an orchestrator Agent call remote Agents as if they were local functions:

package tools

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "google.golang.org/adk/a2a/client"
    "google.golang.org/adk/tool"
)

// ExternalAgentTool wraps a remote A2A Agent as a local Tool
type ExternalAgentTool struct {
    name        string
    description string
    client      *a2aclient.Client
    skill       string
    timeout     time.Duration
    maxRetries  int
}

// NewExternalAgentTool creates an external Agent Tool
func NewExternalAgentTool(config ExternalAgentConfig) (*ExternalAgentTool, error) {
    ctx := context.Background()

    client, err := a2aclient.New(ctx,
        a2aclient.WithURL(config.URL),
        a2aclient.WithAPIKey(config.APIKey),
        a2aclient.WithTimeout(config.Timeout),
        a2aclient.WithRetry(config.MaxRetries, time.Second),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to create client: %w", err)
    }

    return &ExternalAgentTool{
        name:        config.Name,
        description: config.Description,
        client:      client,
        skill:       config.Skill,
        timeout:     config.Timeout,
        maxRetries:  config.MaxRetries,
    }, nil
}

type ExternalAgentConfig struct {
    Name        string
    Description string
    URL         string
    APIKey      string
    Skill       string
    Timeout     time.Duration
    MaxRetries  int
}

func (t *ExternalAgentTool) Name() string {
    return t.name
}

func (t *ExternalAgentTool) Description() string {
    return t.description
}

func (t *ExternalAgentTool) Schema() tool.Schema {
    return tool.Schema{
        Type: "object",
        Properties: map[string]tool.Property{
            "query": {
                Type:        "string",
                Description: "The query to send to the external Agent",
            },
        },
        Required: []string{"query"},
    }
}

func (t *ExternalAgentTool) Call(ctx context.Context, input string) (string, error) {
    // Parse the input
    var params struct {
        Query string `json:"query"`
    }
    if err := json.Unmarshal([]byte(input), &params); err != nil {
        return "", fmt.Errorf("invalid input: %w", err)
    }

    // Create a context with a timeout
    callCtx, cancel := context.WithTimeout(ctx, t.timeout)
    defer cancel()

    // Call the external Agent
    task, err := t.client.SendTask(callCtx, &a2a.Task{
        Input: map[string]interface{}{
            "skill": t.skill,
            "query": params.Query,
        },
    })
    if err != nil {
        return "", fmt.Errorf("failed to send task: %w", err)
    }

    // Poll for the result
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            currentTask, err := t.client.GetTask(callCtx, task.ID)
            if err != nil {
                return "", fmt.Errorf("failed to get task: %w", err)
            }

            switch currentTask.Status {
            case a2a.TaskStatusCompleted:
                result, _ := currentTask.Output["result"].(string)
                return result, nil
            case a2a.TaskStatusFailed:
                errMsg, _ := currentTask.Output["error"].(string)
                return "", fmt.Errorf("external agent failed: %s", errMsg)
            case a2a.TaskStatusCancelled:
                return "", fmt.Errorf("task was cancelled")
            }

        case <-callCtx.Done():
            // Cancel the task on timeout
            t.client.CancelTask(ctx, task.ID)
            return "", fmt.Errorf("external agent call timeout")
        }
    }
}

// Usage example
func setupOrchestratorAgent() (*llmagent.Agent, error) {
    ctx := context.Background()

    // Create external Agent Tools
    weatherTool, err := NewExternalAgentTool(ExternalAgentConfig{
        Name:        "weather_query",
        Description: "Query weather information for a specified city",
        URL:         "https://weather-agent.example.com/a2a",
        APIKey:      "weath...ey",
        Skill:       "current-weather",
        Timeout:     10 * time.Second,
        MaxRetries:  2,
    })
    if err != nil {
        return nil, err
    }

    dataTool, err := NewExternalAgentTool(ExternalAgentConfig{
        Name:        "data_analysis",
        Description: "Analyze datasets and return statistical summaries",
        URL:         "https://data-agent.example.com/a2a",
        APIKey:      ***
        Skill:       "csv-analysis",
        Timeout:     60 * time.Second,
        MaxRetries:  3,
    })
    if err != nil {
        return nil, err
    }

    // Create the orchestrator Agent
    agent, err := llmagent.New(llmagent.Config{
        Name:        "orchestrator",
        Model:       model,
        Instruction: `You are a task dispatch expert. Based on user needs, call the appropriate tools to complete the task.
Available tools:
- weather_query: query weather
- data_analysis: data analysis

If a user request involves multiple steps, call the tools in sequence.`,
        Tools: []tool.Tool{weatherTool, dataTool},
    })
    if err != nil {
        return nil, err
    }

    return agent, nil
}

Advanced Patterns: Agent Orchestration

Calling Multiple Agents in Parallel

func parallelAgentCalls(ctx context.Context, inputs map[string]string) (map[string]string, error) {
    type agentCall struct {
        name   string
        client *a2aclient.Client
        input  string
    }

    calls := []agentCall{
        {"weather", weatherClient, inputs["weather"]},
        {"news", newsClient, inputs["news"]},
        {"stock", stockClient, inputs["stock"]},
    }

    results := make(map[string]string)
    errCh := make(chan error, len(calls))
    resultCh := make(chan struct {
        name   string
        result string
        err    error
    }, len(calls))

    // Call in parallel
    for _, call := range calls {
        go func(c agentCall) {
            task, err := c.client.SendTask(ctx, &a2a.Task{
                Input: map[string]interface{}{
                    "query": c.input,
                },
            })
            if err != nil {
                resultCh <- struct {
                    name   string
                    result string
                    err    error
                }{c.name, "", err}
                return
            }

            // Wait for completion (simplified; real code should poll)
            result, err := waitForTask(ctx, c.client, task.ID)
            resultCh <- struct {
                name   string
                result string
                err    error
            }{c.name, result, err}
        }(call)
    }

    // Collect results
    var errs []error
    for i := 0; i < len(calls); i++ {
        r := <-resultCh
        if r.err != nil {
            errs = append(errs, fmt.Errorf("%s: %w", r.name, r.err))
            continue
        }
        results[r.name] = r.result
    }

    if len(errs) > 0 {
        return results, fmt.Errorf("partial failures: %v", errs)
    }

    return results, nil
}

Pipeline Calls (Chained Processing)

func pipelineAgentCalls(ctx context.Context, initialInput string) (string, error) {
    // Step 1: data fetching
    step1Result, err := callAgent(ctx, dataClient, "fetch-data", initialInput)
    if err != nil {
        return "", fmt.Errorf("step 1 failed: %w", err)
    }

    // Step 2: data processing
    step2Result, err := callAgent(ctx, processClient, "process-data", step1Result)
    if err != nil {
        return "", fmt.Errorf("step 2 failed: %w", err)
    }

    // Step 3: result formatting
    step3Result, err := callAgent(ctx, formatClient, "format-result", step2Result)
    if err != nil {
        return "", fmt.Errorf("step 3 failed: %w", err)
    }

    return step3Result, nil
}

Error Handling and Retry Strategies

Smart Retry

type RetryPolicy struct {
    MaxRetries  int
    BaseDelay   time.Duration
    MaxDelay    time.Duration
    Multiplier  float64
    RetryableErrors []string // retryable error types
}

func (p *RetryPolicy) Execute(ctx context.Context, operation func() error) error {
    var lastErr error

    for attempt := 0; attempt <= p.MaxRetries; attempt++ {
        if attempt > 0 {
            // Compute the backoff delay
            delay := p.BaseDelay * time.Duration(math.Pow(p.Multiplier, float64(attempt-1)))
            if delay > p.MaxDelay {
                delay = p.MaxDelay
            }

            // Add jitter to prevent thundering herd
            jitter := time.Duration(rand.Int63n(int64(delay) / 2))
            delay = delay + jitter

            log.Printf("retry attempt %d/%d after %v", attempt, p.MaxRetries, delay)

            select {
            case <-time.After(delay):
            case <-ctx.Done():
                return ctx.Err()
            }
        }

        err := operation()
        if err == nil {
            return nil
        }

        lastErr = err

        // Check whether it's retryable
        if !p.isRetryable(err) {
            return err
        }

        log.Printf("attempt %d failed: %v", attempt+1, err)
    }

    return fmt.Errorf("max retries exceeded: %w", lastErr)
}

func (p *RetryPolicy) isRetryable(err error) bool {
    errStr := err.Error()
    for _, retryable := range p.RetryableErrors {
        if strings.Contains(errStr, retryable) {
            return true
        }
    }
    return false
}

// Usage example
policy := RetryPolicy{
    MaxRetries:  3,
    BaseDelay:   time.Second,
    MaxDelay:    30 * time.Second,
    Multiplier:  2.0,
    RetryableErrors: []string{
        "connection refused",
        "timeout",
        "temporary",
        "rate limit",
        "service unavailable",
    },
}

err := policy.Execute(ctx, func() error {
    _, err := client.SendTask(ctx, task)
    return err
})

Deep Dive: Common Questions

Q: What if an external Agent becomes unresponsive?

Root causes:

Network partition or latency
The external Agent is overloaded
The external Agent crashed
The request itself triggers an infinite loop or long-running execution

Solutions:

// 1. Configure reasonable timeouts
client, _ := a2aclient.New(ctx,
    a2aclient.WithTimeout(30*time.Second),        // per-request timeout
    a2aclient.WithRequestTimeout(5*time.Minute),  // total task timeout
)

// 2. Implement circuit breaking
cbClient := NewCircuitBreakerClient(client)

// 3. Graceful degradation
func callWithFallback(ctx context.Context, client *a2aclient.Client, input string) (string, error) {
    result, err := client.Call(ctx, input)
    if err != nil {
        log.Printf("primary agent failed: %v", err)

        // Try a backup Agent
        result, err = fallbackClient.Call(ctx, input)
        if err != nil {
            // Return a cached result
            return getCachedResult(input), nil
        }
    }
    return result, nil
}

// 4. Async processing + callback
func callAsyncWithCallback(ctx context.Context, client *a2aclient.Client, input string, callbackURL string) error {
    _, err := client.SendTask(ctx, &a2a.Task{
        Input: map[string]interface{}{
            "query":        input,
            "callback_url": callbackURL,
        },
    })
    return err
}

Q: How do you manage multiple external Agents?

Agent Registry pattern:

type AgentRegistry struct {
    agents map[string]*AgentConnection
    mu     sync.RWMutex
}

type AgentConnection struct {
    Name        string
    Client      *a2aclient.Client
    Health      HealthStatus
    LastUsed    time.Time
    CallCount   int64
    ErrorCount  int64
    AvgLatency  time.Duration
}

func (r *AgentRegistry) GetHealthyAgent(skill string) (*AgentConnection, error) {
    r.mu.RLock()
    defer r.mu.RUnlock()

    var candidates []*AgentConnection
    for _, agent := range r.agents {
        if agent.HasSkill(skill) && agent.Health == HealthHealthy {
            candidates = append(candidates, agent)
        }
    }

    if len(candidates) == 0 {
        return nil, fmt.Errorf("no healthy agent found for skill: %s", skill)
    }

    // Pick the one with the lowest latency
    sort.Slice(candidates, func(i, j int) bool {
        return candidates[i].AvgLatency < candidates[j].AvgLatency
    })

    return candidates[0], nil
}

func (r *AgentRegistry) HealthCheck(ctx context.Context) {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            r.mu.RLock()
            agents := make([]*AgentConnection, 0, len(r.agents))
            for _, agent := range r.agents {
                agents = append(agents, agent)
            }
            r.mu.RUnlock()

            for _, agent := range agents {
                go func(a *AgentConnection) {
                    start := time.Now()
                    _, err := a.Client.GetAgentCard(ctx)
                    latency := time.Since(start)

                    r.mu.Lock()
                    defer r.mu.Unlock()

                    a.LastUsed = time.Now()
                    a.AvgLatency = (a.AvgLatency + latency) / 2

                    if err != nil {
                        a.ErrorCount++
                        if a.ErrorCount > 5 {
                            a.Health = HealthUnhealthy
                        }
                    } else {
                        a.ErrorCount = 0
                        a.Health = HealthHealthy
                    }
                }(agent)
            }
        case <-ctx.Done():
            return
        }
    }
}

Next Steps

Consuming is done—next up is cross-language collaboration: a hands-on Python + Go case study.

← Exposing | Cross-Language Collaboration →

Follow “Mengshou Programming” on WeChat for more hands-on Go ADK tutorials—weekly updates on practical Go / AI programming content.

Frequently Asked Questions

How is consuming an A2A Agent different from a regular HTTP API?

It requires handling async tasks, streaming responses, retries, timeouts, and task lifecycle management.

How do you configure a production-grade A2A Client?

Customize http.Transport with connection pooling, TLS, timeouts, retries, and API key settings.

What options does a2aclient.New support?

WithURL, WithHTTPClient, WithAPIKey, WithRetry, and WithRequestTimeout.

What is the basic flow for synchronous calls?

Build a task, send it with client.SendTask, and wait for the final result.

How do you optimize the connection pool for high concurrency?

Increase MaxIdleConns, MaxIdleConnsPerHost, MaxConnsPerHost, and extend IdleConnTimeout.

Creating the Client: The Foundation of Connection Management#

Basic Client Configuration#

Connection Pool Optimization#

Calling External Agents: Task Lifecycle Management#

Synchronous Calls (Simple Scenarios)#

Asynchronous Calls (Recommended for Production)#

Streaming Calls (Real-Time Responses)#

Wrapping as a Tool: Seamless Integration into Local Agents#

Advanced Patterns: Agent Orchestration#

Calling Multiple Agents in Parallel#

Pipeline Calls (Chained Processing)#

Error Handling and Retry Strategies#

Smart Retry#

Deep Dive: Common Questions#

Q: What if an external Agent becomes unresponsive?#

Q: How do you manage multiple external Agents?#

Next Steps#