protocol

package
v0.15.0 Latest Latest
Warning

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

 Go to latest Published: Sep 25, 2026 License: MIT Imports: 6 Imported by: 0

Documentation

Overview

Package protocol defines the agent-to-agent communication protocol.

The Bus is a mutex-protected message router. Replies to a SendAndWait call are matched by correlation id (the request's message id): each in-flight request registers a private, single-slot reply channel keyed by that id, and the delivery path dispatches the reply straight to the matching waiter. This means concurrent senders that share a single inbox can never receive one another's replies.

Handler invocations run on a bounded pool of goroutines. When the pool is saturated, delivery fails with a back-pressure error instead of spawning unbounded goroutines. Per-peer inboxes are bounded buffered channels that likewise apply back-pressure once full. Direct agent-to-agent channels bypass the central router for point-to-point communication.

Envelope values may be reused via the AcquireEnvelope/ReleaseEnvelope pool to reduce allocations in high-throughput senders; the bus itself does not pool envelopes internally.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ReleaseEnvelope 

func ReleaseEnvelope(e *Envelope)

ReleaseEnvelope returns an envelope to the pool after clearing it.

WARNING: do NOT release an envelope after passing it to Bus.Send/SendAndWait. The bus retains that pointer (in message history and in the recipient's inbox), so releasing it zeroes state still in use and corrupts delivery. Release is only safe for envelopes you built but did not send.

Types

type Bus 

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

Bus is the central message router for agent-to-agent communication. Delivery is non-blocking: if a peer's inbox is full the send fails with an error rather than blocking the sender (back-pressure).

func NewBus 

func NewBus() *Bus

NewBus creates a new communication bus with default settings.

func NewBusWithConfig 

func NewBusWithConfig(cfg BusConfig) *Bus

NewBusWithConfig creates a bus with explicit resource limits.

func (*Bus) Ask 

func (b *Bus) Ask(ctx context.Context, from, to, question string) (string, error)

Ask sends a question and waits for the answer.

func (*Bus) Close 

func (b *Bus) Close()

Close shuts down the bus and all direct channels.

func (*Bus) DelegateTask 

func (b *Bus) DelegateTask(ctx context.Context, from, to, subject string, task TaskPayload) (*ResultPayload, error)

DelegateTask sends a task request and waits for the result.

func (*Bus) DirectChannelBetween 

func (b *Bus) DirectChannelBetween(agentA, agentB string, bufSize int) *DirectChannel

DirectChannelBetween returns (or creates) a dedicated channel between two agents.

func (*Bus) FindByCapability 

func (b *Bus) FindByCapability(capability string) []*Peer

FindByCapability returns agents that advertise the given capability.

func (*Bus) History 

func (b *Bus) History() []*Envelope

History returns all messages recorded by the bus (for observability/debugging).

func (*Bus) Peers 

func (b *Bus) Peers() []*Peer

Peers returns all registered agents.

func (*Bus) Register 

func (b *Bus) Register(id, name, description string, capabilities []string, handler Handler) error

Register adds an agent to the bus so it can send and receive messages.

func (*Bus) Send 

func (b *Bus) Send(ctx context.Context, env *Envelope) error

Send delivers an envelope to its recipient. For broadcasts (To=="*") the message is delivered to all peers except the sender. Returns an error immediately if the recipient's inbox is full (back-pressure).

func (*Bus) SendAndWait 

func (b *Bus) SendAndWait(ctx context.Context, env *Envelope) (*Envelope, error)

SendAndWait sends an envelope and blocks until the matching reply is received or the context is canceled.

The reply is matched by correlation id: before sending, the caller registers a private reply channel keyed by the request's message id. The delivery path dispatches the reply to that channel, so many concurrent SendAndWait calls on the same sender inbox each receive exactly their own reply with no mis-delivery and no requeue races.

func (*Bus) Unregister 

func (b *Bus) Unregister(id string)

Unregister removes an agent from the bus.

type BusConfig 

type BusConfig struct {
	InboxSize  int // per-peer inbox buffer; 0 = defaultInboxSize
	HistoryCap int // max retained history entries; 0 = defaultHistoryCap
	// MaxConcurrentHandlers caps the number of handler goroutines that may run
	// at once across the whole bus; 0 = defaultMaxHandlers. When the cap is
	// reached, delivery to a handler-backed peer fails with a back-pressure
	// error rather than spawning an unbounded number of goroutines.
	MaxConcurrentHandlers int
}

BusConfig tunes Bus resource limits.

type ChatMessage 

type ChatMessage struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

ChatMessage is a message in a conversation being handed off.

type DirectChannel 

type DirectChannel struct {
	AtoB chan *Envelope
	BtoA chan *Envelope
}

DirectChannel is a dedicated bidirectional pipe between two specific agents that bypasses the central bus for minimal-latency point-to-point messaging. Each direction uses a separate buffered channel to avoid head-of-line blocking.

func NewDirectChannel 

func NewDirectChannel(bufSize int) *DirectChannel

NewDirectChannel creates a channel pair with the given buffer capacity.

func (*DirectChannel) Close 

func (dc *DirectChannel) Close()

Close drains and closes both directions.

type Envelope 

type Envelope struct {
	ID        string            `json:"id"`
	Type      MessageType       `json:"type"`
	From      string            `json:"from"`
	To        string            `json:"to"`
	ReplyTo   string            `json:"reply_to,omitempty"`
	Subject   string            `json:"subject"`
	Body      json.RawMessage   `json:"body"`
	Priority  Priority          `json:"priority"`
	Metadata  map[string]string `json:"metadata,omitempty"`
	CreatedAt time.Time         `json:"created_at"`
	ExpiresAt time.Time         `json:"expires_at,omitempty"`
}

Envelope is the unit of communication between agents.

func AcquireEnvelope 

func AcquireEnvelope() *Envelope

AcquireEnvelope returns a pooled envelope for building an outgoing message. Call ReleaseEnvelope only for envelopes you did NOT hand to a bus (see the ReleaseEnvelope warning).

type Handler 

type Handler func(ctx context.Context, env *Envelope) (*Envelope, error)

Handler processes an incoming envelope and optionally returns a reply.

type HandoffPayload 

type HandoffPayload struct {
	Reason       string         `json:"reason"`
	Conversation []ChatMessage  `json:"conversation,omitempty"`
	Context      map[string]any `json:"context,omitempty"`
}

HandoffPayload is the body of a TypeHandoff envelope.

type MessageType 

type MessageType string

MessageType classifies the intent of an inter-agent message. 

const (
	TypeTaskRequest MessageType = "task_request"
	TypeTaskResult  MessageType = "task_result"
	TypeQuestion    MessageType = "question"
	TypeAnswer      MessageType = "answer"
	TypeBroadcast   MessageType = "broadcast"
	TypeAck         MessageType = "ack"
	TypeError       MessageType = "error"
	TypeHandoff     MessageType = "handoff"
	TypeStatus      MessageType = "status"
)

type Peer 

type Peer struct {
	ID           string   `json:"id"`
	Name         string   `json:"name"`
	Description  string   `json:"description,omitempty"`
	Capabilities []string `json:"capabilities,omitempty"`
	// contains filtered or unexported fields
}

Peer represents a registered agent in the communication bus.

type Priority 

type Priority int

Priority controls message ordering when an agent's inbox has multiple pending messages. 

const (
	PriorityLow    Priority = 0
	PriorityNormal Priority = 1
	PriorityHigh   Priority = 2
	PriorityUrgent Priority = 3
)

type ResultPayload 

type ResultPayload struct {
	TaskID  string         `json:"task_id"`
	Success bool           `json:"success"`
	Output  map[string]any `json:"output,omitempty"`
	Error   string         `json:"error,omitempty"`
	Summary string         `json:"summary,omitempty"`
}

ResultPayload is the body of a TypeTaskResult envelope.

type StatusPayload 

type StatusPayload struct {
	TaskID   string  `json:"task_id"`
	Progress float64 `json:"progress"`
	Message  string  `json:"message"`
}

StatusPayload is the body of a TypeStatus envelope.

type TaskPayload 

type TaskPayload struct {
	Description string         `json:"description"`
	Input       map[string]any `json:"input,omitempty"`
	Constraints []string       `json:"constraints,omitempty"`
	Deadline    time.Time      `json:"deadline,omitempty"`
}

TaskPayload is the body of a TypeTaskRequest envelope.

Source Files

View all Source files

Directories

Path Synopsis
Package a2a implements the Agent-to-Agent (A2A) protocol for cross-framework agent communication.
 Package a2a implements the Agent-to-Agent (A2A) protocol for cross-framework agent communication.

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.