a2a

package
v0.50.105 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	PolicySigningSecretEnv              = controlplane.PolicySigningSecretEnv
	AllowUnsignedControlPlaneEnv        = controlplane.AllowUnsignedControlPlaneEnv
	CapabilityServerModelV1             = "server_model_v1"
	CapabilityPipelinePhasesV1          = "pipeline_phases_v1"
	CapabilityPipelineInstructionsV1    = "pipeline_instructions_v1"
	CapabilityPipelinePromptTemplatesV1 = "pipeline_prompt_templates_v1"
	CapabilityIterationBudgetV1         = "iteration_budget_v1"
	CapabilitySignedPolicyV1            = "signed_policy_v1"
	CapabilitySignedControlPlaneV1      = "signed_control_plane_v1"
)
View Source 
const (
	MethodSendMessage      = "tasks/send"
	MethodSendSubscribe    = "tasks/sendSubscribe"
	MethodGetTask          = "tasks/get"
	MethodCancelTask       = "tasks/cancel"
	MethodRegisterCard     = "agent/register"
	MethodHeartbeat        = "agent/heartbeat"
	MethodStatusUpdate     = "tasks/statusUpdate"
	MethodApproval         = "tasks/approval"
	MethodApprovalResponse = "tasks/approvalResponse"
)

A2A JSON-RPC method constants.

Variables

This section is empty.

Functions

func DefaultCapabilities 

func DefaultCapabilities() []string

DefaultCapabilities returns the worker control-plane capabilities advertised at registration. @AX:ANCHOR [AUTO] registration capability contract; keep ordering and exported values stable for agent-card generation and control-plane verification. @AX:REASON: Used by server registration, agent-card construction, and control-plane tests to assert the worker's advertised feature set.

func SignedControlPlaneEnforced 

func SignedControlPlaneEnforced() bool

SignedControlPlaneEnforced returns true when the worker is running in a mode where server-issued control-plane metadata must be trusted over local fallback.

func VerifyCachedPolicyFile 

func VerifyCachedPolicyFile(policyPath string, policy any) error

VerifyCachedPolicyFile verifies the sidecar signature for a cached policy file when signature validation is enabled via AUTOPUS_A2A_POLICY_SIGNING_SECRET.

Types

type AgentCard 

type AgentCard struct {
	Name                      string   `json:"name"`
	Description               string   `json:"description"`
	URL                       string   `json:"url"`
	WorkspaceID               string   `json:"workspace_id,omitempty"`
	Skills                    []string `json:"skills"`
	Providers                 []string `json:"providers,omitempty"`
	ExecutionLanes            []string `json:"execution_lanes,omitempty"`
	Capabilities              []string `json:"capabilities,omitempty"`
	UnsupportedModelOverrides []string `json:"unsupported_model_overrides,omitempty"`
	SupportedInputModes       []string `json:"supported_input_modes"`
}

AgentCard describes the worker's capabilities for registration.

type ApprovalRequestParams 

type ApprovalRequestParams struct {
	TaskID     string `json:"task_id"`
	ApprovalID string `json:"approval_id,omitempty"`
	TraceID    string `json:"trace_id,omitempty"`
	Action     string `json:"action"`
	RiskLevel  string `json:"risk_level"`
	Context    string `json:"context"`
}

ApprovalRequestParams holds approval request payload from the backend.

type ApprovalResponseParams 

type ApprovalResponseParams struct {
	TaskID     string `json:"task_id"`
	ApprovalID string `json:"approval_id,omitempty"`
	TraceID    string `json:"trace_id,omitempty"`
	Decision   string `json:"decision"` // "approve", "deny", "skip"
}

ApprovalResponseParams holds the user's approval decision.

type Artifact 

type Artifact struct {
	Name     string `json:"name"`
	MimeType string `json:"mime_type,omitempty"`
	Data     string `json:"data"`
}

Artifact holds a single result artifact from task execution.

type CardBuilder 

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

CardBuilder constructs an AgentCard from worker configuration.

func NewCardBuilder 

func NewCardBuilder(name, backendURL string) *CardBuilder

NewCardBuilder creates a CardBuilder with the given worker name and backend URL.

func (*CardBuilder) Build 

func (b *CardBuilder) Build() AgentCard

Build assembles an AgentCard with deduplicated skills from all providers.

func (*CardBuilder) WithExecutionLanes 

func (b *CardBuilder) WithExecutionLanes(lanes []string) *CardBuilder

WithExecutionLanes sets the explicitly advertised execution lanes.

func (*CardBuilder) WithProviders 

func (b *CardBuilder) WithProviders(providers []string) *CardBuilder

WithProviders sets the provider list for skill resolution.

func (*CardBuilder) WithVersion 

func (b *CardBuilder) WithVersion(version string) *CardBuilder

WithVersion sets the worker version string.

type Client 

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

Client wraps Transport with automatic reconnection and state recovery.

func NewClient 

func NewClient(config ClientConfig) *Client

NewClient creates a new A2A WebSocket client.

func (*Client) Close 

func (c *Client) Close() error

Close gracefully shuts down the client and its transport.

func (*Client) Connect 

func (c *Client) Connect(ctx context.Context) error

Connect establishes the initial WebSocket connection, registers the agent card, and transitions to the connected state.

func (*Client) HandleConnectionLoss 

func (c *Client) HandleConnectionLoss(ctx context.Context, cause error) error

HandleConnectionLoss should be called when a receive or send error indicates the connection is broken. It triggers automatic reconnection with state recovery. Returns nil on successful reconnect.

func (*Client) OnConnected 

func (c *Client) OnConnected(fn func())

OnConnected registers a callback fired on initial connect and each reconnect.

func (*Client) OnDisconnected 

func (c *Client) OnDisconnected(fn func(error))

OnDisconnected registers a callback fired when the connection is lost.

func (*Client) OnReconnectFailed 

func (c *Client) OnReconnectFailed(fn func(error))

OnReconnectFailed registers a callback fired when all retry attempts are exhausted.

func (*Client) State 

func (c *Client) State() ConnectionState

State returns the current connection state.

func (*Client) Transport 

func (c *Client) Transport() *Transport

Transport returns the underlying transport for direct message I/O.

type ClientConfig 

type ClientConfig struct {
	Transport      TransportConfig
	AgentCard      AgentCard
	StateRecoverer StateRecoverer
}

ClientConfig holds configuration for the A2A WebSocket client.

type ConnectionState 

type ConnectionState string

ConnectionState represents the client's connection lifecycle. 

const (
	StateConnected    ConnectionState = "connected"
	StateConnecting   ConnectionState = "connecting"
	StateDisconnected ConnectionState = "disconnected"
)

type DispatchIssue 

type DispatchIssue struct {
	TaskID  string
	Stage   string
	Message string
}

DispatchIssue captures a transport failure during platform-facing task reconciliation.

type Heartbeat 

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

Heartbeat sends periodic heartbeat messages over the A2A WebSocket and detects connection loss via timeout.

func NewHeartbeat 

func NewHeartbeat(sendFn func() error, onTimeout func()) *Heartbeat

NewHeartbeat creates a Heartbeat that calls sendFn every 30s. If no Ack is received within 60s, onTimeout is called.

func NewHeartbeatWithJSONRPC 

func NewHeartbeatWithJSONRPC(sendFn func(msg []byte) error, onTimeout func()) *Heartbeat

@AX:ANCHOR [AUTO] protocol entry point — sole constructor for JSON-RPC heartbeat; server.go wires this at startup — fan_in: 2 NewHeartbeatWithJSONRPC creates a Heartbeat that sends agent/heartbeat JSON-RPC 2.0 messages instead of raw WebSocket ping frames (S1 requirement).

func (*Heartbeat) Ack 

func (h *Heartbeat) Ack()

Ack records that a heartbeat response was received.

func (*Heartbeat) HandleHeartbeatResponse 

func (h *Heartbeat) HandleHeartbeatResponse(result map[string]string)

HandleHeartbeatResponse records an acknowledgement from a backend heartbeat response. Called when the backend sends {"status":"ok"} in reply to an agent/heartbeat request.

func (*Heartbeat) Start 

func (h *Heartbeat) Start(ctx context.Context)

Start runs the heartbeat loop in a goroutine until ctx is cancelled.

type IterationBudget 

type IterationBudget struct {
	Limit           int     `json:"limit"`
	WarnThreshold   float64 `json:"warn_threshold,omitempty"`
	DangerThreshold float64 `json:"danger_threshold,omitempty"`
}

IterationBudget defines a server-issued tool-call budget for the task.

type JSONRPCError 

type JSONRPCError struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
	Data    any    `json:"data,omitempty"`
}

JSONRPCError represents a JSON-RPC error object.

type JSONRPCNotification 

type JSONRPCNotification struct {
	JSONRPC string `json:"jsonrpc"`
	Method  string `json:"method"`
	Params  any    `json:"params,omitempty"`
}

JSONRPCNotification is a JSON-RPC 2.0 notification (no id).

type JSONRPCRequest 

type JSONRPCRequest struct {
	JSONRPC string          `json:"jsonrpc"`
	ID      json.RawMessage `json:"id,omitempty"`
	Method  string          `json:"method"`
	Params  json.RawMessage `json:"params,omitempty"`
}

JSONRPCRequest is the inbound JSON-RPC 2.0 request envelope.

type JSONRPCResponse 

type JSONRPCResponse struct {
	JSONRPC string          `json:"jsonrpc"`
	ID      json.RawMessage `json:"id,omitempty"`
	Result  any             `json:"result,omitempty"`
	Error   *JSONRPCError   `json:"error,omitempty"`
}

JSONRPCResponse is the outbound JSON-RPC 2.0 response envelope.

type LifecycleManager 

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

LifecycleManager tracks lifecycles for all active A2A tasks.

func NewLifecycleManager 

func NewLifecycleManager() *LifecycleManager

NewLifecycleManager creates a new lifecycle manager.

func (*LifecycleManager) ActiveTasks 

func (m *LifecycleManager) ActiveTasks() []*Task

ActiveTasks returns all tasks that are not in a terminal state.

func (*LifecycleManager) AddListener 

func (m *LifecycleManager) AddListener(fn TransitionListener)

AddListener registers a listener that will be added to all currently tracked and future task lifecycles.

func (*LifecycleManager) Get 

func (m *LifecycleManager) Get(taskID string) (*TaskLifecycle, bool)

Get retrieves the lifecycle for the given task ID.

func (*LifecycleManager) Remove 

func (m *LifecycleManager) Remove(taskID string)

Remove stops tracking a task lifecycle.

func (*LifecycleManager) Track 

func (m *LifecycleManager) Track(task *Task) *TaskLifecycle

Track begins tracking the lifecycle of a task. Manager-level listeners are automatically added to the new lifecycle.

type PollResult 

type PollResult struct {
	ID                       string            `json:"id"`
	Type                     string            `json:"type"`
	Model                    string            `json:"model,omitempty"`
	PipelinePhases           []string          `json:"pipeline_phases,omitempty"`
	PipelineInstructions     map[string]string `json:"pipeline_instructions,omitempty"`
	PipelinePromptTemplates  map[string]string `json:"pipeline_prompt_templates,omitempty"`
	IterationBudget          *IterationBudget  `json:"iteration_budget,omitempty"`
	ControlPlaneCapabilities []string          `json:"control_plane_capabilities,omitempty"`
	ControlPlaneSignature    string            `json:"control_plane_signature,omitempty"`
	PolicySignature          string            `json:"policy_signature,omitempty"`
	Payload                  json.RawMessage   `json:"payload"`
}

PollResult represents a single task returned from the REST poll endpoint.

type RESTPoller 

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

RESTPoller polls the backend REST endpoint as a fallback when WebSocket is unavailable.

func NewRESTPoller 

func NewRESTPoller(config RESTPollerConfig) *RESTPoller

NewRESTPoller creates a new RESTPoller with the given configuration.

func (*RESTPoller) SetAuthToken 

func (p *RESTPoller) SetAuthToken(token string)

SetAuthToken updates the bearer token used by the poller.

func (*RESTPoller) Start 

func (p *RESTPoller) Start(ctx context.Context)

@AX:ANCHOR [AUTO] fallback activation contract — called by messageLoop on connection exhaustion; Stop must be called when WebSocket recovers — fan_in: 3 (messageLoop, Server.Close, test) Start begins the polling loop in a goroutine. Polls until ctx is cancelled or Stop is called.

func (*RESTPoller) Stop 

func (p *RESTPoller) Stop()

Stop cancels the polling goroutine (e.g., when WebSocket recovers).

type RESTPollerConfig 

type RESTPollerConfig struct {
	BackendURL   string
	WorkerID     string
	AuthToken    string
	PollInterval time.Duration
	PollTimeout  time.Duration
	TaskHandler  func(task PollResult) error
	// OnAuthError is called when the poll endpoint returns 401.
	OnAuthError func(statusCode int)
}

RESTPollerConfig holds configuration for the REST fallback poller.

type RegistrationResult 

type RegistrationResult struct {
	Success  bool   `json:"success"`
	WorkerID string `json:"worker_id,omitempty"`
	Error    string `json:"error,omitempty"`
}

RegistrationResult holds the parsed response from agent/register.

func ParseRegistrationResponse 

func ParseRegistrationResponse(data []byte) (*RegistrationResult, error)

ParseRegistrationResponse unmarshals a registration response payload.

type SecurityPolicy 

type SecurityPolicy struct {
	AllowNetwork bool     `json:"allow_network"`
	AllowFS      bool     `json:"allow_fs"`
	AllowedPaths []string `json:"allowed_paths,omitempty"`
	TimeoutSec   int      `json:"timeout_sec,omitempty"`
}

SecurityPolicy defines the security constraints for task execution.

type SendMessageParams 

type SendMessageParams struct {
	TaskID                   string            `json:"task_id"`
	Payload                  json.RawMessage   `json:"payload"`
	RequiredLane             string            `json:"required_lane,omitempty"`
	Model                    string            `json:"model,omitempty"`
	Provider                 string            `json:"provider,omitempty"`
	PipelinePhases           []string          `json:"pipeline_phases,omitempty"`
	PipelineInstructions     map[string]string `json:"pipeline_instructions,omitempty"`
	PipelinePromptTemplates  map[string]string `json:"pipeline_prompt_templates,omitempty"`
	IterationBudget          *IterationBudget  `json:"iteration_budget,omitempty"`
	ControlPlaneCapabilities []string          `json:"control_plane_capabilities,omitempty"`
	ControlPlaneSignature    string            `json:"control_plane_signature,omitempty"`
	PolicySignature          string            `json:"policy_signature,omitempty"`
	SecurityPolicy           SecurityPolicy    `json:"security_policy"`
}

SendMessageParams is the payload for the tasks/send method.

type Server 

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

Server manages the A2A JSON-RPC protocol over WebSocket transport.

func NewServer 

func NewServer(config ServerConfig) *Server

NewServer creates a new A2A server with the given configuration.

func (*Server) Close 

func (s *Server) Close() error

Close shuts down the server and its transport.

func (*Server) HandlePolledTask 

func (s *Server) HandlePolledTask(ctx context.Context, task PollResult) error

HandlePolledTask routes a REST-polled task through the same dispatch path used by WebSocket-delivered tasks.

func (*Server) ReconnectTransport 

func (s *Server) ReconnectTransport(ctx context.Context) error

ReconnectTransport attempts to reconnect the WebSocket transport.

func (*Server) RegisterAgentCard 

func (s *Server) RegisterAgentCard(card AgentCard) error

RegisterAgentCard sends the agent card to the backend.

func (*Server) SendApprovalResponse 

func (s *Server) SendApprovalResponse(params ApprovalResponseParams) error

SendApprovalResponse sends the user's approval decision back to the backend.

func (*Server) SetAuthToken 

func (s *Server) SetAuthToken(token string)

SetAuthToken updates the auth token used for backend communication.

func (*Server) SetRESTPoller 

func (s *Server) SetRESTPoller(p *RESTPoller)

SetRESTPoller attaches a REST poller that activates when WebSocket connection is exhausted.

func (*Server) Start 

func (s *Server) Start(ctx context.Context) error

@AX:ANCHOR [AUTO] lifecycle entry point — connects transport, wires heartbeat, spawns messageLoop goroutine — fan_in: 3 (cmd/worker, integration tests, reload path) Start connects to the backend, registers an Agent Card, and enters the message loop.

func (*Server) UpdateTaskStatus 

func (s *Server) UpdateTaskStatus(taskID string, status TaskStatus, result *TaskResult) error

UpdateTaskStatus sends a task status update to the backend via WebSocket.

type ServerConfig 

type ServerConfig struct {
	BackendURL            string
	WorkerName            string
	WorkspaceID           string
	Skills                []string
	Providers             []string
	ExecutionLanes        []string
	Handler               TaskHandler
	AuthToken             string // Bearer token for backend auth (SEC-005)
	ApprovalCallback      func(ApprovalRequestParams)
	DispatchIssueCallback func(DispatchIssue)
	OnConnectionExhausted func() // called once when reconnect backoff reaches maxBackoff
}

ServerConfig holds configuration for the A2A server.

type StateRecoverer 

type StateRecoverer interface {
	// InFlightTasks returns tasks with status "working" or "input-required".
	InFlightTasks() []Task
	// OnStateRecovered is called after reconnect with the recovered task list.
	OnStateRecovered(tasks []Task)
}

StateRecoverer is implemented by components that need to recover in-flight task state after a reconnection (e.g., the Server).

type StatusUpdateParams 

type StatusUpdateParams struct {
	TaskID string      `json:"task_id"`
	Status TaskStatus  `json:"status"`
	Result *TaskResult `json:"result,omitempty"`
}

StatusUpdateParams is sent to the backend to update task state.

type Task 

type Task struct {
	ID        string     `json:"id"`
	Status    TaskStatus `json:"status"`
	Artifacts []Artifact `json:"artifacts,omitempty"`
	Metadata  TaskMeta   `json:"metadata,omitempty"`
}

Task represents an A2A task with status tracking.

type TaskHandler 

type TaskHandler func(ctx context.Context, taskID string, payload json.RawMessage) (*TaskResult, error)

TaskHandler is invoked when a new task is received.

type TaskLifecycle 

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

TaskLifecycle manages state transitions for a single A2A task.

func NewTaskLifecycle 

func NewTaskLifecycle(task *Task) *TaskLifecycle

NewTaskLifecycle creates a lifecycle tracker for the given task.

func (*TaskLifecycle) AddListener 

func (l *TaskLifecycle) AddListener(fn TransitionListener)

AddListener registers a transition listener.

func (*TaskLifecycle) IsTerminal 

func (l *TaskLifecycle) IsTerminal() bool

IsTerminal returns true if the task is in a terminal state.

func (*TaskLifecycle) Status 

func (l *TaskLifecycle) Status() TaskStatus

Status returns the current task status.

func (*TaskLifecycle) Task 

func (l *TaskLifecycle) Task() *Task

Task returns the tracked task (snapshot under lock).

func (*TaskLifecycle) Transition 

func (l *TaskLifecycle) Transition(to TaskStatus) error

Transition validates and executes a state transition. Returns an error if the transition is not allowed.

type TaskMeta 

type TaskMeta struct {
	CreatedAt string `json:"created_at,omitempty"`
	UpdatedAt string `json:"updated_at,omitempty"`
}

TaskMeta contains optional metadata attached to a task.

type TaskResult 

type TaskResult struct {
	Status        TaskStatus `json:"status"`
	Artifacts     []Artifact `json:"artifacts,omitempty"`
	Error         string     `json:"error,omitempty"`
	SessionID     string     `json:"session_id,omitempty"`
	TraceID       string     `json:"trace_id,omitempty"`
	CorrelationID string     `json:"correlation_id,omitempty"`
}

TaskResult holds the outcome of a completed or failed task.

type TaskStatus 

type TaskStatus string

TaskStatus represents the lifecycle state of an A2A task. 

const (
	StatusWorking       TaskStatus = "working"
	StatusInputRequired TaskStatus = "input-required"
	StatusCompleted     TaskStatus = "completed"
	StatusFailed        TaskStatus = "failed"
	StatusCanceled      TaskStatus = "canceled"
)

type TransitionListener 

type TransitionListener func(taskID string, from, to TaskStatus)

TransitionListener is called when a task transitions between states.

type Transport 

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

Transport wraps a gorilla/websocket connection with heartbeat and reconnect.

func NewTransport 

func NewTransport(config TransportConfig) *Transport

NewTransport creates a new WebSocket transport.

func (*Transport) Close 

func (t *Transport) Close() error

Close gracefully shuts down the transport.

func (*Transport) Connect 

func (t *Transport) Connect(ctx context.Context) error

Connect dials the WebSocket endpoint and starts the heartbeat loop.

func (*Transport) Receive 

func (t *Transport) Receive() ([]byte, error)

Receive reads the next message from the WebSocket connection.

func (*Transport) Reconnect 

func (t *Transport) Reconnect(ctx context.Context) error

Reconnect attempts to re-establish the WebSocket connection with exponential backoff. base=3s, factor=2, max retries=4.

func (*Transport) Send 

func (t *Transport) Send(msg []byte) error

Send writes a message to the WebSocket connection.

func (*Transport) SetAuthToken 

func (t *Transport) SetAuthToken(token string)

SetAuthToken updates the bearer token used for future connects/reconnects.

type TransportConfig 

type TransportConfig struct {
	URL              string
	AuthToken        string // Bearer token for WebSocket auth (SEC-005)
	HeartbeatSec     int
	ReconnectBaseSec int
	ReconnectFactor  int
	MaxRetries       int
	// Dialer overrides the default WebSocket dialer (used in tests for TLS certs).
	Dialer *websocket.Dialer
}

TransportConfig holds WebSocket transport settings.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.