agents

Documentation

Overview

Package agents provides base infrastructure for AI trading agents

Package agents provides base infrastructure for AI trading agents

Index

• Constants
• type AgentConfig
• type AgentMetrics
• type BaseAgent
  • func NewBaseAgent(config *AgentConfig, log zerolog.Logger, metricsPort int) *BaseAgent
  • func (a *BaseAgent) AddInFlightOperation()
  • func (a *BaseAgent) CallMCPTool(ctx context.Context, serverName string, toolName string, ...) (*mcp.CallToolResult, error)
  • func (a *BaseAgent) CheckPausedAndSkip() bool
  • func (a *BaseAgent) DoneInFlightOperation()
  • func (a *BaseAgent) GetConfig() *AgentConfig
  • func (a *BaseAgent) GetNATSConnection() *nats.Conn
  • func (a *BaseAgent) GetName() string
  • func (a *BaseAgent) GetShutdownConfig() ShutdownConfig
  • func (a *BaseAgent) GetType() string
  • func (a *BaseAgent) GetVersion() string
  • func (a *BaseAgent) Initialize(ctx context.Context) error
  • func (a *BaseAgent) IsPaused() bool
  • func (a *BaseAgent) ListMCPTools(ctx context.Context, serverName string) (*mcp.ListToolsResult, error)
  • func (a *BaseAgent) Run(ctx context.Context) error
  • func (a *BaseAgent) SetShutdownConfig(config ShutdownConfig)
  • func (a *BaseAgent) SetStepFn(fn func(ctx context.Context) error)
  • func (a *BaseAgent) SetupControlSubscription(natsURL, controlTopic string) error
  • func (a *BaseAgent) Shutdown(ctx context.Context) error
  • func (a *BaseAgent) Step(ctx context.Context) error
• type HeartbeatConfig
  • func DefaultHeartbeatConfig() HeartbeatConfig
• type HeartbeatMessage
• type HeartbeatPublisher
  • func NewHeartbeatPublisher(agentName, agentType string, config HeartbeatConfig, log zerolog.Logger) *HeartbeatPublisher
  • func (h *HeartbeatPublisher) IsRunning() bool
  • func (h *HeartbeatPublisher) PublishNow()
  • func (h *HeartbeatPublisher) PublishWithStatus(status string)
  • func (h *HeartbeatPublisher) SetNATSConn(conn *nats.Conn)
  • func (h *HeartbeatPublisher) Start()
  • func (h *HeartbeatPublisher) Stop()
• type MCPServerConfig
• type ShutdownConfig
  • func DefaultShutdownConfig() ShutdownConfig

Constants

const (
	// DefaultShutdownTimeout is the default timeout for graceful agent shutdown operations
	DefaultShutdownTimeout = 10 * time.Second

	// DefaultNATSDrainTimeout is the timeout for draining NATS subscriptions during shutdown
	DefaultNATSDrainTimeout = 5 * time.Second
)

Types

type AgentConfig

type AgentConfig struct {
	// Identity
	Name    string `json:"name" yaml:"name"`
	Type    string `json:"type" yaml:"type"`
	Version string `json:"version" yaml:"version"`

	// MCP Server Connections (multiple servers supported)
	MCPServers []MCPServerConfig `json:"mcp_servers" yaml:"mcp_servers"`

	// Agent-specific configuration
	Config map[string]interface{} `json:"config" yaml:"config"`

	// Behavior
	StepInterval time.Duration `json:"step_interval" yaml:"step_interval"` // Time between decision cycles
	Enabled      bool          `json:"enabled" yaml:"enabled"`
}

AgentConfig holds configuration for an agent

type AgentMetrics

type AgentMetrics struct {
	StepsTotal      prometheus.Counter
	StepDuration    prometheus.Histogram
	MCPCallsTotal   prometheus.Counter
	MCPErrorsTotal  prometheus.Counter
	MCPCallDuration prometheus.Histogram
	AgentStatus     prometheus.Gauge
}

AgentMetrics holds Prometheus metrics for an agent

type BaseAgent

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

BaseAgent provides common functionality for all agents

func NewBaseAgent

func NewBaseAgent(config *AgentConfig, log zerolog.Logger, metricsPort int) *BaseAgent

NewBaseAgent creates a new base agent

func (*BaseAgent) AddInFlightOperation

func (a *BaseAgent) AddInFlightOperation()

AddInFlightOperation increments the WaitGroup counter for tracking in-flight operations Call this before starting an async operation that should be tracked for shutdown

func (*BaseAgent) CallMCPTool

func (a *BaseAgent) CallMCPTool(ctx context.Context, serverName string, toolName string, arguments map[string]interface{}) (*mcp.CallToolResult, error)

CallMCPTool calls a tool on a specific MCP server with a 60-second timeout

func (*BaseAgent) CheckPausedAndSkip

func (a *BaseAgent) CheckPausedAndSkip() bool

CheckPausedAndSkip checks if trading is paused and logs if skipping Returns true if paused (should skip), false if not paused (should continue)

func (*BaseAgent) DoneInFlightOperation

func (a *BaseAgent) DoneInFlightOperation()

DoneInFlightOperation decrements the WaitGroup counter Call this when an async operation completes (typically in a defer statement)

func (*BaseAgent) GetConfig

func (a *BaseAgent) GetConfig() *AgentConfig

GetConfig returns the agent's configuration

func (*BaseAgent) GetNATSConnection

func (a *BaseAgent) GetNATSConnection() *nats.Conn

GetNATSConnection returns the NATS connection for direct access if needed

func (*BaseAgent) GetName

func (a *BaseAgent) GetName() string

GetName returns the agent's name

func (*BaseAgent) GetShutdownConfig

func (a *BaseAgent) GetShutdownConfig() ShutdownConfig

GetShutdownConfig returns the current shutdown configuration

func (*BaseAgent) GetType

func (a *BaseAgent) GetType() string

GetType returns the agent's type

func (*BaseAgent) GetVersion

func (a *BaseAgent) GetVersion() string

GetVersion returns the agent's version

func (*BaseAgent) Initialize

func (a *BaseAgent) Initialize(ctx context.Context) error

Initialize sets up the agent and connects to MCP servers

func (*BaseAgent) IsPaused

func (a *BaseAgent) IsPaused() bool

IsPaused returns whether trading is currently paused

func (*BaseAgent) ListMCPTools

func (a *BaseAgent) ListMCPTools(ctx context.Context, serverName string) (*mcp.ListToolsResult, error)

ListMCPTools lists available tools from a specific MCP server

func (*BaseAgent) Run

func (a *BaseAgent) Run(ctx context.Context) error

Run starts the agent's main loop

func (*BaseAgent) SetShutdownConfig

func (a *BaseAgent) SetShutdownConfig(config ShutdownConfig)

SetShutdownConfig allows customizing the shutdown configuration

func (*BaseAgent) SetStepFn

func (a *BaseAgent) SetStepFn(fn func(ctx context.Context) error)

SetStepFn registers a step function that Run will call instead of BaseAgent.Step. Use this in subagents to ensure their Step implementation is called correctly.

func (*BaseAgent) SetupControlSubscription

func (a *BaseAgent) SetupControlSubscription(natsURL, controlTopic string) error

SetupControlSubscription connects to NATS and subscribes to orchestrator control events This should be called by agents that need to respond to pause/resume commands

func (*BaseAgent) Shutdown

func (a *BaseAgent) Shutdown(ctx context.Context) error

Shutdown gracefully stops the agent with proper cleanup and logging The shutdown process follows this order: 1. Cancel internal context to stop agent operations 2. Drain NATS subscriptions (allows in-flight messages to complete) 3. Close NATS connection 4. Close MCP sessions 5. Shutdown metrics server 6. Wait for all goroutines to complete

func (*BaseAgent) Step

func (a *BaseAgent) Step(ctx context.Context) error

Step performs a single decision cycle

type HeartbeatConfig

type HeartbeatConfig struct {
	// Interval between heartbeat messages (default: 30 seconds)
	Interval time.Duration
	// Topic is the NATS topic to publish heartbeats to (e.g., "cryptofunk.agent.heartbeat")
	Topic string
}

HeartbeatConfig holds configuration for heartbeat publishing

func DefaultHeartbeatConfig

func DefaultHeartbeatConfig() HeartbeatConfig

DefaultHeartbeatConfig returns the default heartbeat configuration

type HeartbeatMessage

type HeartbeatMessage struct {
	AgentName string    `json:"agent_name"`
	AgentType string    `json:"agent_type"`
	Timestamp time.Time `json:"timestamp"`
	Status    string    `json:"status"`
}

HeartbeatMessage represents a heartbeat message published by agents

type HeartbeatPublisher

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

HeartbeatPublisher handles periodic heartbeat publishing for agents

func NewHeartbeatPublisher

func NewHeartbeatPublisher(agentName, agentType string, config HeartbeatConfig, log zerolog.Logger) *HeartbeatPublisher

NewHeartbeatPublisher creates a new heartbeat publisher The natsConn can be nil initially and set later with SetNATSConn

func (*HeartbeatPublisher) IsRunning

func (h *HeartbeatPublisher) IsRunning() bool

IsRunning returns whether the heartbeat publisher is currently running

func (*HeartbeatPublisher) PublishNow

func (h *HeartbeatPublisher) PublishNow()

PublishNow immediately publishes a heartbeat message (useful for status updates)

func (*HeartbeatPublisher) PublishWithStatus

func (h *HeartbeatPublisher) PublishWithStatus(status string)

PublishWithStatus publishes a heartbeat with a custom status

func (*HeartbeatPublisher) SetNATSConn

func (h *HeartbeatPublisher) SetNATSConn(conn *nats.Conn)

SetNATSConn sets the NATS connection for the heartbeat publisher This allows setting the connection after initialization

func (*HeartbeatPublisher) Start

func (h *HeartbeatPublisher) Start()

Start begins publishing heartbeat messages at the configured interval The goroutine will publish immediately on start, then at the configured interval

func (*HeartbeatPublisher) Stop

func (h *HeartbeatPublisher) Stop()

Stop stops the heartbeat publisher

type MCPServerConfig

type MCPServerConfig struct {
	Name string `json:"name" yaml:"name"` // Server identifier (e.g., "coingecko", "technical_indicators")
	URL  string `json:"url" yaml:"url"`   // HTTP URL of the MCP server (e.g., "http://localhost:8093/mcp")
	// Type controls the client transport: "http" uses Streamable HTTP (default, for our servers);
	// "sse" uses the legacy SSE transport (for external providers like CoinGecko).
	Type string `json:"type" yaml:"type"`
	// Optional, if true, connection failures are logged as warnings and the agent continues without this server.
	Optional bool `json:"optional" yaml:"optional"`
}

MCPServerConfig holds configuration for a single MCP server

type ShutdownConfig

type ShutdownConfig struct {
	// Timeout is the maximum time to wait for graceful shutdown (default: 10 seconds)
	Timeout time.Duration
	// NATSDrainTimeout is the timeout for draining NATS subscriptions (default: 5 seconds)
	NATSDrainTimeout time.Duration
}

ShutdownConfig holds configuration for graceful shutdown behavior

func DefaultShutdownConfig

func DefaultShutdownConfig() ShutdownConfig

DefaultShutdownConfig returns the default shutdown configuration