a2a

package
v0.12.2 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 19, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package a2a implements the Agent-to-Agent (A2A) protocol for cross-framework agent communication. It provides both server and client implementations.

Index

Constants

This section is empty.

Variables

View Source
var ErrTaskNotFound = errors.New("a2a: task not found")

ErrTaskNotFound is returned by a TaskStore when a task id is unknown or is not visible to the caller's tenant. The server maps it to HTTP 404.

Functions

func NewRemoteAgentTool added in v0.10.0

func NewRemoteAgentTool(name, description string, client *Client, opts ...RemoteToolOption) *tool.Definition

NewRemoteAgentTool adapts a remote A2A agent into a Chronos tool so an agent can delegate a sub-task to it — the same way it would spawn a local subagent (composes with the harness spawn/subagent model). The tool submits the task, waits for the remote agent to finish, and returns only its final output, so the remote agent's intermediate work never enters the caller's context.

name/description are what the calling model sees; client points at the remote A2A endpoint.

Types

type AgentCard

type AgentCard struct {
	Name         string   `json:"name"`
	Description  string   `json:"description"`
	Version      string   `json:"version"`
	Capabilities []string `json:"capabilities"`
	InputSchema  any      `json:"input_schema,omitempty"`
	OutputSchema any      `json:"output_schema,omitempty"`
}

AgentCard describes an A2A agent's capabilities.

func CardFromSkills added in v0.10.0

func CardFromSkills(name, description, version string, reg *skill.Registry) AgentCard

CardFromSkills builds an AgentCard whose Capabilities are the names of the skills installed in reg, so an A2A peer discovers exactly what the agent can do. A nil registry yields a card with no capabilities.

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client connects to an external A2A agent.

func NewClient

func NewClient(baseURL string) *Client

NewClient creates an A2A client for connecting to an external agent.

func (*Client) CreateTask

func (c *Client) CreateTask(ctx context.Context, input string, metadata map[string]any) (*Task, error)

CreateTask submits a task to the remote agent.

func (*Client) GetAgentCard

func (c *Client) GetAgentCard(ctx context.Context) (*AgentCard, error)

GetAgentCard retrieves the agent's capability card.

func (*Client) GetTask

func (c *Client) GetTask(ctx context.Context, taskID string) (*Task, error)

GetTask polls the status of a task.

func (*Client) StreamTask added in v0.10.0

func (c *Client) StreamTask(ctx context.Context, taskID string) (snapshots <-chan Task, errc <-chan error)

StreamTask consumes the server's Server-Sent Events for a task, delivering a snapshot on each state transition until the task reaches a terminal state, the server closes the stream, or ctx is done. Exactly one of the two returned channels produces a value before both are closed: the task channel yields snapshots (the last being terminal), the error channel yields a single error.

func (*Client) WaitForCompletion

func (c *Client) WaitForCompletion(ctx context.Context, taskID string, pollInterval time.Duration) (*Task, error)

WaitForCompletion polls a task until it reaches a terminal state.

type DurableStore added in v0.10.0

type DurableStore struct {
	// contains filtered or unexported fields
}

DurableStore is a TaskStore backed by the durable queue (engine/queue) for restart-resumable scheduling and by checkpoint storage (storage.Storage) for the task record. Execution is driven by a queue.Worker wired to Executor, so a task whose worker dies is re-leased and resumed once its lease expires (orphan recovery), and tasks survive a process restart. The task record is a single checkpoint keyed by the task id (overwritten on each transition), which inherits storage's per-tenant scoping: a Get for another tenant's task resolves to ErrTaskNotFound.

func NewDurableStore added in v0.10.0

func NewDurableStore(q *queue.Queue, store storage.Storage, handler Handler) *DurableStore

NewDurableStore builds a queue-backed TaskStore. Wire Executor into a queue.Worker (and run a queue.Reaper for orphan recovery) to execute tasks:

ds := a2a.NewDurableStore(q, store, handler)
w, _ := queue.NewWorker(q, ds.Executor, queue.WorkerConfig{ID: "a2a-worker-1"})
go w.Run(ctx)
srv := a2a.NewServerWithStore(card, ds)

func (*DurableStore) Cancel added in v0.10.0

func (d *DurableStore) Cancel(ctx context.Context, id string) (*Task, error)

Cancel marks a pending or running task canceled. The running worker observes the cancellation via a checkpoint reload and stops recording further state.

func (*DurableStore) Executor added in v0.10.0

func (d *DurableStore) Executor(ctx context.Context, run *queue.Run) queue.Result

Executor performs the work for a claimed run. Wire it into a queue.Worker. It re-establishes the task's tenant, runs the handler, and records the terminal state. Returning a non-nil Result.Err lets the queue retry with backoff up to the run's MaxAttempts; the task is marked failed only once the budget is spent.

func (*DurableStore) Get added in v0.10.0

func (d *DurableStore) Get(ctx context.Context, id string) (*Task, error)

Get returns the task record for the context tenant, or ErrTaskNotFound. A genuine miss (including a cross-tenant miss, which the tenant-scoped query renders as no rows) maps to ErrTaskNotFound; a transient read failure (DB down, context canceled) is surfaced wrapped so callers do not mistake an outage for a missing task and, critically, so Executor does not reconstruct-and-rerun a task whose record merely failed to load.

func (*DurableStore) Submit added in v0.10.0

func (d *DurableStore) Submit(ctx context.Context, input string, metadata map[string]any) (*Task, error)

Submit persists a pending task record and enqueues a run for a worker to pick up. The task is scoped to the context tenant.

type Handler

type Handler func(ctx context.Context, task *Task) error

Handler processes A2A tasks. It should be idempotent: on the durable backend a task may be executed more than once (queue retries and orphan recovery are at-least-once), and re-running simply overwrites the task's output.

type RemoteToolOption added in v0.10.0

type RemoteToolOption func(*remoteToolConfig)

RemoteToolOption configures a remote-agent tool.

func WithPermission added in v0.10.0

func WithPermission(p tool.Permission) RemoteToolOption

WithPermission overrides the tool's permission (default PermAllow). Use PermRequireApproval to gate outbound delegation to an external agent behind human approval.

func WithPollInterval added in v0.10.0

func WithPollInterval(d time.Duration) RemoteToolOption

WithPollInterval sets how often the tool polls the remote task for completion.

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server exposes an agent as an A2A endpoint. Task lifecycle is delegated to a TaskStore, so the same HTTP surface serves both the in-memory default and the durable, restart-resumable queue backend (see NewDurableStore).

func NewServer

func NewServer(card AgentCard, handler Handler) *Server

NewServer creates an A2A server backed by the default in-memory store, which runs handler in a goroutine per task. Tasks do not survive a restart; use NewServerWithStore with a durable store for resumable tasks.

func NewServerWithStore added in v0.10.0

func NewServerWithStore(card AgentCard, store TaskStore) *Server

NewServerWithStore creates an A2A server backed by an explicit TaskStore (e.g. NewDurableStore for queue-backed, restart-resumable tasks).

func (*Server) ServeHTTP

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP handles A2A protocol requests.

type Task

type Task struct {
	ID        string         `json:"id"`
	Status    TaskStatus     `json:"status"`
	Input     string         `json:"input"`
	Output    string         `json:"output,omitempty"`
	Error     string         `json:"error,omitempty"`
	Metadata  map[string]any `json:"metadata,omitempty"`
	CreatedAt time.Time      `json:"created_at"`
	UpdatedAt time.Time      `json:"updated_at"`
}

Task represents an A2A task.

type TaskStatus

type TaskStatus string

TaskStatus represents the status of an A2A task.

const (
	TaskStatusPending   TaskStatus = "pending"
	TaskStatusRunning   TaskStatus = "running"
	TaskStatusCompleted TaskStatus = "completed"
	TaskStatusFailed    TaskStatus = "failed"
	TaskStatusCancelled TaskStatus = "cancelled" //nolint:misspell // wire value; external clients may expect British spelling
)

type TaskStore added in v0.10.0

type TaskStore interface {
	// Submit persists a new task and begins (or schedules) its execution,
	// returning a snapshot of the created task.
	Submit(ctx context.Context, input string, metadata map[string]any) (*Task, error)
	// Get returns a snapshot of the task, or ErrTaskNotFound.
	Get(ctx context.Context, id string) (*Task, error)
	// Cancel transitions a pending or running task to the canceled state and
	// returns a snapshot. Terminal tasks are returned unchanged.
	//
	// Backends differ in whether Cancel preempts in-flight work: the in-memory
	// store aborts the handler via context cancellation; the durable store records
	// the cancellation and discards the handler's result when it finishes, but does
	// not interrupt a running handler.
	Cancel(ctx context.Context, id string) (*Task, error)
}

TaskStore persists A2A tasks and drives their execution. The Server delegates every task-lifecycle operation to a TaskStore so the backend — the in-memory default or the durable, restart-resumable queue (see NewDurableStore) — is swappable without touching the HTTP layer.

Implementations must be safe for concurrent use and honor the context tenant (see storage.WithTenant): a Get/Cancel for a task created under another tenant must return ErrTaskNotFound.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL