events

Documentation

Overview

Package events provides a pub/sub event system for AI agent interactions, supporting type-safe event handling with rich metadata and serialization. It builds on top of the provider package's streaming events, adding sender tracking and pub/sub capabilities.

Design decisions:

• Type safety: Generic event types ensure compile-time correctness
• Rich metadata: Every event includes run/turn IDs, timestamps, and optional metadata
• Sender tracking: Events maintain origin information through the system
• Efficient JSON: Custom marshaling with pre-allocated type markers
• Error context: Errors preserve full execution context for debugging
• Provider integration: Seamless conversion from provider.StreamEvent

Event hierarchy:

• Event: Base interface for all pub/sub events ├── Delim: Stream boundary markers ├── Chunk[T]: Incremental response fragments ├── Request[T]: Incoming requests (user prompts, tool calls) ├── Response[T]: Complete responses with context ├── Result[T]: Final computation results └── Error: Error events with preserved context

Each event type includes:

• RunID: Unique identifier for the execution run
• TurnID: Identifier for the specific interaction turn
• Timestamp: When the event occurred
• Sender: Origin of the event (agent, tool, etc.)
• Meta: Optional structured metadata

Example usage:

// Convert provider events to pub/sub events
providerEvent := provider.Chunk[messages.AssistantMessage]{...}
pubsubEvent := events.FromStreamEvent(providerEvent, "openai")

// Create and handle events
event := events.Request[messages.UserMessage]{
    RunID:  uuid.New(),
    TurnID: uuid.New(),
    Message: messages.UserMessage{
        Content: messages.ContentOrParts{Content: "Hello"},
    },
    Sender: "user",
}

// Type-safe event handling
switch e := event.(type) {
case events.Request[messages.UserMessage]:
    // Handle user request
case events.Response[messages.AssistantMessage]:
    // Handle assistant response
case events.Error:
    // Handle error with context
}

The package is designed to work seamlessly with the provider and messages packages, providing a complete system for handling AI agent interactions with proper typing, context preservation, and error handling.

Index

• func ToJSON(event Event) ([]byte, error)
• type Chunk
  • func (c Chunk[T]) MarshalJSON() ([]byte, error)
  • func (c *Chunk[T]) UnmarshalJSON(data []byte) error
• type CompositeHook
  • func (c CompositeHook) OnAssistantChunk(ctx context.Context, ac messages.Message[messages.AssistantMessage])
  • func (c CompositeHook) OnAssistantMessage(ctx context.Context, am messages.Message[messages.AssistantMessage])
  • func (c CompositeHook) OnError(ctx context.Context, err error)
  • func (c CompositeHook) OnToolCallChunk(ctx context.Context, tc messages.Message[messages.ToolCallMessage])
  • func (c CompositeHook) OnToolCallMessage(ctx context.Context, tm messages.Message[messages.ToolCallMessage])
  • func (c CompositeHook) OnToolCallResponse(ctx context.Context, tr messages.Message[messages.ToolResponse])
  • func (c CompositeHook) OnUserPrompt(ctx context.Context, up messages.Message[messages.UserMessage])
• type Delim
  • func (d Delim) MarshalJSON() ([]byte, error)
  • func (d *Delim) UnmarshalJSON(data []byte) error
• type Error
  • func (e Error) Error() string
  • func (e Error) MarshalJSON() ([]byte, error)
  • func (e *Error) UnmarshalJSON(data []byte) error
• type Event
  • func FromJSON(jsonData []byte) (Event, error)
  • func FromStreamEvent(e provider.StreamEvent, sender string) Event
• type Hook
  • func NewCompositeHook(hooks ...Hook) Hook
• type Request
  • func (r Request[T]) MarshalJSON() ([]byte, error)
  • func (r *Request[T]) UnmarshalJSON(data []byte) error
• type Response
  • func (r Response[T]) MarshalJSON() ([]byte, error)
  • func (r *Response[T]) UnmarshalJSON(data []byte) error
• type Result
  • func (r Result[T]) MarshalJSON() ([]byte, error)
  • func (r *Result[T]) UnmarshalJSON(data []byte) error

Functions

func ToJSON

func ToJSON(event Event) ([]byte, error)

Types

type Chunk

type Chunk[T messages.Response] struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Chunk     T               `json:"chunk"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Chunk[T]) MarshalJSON

func (c Chunk[T]) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Chunk[T]

func (*Chunk[T]) UnmarshalJSON

func (c *Chunk[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Chunk[T]

type CompositeHook

type CompositeHook []Hook

CompositeHook allows combining multiple hooks into a single hook implementation. Note: This is provided as a utility for combining hooks, not as a way to avoid implementing the full interface.

func (CompositeHook) OnAssistantChunk

func (c CompositeHook) OnAssistantChunk(ctx context.Context, ac messages.Message[messages.AssistantMessage])

func (CompositeHook) OnAssistantMessage

func (c CompositeHook) OnAssistantMessage(ctx context.Context, am messages.Message[messages.AssistantMessage])

func (CompositeHook) OnError

func (c CompositeHook) OnError(ctx context.Context, err error)

func (CompositeHook) OnToolCallChunk

func (c CompositeHook) OnToolCallChunk(ctx context.Context, tc messages.Message[messages.ToolCallMessage])

func (CompositeHook) OnToolCallMessage

func (c CompositeHook) OnToolCallMessage(ctx context.Context, tm messages.Message[messages.ToolCallMessage])

func (CompositeHook) OnToolCallResponse

func (c CompositeHook) OnToolCallResponse(ctx context.Context, tr messages.Message[messages.ToolResponse])

func (CompositeHook) OnUserPrompt

func (c CompositeHook) OnUserPrompt(ctx context.Context, up messages.Message[messages.UserMessage])

type Delim

type Delim struct {
	RunID  uuid.UUID `json:"run_id"`
	TurnID uuid.UUID `json:"turn_id"`
	Delim  string    `json:"delim"`
}

func (Delim) MarshalJSON

func (d Delim) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Delim

func (*Delim) UnmarshalJSON

func (d *Delim) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Delim

type Error

type Error struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Err       error           `json:"error"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Error) Error

func (e Error) Error() string

func (Error) MarshalJSON

func (e Error) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Error

func (*Error) UnmarshalJSON

func (e *Error) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Error

type Event

type Event interface {
	// contains filtered or unexported methods
}

func FromJSON

func FromJSON(jsonData []byte) (Event, error)

func FromStreamEvent

func FromStreamEvent(e provider.StreamEvent, sender string) Event

type Hook

type Hook interface {
	OnUserPrompt(context.Context, messages.Message[messages.UserMessage])

	OnAssistantChunk(context.Context, messages.Message[messages.AssistantMessage])

	OnToolCallChunk(context.Context, messages.Message[messages.ToolCallMessage])

	OnAssistantMessage(context.Context, messages.Message[messages.AssistantMessage])

	OnToolCallMessage(context.Context, messages.Message[messages.ToolCallMessage])

	OnToolCallResponse(context.Context, messages.Message[messages.ToolResponse])

	OnError(context.Context, error)
}

Hook defines the interface for handling all possible event types in the execution flow. This interface is deliberately designed without a base "no-op" implementation to ensure consumers make explicit decisions about handling each event type.

Design decisions:

1. All methods must be implemented: This is a conscious choice to ensure compile-time safety. When new event types are added, all implementations will need to be updated.
2. No provided no-op implementation: While it might be convenient to provide a NoOpHook, doing so would undermine the interface's primary benefit of forcing conscious decisions about event handling.
3. Complete coverage: The interface covers all possible event types to ensure no events can be accidentally missed in implementations.

Implementation guidelines:

• Implement all methods explicitly, even if some events don't require handling
• Consider logging or monitoring for events that aren't actively handled
• Be prepared for new methods to be added as the system evolves

Example implementation:

type MyHandler struct{}

func (h *MyHandler) OnUserPrompt(ctx context.Context, msg messages.Message[messages.UserMessage]) {
    // Explicit handling of user prompts
}
func (h *MyHandler) OnAssistantChunk(...) {
    // Explicit decision to not handle chunks
    log.Debug("ignoring assistant chunk")
}
// ... must implement all other methods

func NewCompositeHook

func NewCompositeHook(hooks ...Hook) Hook

type Request

type Request[T messages.Request] struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Message   T               `json:"message"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Request[T]) MarshalJSON

func (r Request[T]) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Request[T]

func (*Request[T]) UnmarshalJSON

func (r *Request[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Request[T]

type Response

type Response[T messages.Response] struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Response  T               `json:"response"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Response[T]) MarshalJSON

func (r Response[T]) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Response[T]

func (*Response[T]) UnmarshalJSON

func (r *Response[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Response[T]

type Result

type Result[T any] struct {
	RunID     uuid.UUID       `json:"run_id"`
	TurnID    uuid.UUID       `json:"turn_id"`
	Result    T               `json:"result"`
	Sender    string          `json:"sender,omitempty"`
	Timestamp strfmt.DateTime `json:"timestamp,omitempty"`
	Meta      gjson.Result    `json:"meta,omitempty"`
}

func (Result[T]) MarshalJSON

func (r Result[T]) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for Result[T]

func (*Result[T]) UnmarshalJSON

func (r *Result[T]) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for Result[T]