refactor(ai): Replace mcp package with tools package for direct tool execution

This refactoring removes the MCP (Model Context Protocol) server layer and
converts AI tools to be called directly by the chat service.

Key changes:
- Rename package from internal/ai/mcp to internal/ai/tools
- Remove server.go - tools no longer exposed via MCP server
- Tools are now called directly by the chat service via ExecuteTool()

New tools added:
- Kubernetes: clusters, nodes, pods, deployments (4 tools)
- PMG: mail gateway status, mail stats, queues, spam stats (4 tools)
- Infrastructure: snapshots, PBS jobs, backup tasks, network stats,
  disk I/O, cluster status, swarm, services, tasks, recent tasks,
  physical disks, RAID status, host Ceph, resource disks (14 tools)
- Patrol: connection health, resolved alerts (2 tools)

Test coverage:
- Added comprehensive test files for adapters, infrastructure,
  patrol, profiles, and query tools

Total tools: 50 (was ~25)
This commit is contained in:
rcourtman
2026-01-19 19:17:24 +00:00
parent 5ff4f97a0d
commit 17ca31a557
21 changed files with 6492 additions and 1539 deletions
-620
View File
@@ -1,620 +0,0 @@
package mcp
import "time"
// MetricPoint represents a single metric data point
type MetricPoint struct {
Timestamp time.Time `json:"timestamp"`
CPU float64 `json:"cpu"`
Memory float64 `json:"memory"`
Disk float64 `json:"disk,omitempty"`
}
// ResourceMetricsSummary summarizes metrics for a resource over a period
type ResourceMetricsSummary struct {
ResourceID string `json:"resource_id"`
ResourceName string `json:"resource_name"`
ResourceType string `json:"resource_type"`
AvgCPU float64 `json:"avg_cpu"`
MaxCPU float64 `json:"max_cpu"`
AvgMemory float64 `json:"avg_memory"`
MaxMemory float64 `json:"max_memory"`
AvgDisk float64 `json:"avg_disk,omitempty"`
MaxDisk float64 `json:"max_disk,omitempty"`
Trend string `json:"trend"` // "stable", "growing", "declining"
}
// MetricBaseline represents learned normal behavior for a metric
type MetricBaseline struct {
Mean float64 `json:"mean"`
StdDev float64 `json:"std_dev"`
Min float64 `json:"min"`
Max float64 `json:"max"`
}
// Pattern represents a detected operational pattern
type Pattern struct {
ResourceID string `json:"resource_id"`
ResourceName string `json:"resource_name"`
PatternType string `json:"pattern_type"` // "recurring_spike", "gradual_growth", "weekly_cycle"
Description string `json:"description"`
Confidence float64 `json:"confidence"`
LastSeen time.Time `json:"last_seen"`
}
// Prediction represents a predicted future issue
type Prediction struct {
ResourceID string `json:"resource_id"`
ResourceName string `json:"resource_name"`
IssueType string `json:"issue_type"` // "disk_full", "memory_exhaustion", etc.
PredictedTime time.Time `json:"predicted_time"`
Confidence float64 `json:"confidence"`
Recommendation string `json:"recommendation"`
}
// ActiveAlert represents an active alert
type ActiveAlert struct {
ID string `json:"id"`
ResourceID string `json:"resource_id"`
ResourceName string `json:"resource_name"`
Type string `json:"type"` // "cpu", "memory", "disk", "offline"
Severity string `json:"severity"`
Value float64 `json:"value"`
Threshold float64 `json:"threshold"`
StartTime time.Time `json:"start_time"`
Message string `json:"message"`
}
// Finding represents a patrol finding
type Finding struct {
ID string `json:"id"`
Key string `json:"key"`
Severity string `json:"severity"`
Category string `json:"category"`
ResourceID string `json:"resource_id"`
ResourceName string `json:"resource_name"`
ResourceType string `json:"resource_type"`
Title string `json:"title"`
Description string `json:"description"`
Recommendation string `json:"recommendation"`
Evidence string `json:"evidence"`
DetectedAt time.Time `json:"detected_at"`
LastSeenAt time.Time `json:"last_seen_at"`
TimesRaised int `json:"times_raised"`
}
// GuestInfo represents resolved guest information
type GuestInfo struct {
VMID int
Name string
Node string
Type string // "vm" or "lxc"
Status string
Instance string
}
// ========== JSON Response Types ==========
// CapabilitiesResponse is returned by pulse_get_capabilities
// AgentInfo represents a connected execution agent
type AgentInfo struct {
Hostname string `json:"hostname"`
Version string `json:"version,omitempty"`
Platform string `json:"platform,omitempty"`
ConnectedAt string `json:"connected_at,omitempty"`
}
type CapabilitiesResponse struct {
ControlLevel string `json:"control_level"`
Features FeatureFlags `json:"features"`
ProtectedGuests []string `json:"protected_guests,omitempty"`
ConnectedAgents int `json:"connected_agents"`
Agents []AgentInfo `json:"agents,omitempty"` // List of connected agents with hostnames
Version string `json:"version"`
}
// FeatureFlags indicates which features are available
type FeatureFlags struct {
MetricsHistory bool `json:"metrics_history"`
Baselines bool `json:"baselines"`
Patterns bool `json:"patterns"`
Alerts bool `json:"alerts"`
Findings bool `json:"findings"`
Backups bool `json:"backups"`
Storage bool `json:"storage"`
DiskHealth bool `json:"disk_health"`
AgentProfiles bool `json:"agent_profiles"`
Control bool `json:"control"`
}
// InfrastructureResponse is returned by pulse_list_infrastructure
type InfrastructureResponse struct {
Nodes []NodeSummary `json:"nodes,omitempty"`
VMs []VMSummary `json:"vms,omitempty"`
Containers []ContainerSummary `json:"containers,omitempty"`
DockerHosts []DockerHostSummary `json:"docker_hosts,omitempty"`
Total TotalCounts `json:"total"`
Pagination *PaginationInfo `json:"pagination,omitempty"`
}
// NodeSummary is a summarized node for list responses
type NodeSummary struct {
Name string `json:"name"`
Status string `json:"status"`
ID string `json:"id,omitempty"`
AgentConnected bool `json:"agent_connected"` // True if an execution agent is connected for this node
}
// VMSummary is a summarized VM for list responses
type VMSummary struct {
VMID int `json:"vmid"`
Name string `json:"name"`
Status string `json:"status"`
Node string `json:"node"`
CPU float64 `json:"cpu_percent,omitempty"`
Memory float64 `json:"memory_percent,omitempty"`
}
// ContainerSummary is a summarized LXC container for list responses
type ContainerSummary struct {
VMID int `json:"vmid"`
Name string `json:"name"`
Status string `json:"status"`
Node string `json:"node"`
CPU float64 `json:"cpu_percent,omitempty"`
Memory float64 `json:"memory_percent,omitempty"`
}
// DockerHostSummary is a summarized Docker host for list responses
type DockerHostSummary struct {
ID string `json:"id"`
Hostname string `json:"hostname"`
DisplayName string `json:"display_name,omitempty"`
ContainerCount int `json:"container_count"`
AgentConnected bool `json:"agent_connected"` // True if an execution agent is connected for this host
Containers []DockerContainerSummary `json:"containers,omitempty"`
}
// DockerContainerSummary is a summarized Docker container
type DockerContainerSummary struct {
ID string `json:"id"`
Name string `json:"name"`
State string `json:"state"`
Image string `json:"image"`
Health string `json:"health,omitempty"`
}
// TotalCounts for infrastructure response
type TotalCounts struct {
Nodes int `json:"nodes"`
VMs int `json:"vms"`
Containers int `json:"containers"`
DockerHosts int `json:"docker_hosts"`
}
// PaginationInfo describes pagination state
type PaginationInfo struct {
Total int `json:"total"`
Limit int `json:"limit"`
Offset int `json:"offset"`
}
// ========== Topology Response Types (Hierarchical View) ==========
// TopologyResponse provides a fully hierarchical view of infrastructure
// This is the recommended tool for understanding infrastructure relationships
type TopologyResponse struct {
Proxmox ProxmoxTopology `json:"proxmox"`
Docker DockerTopology `json:"docker"`
Summary TopologySummary `json:"summary"`
}
// ProxmoxTopology shows Proxmox nodes with their nested VMs and containers
type ProxmoxTopology struct {
Nodes []ProxmoxNodeTopology `json:"nodes"`
}
// ProxmoxNodeTopology represents a Proxmox node with its guests
type ProxmoxNodeTopology struct {
Name string `json:"name"`
ID string `json:"id,omitempty"`
Status string `json:"status"`
AgentConnected bool `json:"agent_connected"`
CanExecute bool `json:"can_execute"` // True if commands can be executed on this node
VMs []TopologyVM `json:"vms,omitempty"`
Containers []TopologyLXC `json:"containers,omitempty"`
VMCount int `json:"vm_count"`
ContainerCount int `json:"container_count"`
}
// TopologyVM represents a VM in the topology
type TopologyVM struct {
VMID int `json:"vmid"`
Name string `json:"name"`
Status string `json:"status"`
CPU float64 `json:"cpu_percent,omitempty"`
Memory float64 `json:"memory_percent,omitempty"`
OS string `json:"os,omitempty"`
Tags []string `json:"tags,omitempty"`
}
// TopologyLXC represents an LXC container in the topology
type TopologyLXC struct {
VMID int `json:"vmid"`
Name string `json:"name"`
Status string `json:"status"`
CPU float64 `json:"cpu_percent,omitempty"`
Memory float64 `json:"memory_percent,omitempty"`
OS string `json:"os,omitempty"`
Tags []string `json:"tags,omitempty"`
HasDocker bool `json:"has_docker,omitempty"` // True if Docker is installed inside this container
}
// DockerTopology shows Docker hosts with their nested containers
type DockerTopology struct {
Hosts []DockerHostTopology `json:"hosts"`
}
// DockerHostTopology represents a Docker host with its containers
type DockerHostTopology struct {
Hostname string `json:"hostname"`
DisplayName string `json:"display_name,omitempty"`
AgentConnected bool `json:"agent_connected"`
CanExecute bool `json:"can_execute"` // True if commands can be executed on this host
Containers []DockerContainerSummary `json:"containers,omitempty"`
ContainerCount int `json:"container_count"`
RunningCount int `json:"running_count"`
}
// TopologySummary provides aggregate counts and status
type TopologySummary struct {
TotalNodes int `json:"total_nodes"`
TotalVMs int `json:"total_vms"`
TotalLXCContainers int `json:"total_lxc_containers"`
TotalDockerHosts int `json:"total_docker_hosts"`
TotalDockerContainers int `json:"total_docker_containers"`
NodesWithAgents int `json:"nodes_with_agents"`
DockerHostsWithAgents int `json:"docker_hosts_with_agents"`
RunningVMs int `json:"running_vms"`
RunningLXC int `json:"running_lxc"`
RunningDocker int `json:"running_docker"`
}
// ResourceResponse is returned by pulse_get_resource
type ResourceResponse struct {
Type string `json:"type"` // "vm", "container", "docker"
ID string `json:"id"`
Name string `json:"name"`
Status string `json:"status"`
Node string `json:"node,omitempty"`
Host string `json:"host,omitempty"`
CPU ResourceCPU `json:"cpu"`
Memory ResourceMemory `json:"memory"`
Disk *ResourceDisk `json:"disk,omitempty"`
OS string `json:"os,omitempty"`
Tags []string `json:"tags,omitempty"`
Networks []NetworkInfo `json:"networks,omitempty"`
Ports []PortInfo `json:"ports,omitempty"`
Mounts []MountInfo `json:"mounts,omitempty"`
Labels map[string]string `json:"labels,omitempty"`
LastBackup *time.Time `json:"last_backup,omitempty"`
Image string `json:"image,omitempty"`
Health string `json:"health,omitempty"`
RestartCount int `json:"restart_count,omitempty"`
UpdateAvailable bool `json:"update_available,omitempty"`
}
// ResourceCPU describes CPU usage
type ResourceCPU struct {
Percent float64 `json:"percent"`
Cores int `json:"cores"`
}
// ResourceMemory describes memory usage
type ResourceMemory struct {
Percent float64 `json:"percent"`
UsedGB float64 `json:"used_gb"`
TotalGB float64 `json:"total_gb"`
}
// ResourceDisk describes disk usage
type ResourceDisk struct {
Percent float64 `json:"percent"`
UsedGB float64 `json:"used_gb"`
TotalGB float64 `json:"total_gb"`
}
// NetworkInfo describes a network interface
type NetworkInfo struct {
Name string `json:"name"`
Addresses []string `json:"addresses"`
}
// PortInfo describes a port mapping
type PortInfo struct {
Private int `json:"private"`
Public int `json:"public,omitempty"`
Protocol string `json:"protocol"`
IP string `json:"ip,omitempty"`
}
// MountInfo describes a volume mount
type MountInfo struct {
Source string `json:"source"`
Destination string `json:"destination"`
ReadWrite bool `json:"rw"`
}
// URLFetchResponse is returned by pulse_get_url_content
type URLFetchResponse struct {
URL string `json:"url"`
StatusCode int `json:"status_code"`
Headers map[string]string `json:"headers"`
Body string `json:"body"`
Error string `json:"error,omitempty"`
}
// AlertsResponse is returned by pulse_list_alerts
type AlertsResponse struct {
Alerts []ActiveAlert `json:"alerts"`
Count int `json:"count"`
Pagination *PaginationInfo `json:"pagination,omitempty"`
}
// FindingsResponse is returned by pulse_list_findings
type FindingsResponse struct {
Active []Finding `json:"active"`
Dismissed []Finding `json:"dismissed,omitempty"`
Counts FindingCounts `json:"counts"`
Pagination *PaginationInfo `json:"pagination,omitempty"`
}
// FindingCounts for findings response
type FindingCounts struct {
Active int `json:"active"`
Dismissed int `json:"dismissed"`
}
// MetricsResponse is returned by pulse_get_metrics
type MetricsResponse struct {
ResourceID string `json:"resource_id,omitempty"`
Period string `json:"period"`
Points []MetricPoint `json:"points,omitempty"`
Summary map[string]ResourceMetricsSummary `json:"summary,omitempty"`
}
// BaselinesResponse is returned by pulse_get_baselines
type BaselinesResponse struct {
ResourceID string `json:"resource_id,omitempty"`
Baselines map[string]map[string]*MetricBaseline `json:"baselines"` // resourceID -> metric -> baseline
}
// PatternsResponse is returned by pulse_get_patterns
type PatternsResponse struct {
Patterns []Pattern `json:"patterns"`
Predictions []Prediction `json:"predictions"`
}
// BackupsResponse is returned by pulse_list_backups
type BackupsResponse struct {
PBS []PBSBackupSummary `json:"pbs,omitempty"`
PVE []PVEBackupSummary `json:"pve,omitempty"`
PBSServers []PBSServerSummary `json:"pbs_servers,omitempty"`
RecentTasks []BackupTaskSummary `json:"recent_tasks,omitempty"`
Pagination *PaginationInfo `json:"pagination,omitempty"`
}
// PBSBackupSummary is a summarized PBS backup
type PBSBackupSummary struct {
VMID string `json:"vmid"`
BackupType string `json:"backup_type"`
BackupTime time.Time `json:"backup_time"`
Instance string `json:"instance"`
Datastore string `json:"datastore"`
SizeGB float64 `json:"size_gb"`
Verified bool `json:"verified"`
Protected bool `json:"protected"`
}
// PVEBackupSummary is a summarized PVE backup
type PVEBackupSummary struct {
VMID int `json:"vmid"`
BackupTime time.Time `json:"backup_time"`
SizeGB float64 `json:"size_gb"`
Storage string `json:"storage"`
}
// PBSServerSummary is a summarized PBS server
type PBSServerSummary struct {
Name string `json:"name"`
Host string `json:"host"`
Status string `json:"status"`
Datastores []DatastoreSummary `json:"datastores"`
}
// DatastoreSummary is a summarized datastore
type DatastoreSummary struct {
Name string `json:"name"`
UsagePercent float64 `json:"usage_percent"`
FreeGB float64 `json:"free_gb"`
}
// BackupTaskSummary is a summarized backup task
type BackupTaskSummary struct {
VMID int `json:"vmid"`
Node string `json:"node"`
Status string `json:"status"`
StartTime time.Time `json:"start_time"`
}
// StorageResponse is returned by pulse_list_storage
type StorageResponse struct {
Pools []StoragePoolSummary `json:"pools,omitempty"`
CephClusters []CephClusterSummary `json:"ceph_clusters,omitempty"`
Pagination *PaginationInfo `json:"pagination,omitempty"`
}
// StoragePoolSummary is a summarized storage pool
type StoragePoolSummary struct {
ID string `json:"id"`
Name string `json:"name"`
Type string `json:"type"`
Status string `json:"status"`
UsagePercent float64 `json:"usage_percent"`
UsedGB float64 `json:"used_gb"`
TotalGB float64 `json:"total_gb"`
FreeGB float64 `json:"free_gb"`
Content string `json:"content"`
Shared bool `json:"shared"`
ZFS *ZFSPoolSummary `json:"zfs,omitempty"`
}
// ZFSPoolSummary is a summarized ZFS pool
type ZFSPoolSummary struct {
Name string `json:"name"`
State string `json:"state"`
ReadErrors int64 `json:"read_errors"`
WriteErrors int64 `json:"write_errors"`
ChecksumErrors int64 `json:"checksum_errors"`
Scan string `json:"scan,omitempty"`
}
// CephClusterSummary is a summarized Ceph cluster
type CephClusterSummary struct {
Name string `json:"name"`
Health string `json:"health"`
HealthMessage string `json:"health_message,omitempty"`
UsagePercent float64 `json:"usage_percent"`
UsedTB float64 `json:"used_tb"`
TotalTB float64 `json:"total_tb"`
NumOSDs int `json:"num_osds"`
NumOSDsUp int `json:"num_osds_up"`
NumOSDsIn int `json:"num_osds_in"`
NumMons int `json:"num_mons"`
NumMgrs int `json:"num_mgrs"`
}
// DiskHealthResponse is returned by pulse_get_disk_health
type DiskHealthResponse struct {
Hosts []HostDiskHealth `json:"hosts"`
}
// HostDiskHealth is disk health for a single host
type HostDiskHealth struct {
Hostname string `json:"hostname"`
SMART []SMARTDiskSummary `json:"smart,omitempty"`
RAID []RAIDArraySummary `json:"raid,omitempty"`
Ceph *CephStatusSummary `json:"ceph,omitempty"`
}
// SMARTDiskSummary is a summarized SMART disk
type SMARTDiskSummary struct {
Device string `json:"device"`
Model string `json:"model"`
Health string `json:"health"`
Temperature int `json:"temperature,omitempty"`
}
// RAIDArraySummary is a summarized RAID array
type RAIDArraySummary struct {
Device string `json:"device"`
Level string `json:"level"`
State string `json:"state"`
ActiveDevices int `json:"active_devices"`
WorkingDevices int `json:"working_devices"`
FailedDevices int `json:"failed_devices"`
SpareDevices int `json:"spare_devices"`
RebuildPercent float64 `json:"rebuild_percent,omitempty"`
}
// CephStatusSummary is a summarized Ceph status from agent
type CephStatusSummary struct {
Health string `json:"health"`
NumOSDs int `json:"num_osds"`
NumOSDsUp int `json:"num_osds_up"`
NumOSDsIn int `json:"num_osds_in"`
NumPGs int `json:"num_pgs"`
UsagePercent float64 `json:"usage_percent"`
}
// AgentScopeResponse is returned by pulse_get_agent_scope
type AgentScopeResponse struct {
AgentID string `json:"agent_id"`
AgentLabel string `json:"agent_label"`
ProfileID string `json:"profile_id,omitempty"`
ProfileName string `json:"profile_name,omitempty"`
ProfileVersion int `json:"profile_version,omitempty"`
Settings map[string]interface{} `json:"settings,omitempty"`
ObservedModules []string `json:"observed_modules,omitempty"`
CommandsEnabled *bool `json:"commands_enabled,omitempty"`
}
// CommandResponse is returned by control tools
type CommandResponse struct {
Success bool `json:"success"`
Output string `json:"output,omitempty"`
ExitCode int `json:"exit_code,omitempty"`
Error string `json:"error,omitempty"`
}
// ControlActionResponse is returned by pulse_control_guest and pulse_control_docker
type ControlActionResponse struct {
Success bool `json:"success"`
Action string `json:"action"`
Target string `json:"target"`
TargetType string `json:"target_type"` // "vm", "lxc", "docker"
Output string `json:"output,omitempty"`
Error string `json:"error,omitempty"`
}
// ========== Docker Updates Types ==========
// ContainerUpdateInfo represents a container with an available update
type ContainerUpdateInfo struct {
HostID string `json:"host_id"`
HostName string `json:"host_name"`
ContainerID string `json:"container_id"`
ContainerName string `json:"container_name"`
Image string `json:"image"`
CurrentDigest string `json:"current_digest,omitempty"`
LatestDigest string `json:"latest_digest,omitempty"`
UpdateAvailable bool `json:"update_available"`
LastChecked int64 `json:"last_checked,omitempty"`
Error string `json:"error,omitempty"`
}
// DockerCommandStatus represents the status of a queued Docker command
type DockerCommandStatus struct {
ID string `json:"id"`
Type string `json:"type"`
Status string `json:"status"`
Message string `json:"message"`
}
// DockerUpdatesResponse is returned by pulse_list_docker_updates
type DockerUpdatesResponse struct {
Updates []ContainerUpdateInfo `json:"updates"`
Total int `json:"total"`
HostID string `json:"host_id,omitempty"`
}
// DockerCheckUpdatesResponse is returned by pulse_check_docker_updates
type DockerCheckUpdatesResponse struct {
Success bool `json:"success"`
HostID string `json:"host_id"`
HostName string `json:"host_name"`
CommandID string `json:"command_id"`
Message string `json:"message"`
Command DockerCommandStatus `json:"command"`
}
// DockerUpdateContainerResponse is returned by pulse_update_docker_container
type DockerUpdateContainerResponse struct {
Success bool `json:"success"`
HostID string `json:"host_id"`
ContainerID string `json:"container_id"`
ContainerName string `json:"container_name"`
CommandID string `json:"command_id"`
Message string `json:"message"`
Command DockerCommandStatus `json:"command"`
}
-258
View File
@@ -1,258 +0,0 @@
package mcp
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"sync"
"github.com/rs/zerolog/log"
)
const (
ProtocolVersion = "2024-11-05"
ServerName = "pulse-mcp"
ServerVersion = "1.0.0"
)
// ToolExecutor executes tools on behalf of the MCP server
type ToolExecutor interface {
ExecuteTool(ctx context.Context, name string, args map[string]interface{}) (CallToolResult, error)
ListTools() []Tool
}
// Server implements an MCP server over HTTP
type Server struct {
mu sync.RWMutex
executor ToolExecutor
addr string
server *http.Server
}
// NewServer creates a new MCP server
func NewServer(addr string, executor ToolExecutor) *Server {
return &Server{
addr: addr,
executor: executor,
}
}
// Start starts the MCP server
func (s *Server) Start() error {
mux := http.NewServeMux()
mux.HandleFunc("/", s.handleRequest)
mux.HandleFunc("/health", s.handleHealth)
s.server = &http.Server{
Addr: s.addr,
Handler: mux,
}
log.Info().Str("addr", s.addr).Msg("Starting MCP server")
return s.server.ListenAndServe()
}
// Stop stops the MCP server
func (s *Server) Stop(ctx context.Context) error {
if s.server != nil {
return s.server.Shutdown(ctx)
}
return nil
}
// Addr returns the server address
func (s *Server) Addr() string {
return s.addr
}
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte(`{"status": "ok"}`))
}
func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
body, err := io.ReadAll(r.Body)
if err != nil {
s.writeError(w, nil, ErrParse, "Failed to read request body")
return
}
defer r.Body.Close()
var req Request
if err := json.Unmarshal(body, &req); err != nil {
s.writeError(w, nil, ErrParse, "Failed to parse JSON-RPC request")
return
}
if req.JSONRPC != "2.0" {
s.writeError(w, req.ID, ErrInvalidRequest, "Invalid JSON-RPC version")
return
}
log.Debug().
Str("method", req.Method).
Interface("id", req.ID).
Msg("MCP request received")
result, mcpErr := s.handleMethod(r.Context(), req)
if mcpErr != nil {
s.writeErrorResponse(w, req.ID, mcpErr)
return
}
s.writeResult(w, req.ID, result)
}
func (s *Server) handleMethod(ctx context.Context, req Request) (interface{}, *Error) {
switch req.Method {
case "initialize":
return s.handleInitialize(req.Params)
case "initialized":
return nil, nil
case "tools/list":
return s.handleListTools()
case "tools/call":
return s.handleCallTool(ctx, req.Params)
case "resources/list":
// Return empty list - resources not implemented
return &ListResourcesResult{Resources: []Resource{}}, nil
case "prompts/list":
// Return empty list - prompts not implemented
return &ListPromptsResult{Prompts: []Prompt{}}, nil
case "ping":
return map[string]interface{}{}, nil
default:
return nil, &Error{
Code: ErrMethodNotFound,
Message: fmt.Sprintf("Method not found: %s", req.Method),
}
}
}
func (s *Server) handleInitialize(params json.RawMessage) (*InitializeResult, *Error) {
var initParams InitializeParams
if params != nil {
if err := json.Unmarshal(params, &initParams); err != nil {
return nil, &Error{
Code: ErrInvalidParams,
Message: "Failed to parse initialize params",
}
}
}
log.Info().
Str("client", initParams.ClientInfo.Name).
Str("clientVersion", initParams.ClientInfo.Version).
Str("protocolVersion", initParams.ProtocolVersion).
Msg("MCP client connected")
return &InitializeResult{
ProtocolVersion: ProtocolVersion,
Capabilities: Capabilities{
Tools: &ToolsCapability{
ListChanged: false,
},
// Resources and Prompts not advertised - not implemented
},
ServerInfo: ServerInfo{
Name: ServerName,
Version: ServerVersion,
},
}, nil
}
func (s *Server) handleListTools() (*ListToolsResult, *Error) {
s.mu.RLock()
executor := s.executor
s.mu.RUnlock()
if executor == nil {
return &ListToolsResult{Tools: []Tool{}}, nil
}
tools := executor.ListTools()
return &ListToolsResult{Tools: tools}, nil
}
func (s *Server) handleCallTool(ctx context.Context, params json.RawMessage) (*CallToolResult, *Error) {
var callParams CallToolParams
if err := json.Unmarshal(params, &callParams); err != nil {
return nil, &Error{
Code: ErrInvalidParams,
Message: "Failed to parse tool call params",
}
}
s.mu.RLock()
executor := s.executor
s.mu.RUnlock()
if executor == nil {
return nil, &Error{
Code: ErrInternal,
Message: "No tool executor configured",
}
}
log.Debug().
Str("tool", callParams.Name).
Interface("args", callParams.Arguments).
Msg("Executing tool")
result, err := executor.ExecuteTool(ctx, callParams.Name, callParams.Arguments)
if err != nil {
log.Error().Err(err).Str("tool", callParams.Name).Msg("Tool execution failed")
return &CallToolResult{
Content: []Content{NewTextContent(err.Error())},
IsError: true,
}, nil
}
return &result, nil
}
func (s *Server) writeResult(w http.ResponseWriter, id interface{}, result interface{}) {
resultJSON, err := json.Marshal(result)
if err != nil {
s.writeError(w, id, ErrInternal, "Failed to marshal result")
return
}
resp := Response{
JSONRPC: "2.0",
ID: id,
Result: resultJSON,
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
}
func (s *Server) writeError(w http.ResponseWriter, id interface{}, code int, message string) {
s.writeErrorResponse(w, id, &Error{Code: code, Message: message})
}
func (s *Server) writeErrorResponse(w http.ResponseWriter, id interface{}, err *Error) {
resp := Response{
JSONRPC: "2.0",
ID: id,
Error: err,
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
}
// SetExecutor updates the tool executor
func (s *Server) SetExecutor(executor ToolExecutor) {
s.mu.Lock()
defer s.mu.Unlock()
s.executor = executor
}
-615
View File
@@ -1,615 +0,0 @@
package mcp
import (
"context"
"encoding/json"
"fmt"
)
// registerInfrastructureTools registers infrastructure context tools (backup, storage, disk health)
func (e *PulseToolExecutor) registerInfrastructureTools() {
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_list_backups",
Description: "List backup status for VMs and containers. Shows last backup times, backup jobs, and identifies resources without recent backups.",
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"resource_id": {
Type: "string",
Description: "Optional: filter by specific VM or container ID",
},
"limit": {
Type: "integer",
Description: "Maximum number of results (default: 100)",
},
"offset": {
Type: "integer",
Description: "Number of results to skip",
},
},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeListBackups(ctx, args)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_list_storage",
Description: "List storage pool information including usage, ZFS pool health, and Ceph cluster status.",
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"storage_id": {
Type: "string",
Description: "Optional: specific storage ID for detailed info",
},
"limit": {
Type: "integer",
Description: "Maximum number of results (default: 100)",
},
"offset": {
Type: "integer",
Description: "Number of results to skip",
},
},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeListStorage(ctx, args)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_disk_health",
Description: "Get disk health information including SMART data, RAID array status, and Ceph cluster health from host agents.",
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetDiskHealth(ctx, args)
},
})
// Docker Updates Tools
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_list_docker_updates",
Description: `List Docker containers with pending image updates.
Returns: JSON with containers that have newer images available in their registry, including image names, current/latest digests, and any check errors.
Use when: User asks about available Docker updates, which containers need updating, or wants to see update status across Docker hosts.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"host": {
Type: "string",
Description: "Optional: filter by Docker host name or ID",
},
},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeListDockerUpdates(ctx, args)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_check_docker_updates",
Description: `Trigger an update check for Docker containers on a host.
The Docker agent will check registries for newer images and report back. Results appear in pulse_list_docker_updates after the next agent report cycle (~30 seconds).
Use when: User wants to refresh/rescan for available Docker updates on a specific host.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"host": {
Type: "string",
Description: "Docker host name or ID to check for updates",
},
},
Required: []string{"host"},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeCheckDockerUpdates(ctx, args)
},
RequireControl: true,
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_update_docker_container",
Description: `Update a Docker container to its latest image.
This pulls the latest image, stops the container, recreates it with the same configuration, and starts it. The old container is kept as a backup and automatically cleaned up after 5 minutes if the new container is stable.
Use when: User explicitly asks to update a specific Docker container to its latest version.
Do NOT use for: Checking what updates are available (use pulse_list_docker_updates), or just restarting a container (use pulse_control_docker).`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"container": {
Type: "string",
Description: "Container name or ID to update",
},
"host": {
Type: "string",
Description: "Docker host name or ID where the container runs",
},
},
Required: []string{"container", "host"},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeUpdateDockerContainer(ctx, args)
},
RequireControl: true,
})
}
func (e *PulseToolExecutor) executeListBackups(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
resourceID, _ := args["resource_id"].(string)
limit := intArg(args, "limit", 100)
offset := intArg(args, "offset", 0)
if e.backupProvider == nil {
return NewTextResult("Backup information not available."), nil
}
backups := e.backupProvider.GetBackups()
pbsInstances := e.backupProvider.GetPBSInstances()
response := BackupsResponse{}
// PBS Backups
count := 0
for _, b := range backups.PBS {
if resourceID != "" && b.VMID != resourceID {
continue
}
if count < offset {
count++
continue
}
if len(response.PBS) >= limit {
break
}
response.PBS = append(response.PBS, PBSBackupSummary{
VMID: b.VMID,
BackupType: b.BackupType,
BackupTime: b.BackupTime,
Instance: b.Instance,
Datastore: b.Datastore,
SizeGB: float64(b.Size) / (1024 * 1024 * 1024),
Verified: b.Verified,
Protected: b.Protected,
})
count++
}
// PVE Backups
count = 0
for _, b := range backups.PVE.StorageBackups {
if resourceID != "" && string(rune(b.VMID)) != resourceID {
continue
}
if count < offset {
count++
continue
}
if len(response.PVE) >= limit {
break
}
response.PVE = append(response.PVE, PVEBackupSummary{
VMID: b.VMID,
BackupTime: b.Time,
SizeGB: float64(b.Size) / (1024 * 1024 * 1024),
Storage: b.Storage,
})
count++
}
// PBS Servers
for _, pbs := range pbsInstances {
server := PBSServerSummary{
Name: pbs.Name,
Host: pbs.Host,
Status: pbs.Status,
}
for _, ds := range pbs.Datastores {
server.Datastores = append(server.Datastores, DatastoreSummary{
Name: ds.Name,
UsagePercent: ds.Usage * 100,
FreeGB: float64(ds.Free) / (1024 * 1024 * 1024),
})
}
response.PBSServers = append(response.PBSServers, server)
}
// Recent tasks
for _, t := range backups.PVE.BackupTasks {
if len(response.RecentTasks) >= 20 {
break
}
response.RecentTasks = append(response.RecentTasks, BackupTaskSummary{
VMID: t.VMID,
Node: t.Node,
Status: t.Status,
StartTime: t.StartTime,
})
}
// Ensure non-nil slices
if response.PBS == nil {
response.PBS = []PBSBackupSummary{}
}
if response.PVE == nil {
response.PVE = []PVEBackupSummary{}
}
if response.PBSServers == nil {
response.PBSServers = []PBSServerSummary{}
}
if response.RecentTasks == nil {
response.RecentTasks = []BackupTaskSummary{}
}
return NewJSONResult(response), nil
}
func (e *PulseToolExecutor) executeListStorage(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
storageID, _ := args["storage_id"].(string)
limit := intArg(args, "limit", 100)
offset := intArg(args, "offset", 0)
if e.storageProvider == nil {
return NewTextResult("Storage information not available."), nil
}
storage := e.storageProvider.GetStorage()
cephClusters := e.storageProvider.GetCephClusters()
response := StorageResponse{}
// Storage pools
count := 0
for _, s := range storage {
if storageID != "" && s.ID != storageID && s.Name != storageID {
continue
}
if count < offset {
count++
continue
}
if len(response.Pools) >= limit {
break
}
pool := StoragePoolSummary{
ID: s.ID,
Name: s.Name,
Type: s.Type,
Status: s.Status,
UsagePercent: s.Usage * 100,
UsedGB: float64(s.Used) / (1024 * 1024 * 1024),
TotalGB: float64(s.Total) / (1024 * 1024 * 1024),
FreeGB: float64(s.Free) / (1024 * 1024 * 1024),
Content: s.Content,
Shared: s.Shared,
}
if s.ZFSPool != nil {
pool.ZFS = &ZFSPoolSummary{
Name: s.ZFSPool.Name,
State: s.ZFSPool.State,
ReadErrors: s.ZFSPool.ReadErrors,
WriteErrors: s.ZFSPool.WriteErrors,
ChecksumErrors: s.ZFSPool.ChecksumErrors,
Scan: s.ZFSPool.Scan,
}
}
response.Pools = append(response.Pools, pool)
count++
}
// Ceph clusters
for _, c := range cephClusters {
response.CephClusters = append(response.CephClusters, CephClusterSummary{
Name: c.Name,
Health: c.Health,
HealthMessage: c.HealthMessage,
UsagePercent: c.UsagePercent,
UsedTB: float64(c.UsedBytes) / (1024 * 1024 * 1024 * 1024),
TotalTB: float64(c.TotalBytes) / (1024 * 1024 * 1024 * 1024),
NumOSDs: c.NumOSDs,
NumOSDsUp: c.NumOSDsUp,
NumOSDsIn: c.NumOSDsIn,
NumMons: c.NumMons,
NumMgrs: c.NumMgrs,
})
}
// Ensure non-nil slices
if response.Pools == nil {
response.Pools = []StoragePoolSummary{}
}
if response.CephClusters == nil {
response.CephClusters = []CephClusterSummary{}
}
return NewJSONResult(response), nil
}
func (e *PulseToolExecutor) executeGetDiskHealth(_ context.Context, _ map[string]interface{}) (CallToolResult, error) {
if e.diskHealthProvider == nil && e.storageProvider == nil {
return NewTextResult("Disk health information not available."), nil
}
response := DiskHealthResponse{
Hosts: []HostDiskHealth{},
}
// SMART and RAID data from host agents
if e.diskHealthProvider != nil {
hosts := e.diskHealthProvider.GetHosts()
for _, host := range hosts {
hostHealth := HostDiskHealth{
Hostname: host.Hostname,
}
// SMART data
for _, disk := range host.Sensors.SMART {
hostHealth.SMART = append(hostHealth.SMART, SMARTDiskSummary{
Device: disk.Device,
Model: disk.Model,
Health: disk.Health,
Temperature: disk.Temperature,
})
}
// RAID arrays
for _, raid := range host.RAID {
hostHealth.RAID = append(hostHealth.RAID, RAIDArraySummary{
Device: raid.Device,
Level: raid.Level,
State: raid.State,
ActiveDevices: raid.ActiveDevices,
WorkingDevices: raid.WorkingDevices,
FailedDevices: raid.FailedDevices,
SpareDevices: raid.SpareDevices,
RebuildPercent: raid.RebuildPercent,
})
}
// Ceph from agent
if host.Ceph != nil {
hostHealth.Ceph = &CephStatusSummary{
Health: host.Ceph.Health.Status,
NumOSDs: host.Ceph.OSDMap.NumOSDs,
NumOSDsUp: host.Ceph.OSDMap.NumUp,
NumOSDsIn: host.Ceph.OSDMap.NumIn,
NumPGs: host.Ceph.PGMap.NumPGs,
UsagePercent: host.Ceph.PGMap.UsagePercent,
}
}
// Only add if there's data
if len(hostHealth.SMART) > 0 || len(hostHealth.RAID) > 0 || hostHealth.Ceph != nil {
// Ensure non-nil slices
if hostHealth.SMART == nil {
hostHealth.SMART = []SMARTDiskSummary{}
}
if hostHealth.RAID == nil {
hostHealth.RAID = []RAIDArraySummary{}
}
response.Hosts = append(response.Hosts, hostHealth)
}
}
}
return NewJSONResult(response), nil
}
// ========== Docker Updates Tool Implementations ==========
func (e *PulseToolExecutor) executeListDockerUpdates(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.updatesProvider == nil {
return NewTextResult("Docker update information not available. Ensure updates provider is configured."), nil
}
hostFilter, _ := args["host"].(string)
// Resolve host name to ID if needed
hostID := e.resolveDockerHostID(hostFilter)
updates := e.updatesProvider.GetPendingUpdates(hostID)
// Ensure non-nil slice
if updates == nil {
updates = []ContainerUpdateInfo{}
}
response := DockerUpdatesResponse{
Updates: updates,
Total: len(updates),
HostID: hostID,
}
return NewJSONResult(response), nil
}
func (e *PulseToolExecutor) executeCheckDockerUpdates(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.updatesProvider == nil {
return NewTextResult("Docker update checking not available. Ensure updates provider is configured."), nil
}
hostArg, _ := args["host"].(string)
if hostArg == "" {
return NewErrorResult(fmt.Errorf("host is required")), nil
}
// Resolve host name to ID
hostID := e.resolveDockerHostID(hostArg)
if hostID == "" {
return NewTextResult(fmt.Sprintf("Docker host '%s' not found.", hostArg)), nil
}
hostName := e.getDockerHostName(hostID)
// Control level check - suggest mode just returns the suggestion
if e.controlLevel == ControlLevelSuggest {
return NewTextResult(fmt.Sprintf("To check for Docker updates on host '%s', use the UI or API:\n\nPOST /api/agents/docker/hosts/%s/check-updates", hostName, hostID)), nil
}
// Trigger the update check
cmdStatus, err := e.updatesProvider.TriggerUpdateCheck(hostID)
if err != nil {
return NewTextResult(fmt.Sprintf("Failed to trigger update check: %v", err)), nil
}
response := DockerCheckUpdatesResponse{
Success: true,
HostID: hostID,
HostName: hostName,
CommandID: cmdStatus.ID,
Message: "Update check command queued. Results will be available after the next agent report cycle (~30 seconds).",
Command: cmdStatus,
}
return NewJSONResult(response), nil
}
func (e *PulseToolExecutor) executeUpdateDockerContainer(ctx context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.updatesProvider == nil {
return NewTextResult("Docker update functionality not available. Ensure updates provider is configured."), nil
}
containerArg, _ := args["container"].(string)
hostArg, _ := args["host"].(string)
if containerArg == "" {
return NewErrorResult(fmt.Errorf("container is required")), nil
}
if hostArg == "" {
return NewErrorResult(fmt.Errorf("host is required")), nil
}
// Check if update actions are enabled
if !e.updatesProvider.IsUpdateActionsEnabled() {
return NewTextResult("Docker container updates are disabled by server configuration. Set PULSE_DISABLE_DOCKER_UPDATE_ACTIONS=false or enable in Settings to allow updates."), nil
}
// Resolve container and host
container, dockerHost, err := e.resolveDockerContainer(containerArg, hostArg)
if err != nil {
return NewTextResult(fmt.Sprintf("Could not find container '%s' on host '%s': %v", containerArg, hostArg, err)), nil
}
containerName := trimContainerName(container.Name)
// Control level handling
if e.controlLevel == ControlLevelSuggest {
return NewTextResult(fmt.Sprintf(`To update container '%s' on host '%s', use the UI or run:
POST /api/agents/docker/containers/update
{
"hostId": "%s",
"containerId": "%s",
"containerName": "%s"
}`, containerName, dockerHost.Hostname, dockerHost.ID, container.ID, containerName)), nil
}
// Controlled mode - require approval
if e.controlLevel == ControlLevelControlled {
command := fmt.Sprintf("docker update %s", containerName)
agentHostname := e.getAgentHostnameForDockerHost(dockerHost)
approvalID := createApprovalRecord(command, "docker", container.ID, agentHostname, fmt.Sprintf("Update container %s to latest image", containerName))
return NewTextResult(formatDockerUpdateApprovalNeeded(containerName, dockerHost.Hostname, approvalID)), nil
}
// Autonomous mode - execute directly
cmdStatus, err := e.updatesProvider.UpdateContainer(dockerHost.ID, container.ID, containerName)
if err != nil {
return NewTextResult(fmt.Sprintf("Failed to queue update command: %v", err)), nil
}
response := DockerUpdateContainerResponse{
Success: true,
HostID: dockerHost.ID,
ContainerID: container.ID,
ContainerName: containerName,
CommandID: cmdStatus.ID,
Message: fmt.Sprintf("Update command queued for container '%s'. The agent will pull the latest image and recreate the container.", containerName),
Command: cmdStatus,
}
return NewJSONResult(response), nil
}
// Helper methods for Docker updates
func (e *PulseToolExecutor) resolveDockerHostID(hostArg string) string {
if hostArg == "" {
return ""
}
if e.stateProvider == nil {
return hostArg
}
state := e.stateProvider.GetState()
for _, host := range state.DockerHosts {
if host.ID == hostArg || host.Hostname == hostArg || host.DisplayName == hostArg {
return host.ID
}
}
return hostArg // Return as-is if not found (provider will handle error)
}
func (e *PulseToolExecutor) getDockerHostName(hostID string) string {
if e.stateProvider == nil {
return hostID
}
state := e.stateProvider.GetState()
for _, host := range state.DockerHosts {
if host.ID == hostID {
if host.DisplayName != "" {
return host.DisplayName
}
return host.Hostname
}
}
return hostID
}
func formatDockerUpdateApprovalNeeded(containerName, hostName, approvalID string) string {
payload := map[string]interface{}{
"type": "approval_required",
"approval_id": approvalID,
"container_name": containerName,
"docker_host": hostName,
"action": "update",
"command": fmt.Sprintf("docker update %s (pull latest + recreate)", containerName),
"how_to_approve": "Click the approval button in the chat to execute this update.",
"do_not_retry": true,
}
b, _ := json.Marshal(payload)
return "APPROVAL_REQUIRED: " + string(b)
}
func trimLeadingSlash(name string) string {
if len(name) > 0 && name[0] == '/' {
return name[1:]
}
return name
}
@@ -1,4 +1,4 @@
package mcp
package tools
import (
"fmt"
+512
View File
@@ -0,0 +1,512 @@
package tools
import (
"errors"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
type fakeStateGetter struct {
state models.StateSnapshot
}
func (f fakeStateGetter) GetState() models.StateSnapshot {
return f.state
}
type fakeAlertManager struct {
alerts []alerts.Alert
}
func (f fakeAlertManager) GetActiveAlerts() []alerts.Alert {
return f.alerts
}
type fakeMetricsSource struct {
allGuest map[string]map[string][]RawMetricPoint
guest map[string]map[string][]RawMetricPoint
node map[string]map[string][]RawMetricPoint
}
func (f *fakeMetricsSource) GetGuestMetrics(guestID string, metricType string, _ time.Duration) []RawMetricPoint {
if f.guest == nil {
return nil
}
if byMetric, ok := f.guest[guestID]; ok {
return byMetric[metricType]
}
return nil
}
func (f *fakeMetricsSource) GetNodeMetrics(nodeID string, metricType string, _ time.Duration) []RawMetricPoint {
if f.node == nil {
return nil
}
if byMetric, ok := f.node[nodeID]; ok {
return byMetric[metricType]
}
return nil
}
func (f *fakeMetricsSource) GetAllGuestMetrics(guestID string, _ time.Duration) map[string][]RawMetricPoint {
if f.allGuest == nil {
return nil
}
return f.allGuest[guestID]
}
type fakeBaselineSource struct {
mean float64
stddev float64
ok bool
all map[string]map[string]BaselineData
}
func (f *fakeBaselineSource) GetBaseline(_ string, _ string) (float64, float64, int, bool) {
return f.mean, f.stddev, 10, f.ok
}
func (f *fakeBaselineSource) GetAllBaselines() map[string]map[string]BaselineData {
return f.all
}
type fakePatternSource struct {
patterns []PatternData
predictions []PredictionData
}
func (f *fakePatternSource) GetPatterns() []PatternData {
return f.patterns
}
func (f *fakePatternSource) GetPredictions() []PredictionData {
return f.predictions
}
type fakeFindingsManager struct {
resolveArgs []string
dismissArgs []string
resolveErr error
dismissErr error
}
func (f *fakeFindingsManager) ResolveFinding(findingID, note string) error {
f.resolveArgs = []string{findingID, note}
return f.resolveErr
}
func (f *fakeFindingsManager) DismissFinding(findingID, reason, note string) error {
f.dismissArgs = []string{findingID, reason, note}
return f.dismissErr
}
type fakeMetadataUpdater struct {
resourceArgs []string
err error
}
func (f *fakeMetadataUpdater) SetResourceURL(resourceType, resourceID, url string) error {
f.resourceArgs = []string{resourceType, resourceID, url}
return f.err
}
type fakeUpdatesMonitor struct {
state models.StateSnapshot
checkStatus models.DockerHostCommandStatus
updateStatus models.DockerHostCommandStatus
checkErr error
updateErr error
}
func (f *fakeUpdatesMonitor) GetState() models.StateSnapshot {
return f.state
}
func (f *fakeUpdatesMonitor) QueueDockerCheckUpdatesCommand(_ string) (models.DockerHostCommandStatus, error) {
return f.checkStatus, f.checkErr
}
func (f *fakeUpdatesMonitor) QueueDockerContainerUpdateCommand(_ string, _ string, _ string) (models.DockerHostCommandStatus, error) {
return f.updateStatus, f.updateErr
}
type fakeUpdatesConfig struct {
enabled bool
}
func (f *fakeUpdatesConfig) IsDockerUpdateActionsEnabled() bool {
return f.enabled
}
func TestAlertManagerMCPAdapter(t *testing.T) {
if NewAlertManagerMCPAdapter(nil) != nil {
t.Fatal("expected nil adapter for nil manager")
}
ts := time.Now()
manager := fakeAlertManager{
alerts: []alerts.Alert{
{
ID: "a1",
ResourceID: "vm-1",
ResourceName: "vm1",
Type: "cpu",
Level: alerts.AlertLevelWarning,
Value: 80,
Threshold: 70,
StartTime: ts,
Message: "high cpu",
},
},
}
adapter := NewAlertManagerMCPAdapter(manager)
got := adapter.GetActiveAlerts()
if len(got) != 1 {
t.Fatalf("expected 1 alert, got %d", len(got))
}
if got[0].Severity != "warning" || got[0].ResourceName != "vm1" || got[0].Message != "high cpu" {
t.Fatalf("unexpected alert mapping: %+v", got[0])
}
}
func TestStorageBackupDiskAdapters(t *testing.T) {
state := models.StateSnapshot{
Storage: []models.Storage{{ID: "s1"}},
CephClusters: []models.CephCluster{{ID: "c1"}},
Backups: models.Backups{PVE: models.PVEBackups{}},
PBSInstances: []models.PBSInstance{{ID: "pbs1"}},
Hosts: []models.Host{{ID: "h1"}},
}
if NewStorageMCPAdapter(nil) != nil {
t.Fatal("expected nil storage adapter for nil state")
}
emptyStorage := (&StorageMCPAdapter{}).GetStorage()
if emptyStorage != nil {
t.Fatal("expected nil storage when state getter missing")
}
storageAdapter := NewStorageMCPAdapter(fakeStateGetter{state: state})
if len(storageAdapter.GetStorage()) != 1 {
t.Fatal("expected storage data")
}
if len(storageAdapter.GetCephClusters()) != 1 {
t.Fatal("expected ceph data")
}
if NewBackupMCPAdapter(nil) != nil {
t.Fatal("expected nil backup adapter for nil state")
}
backupAdapter := NewBackupMCPAdapter(fakeStateGetter{state: state})
if len(backupAdapter.GetPBSInstances()) != 1 {
t.Fatal("expected pbs instances")
}
if NewDiskHealthMCPAdapter(nil) != nil {
t.Fatal("expected nil disk health adapter for nil state")
}
diskAdapter := NewDiskHealthMCPAdapter(fakeStateGetter{state: state})
if len(diskAdapter.GetHosts()) != 1 {
t.Fatal("expected hosts")
}
}
func TestMetricsHistoryMCPAdapter(t *testing.T) {
now := time.Now()
points := map[string][]RawMetricPoint{
"cpu": {{Value: 10, Timestamp: now}},
"memory": {{Value: 20, Timestamp: now}},
}
source := &fakeMetricsSource{
allGuest: map[string]map[string][]RawMetricPoint{
"100": points,
},
}
adapter := NewMetricsHistoryMCPAdapter(fakeStateGetter{}, source)
got, err := adapter.GetResourceMetrics("100", time.Hour)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(got) != 1 || got[0].CPU != 10 || got[0].Memory != 20 {
t.Fatalf("unexpected merged metrics: %+v", got)
}
// Node fallback when guest metrics empty
source = &fakeMetricsSource{
allGuest: map[string]map[string][]RawMetricPoint{},
node: map[string]map[string][]RawMetricPoint{
"node1": {
"cpu": {{Value: 5, Timestamp: now}},
"memory": {{Value: 15, Timestamp: now}},
},
},
}
adapter = NewMetricsHistoryMCPAdapter(fakeStateGetter{}, source)
got, err = adapter.GetResourceMetrics("node1", time.Hour)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(got) != 1 || got[0].CPU != 5 || got[0].Memory != 15 {
t.Fatalf("unexpected node metrics: %+v", got)
}
adapter = &MetricsHistoryMCPAdapter{}
empty, err := adapter.GetResourceMetrics("missing", time.Hour)
if err != nil || empty != nil {
t.Fatal("expected nil metrics when source missing")
}
}
func TestMetricsSummaryAndHelpers(t *testing.T) {
now := time.Now()
source := &fakeMetricsSource{
guest: map[string]map[string][]RawMetricPoint{
"100": {
"cpu": {{Value: 10, Timestamp: now}, {Value: 20, Timestamp: now.Add(time.Minute)}},
"memory": {{Value: 30, Timestamp: now}},
},
},
node: map[string]map[string][]RawMetricPoint{
"node1": {
"cpu": {{Value: 0, Timestamp: now}, {Value: 10, Timestamp: now.Add(time.Minute)}},
"memory": {{Value: 5, Timestamp: now}},
},
},
}
state := models.StateSnapshot{
VMs: []models.VM{{VMID: 100, Name: "vm1"}},
Containers: []models.Container{{VMID: 200, Name: "ct1"}},
Nodes: []models.Node{{ID: "node1", Name: "node-1"}},
}
adapter := NewMetricsHistoryMCPAdapter(fakeStateGetter{state: state}, source)
summary, err := adapter.GetAllMetricsSummary(time.Hour)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(summary) != 2 {
t.Fatalf("expected summaries for vm and node, got %d", len(summary))
}
if summary["100"].ResourceName != "vm1" || summary["node1"].ResourceName != "node-1" {
t.Fatalf("unexpected summary names: %+v", summary)
}
merged := mergeMetricsByTimestamp(map[string][]RawMetricPoint{
"cpu": {{Value: 1, Timestamp: now}},
"memory": {{Value: 2, Timestamp: now}},
"disk": {{Value: 3, Timestamp: now.Add(time.Minute)}},
})
if len(merged) != 2 {
t.Fatalf("expected 2 merged points, got %d", len(merged))
}
avg, max := computeStats([]RawMetricPoint{{Value: 1}, {Value: 3}})
if avg != 2 || max != 3 {
t.Fatalf("unexpected stats avg=%v max=%v", avg, max)
}
if computeTrend([]RawMetricPoint{{Value: 1}}) != "stable" {
t.Fatal("expected stable trend for short series")
}
if computeTrend([]RawMetricPoint{{Value: 0}, {Value: 0}, {Value: 0}, {Value: 10}}) != "growing" {
t.Fatal("expected growing trend")
}
if computeTrend([]RawMetricPoint{{Value: 10}, {Value: 10}, {Value: 10}, {Value: 0}}) != "declining" {
t.Fatal("expected declining trend")
}
if computeTrend([]RawMetricPoint{{Value: 1}, {Value: 2}, {Value: 2}, {Value: 2}}) != "stable" {
t.Fatal("expected stable trend within threshold")
}
}
func TestBaselineMCPAdapter(t *testing.T) {
if NewBaselineMCPAdapter(nil) != nil {
t.Fatal("expected nil baseline adapter for nil source")
}
adapter := NewBaselineMCPAdapter(&fakeBaselineSource{mean: 10, stddev: 2, ok: true})
baseline := adapter.GetBaseline("vm1", "cpu")
if baseline == nil || baseline.Min != 6 || baseline.Max != 14 {
t.Fatalf("unexpected baseline: %+v", baseline)
}
adapter = &BaselineMCPAdapter{}
if adapter.GetBaseline("vm1", "cpu") != nil {
t.Fatal("expected nil baseline when source missing")
}
adapter = NewBaselineMCPAdapter(&fakeBaselineSource{ok: false})
if adapter.GetBaseline("vm1", "cpu") != nil {
t.Fatal("expected nil baseline when not found")
}
adapter = NewBaselineMCPAdapter(&fakeBaselineSource{all: nil})
if adapter.GetAllBaselines() != nil {
t.Fatal("expected nil baselines when source returns nil")
}
all := map[string]map[string]BaselineData{
"100": {"cpu": {Mean: 5, StdDev: 1}},
}
adapter = NewBaselineMCPAdapter(&fakeBaselineSource{all: all})
allBaselines := adapter.GetAllBaselines()
if allBaselines["100"]["cpu"].Min != 3 || allBaselines["100"]["cpu"].Max != 7 {
t.Fatalf("unexpected all baselines: %+v", allBaselines)
}
}
func TestPatternMCPAdapter(t *testing.T) {
state := models.StateSnapshot{
VMs: []models.VM{{VMID: 100, Name: "vm1"}},
Nodes: []models.Node{{ID: "node1", Name: "node-1"}},
Containers: []models.Container{{VMID: 200, Name: "ct1"}},
}
source := &fakePatternSource{
patterns: []PatternData{
{ResourceID: "100", PatternType: "cpu", Description: "spike"},
{ResourceID: "node1", PatternType: "disk", Description: "trend"},
},
predictions: []PredictionData{
{ResourceID: "200", IssueType: "memory", Recommendation: "scale"},
},
}
adapter := NewPatternMCPAdapter(source, fakeStateGetter{state: state})
patterns := adapter.GetPatterns()
if len(patterns) != 2 || patterns[0].ResourceName != "vm1" || patterns[1].ResourceName != "node-1" {
t.Fatalf("unexpected patterns: %+v", patterns)
}
predictions := adapter.GetPredictions()
if len(predictions) != 1 || predictions[0].ResourceName != "ct1" {
t.Fatalf("unexpected predictions: %+v", predictions)
}
adapter = NewPatternMCPAdapter(source, nil)
patterns = adapter.GetPatterns()
if patterns[0].ResourceName != "100" {
t.Fatal("expected resource ID when state missing")
}
}
func TestFindingsAndMetadataAdapters(t *testing.T) {
manager := &fakeFindingsManager{resolveErr: errors.New("resolve"), dismissErr: errors.New("dismiss")}
adapter := NewFindingsManagerMCPAdapter(manager)
if err := adapter.ResolveFinding("f1", "note"); err == nil {
t.Fatal("expected resolve error")
}
if err := adapter.DismissFinding("f1", "reason", "note"); err == nil {
t.Fatal("expected dismiss error")
}
if len(manager.resolveArgs) != 2 || len(manager.dismissArgs) != 3 {
t.Fatal("expected args to be captured")
}
adapter = &FindingsManagerMCPAdapter{}
if err := adapter.ResolveFinding("f1", "note"); err == nil {
t.Fatal("expected error when manager missing")
}
updater := &fakeMetadataUpdater{err: errors.New("update")}
meta := NewMetadataUpdaterMCPAdapter(updater)
if err := meta.SetResourceURL("vm", "1", "http://x"); err == nil {
t.Fatal("expected update error")
}
if len(updater.resourceArgs) != 3 {
t.Fatal("expected resource args captured")
}
meta = &MetadataUpdaterMCPAdapter{}
if err := meta.SetResourceURL("vm", "1", "http://x"); err == nil {
t.Fatal("expected error when metadata updater missing")
}
}
func TestUpdatesMCPAdapter(t *testing.T) {
if NewUpdatesMCPAdapter(nil, nil) != nil {
t.Fatal("expected nil updates adapter for nil monitor")
}
now := time.Now()
state := models.StateSnapshot{
DockerHosts: []models.DockerHost{
{
ID: "host1",
Hostname: "h1",
DisplayName: "Host 1",
Containers: []models.DockerContainer{
{
ID: "c1",
Name: "/nginx",
UpdateStatus: &models.DockerContainerUpdateStatus{
UpdateAvailable: true,
CurrentDigest: "old",
LatestDigest: "new",
LastChecked: now,
},
},
},
},
{
ID: "host2",
Hostname: "h2",
DisplayName: "Host 2",
Containers: []models.DockerContainer{
{
ID: "c2",
Name: "redis",
UpdateStatus: &models.DockerContainerUpdateStatus{
Error: "rate limited",
},
},
},
},
},
}
monitor := &fakeUpdatesMonitor{state: state}
adapter := NewUpdatesMCPAdapter(monitor, &fakeUpdatesConfig{enabled: false})
updates := adapter.GetPendingUpdates("host1")
if len(updates) != 1 || updates[0].ContainerName != "nginx" {
t.Fatalf("unexpected updates: %+v", updates)
}
if adapter.IsUpdateActionsEnabled() {
t.Fatal("expected updates disabled")
}
if (&UpdatesMCPAdapter{}).IsUpdateActionsEnabled() != true {
t.Fatal("expected updates enabled by default")
}
monitor.checkErr = errors.New("check")
if _, err := adapter.TriggerUpdateCheck("host1"); err == nil {
t.Fatal("expected check error")
}
monitor.checkErr = nil
monitor.checkStatus = models.DockerHostCommandStatus{ID: "cmd1", Type: "check", Status: "queued"}
status, err := adapter.TriggerUpdateCheck("host1")
if err != nil || status.ID != "cmd1" {
t.Fatalf("unexpected status: %+v err=%v", status, err)
}
monitor.updateErr = errors.New("update")
if _, err := adapter.UpdateContainer("host1", "c1", "nginx"); err == nil {
t.Fatal("expected update error")
}
monitor.updateErr = nil
monitor.updateStatus = models.DockerHostCommandStatus{ID: "cmd2", Type: "update", Status: "queued"}
status, err = adapter.UpdateContainer("host1", "c1", "nginx")
if err != nil || status.ID != "cmd2" {
t.Fatalf("unexpected update status: %+v err=%v", status, err)
}
if trimContainerName("/redis") != "redis" || trimContainerName("plain") != "plain" {
t.Fatal("unexpected trim result")
}
}
File diff suppressed because it is too large Load Diff
@@ -1,4 +1,4 @@
package mcp
package tools
import (
"context"
@@ -9,6 +9,9 @@ import (
"github.com/rs/zerolog/log"
)
// ServerVersion is the version of the MCP tool implementation
const ServerVersion = "1.0.0"
// StateProvider provides access to infrastructure state
type StateProvider interface {
GetState() models.StateSnapshot
@@ -299,12 +302,18 @@ func (e *PulseToolExecutor) registerTools() {
// Query tools (always available)
e.registerQueryTools()
// Kubernetes tools (always available)
e.registerKubernetesTools()
// Patrol context tools (always available)
e.registerPatrolTools()
// Infrastructure tools (always available)
e.registerInfrastructureTools()
// PMG (Mail Gateway) tools (always available)
e.registerPMGTools()
// Profile tools - read operations always available
e.registerProfileTools()
@@ -1,7 +1,8 @@
package mcp
package tools
import (
"context"
"encoding/json"
"testing"
"time"
@@ -171,10 +172,10 @@ func TestExecuteGetInfrastructureState(t *testing.T) {
{Name: "pve1", Status: "online"},
},
VMs: []models.VM{
{Name: "test-vm", VMID: 100, Status: "running"},
{Name: "test-vm", VMID: 100, Status: "running", Node: "pve1"},
},
Containers: []models.Container{
{Name: "test-ct", VMID: 101, Status: "running"},
{Name: "test-ct", VMID: 101, Status: "running", Node: "pve1"},
},
}
@@ -183,7 +184,7 @@ func TestExecuteGetInfrastructureState(t *testing.T) {
}
executor := NewPulseToolExecutor(cfg)
result, err := executor.ExecuteTool(context.Background(), "pulse_get_infrastructure_state", nil)
result, err := executor.ExecuteTool(context.Background(), "pulse_get_topology", nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
@@ -195,15 +196,21 @@ func TestExecuteGetInfrastructureState(t *testing.T) {
t.Fatal("expected content in result")
}
text := result.Content[0].Text
if !contains(text, "pve1") {
t.Error("expected node name in output")
var response TopologyResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &response); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if !contains(text, "test-vm") {
t.Error("expected VM name in output")
if response.Summary.TotalNodes != 1 || response.Summary.TotalVMs != 1 || response.Summary.TotalLXCContainers != 1 {
t.Fatalf("unexpected summary totals: %+v", response.Summary)
}
if !contains(text, "test-ct") {
t.Error("expected container name in output")
if len(response.Proxmox.Nodes) != 1 || response.Proxmox.Nodes[0].Name != "pve1" {
t.Fatalf("expected node pve1, got %+v", response.Proxmox.Nodes)
}
if len(response.Proxmox.Nodes[0].VMs) != 1 || response.Proxmox.Nodes[0].VMs[0].Name != "test-vm" {
t.Fatalf("expected VM test-vm, got %+v", response.Proxmox.Nodes[0].VMs)
}
if len(response.Proxmox.Nodes[0].Containers) != 1 || response.Proxmox.Nodes[0].Containers[0].Name != "test-ct" {
t.Fatalf("expected container test-ct, got %+v", response.Proxmox.Nodes[0].Containers)
}
}
@@ -318,7 +325,7 @@ func TestExecuteGetActiveAlerts(t *testing.T) {
}
executor := NewPulseToolExecutor(cfg)
result, err := executor.ExecuteTool(context.Background(), "pulse_get_active_alerts", nil)
result, err := executor.ExecuteTool(context.Background(), "pulse_list_alerts", nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
@@ -327,12 +334,15 @@ func TestExecuteGetActiveAlerts(t *testing.T) {
t.Fatal("expected successful result")
}
text := result.Content[0].Text
if !contains(text, "test-vm") {
t.Error("expected resource name in output")
var response AlertsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &response); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if !contains(text, "95.0%") {
t.Error("expected value in output")
if response.Count != 1 || len(response.Alerts) != 1 {
t.Fatalf("expected 1 alert, got %+v", response)
}
if response.Alerts[0].ResourceName != "test-vm" || response.Alerts[0].Value != 95.0 {
t.Fatalf("unexpected alert: %+v", response.Alerts[0])
}
}
@@ -352,24 +362,30 @@ func TestExecuteGetFindingsWithDismissed(t *testing.T) {
executor := NewPulseToolExecutor(cfg)
// Without dismissed
result, _ := executor.ExecuteTool(context.Background(), "pulse_get_findings", map[string]interface{}{
result, _ := executor.ExecuteTool(context.Background(), "pulse_list_findings", map[string]interface{}{
"include_dismissed": false,
})
text := result.Content[0].Text
if !contains(text, "Active Issue") {
t.Error("expected active finding")
var response FindingsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &response); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if contains(text, "Dismissed Issue") {
t.Error("should not include dismissed findings")
if len(response.Active) != 1 || response.Active[0].Title != "Active Issue" {
t.Fatalf("expected active finding, got %+v", response.Active)
}
if len(response.Dismissed) != 0 {
t.Fatalf("expected no dismissed findings, got %+v", response.Dismissed)
}
// With dismissed
result, _ = executor.ExecuteTool(context.Background(), "pulse_get_findings", map[string]interface{}{
result, _ = executor.ExecuteTool(context.Background(), "pulse_list_findings", map[string]interface{}{
"include_dismissed": true,
})
text = result.Content[0].Text
if !contains(text, "Dismissed Issue") {
t.Error("expected dismissed findings when included")
response = FindingsResponse{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &response); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if len(response.Dismissed) != 1 || response.Dismissed[0].Title != "Dismissed Issue" {
t.Fatalf("expected dismissed findings, got %+v", response.Dismissed)
}
}
@@ -1,4 +1,4 @@
package mcp
package tools
import (
"encoding/json"
@@ -1,4 +1,4 @@
package mcp
package tools
import (
"context"
@@ -1,4 +1,4 @@
package mcp
package tools
import (
"context"
@@ -140,7 +140,10 @@ func (e *PulseToolExecutor) executeRunCommand(ctx context.Context, args map[stri
// Note: Control level read_only check is now centralized in registry.Execute()
// Check security policy
// Check if this is a pre-approved execution (agentic loop re-executing after user approval)
preApproved := isPreApproved(args)
// Check security policy (skip block check - blocks cannot be pre-approved)
decision := agentexec.PolicyAllow
if e.policy != nil {
decision = e.policy.Evaluate(command)
@@ -153,7 +156,8 @@ func (e *PulseToolExecutor) executeRunCommand(ctx context.Context, args map[stri
return NewTextResult(formatCommandSuggestion(command, runOnHost, targetHost)), nil
}
if e.controlLevel == ControlLevelControlled {
// Skip approval checks if pre-approved
if !preApproved && e.controlLevel == ControlLevelControlled {
targetType := "container"
if runOnHost {
targetType = "host"
@@ -161,7 +165,7 @@ func (e *PulseToolExecutor) executeRunCommand(ctx context.Context, args map[stri
approvalID := createApprovalRecord(command, targetType, e.targetID, targetHost, "Control level requires approval")
return NewTextResult(formatApprovalNeeded(command, "Control level requires approval", approvalID)), nil
}
if decision == agentexec.PolicyRequireApproval && !e.isAutonomous {
if !preApproved && decision == agentexec.PolicyRequireApproval && !e.isAutonomous {
targetType := "container"
if runOnHost {
targetType = "host"
@@ -263,8 +267,11 @@ func (e *PulseToolExecutor) executeControlGuest(ctx context.Context, args map[st
command = fmt.Sprintf("%s stop %d --skiplock", cmdTool, guest.VMID)
}
// Check security policy
if e.policy != nil {
// Check if this is a pre-approved execution (agentic loop re-executing after user approval)
preApproved := isPreApproved(args)
// Check security policy (skip if pre-approved)
if !preApproved && e.policy != nil {
decision := e.policy.Evaluate(command)
if decision == agentexec.PolicyBlock {
return NewTextResult(formatPolicyBlocked(command, "This command is blocked by security policy")), nil
@@ -276,8 +283,8 @@ func (e *PulseToolExecutor) executeControlGuest(ctx context.Context, args map[st
}
}
// Check control level - this must be outside policy check since policy may be nil
if e.controlLevel == ControlLevelControlled {
// Check control level - this must be outside policy check since policy may be nil (skip if pre-approved)
if !preApproved && e.controlLevel == ControlLevelControlled {
// Use guest.Node (the Proxmox host) as targetName so approval execution can find the correct agent
approvalID := createApprovalRecord(command, guest.Type, fmt.Sprintf("%d", guest.VMID), guest.Node, fmt.Sprintf("%s guest %s", action, guest.Name))
return NewTextResult(formatControlApprovalNeeded(guest.Name, guest.VMID, action, command, approvalID)), nil
@@ -336,6 +343,9 @@ func (e *PulseToolExecutor) executeControlDocker(ctx context.Context, args map[s
// Note: Control level read_only check is now centralized in registry.Execute()
// Check if this is a pre-approved execution (agentic loop re-executing after user approval)
preApproved := isPreApproved(args)
container, dockerHost, err := e.resolveDockerContainer(containerName, hostName)
if err != nil {
return NewTextResult(fmt.Sprintf("Could not find Docker container '%s': %v", containerName, err)), nil
@@ -346,7 +356,8 @@ func (e *PulseToolExecutor) executeControlDocker(ctx context.Context, args map[s
// Get the agent hostname for approval records (may differ from docker host display name)
agentHostname := e.getAgentHostnameForDockerHost(dockerHost)
if e.policy != nil {
// Skip approval checks if pre-approved
if !preApproved && e.policy != nil {
decision := e.policy.Evaluate(command)
if decision == agentexec.PolicyBlock {
return NewTextResult(formatPolicyBlocked(command, "This command is blocked by security policy")), nil
@@ -357,8 +368,8 @@ func (e *PulseToolExecutor) executeControlDocker(ctx context.Context, args map[s
}
}
// Check control level - this must be outside policy check since policy may be nil
if e.controlLevel == ControlLevelControlled {
// Check control level - this must be outside policy check since policy may be nil (skip if pre-approved)
if !preApproved && e.controlLevel == ControlLevelControlled {
approvalID := createApprovalRecord(command, "docker", container.Name, agentHostname, fmt.Sprintf("%s Docker container %s", action, container.Name))
return NewTextResult(formatDockerApprovalNeeded(container.Name, dockerHost.Hostname, action, command, approvalID)), nil
}
@@ -589,6 +600,34 @@ func createApprovalRecord(command, targetType, targetID, targetName, context str
return req.ID
}
// isPreApproved checks if the args contain a valid, approved approval_id.
// This is used when the agentic loop re-executes a tool after user approval.
func isPreApproved(args map[string]interface{}) bool {
approvalID, ok := args["_approval_id"].(string)
if !ok || approvalID == "" {
return false
}
store := approval.GetStore()
if store == nil {
return false
}
req, found := store.GetApproval(approvalID)
if !found {
log.Debug().Str("approval_id", approvalID).Msg("Pre-approval check: approval not found")
return false
}
if req.Status == approval.StatusApproved {
log.Debug().Str("approval_id", approvalID).Msg("Pre-approval check: approved, skipping approval flow")
return true
}
log.Debug().Str("approval_id", approvalID).Str("status", string(req.Status)).Msg("Pre-approval check: not approved")
return false
}
// Formatting helpers for control tools
func formatApprovalNeeded(command, reason, approvalID string) string {
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,310 @@
package tools
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
type stubBackupProvider struct {
backups models.Backups
pbs []models.PBSInstance
}
func (s *stubBackupProvider) GetBackups() models.Backups {
return s.backups
}
func (s *stubBackupProvider) GetPBSInstances() []models.PBSInstance {
return s.pbs
}
type stubStorageProvider struct {
storage []models.Storage
ceph []models.CephCluster
}
func (s *stubStorageProvider) GetStorage() []models.Storage {
return s.storage
}
func (s *stubStorageProvider) GetCephClusters() []models.CephCluster {
return s.ceph
}
type stubDiskHealthProvider struct {
hosts []models.Host
}
func (s *stubDiskHealthProvider) GetHosts() []models.Host {
return s.hosts
}
type stubUpdatesProvider struct {
pending []ContainerUpdateInfo
enabled bool
triggerCalled bool
lastTriggerHost string
lastUpdateHost string
lastUpdateID string
lastUpdateName string
triggerStatus DockerCommandStatus
triggerErr error
updateStatus DockerCommandStatus
updateErr error
updateCheckEnabled bool
}
func (s *stubUpdatesProvider) GetPendingUpdates(hostID string) []ContainerUpdateInfo {
s.lastTriggerHost = hostID
return s.pending
}
func (s *stubUpdatesProvider) TriggerUpdateCheck(hostID string) (DockerCommandStatus, error) {
s.triggerCalled = true
s.lastTriggerHost = hostID
return s.triggerStatus, s.triggerErr
}
func (s *stubUpdatesProvider) UpdateContainer(hostID, containerID, containerName string) (DockerCommandStatus, error) {
s.lastUpdateHost = hostID
s.lastUpdateID = containerID
s.lastUpdateName = containerName
return s.updateStatus, s.updateErr
}
func (s *stubUpdatesProvider) IsUpdateActionsEnabled() bool {
return s.enabled
}
func TestExecuteListBackupsAndStorage(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
executor.backupProvider = &stubBackupProvider{
backups: models.Backups{
PBS: []models.PBSBackup{
{
VMID: "100",
BackupType: "vm",
BackupTime: time.Unix(1000, 0),
Instance: "pbs1",
Datastore: "ds1",
Size: 1024 * 1024 * 1024,
Verified: true,
Protected: true,
},
},
PVE: models.PVEBackups{
StorageBackups: []models.StorageBackup{
{
VMID: 101,
Time: time.Unix(1100, 0),
Size: 2 * 1024 * 1024 * 1024,
Storage: "local",
},
},
BackupTasks: []models.BackupTask{
{
VMID: 101,
Node: "node1",
Status: "OK",
StartTime: time.Unix(1200, 0),
},
},
},
},
pbs: []models.PBSInstance{
{
Name: "pbs1",
Host: "10.0.0.1",
Status: "online",
Datastores: []models.PBSDatastore{
{
Name: "ds1",
Usage: 0.5,
Free: 1024 * 1024 * 1024,
},
},
},
},
}
result, _ := executor.executeListBackups(context.Background(), map[string]interface{}{})
var backupsResp BackupsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &backupsResp); err != nil {
t.Fatalf("decode backups response: %v", err)
}
if len(backupsResp.PBS) != 1 || backupsResp.PBS[0].SizeGB != 1 {
t.Fatalf("unexpected PBS backups: %+v", backupsResp.PBS)
}
if len(backupsResp.PVE) != 1 || backupsResp.PVE[0].SizeGB != 2 {
t.Fatalf("unexpected PVE backups: %+v", backupsResp.PVE)
}
if len(backupsResp.PBSServers) != 1 || len(backupsResp.PBSServers[0].Datastores) != 1 {
t.Fatalf("unexpected PBS servers: %+v", backupsResp.PBSServers)
}
if len(backupsResp.RecentTasks) != 1 {
t.Fatalf("unexpected recent tasks: %+v", backupsResp.RecentTasks)
}
executor.storageProvider = &stubStorageProvider{
storage: []models.Storage{
{
ID: "store1",
Name: "store1",
Type: "zfs",
Status: "active",
Usage: 0.25,
Used: 1024 * 1024 * 1024,
Total: 4 * 1024 * 1024 * 1024,
Free: 3 * 1024 * 1024 * 1024,
Content: "images",
Shared: false,
ZFSPool: &models.ZFSPool{
Name: "tank",
State: "ONLINE",
ReadErrors: 0,
WriteErrors: 0,
ChecksumErrors: 0,
Scan: "scrub",
},
},
},
ceph: []models.CephCluster{
{
Name: "ceph1",
Health: "HEALTH_OK",
HealthMessage: "ok",
UsagePercent: 12.5,
UsedBytes: 2 * 1024 * 1024 * 1024 * 1024,
TotalBytes: 4 * 1024 * 1024 * 1024 * 1024,
NumOSDs: 3,
NumOSDsUp: 3,
NumOSDsIn: 3,
NumMons: 1,
NumMgrs: 1,
},
},
}
result, _ = executor.executeListStorage(context.Background(), map[string]interface{}{})
var storageResp StorageResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &storageResp); err != nil {
t.Fatalf("decode storage response: %v", err)
}
if len(storageResp.Pools) != 1 || storageResp.Pools[0].ZFS == nil {
t.Fatalf("unexpected storage pools: %+v", storageResp.Pools)
}
if len(storageResp.CephClusters) != 1 || storageResp.CephClusters[0].UsedTB != 2 {
t.Fatalf("unexpected ceph clusters: %+v", storageResp.CephClusters)
}
}
func TestExecuteGetDiskHealth(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
executor.diskHealthProvider = &stubDiskHealthProvider{
hosts: []models.Host{
{
Hostname: "host1",
Sensors: models.HostSensorSummary{
SMART: []models.HostDiskSMART{
{Device: "/dev/sda", Model: "disk", Health: "PASSED", Temperature: 30},
},
},
RAID: []models.HostRAIDArray{
{Device: "/dev/md0", Level: "raid1", State: "clean", ActiveDevices: 2, WorkingDevices: 2},
},
Ceph: &models.HostCephCluster{
Health: models.HostCephHealth{Status: "HEALTH_OK"},
OSDMap: models.HostCephOSDMap{NumOSDs: 3, NumUp: 3, NumIn: 3},
PGMap: models.HostCephPGMap{NumPGs: 128, UsagePercent: 10.5},
},
},
},
}
result, _ := executor.executeGetDiskHealth(context.Background(), map[string]interface{}{})
var resp DiskHealthResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &resp); err != nil {
t.Fatalf("decode disk health: %v", err)
}
if len(resp.Hosts) != 1 || len(resp.Hosts[0].SMART) != 1 || len(resp.Hosts[0].RAID) != 1 {
t.Fatalf("unexpected disk health response: %+v", resp)
}
}
func TestDockerUpdateTools(t *testing.T) {
state := models.StateSnapshot{
DockerHosts: []models.DockerHost{
{
ID: "host1",
Hostname: "docker1",
DisplayName: "Docker One",
Containers: []models.DockerContainer{
{ID: "c1", Name: "/nginx"},
},
},
},
}
executor := NewPulseToolExecutor(ExecutorConfig{
StateProvider: &mockStateProvider{state: state},
ControlLevel: ControlLevelSuggest,
})
updates := &stubUpdatesProvider{
pending: []ContainerUpdateInfo{
{HostID: "host1", ContainerID: "c1", ContainerName: "nginx", UpdateAvailable: true},
},
enabled: true,
triggerStatus: DockerCommandStatus{
ID: "cmd1",
Type: "check",
Status: "queued",
},
updateStatus: DockerCommandStatus{
ID: "cmd2",
Type: "update",
Status: "queued",
},
}
executor.updatesProvider = updates
result, _ := executor.executeListDockerUpdates(context.Background(), map[string]interface{}{"host": "Docker One"})
var listResp DockerUpdatesResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &listResp); err != nil {
t.Fatalf("decode docker updates: %v", err)
}
if listResp.HostID != "host1" || listResp.Total != 1 {
t.Fatalf("unexpected docker updates response: %+v", listResp)
}
result, _ = executor.executeCheckDockerUpdates(context.Background(), map[string]interface{}{"host": "Docker One"})
if !strings.Contains(result.Content[0].Text, "check-updates") {
t.Fatalf("unexpected suggest response: %s", result.Content[0].Text)
}
executor.controlLevel = ControlLevelAutonomous
result, _ = executor.executeUpdateDockerContainer(context.Background(), map[string]interface{}{
"host": "Docker One",
"container": "c1",
})
var updateResp DockerUpdateContainerResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &updateResp); err != nil {
t.Fatalf("decode update response: %v", err)
}
if updates.lastUpdateName != "nginx" || updateResp.CommandID != "cmd2" {
t.Fatalf("unexpected update response: %+v", updateResp)
}
updates.enabled = false
result, _ = executor.executeUpdateDockerContainer(context.Background(), map[string]interface{}{
"host": "Docker One",
"container": "c1",
})
if !strings.Contains(result.Content[0].Text, "updates are disabled") {
t.Fatalf("unexpected disabled response: %s", result.Content[0].Text)
}
}
@@ -1,8 +1,9 @@
package mcp
package tools
import (
"context"
"fmt"
"strings"
"time"
)
@@ -190,6 +191,39 @@ Do NOT use for: Checking if something is running (use pulse_get_topology), or th
return exec.executeDismissFinding(ctx, args)
},
})
// ========== Resolved Alerts Tool ==========
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_list_resolved_alerts",
Description: `List recently resolved alerts (alerts that were active but have since cleared).
Returns: JSON with alerts array containing alert details including type, level, resource info, message, start time, and when it was resolved.
Use when: User asks about alerts that cleared, what issues resolved themselves, or wants to see recent alert history.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"type": {
Type: "string",
Description: "Optional: filter by alert type",
},
"level": {
Type: "string",
Description: "Optional: filter by level (critical, warning)",
},
"limit": {
Type: "integer",
Description: "Maximum number of results (default: 50)",
},
},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeListResolvedAlerts(ctx, args)
},
})
}
func (e *PulseToolExecutor) executeGetMetrics(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
@@ -464,3 +498,63 @@ func (e *PulseToolExecutor) executeDismissFinding(_ context.Context, args map[st
"note": note,
}), nil
}
// ========== Resolved Alerts Tool Implementation ==========
func (e *PulseToolExecutor) executeListResolvedAlerts(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
typeFilter, _ := args["type"].(string)
levelFilter, _ := args["level"].(string)
limit := intArg(args, "limit", 50)
state := e.stateProvider.GetState()
if len(state.RecentlyResolved) == 0 {
return NewTextResult("No recently resolved alerts."), nil
}
var alerts []ResolvedAlertSummary
for _, alert := range state.RecentlyResolved {
// Apply filters
if typeFilter != "" && !strings.EqualFold(alert.Type, typeFilter) {
continue
}
if levelFilter != "" && !strings.EqualFold(alert.Level, levelFilter) {
continue
}
if len(alerts) >= limit {
break
}
alerts = append(alerts, ResolvedAlertSummary{
ID: alert.ID,
Type: alert.Type,
Level: alert.Level,
ResourceID: alert.ResourceID,
ResourceName: alert.ResourceName,
Node: alert.Node,
Instance: alert.Instance,
Message: alert.Message,
Value: alert.Value,
Threshold: alert.Threshold,
StartTime: alert.StartTime,
ResolvedTime: alert.ResolvedTime,
})
}
if alerts == nil {
alerts = []ResolvedAlertSummary{}
}
response := ResolvedAlertsResponse{
Alerts: alerts,
Total: len(state.RecentlyResolved),
}
return NewJSONResult(response), nil
}
+218
View File
@@ -0,0 +1,218 @@
package tools
import (
"context"
"encoding/json"
"errors"
"testing"
)
type stubBaselineProvider struct {
baselines map[string]map[string]*MetricBaseline
}
func (s *stubBaselineProvider) GetBaseline(resourceID, metric string) *MetricBaseline {
if s.baselines == nil {
return nil
}
if metrics, ok := s.baselines[resourceID]; ok {
return metrics[metric]
}
return nil
}
func (s *stubBaselineProvider) GetAllBaselines() map[string]map[string]*MetricBaseline {
return s.baselines
}
type stubPatternProvider struct {
patterns []Pattern
predictions []Prediction
}
func (s *stubPatternProvider) GetPatterns() []Pattern {
return s.patterns
}
func (s *stubPatternProvider) GetPredictions() []Prediction {
return s.predictions
}
type stubFindingsManager struct {
resolveErr error
dismissErr error
}
func (s *stubFindingsManager) ResolveFinding(string, string) error {
return s.resolveErr
}
func (s *stubFindingsManager) DismissFinding(string, string, string) error {
return s.dismissErr
}
func TestExecuteGetMetrics(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
result, _ := executor.executeGetMetrics(context.Background(), map[string]interface{}{"period": "24h"})
if result.IsError || result.Content[0].Text == "" {
t.Fatal("expected metrics not available message")
}
executor.metricsHistory = &mockMetricsHistoryProvider{
metrics: map[string][]MetricPoint{
"res1": {{CPU: 1, Memory: 2}},
},
summary: map[string]ResourceMetricsSummary{
"res1": {ResourceID: "res1"},
},
}
result, _ = executor.executeGetMetrics(context.Background(), map[string]interface{}{
"period": "bad",
"resource_id": "res1",
})
var resp MetricsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &resp); err != nil {
t.Fatalf("decode metrics response: %v", err)
}
if resp.ResourceID != "res1" || len(resp.Points) != 1 {
t.Fatalf("unexpected metrics response: %+v", resp)
}
result, _ = executor.executeGetMetrics(context.Background(), map[string]interface{}{
"period": "7d",
})
if err := json.Unmarshal([]byte(result.Content[0].Text), &resp); err != nil {
t.Fatalf("decode metrics response: %v", err)
}
if resp.Summary == nil || resp.Period != "7d" {
t.Fatalf("unexpected metrics summary: %+v", resp)
}
}
func TestExecuteGetBaselinesAndPatterns(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
result, _ := executor.executeGetBaselines(context.Background(), map[string]interface{}{})
if result.IsError {
t.Fatal("expected baselines not available message")
}
executor.baselineProvider = &stubBaselineProvider{
baselines: map[string]map[string]*MetricBaseline{
"res1": {"cpu": {Mean: 1}},
},
}
result, _ = executor.executeGetBaselines(context.Background(), map[string]interface{}{
"resource_id": "res1",
})
var baselines BaselinesResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &baselines); err != nil {
t.Fatalf("decode baselines: %v", err)
}
if baselines.ResourceID != "res1" || baselines.Baselines["res1"]["cpu"].Mean != 1 {
t.Fatalf("unexpected baselines: %+v", baselines)
}
result, _ = executor.executeGetPatterns(context.Background(), map[string]interface{}{})
if result.IsError {
t.Fatal("expected patterns not available message")
}
executor.patternProvider = &stubPatternProvider{
patterns: []Pattern{{ResourceID: "r1"}},
predictions: []Prediction{{ResourceID: "r2"}},
}
result, _ = executor.executeGetPatterns(context.Background(), map[string]interface{}{})
var patterns PatternsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &patterns); err != nil {
t.Fatalf("decode patterns: %v", err)
}
if len(patterns.Patterns) != 1 || len(patterns.Predictions) != 1 {
t.Fatalf("unexpected patterns: %+v", patterns)
}
}
func TestExecuteListAlertsAndFindings(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
result, _ := executor.executeListAlerts(context.Background(), map[string]interface{}{})
if result.IsError {
t.Fatal("expected alerts not available message")
}
executor.alertProvider = &mockAlertProvider{
alerts: []ActiveAlert{
{ID: "a1", Severity: "warning"},
{ID: "a2", Severity: "critical"},
},
}
result, _ = executor.executeListAlerts(context.Background(), map[string]interface{}{
"severity": "critical",
"limit": float64(1),
})
var alerts AlertsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &alerts); err != nil {
t.Fatalf("decode alerts: %v", err)
}
if alerts.Count != 1 || alerts.Alerts[0].ID != "a2" {
t.Fatalf("unexpected alerts: %+v", alerts)
}
executor.findingsProvider = &mockFindingsProvider{
active: []Finding{{ID: "f1", Severity: "warning"}},
dismissed: []Finding{{ID: "f2", Severity: "info"}},
}
result, _ = executor.executeListFindings(context.Background(), map[string]interface{}{
"include_dismissed": true,
"severity": "warning",
})
var findings FindingsResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &findings); err != nil {
t.Fatalf("decode findings: %v", err)
}
if findings.Counts.Active != 1 || findings.Counts.Dismissed != 1 {
t.Fatalf("unexpected counts: %+v", findings.Counts)
}
if len(findings.Active) != 1 || len(findings.Dismissed) != 0 {
t.Fatalf("unexpected findings: %+v", findings)
}
}
func TestExecuteResolveAndDismissFinding(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
result, _ := executor.executeResolveFinding(context.Background(), map[string]interface{}{})
if !result.IsError {
t.Fatal("expected error without findings manager")
}
executor.findingsManager = &stubFindingsManager{resolveErr: errors.New("resolve")}
result, _ = executor.executeResolveFinding(context.Background(), map[string]interface{}{
"finding_id": "f1",
"resolution_note": "note",
})
if !result.IsError {
t.Fatal("expected resolve error")
}
executor.findingsManager = &stubFindingsManager{dismissErr: errors.New("dismiss")}
result, _ = executor.executeDismissFinding(context.Background(), map[string]interface{}{
"finding_id": "f1",
"reason": "not_an_issue",
"note": "note",
})
if !result.IsError {
t.Fatal("expected dismiss error")
}
executor.findingsManager = &stubFindingsManager{}
result, _ = executor.executeDismissFinding(context.Background(), map[string]interface{}{
"finding_id": "f1",
"reason": "not_an_issue",
"note": "note",
})
var okResp map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &okResp); err != nil {
t.Fatalf("decode dismiss response: %v", err)
}
if okResp["success"] != true {
t.Fatalf("unexpected dismiss response: %+v", okResp)
}
}
+329
View File
@@ -0,0 +1,329 @@
package tools
import (
"context"
"fmt"
)
// registerPMGTools registers Proxmox Mail Gateway query tools
func (e *PulseToolExecutor) registerPMGTools() {
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_pmg_status",
Description: `Get Proxmox Mail Gateway instance status and health.
Returns: JSON with instances array containing id, name, host, status, version, and nodes (with status, role, uptime, load).
Use when: User asks about mail gateway status, PMG health, or mail server infrastructure.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"instance": {
Type: "string",
Description: "Optional: specific PMG instance name or ID",
},
},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetPMGStatus(ctx, args)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_mail_stats",
Description: `Get mail flow statistics (counts, spam, virus, bounces).
Returns: JSON with mail statistics including total in/out, spam in/out, virus counts, bounces, greylist count, and average processing time.
Use when: User asks about mail flow, email statistics, spam counts, or virus detections.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"instance": {
Type: "string",
Description: "Optional: specific PMG instance name or ID",
},
},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetMailStats(ctx, args)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_mail_queues",
Description: `Get mail queue status (active, deferred, hold).
Returns: JSON with queue status for each node including active, deferred, hold, incoming counts and oldest message age.
Use when: User asks about mail queues, deferred messages, mail delivery issues, or queue backlogs.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"instance": {
Type: "string",
Description: "Optional: specific PMG instance name or ID",
},
},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetMailQueues(ctx, args)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_spam_stats",
Description: `Get spam quarantine statistics and distribution.
Returns: JSON with quarantine counts (spam, virus, attachment, blacklisted) and spam score distribution buckets.
Use when: User asks about spam statistics, quarantine status, or spam score distribution.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"instance": {
Type: "string",
Description: "Optional: specific PMG instance name or ID",
},
},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetSpamStats(ctx, args)
},
})
}
func (e *PulseToolExecutor) executeGetPMGStatus(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
instanceFilter, _ := args["instance"].(string)
state := e.stateProvider.GetState()
if len(state.PMGInstances) == 0 {
return NewTextResult("No Proxmox Mail Gateway instances found. PMG monitoring may not be configured."), nil
}
var instances []PMGInstanceSummary
for _, pmg := range state.PMGInstances {
if instanceFilter != "" && pmg.ID != instanceFilter && pmg.Name != instanceFilter {
continue
}
var nodes []PMGNodeSummary
for _, node := range pmg.Nodes {
nodes = append(nodes, PMGNodeSummary{
Name: node.Name,
Status: node.Status,
Role: node.Role,
Uptime: node.Uptime,
LoadAvg: node.LoadAvg,
})
}
instances = append(instances, PMGInstanceSummary{
ID: pmg.ID,
Name: pmg.Name,
Host: pmg.Host,
Status: pmg.Status,
Version: pmg.Version,
Nodes: nodes,
})
}
if len(instances) == 0 && instanceFilter != "" {
return NewTextResult(fmt.Sprintf("PMG instance '%s' not found.", instanceFilter)), nil
}
// Ensure non-nil slices
for i := range instances {
if instances[i].Nodes == nil {
instances[i].Nodes = []PMGNodeSummary{}
}
}
response := PMGStatusResponse{
Instances: instances,
Total: len(instances),
}
return NewJSONResult(response), nil
}
func (e *PulseToolExecutor) executeGetMailStats(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
instanceFilter, _ := args["instance"].(string)
state := e.stateProvider.GetState()
if len(state.PMGInstances) == 0 {
return NewTextResult("No Proxmox Mail Gateway instances found. PMG monitoring may not be configured."), nil
}
// If filtering, find that specific instance
for _, pmg := range state.PMGInstances {
if instanceFilter != "" && pmg.ID != instanceFilter && pmg.Name != instanceFilter {
continue
}
if pmg.MailStats == nil {
if instanceFilter != "" {
return NewTextResult(fmt.Sprintf("No mail statistics available for PMG instance '%s'.", instanceFilter)), nil
}
continue
}
response := MailStatsResponse{
Instance: pmg.Name,
Stats: PMGMailStatsSummary{
Timeframe: pmg.MailStats.Timeframe,
TotalIn: pmg.MailStats.CountIn,
TotalOut: pmg.MailStats.CountOut,
SpamIn: pmg.MailStats.SpamIn,
SpamOut: pmg.MailStats.SpamOut,
VirusIn: pmg.MailStats.VirusIn,
VirusOut: pmg.MailStats.VirusOut,
BouncesIn: pmg.MailStats.BouncesIn,
BouncesOut: pmg.MailStats.BouncesOut,
BytesIn: pmg.MailStats.BytesIn,
BytesOut: pmg.MailStats.BytesOut,
GreylistCount: pmg.MailStats.GreylistCount,
RBLRejects: pmg.MailStats.RBLRejects,
AverageProcessTimeMs: pmg.MailStats.AverageProcessTimeMs,
},
}
return NewJSONResult(response), nil
}
if instanceFilter != "" {
return NewTextResult(fmt.Sprintf("PMG instance '%s' not found.", instanceFilter)), nil
}
return NewTextResult("No mail statistics available from any PMG instance."), nil
}
func (e *PulseToolExecutor) executeGetMailQueues(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
instanceFilter, _ := args["instance"].(string)
state := e.stateProvider.GetState()
if len(state.PMGInstances) == 0 {
return NewTextResult("No Proxmox Mail Gateway instances found. PMG monitoring may not be configured."), nil
}
// Collect queue data from all instances (or filtered instance)
for _, pmg := range state.PMGInstances {
if instanceFilter != "" && pmg.ID != instanceFilter && pmg.Name != instanceFilter {
continue
}
var queues []PMGQueueSummary
for _, node := range pmg.Nodes {
if node.QueueStatus != nil {
queues = append(queues, PMGQueueSummary{
Node: node.Name,
Active: node.QueueStatus.Active,
Deferred: node.QueueStatus.Deferred,
Hold: node.QueueStatus.Hold,
Incoming: node.QueueStatus.Incoming,
Total: node.QueueStatus.Total,
OldestAgeSeconds: node.QueueStatus.OldestAge,
})
}
}
if len(queues) == 0 {
if instanceFilter != "" {
return NewTextResult(fmt.Sprintf("No queue data available for PMG instance '%s'.", instanceFilter)), nil
}
continue
}
response := MailQueuesResponse{
Instance: pmg.Name,
Queues: queues,
}
return NewJSONResult(response), nil
}
if instanceFilter != "" {
return NewTextResult(fmt.Sprintf("PMG instance '%s' not found.", instanceFilter)), nil
}
return NewTextResult("No mail queue data available from any PMG instance."), nil
}
func (e *PulseToolExecutor) executeGetSpamStats(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
instanceFilter, _ := args["instance"].(string)
state := e.stateProvider.GetState()
if len(state.PMGInstances) == 0 {
return NewTextResult("No Proxmox Mail Gateway instances found. PMG monitoring may not be configured."), nil
}
for _, pmg := range state.PMGInstances {
if instanceFilter != "" && pmg.ID != instanceFilter && pmg.Name != instanceFilter {
continue
}
quarantine := PMGQuarantineSummary{}
if pmg.Quarantine != nil {
quarantine = PMGQuarantineSummary{
Spam: pmg.Quarantine.Spam,
Virus: pmg.Quarantine.Virus,
Attachment: pmg.Quarantine.Attachment,
Blacklisted: pmg.Quarantine.Blacklisted,
Total: pmg.Quarantine.Spam + pmg.Quarantine.Virus + pmg.Quarantine.Attachment + pmg.Quarantine.Blacklisted,
}
}
var distribution []PMGSpamBucketSummary
for _, bucket := range pmg.SpamDistribution {
distribution = append(distribution, PMGSpamBucketSummary{
Score: bucket.Score,
Count: bucket.Count,
})
}
response := SpamStatsResponse{
Instance: pmg.Name,
Quarantine: quarantine,
Distribution: distribution,
}
if response.Distribution == nil {
response.Distribution = []PMGSpamBucketSummary{}
}
return NewJSONResult(response), nil
}
if instanceFilter != "" {
return NewTextResult(fmt.Sprintf("PMG instance '%s' not found.", instanceFilter)), nil
}
return NewTextResult("No spam statistics available from any PMG instance."), nil
}
@@ -1,4 +1,4 @@
package mcp
package tools
import (
"context"
+477
View File
@@ -0,0 +1,477 @@
package tools
import (
"context"
"encoding/json"
"errors"
"reflect"
"strings"
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
type mockProfileManager struct {
scope *AgentScope
getErr error
assignErr error
applyErr error
assignName string
applyID string
applyName string
applyNew bool
lastGetAgent string
lastAssignAgent string
lastAssignProfile string
lastApplyAgent string
lastApplyLabel string
lastApplySettings map[string]interface{}
}
func (m *mockProfileManager) GetAgentScope(ctx context.Context, agentID string) (*AgentScope, error) {
m.lastGetAgent = agentID
if m.getErr != nil {
return nil, m.getErr
}
return m.scope, nil
}
func (m *mockProfileManager) AssignProfile(ctx context.Context, agentID, profileID string) (string, error) {
m.lastAssignAgent = agentID
m.lastAssignProfile = profileID
if m.assignErr != nil {
return "", m.assignErr
}
return m.assignName, nil
}
func (m *mockProfileManager) ApplyAgentScope(ctx context.Context, agentID, agentLabel string, settings map[string]interface{}) (string, string, bool, error) {
m.lastApplyAgent = agentID
m.lastApplyLabel = agentLabel
m.lastApplySettings = settings
if m.applyErr != nil {
return "", "", false, m.applyErr
}
return m.applyID, m.applyName, m.applyNew, nil
}
func TestResolveAgentFromHostname(t *testing.T) {
state := models.StateSnapshot{
Hosts: []models.Host{
{ID: "host-1", Hostname: "alpha", DisplayName: "Alpha"},
},
DockerHosts: []models.DockerHost{
{ID: "dock-1", Hostname: "beta", DisplayName: "Beta", CustomDisplayName: "Beta Custom"},
},
}
id, label := resolveAgentFromHostname(state, "ALPHA")
if id != "host-1" || label != "Alpha" {
t.Fatalf("expected host match, got id=%q label=%q", id, label)
}
id, label = resolveAgentFromHostname(state, "beta")
if id != "dock-1" || label != "Beta Custom" {
t.Fatalf("expected docker host match, got id=%q label=%q", id, label)
}
id, label = resolveAgentFromHostname(state, "missing")
if id != "" || label != "" {
t.Fatalf("expected no match, got id=%q label=%q", id, label)
}
}
func TestResolveAgentLabel(t *testing.T) {
state := models.StateSnapshot{
Hosts: []models.Host{
{ID: "host-1", Hostname: "alpha", DisplayName: "Alpha"},
},
DockerHosts: []models.DockerHost{
{ID: "dock-2", AgentID: "agent-2", Hostname: "dock", DisplayName: "Dock", CustomDisplayName: "Docker Host"},
},
}
if label := resolveAgentLabel(state, "host-1"); label != "Alpha" {
t.Fatalf("expected host label, got %q", label)
}
if label := resolveAgentLabel(state, "agent-2"); label != "Docker Host" {
t.Fatalf("expected docker label by agent ID, got %q", label)
}
if label := resolveAgentLabel(state, "dock-2"); label != "Docker Host" {
t.Fatalf("expected docker label by host ID, got %q", label)
}
}
func TestFormatSettingsSummary(t *testing.T) {
if summary := formatSettingsSummary(nil); summary != "none" {
t.Fatalf("expected none for empty settings, got %q", summary)
}
settings := map[string]interface{}{
"beta": true,
"alpha": 1,
}
summary := formatSettingsSummary(settings)
if summary != "alpha=1, beta=true" {
t.Fatalf("unexpected summary: %q", summary)
}
}
func TestDetectAgentModules(t *testing.T) {
state := models.StateSnapshot{
Hosts: []models.Host{
{ID: "agent-1", Hostname: "alpha", CommandsEnabled: true, LinkedNodeID: "node-1"},
},
DockerHosts: []models.DockerHost{
{ID: "dock-1", AgentID: "agent-1", Hostname: "dock"},
},
KubernetesClusters: []models.KubernetesCluster{
{ID: "k8s-1", AgentID: "agent-1", Name: "cluster"},
},
}
modules, commandsEnabled := detectAgentModules(state, "agent-1")
expected := []string{"docker", "host", "kubernetes", "proxmox"}
if !reflect.DeepEqual(modules, expected) {
t.Fatalf("expected modules %v, got %v", expected, modules)
}
if commandsEnabled == nil || !*commandsEnabled {
t.Fatalf("expected commandsEnabled true, got %v", commandsEnabled)
}
modules, commandsEnabled = detectAgentModules(state, "missing")
if modules != nil || commandsEnabled != nil {
t.Fatalf("expected no matches, got modules=%v commandsEnabled=%v", modules, commandsEnabled)
}
}
func TestExecuteGetAgentScopeErrors(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{})
result, err := executor.executeGetAgentScope(context.Background(), map[string]interface{}{})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !result.IsError {
t.Fatal("expected error for missing agent args")
}
result, err = executor.executeGetAgentScope(context.Background(), map[string]interface{}{
"hostname": "alpha",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !result.IsError {
t.Fatal("expected error when state provider missing")
}
}
func TestExecuteGetAgentScopeNotFound(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{
StateProvider: &mockStateProvider{},
})
result, err := executor.executeGetAgentScope(context.Background(), map[string]interface{}{
"hostname": "ghost",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError {
t.Fatal("expected not_found response, got error result")
}
var payload map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &payload); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if payload["error"] != "not_found" {
t.Fatalf("expected not_found error, got %v", payload["error"])
}
}
func TestExecuteGetAgentScopeProfileManagerError(t *testing.T) {
manager := &mockProfileManager{
getErr: errors.New("boom"),
}
executor := NewPulseToolExecutor(ExecutorConfig{
StateProvider: &mockStateProvider{},
AgentProfileManager: manager,
})
result, err := executor.executeGetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError {
t.Fatal("expected JSON error payload, got error result")
}
var payload map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &payload); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if payload["error"] != "failed_to_load" {
t.Fatalf("expected failed_to_load error, got %v", payload["error"])
}
}
func TestExecuteGetAgentScopeWithProfile(t *testing.T) {
state := models.StateSnapshot{
Hosts: []models.Host{
{ID: "agent-1", Hostname: "alpha", DisplayName: "Alpha", CommandsEnabled: true, LinkedNodeID: "node-1"},
},
DockerHosts: []models.DockerHost{
{ID: "dock-1", AgentID: "agent-1", Hostname: "dock"},
},
KubernetesClusters: []models.KubernetesCluster{
{ID: "k8s-1", AgentID: "agent-1", Name: "cluster"},
},
}
manager := &mockProfileManager{
scope: &AgentScope{
AgentID: "agent-1",
ProfileID: "profile-1",
ProfileName: "Default",
ProfileVersion: 2,
Settings: map[string]interface{}{
"enable_docker": true,
},
},
}
executor := NewPulseToolExecutor(ExecutorConfig{
StateProvider: &mockStateProvider{state: state},
AgentProfileManager: manager,
})
result, err := executor.executeGetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError {
t.Fatal("expected successful JSON response")
}
var response AgentScopeResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &response); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if response.ProfileID != "profile-1" || response.ProfileName != "Default" || response.ProfileVersion != 2 {
t.Fatalf("unexpected profile info: %+v", response)
}
if response.AgentLabel != "Alpha" {
t.Fatalf("expected agent label Alpha, got %q", response.AgentLabel)
}
if response.CommandsEnabled == nil || !*response.CommandsEnabled {
t.Fatalf("expected commands enabled true, got %v", response.CommandsEnabled)
}
}
func TestExecuteSetAgentScopeErrors(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{})
result, err := executor.executeSetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError || !strings.Contains(result.Content[0].Text, "not available") {
t.Fatalf("expected not available message, got %+v", result)
}
executor = NewPulseToolExecutor(ExecutorConfig{
AgentProfileManager: &mockProfileManager{},
})
result, err = executor.executeSetAgentScope(context.Background(), map[string]interface{}{})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !result.IsError {
t.Fatal("expected error for missing agent args")
}
result, err = executor.executeSetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
"profile_id": "profile-1",
"settings": map[string]interface{}{
"enable_host": true,
},
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !result.IsError {
t.Fatal("expected error for profile_id + settings")
}
}
func TestExecuteSetAgentScopeSuggestProfile(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{
ControlLevel: ControlLevelSuggest,
AgentProfileManager: &mockProfileManager{},
})
result, err := executor.executeSetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
"profile_id": "profile-1",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError {
t.Fatal("expected suggestion response")
}
var payload map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &payload); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if payload["type"] != "suggestion" || payload["action"] != "assign_profile" {
t.Fatalf("unexpected suggestion payload: %v", payload)
}
}
func TestExecuteSetAgentScopeSuggestSettings(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{
ControlLevel: ControlLevelSuggest,
AgentProfileManager: &mockProfileManager{},
})
result, err := executor.executeSetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
"settings": map[string]interface{}{
"alpha": 1,
"beta": true,
},
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError {
t.Fatal("expected suggestion response")
}
var payload map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &payload); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if payload["type"] != "suggestion" || payload["action"] != "apply_settings" {
t.Fatalf("unexpected suggestion payload: %v", payload)
}
message, _ := payload["message"].(string)
if !strings.Contains(message, "alpha=1, beta=true") {
t.Fatalf("expected settings summary, got %q", message)
}
}
func TestExecuteSetAgentScopeAssignProfile(t *testing.T) {
manager := &mockProfileManager{
assignName: "Gold",
}
executor := NewPulseToolExecutor(ExecutorConfig{
ControlLevel: ControlLevelAutonomous,
AgentProfileManager: manager,
})
result, err := executor.executeSetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
"profile_id": "profile-1",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError {
t.Fatal("expected success response")
}
if manager.lastAssignAgent != "agent-1" || manager.lastAssignProfile != "profile-1" {
t.Fatalf("assign not called with expected values: %+v", manager)
}
var payload map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &payload); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if payload["action"] != "assigned" || payload["profile_name"] != "Gold" {
t.Fatalf("unexpected payload: %v", payload)
}
}
func TestExecuteSetAgentScopeApplySettings(t *testing.T) {
manager := &mockProfileManager{
applyID: "profile-2",
applyName: "Custom",
applyNew: true,
}
executor := NewPulseToolExecutor(ExecutorConfig{
ControlLevel: ControlLevelAutonomous,
AgentProfileManager: manager,
})
result, err := executor.executeSetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
"settings": map[string]interface{}{
"enable_host": true,
},
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError {
t.Fatal("expected success response")
}
if manager.lastApplyAgent != "agent-1" || manager.lastApplySettings == nil {
t.Fatalf("apply not called with expected values: %+v", manager)
}
var payload map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &payload); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if payload["action"] != "created" || payload["profile_name"] != "Custom" {
t.Fatalf("unexpected payload: %v", payload)
}
}
func TestExecuteSetAgentScopeApplySettingsUpdated(t *testing.T) {
manager := &mockProfileManager{
applyID: "profile-3",
applyName: "Existing",
applyNew: false,
}
executor := NewPulseToolExecutor(ExecutorConfig{
ControlLevel: ControlLevelAutonomous,
AgentProfileManager: manager,
})
result, err := executor.executeSetAgentScope(context.Background(), map[string]interface{}{
"agent_id": "agent-1",
"settings": map[string]interface{}{
"enable_docker": true,
},
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if result.IsError {
t.Fatal("expected success response")
}
var payload map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &payload); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if payload["action"] != "updated" || payload["profile_name"] != "Existing" {
t.Fatalf("unexpected payload: %v", payload)
}
}
@@ -1,4 +1,4 @@
package mcp
package tools
import (
"context"
@@ -715,3 +715,402 @@ func intArg(args map[string]interface{}, key string, defaultVal int) int {
}
return defaultVal
}
// ========== Kubernetes Tools ==========
// registerKubernetesTools registers Kubernetes query tools
func (e *PulseToolExecutor) registerKubernetesTools() {
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_kubernetes_clusters",
Description: `List Kubernetes clusters monitored by Pulse with health summary.
Returns: JSON with clusters array containing id, name, status, version, node count, pod count, deployment count.
Use when: User asks about Kubernetes clusters, wants an overview of K8s infrastructure, or needs to find a specific cluster.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetKubernetesClusters(ctx)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_kubernetes_nodes",
Description: `List nodes in a Kubernetes cluster with capacity and status.
Returns: JSON with nodes array containing name, ready status, roles, kubelet version, capacity (CPU, memory, pods), allocatable resources.
Use when: User asks about Kubernetes nodes, node health, or cluster capacity.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"cluster": {
Type: "string",
Description: "Cluster name or ID (required)",
},
},
Required: []string{"cluster"},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetKubernetesNodes(ctx, args)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_kubernetes_pods",
Description: `List pods in a Kubernetes cluster, optionally filtered by namespace or status.
Returns: JSON with pods array containing name, namespace, node, phase, restarts, containers with their states.
Use when: User asks about pods, wants to find a specific pod, or check pod health in a namespace.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"cluster": {
Type: "string",
Description: "Cluster name or ID (required)",
},
"namespace": {
Type: "string",
Description: "Optional: filter by namespace",
},
"status": {
Type: "string",
Description: "Optional: filter by pod phase (Running, Pending, Failed, Succeeded)",
},
"limit": {
Type: "integer",
Description: "Maximum number of results (default: 100)",
},
"offset": {
Type: "integer",
Description: "Number of results to skip",
},
},
Required: []string{"cluster"},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetKubernetesPods(ctx, args)
},
})
e.registry.Register(RegisteredTool{
Definition: Tool{
Name: "pulse_get_kubernetes_deployments",
Description: `List deployments in a Kubernetes cluster with replica status.
Returns: JSON with deployments array containing name, namespace, desired/ready/available/updated replicas.
Use when: User asks about deployments, wants to check deployment health, or find unhealthy deployments.`,
InputSchema: InputSchema{
Type: "object",
Properties: map[string]PropertySchema{
"cluster": {
Type: "string",
Description: "Cluster name or ID (required)",
},
"namespace": {
Type: "string",
Description: "Optional: filter by namespace",
},
"limit": {
Type: "integer",
Description: "Maximum number of results (default: 100)",
},
"offset": {
Type: "integer",
Description: "Number of results to skip",
},
},
Required: []string{"cluster"},
},
},
Handler: func(ctx context.Context, exec *PulseToolExecutor, args map[string]interface{}) (CallToolResult, error) {
return exec.executeGetKubernetesDeployments(ctx, args)
},
})
}
func (e *PulseToolExecutor) executeGetKubernetesClusters(_ context.Context) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
state := e.stateProvider.GetState()
if len(state.KubernetesClusters) == 0 {
return NewTextResult("No Kubernetes clusters found. Kubernetes monitoring may not be configured."), nil
}
var clusters []KubernetesClusterSummary
for _, c := range state.KubernetesClusters {
readyNodes := 0
for _, node := range c.Nodes {
if node.Ready {
readyNodes++
}
}
displayName := c.DisplayName
if c.CustomDisplayName != "" {
displayName = c.CustomDisplayName
}
clusters = append(clusters, KubernetesClusterSummary{
ID: c.ID,
Name: c.Name,
DisplayName: displayName,
Server: c.Server,
Version: c.Version,
Status: c.Status,
NodeCount: len(c.Nodes),
PodCount: len(c.Pods),
DeploymentCount: len(c.Deployments),
ReadyNodes: readyNodes,
})
}
response := KubernetesClustersResponse{
Clusters: clusters,
Total: len(clusters),
}
return NewJSONResult(response), nil
}
func (e *PulseToolExecutor) executeGetKubernetesNodes(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
clusterArg, _ := args["cluster"].(string)
if clusterArg == "" {
return NewErrorResult(fmt.Errorf("cluster is required")), nil
}
state := e.stateProvider.GetState()
// Find the cluster (also match CustomDisplayName)
var cluster *KubernetesClusterSummary
for _, c := range state.KubernetesClusters {
if c.ID == clusterArg || c.Name == clusterArg || c.DisplayName == clusterArg || c.CustomDisplayName == clusterArg {
displayName := c.DisplayName
if c.CustomDisplayName != "" {
displayName = c.CustomDisplayName
}
cluster = &KubernetesClusterSummary{
ID: c.ID,
Name: c.Name,
DisplayName: displayName,
}
var nodes []KubernetesNodeSummary
for _, node := range c.Nodes {
nodes = append(nodes, KubernetesNodeSummary{
UID: node.UID,
Name: node.Name,
Ready: node.Ready,
Unschedulable: node.Unschedulable,
Roles: node.Roles,
KubeletVersion: node.KubeletVersion,
ContainerRuntimeVersion: node.ContainerRuntimeVersion,
OSImage: node.OSImage,
Architecture: node.Architecture,
CapacityCPU: node.CapacityCPU,
CapacityMemoryBytes: node.CapacityMemoryBytes,
CapacityPods: node.CapacityPods,
AllocatableCPU: node.AllocCPU,
AllocatableMemoryBytes: node.AllocMemoryBytes,
AllocatablePods: node.AllocPods,
})
}
response := KubernetesNodesResponse{
Cluster: cluster.DisplayName,
Nodes: nodes,
Total: len(nodes),
}
if response.Nodes == nil {
response.Nodes = []KubernetesNodeSummary{}
}
return NewJSONResult(response), nil
}
}
return NewTextResult(fmt.Sprintf("Kubernetes cluster '%s' not found.", clusterArg)), nil
}
func (e *PulseToolExecutor) executeGetKubernetesPods(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
clusterArg, _ := args["cluster"].(string)
if clusterArg == "" {
return NewErrorResult(fmt.Errorf("cluster is required")), nil
}
namespaceFilter, _ := args["namespace"].(string)
statusFilter, _ := args["status"].(string)
limit := intArg(args, "limit", 100)
offset := intArg(args, "offset", 0)
state := e.stateProvider.GetState()
// Find the cluster (also match CustomDisplayName)
for _, c := range state.KubernetesClusters {
if c.ID == clusterArg || c.Name == clusterArg || c.DisplayName == clusterArg || c.CustomDisplayName == clusterArg {
displayName := c.DisplayName
if c.CustomDisplayName != "" {
displayName = c.CustomDisplayName
}
var pods []KubernetesPodSummary
totalPods := 0
filteredCount := 0
for _, pod := range c.Pods {
// Apply filters
if namespaceFilter != "" && pod.Namespace != namespaceFilter {
continue
}
if statusFilter != "" && !strings.EqualFold(pod.Phase, statusFilter) {
continue
}
filteredCount++
// Apply pagination
if totalPods < offset {
totalPods++
continue
}
if len(pods) >= limit {
totalPods++
continue
}
var containers []KubernetesPodContainerSummary
for _, container := range pod.Containers {
containers = append(containers, KubernetesPodContainerSummary{
Name: container.Name,
Ready: container.Ready,
State: container.State,
RestartCount: container.RestartCount,
Reason: container.Reason,
})
}
pods = append(pods, KubernetesPodSummary{
UID: pod.UID,
Name: pod.Name,
Namespace: pod.Namespace,
NodeName: pod.NodeName,
Phase: pod.Phase,
Reason: pod.Reason,
Restarts: pod.Restarts,
QoSClass: pod.QoSClass,
OwnerKind: pod.OwnerKind,
OwnerName: pod.OwnerName,
Containers: containers,
})
totalPods++
}
response := KubernetesPodsResponse{
Cluster: displayName,
Pods: pods,
Total: len(c.Pods),
Filtered: filteredCount,
}
if response.Pods == nil {
response.Pods = []KubernetesPodSummary{}
}
return NewJSONResult(response), nil
}
}
return NewTextResult(fmt.Sprintf("Kubernetes cluster '%s' not found.", clusterArg)), nil
}
func (e *PulseToolExecutor) executeGetKubernetesDeployments(_ context.Context, args map[string]interface{}) (CallToolResult, error) {
if e.stateProvider == nil {
return NewTextResult("State provider not available."), nil
}
clusterArg, _ := args["cluster"].(string)
if clusterArg == "" {
return NewErrorResult(fmt.Errorf("cluster is required")), nil
}
namespaceFilter, _ := args["namespace"].(string)
limit := intArg(args, "limit", 100)
offset := intArg(args, "offset", 0)
state := e.stateProvider.GetState()
// Find the cluster (also match CustomDisplayName)
for _, c := range state.KubernetesClusters {
if c.ID == clusterArg || c.Name == clusterArg || c.DisplayName == clusterArg || c.CustomDisplayName == clusterArg {
displayName := c.DisplayName
if c.CustomDisplayName != "" {
displayName = c.CustomDisplayName
}
var deployments []KubernetesDeploymentSummary
filteredCount := 0
count := 0
for _, dep := range c.Deployments {
// Apply namespace filter
if namespaceFilter != "" && dep.Namespace != namespaceFilter {
continue
}
filteredCount++
// Apply pagination
if count < offset {
count++
continue
}
if len(deployments) >= limit {
count++
continue
}
deployments = append(deployments, KubernetesDeploymentSummary{
UID: dep.UID,
Name: dep.Name,
Namespace: dep.Namespace,
DesiredReplicas: dep.DesiredReplicas,
ReadyReplicas: dep.ReadyReplicas,
AvailableReplicas: dep.AvailableReplicas,
UpdatedReplicas: dep.UpdatedReplicas,
})
count++
}
response := KubernetesDeploymentsResponse{
Cluster: displayName,
Deployments: deployments,
Total: len(c.Deployments),
Filtered: filteredCount,
}
if response.Deployments == nil {
response.Deployments = []KubernetesDeploymentSummary{}
}
return NewJSONResult(response), nil
}
}
return NewTextResult(fmt.Sprintf("Kubernetes cluster '%s' not found.", clusterArg)), nil
}
+211
View File
@@ -0,0 +1,211 @@
package tools
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
func TestExecuteGetCapabilities(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{
StateProvider: &mockStateProvider{},
AgentServer: &mockAgentServer{
agents: []agentexec.ConnectedAgent{
{Hostname: "host1", Version: "1.0", Platform: "linux"},
},
},
MetricsHistory: &mockMetricsHistoryProvider{},
BaselineProvider: &BaselineMCPAdapter{},
PatternProvider: &PatternMCPAdapter{},
AlertProvider: &mockAlertProvider{},
FindingsProvider: &mockFindingsProvider{},
ControlLevel: ControlLevelControlled,
ProtectedGuests: []string{"100"},
})
result, err := executor.executeGetCapabilities(context.Background())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
var response CapabilitiesResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &response); err != nil {
t.Fatalf("decode response: %v", err)
}
if response.ControlLevel != string(ControlLevelControlled) || response.ConnectedAgents != 1 {
t.Fatalf("unexpected response: %+v", response)
}
if !response.Features.Control || !response.Features.MetricsHistory {
t.Fatalf("unexpected features: %+v", response.Features)
}
}
func TestExecuteGetURLContent(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("X-Test", "ok")
w.WriteHeader(http.StatusOK)
w.Write([]byte("hello"))
}))
defer server.Close()
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
if result, _ := executor.executeGetURLContent(context.Background(), map[string]interface{}{}); !result.IsError {
t.Fatal("expected error when url missing")
}
result, err := executor.executeGetURLContent(context.Background(), map[string]interface{}{
"url": server.URL,
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
var response URLFetchResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &response); err != nil {
t.Fatalf("decode response: %v", err)
}
if response.StatusCode != http.StatusOK || response.Headers["X-Test"] != "ok" {
t.Fatalf("unexpected response: %+v", response)
}
}
func TestExecuteListInfrastructureAndTopology(t *testing.T) {
state := models.StateSnapshot{
Nodes: []models.Node{{ID: "node1", Name: "node1", Status: "online"}},
VMs: []models.VM{
{Name: "vm1", VMID: 100, Status: "running", Node: "node1"},
},
Containers: []models.Container{
{Name: "ct1", VMID: 200, Status: "stopped", Node: "node1"},
},
DockerHosts: []models.DockerHost{
{
ID: "host1",
Hostname: "h1",
DisplayName: "Host 1",
Containers: []models.DockerContainer{
{ID: "c1", Name: "nginx", State: "running", Image: "nginx"},
},
},
},
}
executor := NewPulseToolExecutor(ExecutorConfig{
StateProvider: &mockStateProvider{state: state},
AgentServer: &mockAgentServer{
agents: []agentexec.ConnectedAgent{{Hostname: "node1"}},
},
ControlLevel: ControlLevelControlled,
})
result, err := executor.executeListInfrastructure(context.Background(), map[string]interface{}{
"type": "vms",
"status": "running",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
var infra InfrastructureResponse
if err := json.Unmarshal([]byte(result.Content[0].Text), &infra); err != nil {
t.Fatalf("decode infra: %v", err)
}
if len(infra.VMs) != 1 || infra.VMs[0].Name != "vm1" {
t.Fatalf("unexpected infra response: %+v", infra)
}
// Topology includes derived node for VM reference if missing
state.Nodes = nil
executor.stateProvider = &mockStateProvider{state: state}
topologyResult, err := executor.executeGetTopology(context.Background())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
var topology TopologyResponse
if err := json.Unmarshal([]byte(topologyResult.Content[0].Text), &topology); err != nil {
t.Fatalf("decode topology: %v", err)
}
if topology.Summary.TotalVMs != 1 || len(topology.Proxmox.Nodes) == 0 {
t.Fatalf("unexpected topology: %+v", topology)
}
}
func TestExecuteSetResourceURLAndGetResource(t *testing.T) {
executor := NewPulseToolExecutor(ExecutorConfig{StateProvider: &mockStateProvider{}})
if result, _ := executor.executeSetResourceURL(context.Background(), map[string]interface{}{}); !result.IsError {
t.Fatal("expected error when resource_type missing")
}
updater := &fakeMetadataUpdater{}
executor.metadataUpdater = updater
result, err := executor.executeSetResourceURL(context.Background(), map[string]interface{}{
"resource_type": "guest",
"resource_id": "100",
"url": "http://example",
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(updater.resourceArgs) != 3 || updater.resourceArgs[2] != "http://example" {
t.Fatalf("unexpected updater args: %+v", updater.resourceArgs)
}
var setResp map[string]interface{}
if err := json.Unmarshal([]byte(result.Content[0].Text), &setResp); err != nil {
t.Fatalf("decode set response: %v", err)
}
if setResp["action"] != "set" {
t.Fatalf("unexpected set response: %+v", setResp)
}
state := models.StateSnapshot{
VMs: []models.VM{{ID: "vm1", VMID: 100, Name: "vm1", Status: "running", Node: "node1"}},
Containers: []models.Container{{ID: "ct1", VMID: 200, Name: "ct1", Status: "running", Node: "node1"}},
DockerHosts: []models.DockerHost{{
Hostname: "host",
Containers: []models.DockerContainer{{
ID: "abc123",
Name: "nginx",
State: "running",
Image: "nginx",
}},
}},
}
executor.stateProvider = &mockStateProvider{state: state}
resource, _ := executor.executeGetResource(context.Background(), map[string]interface{}{
"resource_type": "vm",
"resource_id": "100",
})
var res ResourceResponse
if err := json.Unmarshal([]byte(resource.Content[0].Text), &res); err != nil {
t.Fatalf("decode resource: %v", err)
}
if res.Type != "vm" || res.Name != "vm1" {
t.Fatalf("unexpected resource: %+v", res)
}
resource, _ = executor.executeGetResource(context.Background(), map[string]interface{}{
"resource_type": "docker",
"resource_id": "abc",
})
if err := json.Unmarshal([]byte(resource.Content[0].Text), &res); err != nil {
t.Fatalf("decode docker resource: %v", err)
}
if res.Type != "docker" || res.Name != "nginx" {
t.Fatalf("unexpected docker resource: %+v", res)
}
}
func TestIntArg(t *testing.T) {
if got := intArg(map[string]interface{}{}, "limit", 10); got != 10 {
t.Fatalf("unexpected default: %d", got)
}
if got := intArg(map[string]interface{}{"limit": float64(5)}, "limit", 10); got != 5 {
t.Fatalf("unexpected value: %d", got)
}
}
@@ -1,4 +1,4 @@
package mcp
package tools
import "context"