A2A Go SDK

In the default localManager mode, the server manages execution within a single process. A critical feature is that executions are detached from the HTTP request context using context.WithoutCancel(ctx). This ensures that even if a client disconnects, the agent continues its work to completion and the result is safely stored.

Execution Lifecycle

1. Validation: The manager checks for concurrent executions of the same TaskID and verifies concurrency quotas.
2. Pipeline Setup: A runProducerConsumer loop is started using an errgroup.
   • Producer Goroutine: Runs the user's AgentExecutor.Execute method and writes events to the eventpipe.
   • Consumer Goroutine: Reads from the pipe, calls processor.Process to persist the event, and broadcasts it.
3. Termination: The pipeline ends when a terminal event (IsFinal) is processed or if a goroutine panics/errors. If either the producer or consumer hangs, the entire pipeline hangs.
4. Cleanup: The broadcast queue is destroyed, the pipe is closed, and the concurrency quota is released.

The work queue abstraction is used to pass tasks between the Frontend and Backend. There are two primary patterns for implementing or using a queue:

1. Pull Queue Pattern

In a pull queue, the SDK polls a ReadWriter in a loop. It handles retries with exponential backoff, heartbeat attachment, and malformed payload detection.

• Writer interface: Use Write(ctx, *Payload) (TaskID, error) to submit a job. The Payload contains the type ("execute" or "cancel"), TaskID, and the request.
• Queue interface: Use RegisterHandler(HandlerConfig, HandlerFn) to register the backend logic.
• Message Lifecycle: The handler must call msg.Complete(ctx) on success or msg.Return(ctx, error) on failure to re-enqueue the job. The Message interface provides access to the Payload().

2. Push Queue Pattern

Use NewPushQueue(writer) to get a Queue and a HandlerFn. This is intended for environments where an external system (like a load balancer) pushes work directly to the node.

// Pull Queue Message Interface
type Message interface {
    Payload() *Payload
    Complete(ctx context.Context)
    Return(ctx context.Context, err error)
}

To prevent data corruption when multiple instances attempt to update the same task, the taskstore.Store uses version-based Optimistic Concurrency Control (OCC).

How it works

Every store.Update call must include a PrevVersion (the version the caller last observed). If the version currently in the store does not match PrevVersion, the update fails with ErrConcurrentModification.

OCC in Cancellation

When a cancellation request hits an OCC conflict, the system performs a retry loop:

1. Re-fetches the task from the store.
2. If the task is already TaskStateCanceled, the operation succeeds.
3. If the task is in a different terminal state, it returns an error (it is too late to cancel).
4. Otherwise, it retries the update with the new version (up to 10 attempts).

OCC in Execution

If a running agent's processor.Process receives ErrConcurrentModification, it means the task state changed (e.g., was canceled) while the agent was running. The agent will:

1. Re-fetch the task.
2. If the task is terminal, return the result with ExecutionFailureCause set to the OCC error to allow for a graceful shutdown.

A2A Go processes requests through a layered, producer-consumer pipeline. This architecture ensures that agent events are validated and persisted before being delivered to subscribers.

The Pipeline Flow:

1. HTTP Request: Received by the RequestHandler.
2. Manager: Coordinates execution (either localManager for single-process or distributedManager for cluster mode).
3. Factory: Creates matched pairs of Producers (Executors/Cancelers) and Consumers (Processors).
4. Producer-Consumer Pair:
   • The Producer (Agent) generates events and writes them to an eventpipe.
   • The Consumer (Processor) reads from the pipe, validates/persists the event via the TaskUpdate Manager, and writes it to the eventqueue.
5. Event Delivery: The eventqueue fans out processed events to all active Subscriptions.

Key Design Pattern: Producer-Consumer Agents produce events, but they are not sent to users immediately. They must first pass through the consumer, which handles state persistence to the TaskStore. Events only reach subscribers after successful processing.

Understanding how contexts behave is critical for managing long-running tasks and preventing resource leaks.

Detached Execution (Local Mode)

In local mode, Execute and Cancel calls use context.WithoutCancel(ctx) to detach from the incoming HTTP request context.

• Client Disconnection: Does NOT cancel the execution.
• Server Shutdown: Does NOT automatically cancel detached executions unless explicitly managed.

errgroup and runProducerConsumer

When using errgroup.WithContext(ctx) within the producer-consumer pipeline:

• Producer exit: If the producer returns nil, the context is NOT canceled; the consumer continues.
• Consumer exit: If the consumer returns errConsumerStopped, the context IS canceled, signaling the producer to stop.
• Pipe behavior: pipeReader.Read returns ctx.Err() on cancellation. pipeWriter.Write only returns ctx.Err() if the buffer is full; if there is space, the write may succeed even if the context is canceled.

To build with the A2A Go SDK, you must understand these primary abstractions:

• AgentExecutor: The central interface representing the autonomous entity that performs work.
• Task (TaskID): A stateful unit of work or a multi-turn conversation. It maintains its execution state through a History object.
• Context (ContextID): A high-level identifier used to securely link related tasks together, such as a specific user session or a long-running project.
• Message: The fundamental unit of communication between a User and an Agent. Messages are composed of Parts, which can include text, raw bytes, or structured data.
• Artifact: A resource (like a file or a report) that an agent generates incrementally during the execution of a task.
• Event: Streaming primitives (e.g., TaskStatusUpdateEvent, TaskArtifactUpdateEvent) used to synchronize state in real-time over the network.

The --exec mode allows you to run any arbitrary command as an A2A agent. The subprocess does not need to be A2A-aware.

Subprocess Interface:

• stdin: Receives the first text part of the incoming A2A message.
• stdout: Becomes the response content (interpreted based on --chunk).
• stderr: Logged by the CLI at debug level. If the process exits with a non-zero code, stderr content is included in the failure status message.
• Exit code: 0 maps to TaskStateCompleted; non-zero maps to TaskStateFailed.

Output Modes:

1. Default (no --chunk): The entire stdout is collected and emitted as a single text artifact when the process exits.
2. With --chunk=<delimiter>: stdout is read incrementally and split by the delimiter. Each piece is streamed as an artifact chunk event (Append: true) as soon as it's available. This enables streaming without requiring the subprocess to know about the A2A event model.

# Basic execution
a2a serve --exec "python -u a2a_unaware_agent.py"

# Streaming chunks using a newline delimiter
a2a serve --exec "for i in 1 2 3; do echo \$i; sleep 0.5; done" --chunk=$'\n'

# Space-delimited chunks
a2a serve --exec "echo 'alpha beta gamma'" --chunk=' '

# Paragraph-level chunks
a2a serve --exec "cat essay.txt" --chunk='\n\n'