Documentation
¶
Overview ¶
Package a2a implements the v0.1 synchronous A2A client from docs/a2a-design.md ("Mast as A2A client", client-only phasing row): agent-card discovery and caching, JSON-RPC 2.0 message/send on the single A2A v0.3 endpoint (A2A-Version header, bearer auth from an env-var reference), direct-message and task-opened reply handling with bounded tasks/get polling to a terminal state, and tasks/cancel on caller cancellation. Static agent configs load from .agents/a2a/*.yaml (see AgentConfig / LoadDir; wired into pkg/config root scanning).
Build-vs-reuse note (docs/a2a-design.md v0.1 phasing row asks that ADK v2.1.0's agentregistry package be evaluated before hand-building client machinery): evaluated 2026-07-26 and declined for v0.1. That package is a Google Cloud Agent Registry client — ADC-authenticated discovery plus RemoteAgent factories over github.com/a2aproject/ a2a-go/v2, aimed at the registry-discovery story that is explicitly v0.2 here. Using its factories for the static-config path would add a2a-go/v2 as a direct dependency and an ADK RemoteAgent abstraction on top, to obtain three JSON-RPC methods and a card GET that fit in this file against stdlib net/http. Revisit at v0.2, where streaming (message/stream over SSE) and registry discovery make the SDK earn its place.
Layering note: Send returns *federation.Result and wraps the federation sentinel errors directly rather than defining a parallel error/result vocabulary — pkg/federation is protocol-neutral and does not import this package, so the dependency is one-way.
Index ¶
- Constants
- Variables
- type Adapter
- type AgentCard
- type AgentConfig
- type AgentInterface
- type AgentSkill
- type Artifact
- type AuthConfig
- type Backend
- type Capabilities
- type Client
- type ClientOption
- type Config
- type ExposedSkill
- type Message
- type Part
- type Principal
- type RPCError
- type RateLimitRequest
- type RateLimiter
- type SecurityScheme
- type SecuritySchemes
- type Server
- type StaticBearerValidator
- type SubmitParams
- type Task
- type TaskArtifactUpdateEvent
- type TaskInfo
- type TaskMetric
- type TaskState
- type TaskStatus
- type TaskStatusUpdateEvent
- type TokenBucketLimiter
- type TokenValidator
Constants ¶
const AuthTypeBearer = "bearer"
AuthTypeBearer is the only auth type the v0.1 client supports. google-iam (docs/a2a-design.md static-registration example) joins in v0.2 alongside the pluggable token-resolver surface.
const DefaultTimeout = 120 * time.Second
DefaultTimeout bounds a Send when neither the per-agent config nor the caller supplies one. v0.1 calls block to a bounded timeout — long-running remote tasks need programmatic pause, which is v0.2 per docs/durable-execution-design.md phasing.
const ProtocolVersion = "0.3"
ProtocolVersion is the A2A spec line this client implements and advertises in the A2A-Version request header (docs/a2a-design.md resolved open question 1: pin to the v0.3 line, send the header, document tested-against versions per release).
const Scheme = "a2a"
Scheme is the federation reference scheme this adapter serves: a2a://<name>/<skill> (docs/federation-design.md reference grammar).
const TransportJSONRPC = "JSONRPC"
TransportJSONRPC is the card transport identifier for JSON-RPC 2.0 over HTTP — the only transport this client speaks (docs/a2a-design.md endpoint layout: "JSON-RPC only at first"). An absent preferredTransport defaults to JSON-RPC per spec.
const VersionHeader = "A2A-Version"
VersionHeader is the HTTP request header carrying ProtocolVersion.
const WellKnownCardPath = "/.well-known/agent-card.json"
WellKnownCardPath is the spec-defined agent-card discovery path.
Variables ¶
var ( NewStaticBearerValidator = serverauth.NewStaticBearerValidator NewTokenBucketLimiter = serverauth.NewTokenBucketLimiter )
Constructors, re-exported as func values (a Go func cannot be type-aliased). Callers use a2a.NewStaticBearerValidator / a2a.NewTokenBucketLimiter unchanged.
var ErrInvalidToken = serverauth.ErrInvalidToken
ErrInvalidToken marks an unrecognized bearer token; the server maps it to HTTP 401. Re-exported so errors.Is(err, a2a.ErrInvalidToken) keeps matching after the hoist (it is the same error value as serverauth.ErrInvalidToken).
var ErrTaskNotFound = errors.New("a2a: task not found")
ErrTaskNotFound marks an unknown task id; the server maps it to the A2A TaskNotFound JSON-RPC error (-32001).
ErrUnavailable marks a transiently-unavailable backend — e.g. a server draining for shutdown that refuses new work. The server maps it to a server-error JSON-RPC code (-32000) so a caller reads it as retryable, not as an internal fault (-32603). A Backend returns it (wrapped) from SubmitMessage.
Functions ¶
This section is empty.
Types ¶
type Adapter ¶
type Adapter struct {
// contains filtered or unexported fields
}
Adapter is the A2A protocol adapter for the federation registry — the v0.1 slice of docs/federation-design.md's "A2A adapter": static references only; registry-discovered references (a2a://<registry>/ <agent-id>) are v0.2.
func NewAdapter ¶
func NewAdapter(cfgs []AgentConfig, opts ...ClientOption) (*Adapter, error)
NewAdapter builds the adapter from static agent configs (normally pkg/config's .agents/a2a/*.yaml scan). opts apply to every client.
func (*Adapter) Invoke ¶
func (a *Adapter) Invoke(ctx context.Context, ref federation.Reference, inputs map[string]any, opts federation.InvokeOptions) (federation.Handle, error)
Invoke implements federation.Adapter. Per the frozen contract it returns an error only for dispatch-time failures (unknown agent); the synchronous remote call runs inside Invoke — v0.1 blocks to a bounded timeout — and its outcome, success or failure, surfaces from the returned Handle's Wait.
type AgentCard ¶
type AgentCard struct {
Name string `json:"name"`
Description string `json:"description,omitempty"`
URL string `json:"url"`
Version string `json:"version,omitempty"`
ProtocolVersion string `json:"protocolVersion,omitempty"`
PreferredTransport string `json:"preferredTransport,omitempty"`
AdditionalInterfaces []AgentInterface `json:"additionalInterfaces,omitempty"`
Capabilities Capabilities `json:"capabilities,omitempty"`
DefaultInputModes []string `json:"defaultInputModes,omitempty"`
DefaultOutputModes []string `json:"defaultOutputModes,omitempty"`
Skills []AgentSkill `json:"skills,omitempty"`
// SecuritySchemes / Security advertise the auth the endpoint
// requires (mast as A2A server, docs/a2a-design.md "Auth model").
// The client tolerates both absent (unauthenticated remote); the
// server emits them when any exposed skill requires auth.
SecuritySchemes *SecuritySchemes `json:"securitySchemes,omitempty"`
Security []map[string][]string `json:"security,omitempty"`
}
AgentCard is the discovery-time contract served at /.well-known/agent-card.json. Note: spec AgentSkill has NO I/O schemas — only inputModes/outputModes media types. Structured I/O contracts are conveyed in skill descriptions or out of band (docs/a2a-design.md, protocol overview).
type AgentConfig ¶
type AgentConfig struct {
// Name is the reference name: a2a://<name>/<skill>. Must be
// lowercase because the name travels in the URI host position of a
// federation reference, where RFC 3986 case-insensitivity means
// parsers normalize to lowercase (see federation.ParseReference).
Name string `yaml:"name"`
// AgentCardURL locates the agent card. A base URL (empty or "/"
// path) gets WellKnownCardPath appended. The JSON-RPC endpoint is
// then resolved from the card.
AgentCardURL string `yaml:"agent_card_url,omitempty"`
// Endpoint is the JSON-RPC endpoint, bypassing card discovery.
// When both Endpoint and AgentCardURL are set, Endpoint wins for
// transport and the card is still fetched for skill validation.
Endpoint string `yaml:"endpoint,omitempty"`
// Skills is the subset of the agent's skills mast may invoke;
// empty = all (docs/a2a-design.md).
Skills []string `yaml:"skills,omitempty"`
// Auth is optional; absent means unauthenticated calls.
Auth *AuthConfig `yaml:"auth,omitempty"`
// TimeoutSeconds bounds each invocation; 0 = DefaultTimeout.
TimeoutSeconds int `yaml:"timeout_seconds,omitempty"`
// Filename records provenance for error messages.
Filename string `yaml:"-"`
}
AgentConfig is one static A2A agent registration from .agents/a2a/<name>.yaml (docs/a2a-design.md, "Static registration").
func LoadDir ¶
func LoadDir(dir string) ([]AgentConfig, error)
LoadDir loads every *.yaml / *.yml in dir (flat, non-recursive, per the pkg/config scan rules). A missing dir yields zero entries. Any invalid file is a fatal load error. Name-collision checking across files is the caller's job (pkg/config does it alongside its other same-directory collision checks).
func (AgentConfig) Timeout ¶
func (c AgentConfig) Timeout() time.Duration
Timeout returns the effective invocation bound.
func (AgentConfig) Validate ¶
func (c AgentConfig) Validate() error
Validate checks an AgentConfig for load-time errors. All errors are fatal per the config-layout v0.1 rule (fail fast on invalid config).
type AgentInterface ¶
AgentInterface is one (transport, url) alternative from the card.
type AgentSkill ¶
type AgentSkill struct {
ID string `json:"id"`
Name string `json:"name,omitempty"`
Description string `json:"description,omitempty"`
Tags []string `json:"tags,omitempty"`
InputModes []string `json:"inputModes,omitempty"`
OutputModes []string `json:"outputModes,omitempty"`
}
AgentSkill is one named capability. Per the v0.3 spec it carries media types, not JSON Schemas.
type Artifact ¶
type Artifact struct {
ArtifactID string `json:"artifactId,omitempty"`
Name string `json:"name,omitempty"`
Description string `json:"description,omitempty"`
Parts []Part `json:"parts,omitempty"`
}
Artifact is a task output artifact.
type AuthConfig ¶
type AuthConfig struct {
// Type is the auth mechanism; v0.1: "bearer" only.
Type string `yaml:"type"`
// TokenEnv names the environment variable holding the bearer
// token. Resolved at request time, not load time, so rotation
// does not require a config reload.
TokenEnv string `yaml:"token_env"`
}
AuthConfig is the static-registration auth block. Tokens are env-var references, never file-embedded (docs/a2a-design.md).
type Backend ¶ added in v0.2.0
type Backend interface {
// GetTask returns the task snapshot, or ErrTaskNotFound.
GetTask(ctx context.Context, taskID string) (TaskInfo, error)
// CancelTask requests cancellation (idempotent), returning the
// resulting snapshot, or ErrTaskNotFound.
CancelTask(ctx context.Context, taskID, reason string) (TaskInfo, error)
// SubmitMessage runs a message/send turn through the mast turn
// chokepoint and returns the resolved task id and its terminal
// snapshot. Continuing a task whose id is unknown returns
// ErrTaskNotFound; a turn that runs but errors is reported through the
// snapshot's State (failed / canceled / input-required), not an error
// return. The ctx carries any propagated caller trace context.
SubmitMessage(ctx context.Context, p SubmitParams) (taskID string, info TaskInfo, err error)
// StreamMessage runs a message/stream turn through the same chokepoint
// as SubmitMessage, emitting streaming updates through emit as the turn
// progresses. emit is called synchronously and in order on the calling
// goroutine — the SSE handler writes each frame to the wire — so
// implementations need no locking around it. The first emit is the
// initial *Task snapshot; subsequent emits are *TaskStatusUpdateEvent
// progress updates (final=false). The returned task id and terminal
// snapshot let the server emit the closing artifact + a final
// status-update. Error semantics mirror SubmitMessage (ErrTaskNotFound
// / ErrUnavailable before any emit; a turn that runs but fails is
// reported through the snapshot State). A backend must not emit after
// returning.
StreamMessage(ctx context.Context, p SubmitParams, emit func(any)) (taskID string, info TaskInfo, err error)
}
Backend drives task verbs against the mast runtime. The daemon implements it over the transcript store (GetTask), the abort machinery (CancelTask), and — in Stage B — runTurnPre (submit). This package never imports the runtime; the seam mirrors inject.Handler.
type Capabilities ¶
type Capabilities struct {
Streaming bool `json:"streaming,omitempty"`
PushNotifications bool `json:"pushNotifications,omitempty"`
}
Capabilities is the card's capability declaration.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is a synchronous A2A v0.3 client for one configured agent. Safe for concurrent use.
func NewClient ¶
func NewClient(cfg AgentConfig, opts ...ClientOption) (*Client, error)
NewClient validates cfg and returns a Client. No network I/O happens until the first call (card fetch is lazy).
func (*Client) Card ¶
Card returns the agent card, fetching and caching it on first use. Endpoint-only configs (no agent_card_url) return (nil, nil): the card is optional when the operator pinned the endpoint directly.
func (*Client) Send ¶
func (c *Client) Send(ctx context.Context, skill string, inputs map[string]any, timeout time.Duration) (*federation.Result, error)
Send invokes skill on the remote agent with inputs and blocks until a terminal state or the bounded timeout (opts precedence: timeout argument > config timeout_seconds > DefaultTimeout). The reply may be a direct message (returned as a completed Result) or an opened task, which Send polls via tasks/get. If the caller's ctx is canceled or the bound expires mid-task, Send issues tasks/cancel (best effort, on a detached context) before returning.
Skill selection: A2A v0.3 message/send has no first-class skill selector — AgentSkill is card-level metadata and MessageSendParams carries only the message. Mast conveys the chosen skill as message metadata under "skillId" (and validates it against the fetched card and the config's skills allowlist). Servers that route by content ignore the hint harmlessly.
type ClientOption ¶
type ClientOption func(*Client)
ClientOption customizes a Client.
func WithHTTPClient ¶
func WithHTTPClient(h *http.Client) ClientOption
WithHTTPClient substitutes the transport (tests, custom TLS/proxies).
func WithPollInterval ¶
func WithPollInterval(d time.Duration) ClientOption
WithPollInterval overrides the tasks/get polling cadence.
type Config ¶ added in v0.2.0
type Config struct {
// Listen is the bind address, e.g. ":7780". Used by ListenAndServe.
Listen string
// Skills are the exposed workloads. An empty slice serves a card
// with no skills; the daemon only starts the server when at least
// one workload opts in.
Skills []ExposedSkill
// Validator authenticates every /a2a request. Nil disables auth
// (dev only) — like inject's empty BearerToken. When set, a request
// without a valid bearer is refused 401 before any dispatch.
Validator TokenValidator
// Backend is required.
Backend Backend
// Limiter, when non-nil, admits or refuses each turn-driving request
// (message/send) before dispatch — see RateLimiter. Nil disables rate
// limiting. Control-plane verbs (tasks/get, tasks/cancel) are never
// gated.
Limiter RateLimiter
// CardName / CardDescription / CardVersion populate the aggregated
// agent card. CardName defaults to "mast".
CardName string
CardDescription string
CardVersion string
// ExternalURL overrides the card's request-derived url.
ExternalURL string
// Metric, when non-nil, records task outcomes.
Metric TaskMetric
// Logger defaults to slog.Default().
Logger *slog.Logger
// BaseContext, when non-nil, is the context every request derives
// from (the daemon passes its turn lifetime, as for inject).
BaseContext context.Context
}
Config configures the A2A server.
type ExposedSkill ¶ added in v0.2.0
type ExposedSkill struct {
// WorkloadName is the mast workload backing this skill; the task
// registry maps a submitted skill call to it (Stage B), and task
// verbs resolve required scopes through it.
WorkloadName string
// SkillName is the A2A skill id/name (bundle a2a.skill_name).
SkillName string
// Description is rendered into the card skill. The daemon may fold
// the mast-side input/output schema hints into it — spec AgentSkill
// has no schema fields (docs/a2a-design.md note).
Description string
// Tags surface on the card skill; defaults to ["mast"] when empty.
Tags []string
// Scopes are required to invoke this skill. Empty means the skill
// needs authentication only (a valid token, no specific scope) when
// a validator is configured, or open access when it is not.
Scopes []string
}
ExposedSkill is one workload's A2A exposure, projected by the daemon from the bundle's a2a: section (this package does not import pkg/workload). One skill per exposed workload.
type Message ¶
type Message struct {
Kind string `json:"kind"`
MessageID string `json:"messageId"`
Role string `json:"role"`
Parts []Part `json:"parts"`
TaskID string `json:"taskId,omitempty"`
ContextID string `json:"contextId,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
Message is an A2A message (kind: "message").
type Part ¶
type Part struct {
Kind string `json:"kind"`
Text string `json:"text,omitempty"`
Data map[string]any `json:"data,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
Part is a message/artifact content part. Exactly one of Text / Data is populated for the kinds this client produces and consumes ("text", "data"); other kinds ("file") pass through with only Kind set.
type Principal ¶ added in v0.2.0
type Principal = serverauth.Principal
Principal is the authenticated caller a TokenValidator resolves a bearer token to. See serverauth.Principal.
type RPCError ¶
type RPCError struct {
Code int `json:"code"`
Message string `json:"message"`
Data json.RawMessage `json:"data,omitempty"`
}
RPCError is a JSON-RPC 2.0 error object (A2A-specific codes included, e.g. -32001 TaskNotFound).
type RateLimitRequest ¶ added in v0.2.0
type RateLimitRequest = serverauth.RateLimitRequest
RateLimitRequest identifies one inbound call for an admission decision. See serverauth.RateLimitRequest.
type RateLimiter ¶ added in v0.2.0
type RateLimiter = serverauth.RateLimiter
RateLimiter admits or refuses an inbound turn-driving call. See serverauth.RateLimiter.
type SecurityScheme ¶ added in v0.2.0
type SecurityScheme struct {
Type string `json:"type"` // const "http"
Scheme string `json:"scheme"` // const "Bearer"
}
SecurityScheme is one A2A security scheme (HTTP Bearer).
type SecuritySchemes ¶ added in v0.2.0
type SecuritySchemes struct {
Bearer *SecurityScheme `json:"bearer,omitempty"`
}
SecuritySchemes is the card's declared security-scheme set. mast speaks bearer only (docs/a2a-design.md: JWT/JWKS/IAM/OAuth2 validators are pluggable behind TokenValidator, but all present as HTTP Bearer on the wire).
type Server ¶ added in v0.2.0
type Server struct {
// contains filtered or unexported fields
}
Server is the A2A HTTP server. Construct with New; serve with ListenAndServe.
func (*Server) Handler ¶ added in v0.2.0
Handler exposes the server's routes for mounting on an external mux or for tests (httptest.NewServer).
func (*Server) ListenAndServe ¶ added in v0.2.0
ListenAndServe blocks serving requests; returns http.ErrServerClosed on graceful shutdown.
type StaticBearerValidator ¶ added in v0.2.0
type StaticBearerValidator = serverauth.StaticBearerValidator
StaticBearerValidator validates against a fixed token→Principal map. See serverauth.StaticBearerValidator.
type SubmitParams ¶ added in v0.2.0
type SubmitParams struct {
// TaskID continues an existing task (== session id) when set; empty
// mints a fresh task.
TaskID string
// ContextID groups related messages; carried onto the task snapshot.
ContextID string
// Text is the joined text of the inbound message's parts.
Text string
}
SubmitParams is a message/send request projected onto the runtime seam. The server extracts it from the A2A Message; the daemon converts it into a mast turn (Text → user message) and runs it through the turn chokepoint. Data/file parts are text-only for Stage B (docs/a2a-design.md).
type Task ¶
type Task struct {
Kind string `json:"kind"`
ID string `json:"id"`
ContextID string `json:"contextId,omitempty"`
Status TaskStatus `json:"status"`
Artifacts []Artifact `json:"artifacts,omitempty"`
}
Task is an A2A task (kind: "task").
type TaskArtifactUpdateEvent ¶ added in v0.2.0
type TaskArtifactUpdateEvent struct {
Kind string `json:"kind"` // "artifact-update"
TaskID string `json:"taskId"`
ContextID string `json:"contextId,omitempty"`
Artifact Artifact `json:"artifact"`
Append bool `json:"append,omitempty"`
LastChunk bool `json:"lastChunk,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
TaskArtifactUpdateEvent is one streaming artifact update (kind: "artifact-update") in a message/stream SSE response, carrying an output artifact produced during the turn. LastChunk marks the final piece of a chunked artifact — always true for mast's whole-artifact v0.2 emits (docs/a2a-design.md "Stage C").
type TaskInfo ¶ added in v0.2.0
type TaskInfo struct {
// WorkloadName owns the task; it resolves the skill for scope checks.
WorkloadName string
// State is the A2A lifecycle state, mapped from the session's
// log-proven state by the backend. A transcript-only read never
// reports "completed" (the event log cannot prove a turn finished
// versus is in flight) — that state comes from the in-process task
// registry in Stage B.
State TaskState
// ContextID groups related messages; optional in Stage A.
ContextID string
// StatusMessage is an optional human-readable status line surfaced in
// the task's status.message.
StatusMessage string
// Output is the agent's answer for a completed task; when non-empty
// it surfaces as a text artifact on the returned Task (Stage B
// message/send). Empty for read/cancel snapshots.
Output string
}
TaskInfo is the backend's snapshot of a task (== a mast session).
type TaskMetric ¶ added in v0.2.0
type TaskMetric interface {
A2ATask(workload, outcome string)
}
TaskMetric records A2A task lifecycle outcomes. The daemon backs it with observability.Registry.A2ATask; nil disables. The outcome string is a TaskState value (fixed vocabulary — see observability.Prime).
type TaskState ¶
type TaskState string
TaskState is the A2A v0.3 task lifecycle vocabulary.
const ( TaskStateSubmitted TaskState = "submitted" TaskStateWorking TaskState = "working" TaskStateInputRequired TaskState = "input-required" TaskStateAuthRequired TaskState = "auth-required" TaskStateCompleted TaskState = "completed" TaskStateFailed TaskState = "failed" TaskStateCanceled TaskState = "canceled" TaskStateRejected TaskState = "rejected" )
type TaskStatus ¶
type TaskStatus struct {
State TaskState `json:"state"`
Message *Message `json:"message,omitempty"`
}
TaskStatus is the task's current lifecycle position.
type TaskStatusUpdateEvent ¶ added in v0.2.0
type TaskStatusUpdateEvent struct {
Kind string `json:"kind"` // "status-update"
TaskID string `json:"taskId"`
ContextID string `json:"contextId,omitempty"`
Status TaskStatus `json:"status"`
Final bool `json:"final"`
Metadata map[string]any `json:"metadata,omitempty"`
}
TaskStatusUpdateEvent is one streaming status update (kind: "status-update") in a message/stream SSE response. Final marks the terminal update that ends the stream; a server emits exactly one final update per turn (docs/a2a-design.md "Stage C").
type TokenBucketLimiter ¶ added in v0.2.0
type TokenBucketLimiter = serverauth.TokenBucketLimiter
TokenBucketLimiter is the built-in RateLimiter. See serverauth.TokenBucketLimiter.
type TokenValidator ¶ added in v0.2.0
type TokenValidator = serverauth.TokenValidator
TokenValidator resolves a bearer token to a Principal. See serverauth.TokenValidator.