780 lines
29 KiB
Go
780 lines
29 KiB
Go
// Package mcp implements the Oikos MCP interface (plan R3-10).
|
|
// Uses the official MCP Go SDK with Streamable HTTP transport.
|
|
package mcp
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/dtoro/oikos/internal/db"
|
|
"github.com/google/jsonschema-go/jsonschema"
|
|
"github.com/google/uuid"
|
|
"github.com/modelcontextprotocol/go-sdk/mcp"
|
|
"golang.org/x/crypto/ssh"
|
|
)
|
|
|
|
// prop is one input-schema property (name → type + description).
|
|
type prop struct {
|
|
name, typ, desc string
|
|
}
|
|
|
|
// objSchema builds an "object" JSON Schema from a list of properties. The
|
|
// MCP SDK requires every tool to declare an object input schema so tools
|
|
// are self-describing to the agent; a nil schema panics at registration.
|
|
func objSchema(props ...prop) *jsonschema.Schema {
|
|
s := &jsonschema.Schema{Type: "object", Properties: map[string]*jsonschema.Schema{}}
|
|
for _, p := range props {
|
|
s.Properties[p.name] = &jsonschema.Schema{Type: p.typ, Description: p.desc}
|
|
}
|
|
return s
|
|
}
|
|
|
|
// NewHandler creates an http.Handler that serves the Oikos MCP server.
|
|
// agentID is the Hermes agent entity UUID; tool calls are logged to agent_activity.
|
|
func NewHandler(pool *db.Pool, token string, agentID uuid.UUID) http.Handler {
|
|
s := newServer(pool, agentID)
|
|
handler := mcp.NewStreamableHTTPHandler(func(r *http.Request) *mcp.Server {
|
|
if token != "" {
|
|
if r.Header.Get("Authorization") != "Bearer "+token {
|
|
return nil
|
|
}
|
|
}
|
|
return s
|
|
}, nil)
|
|
return handler
|
|
}
|
|
|
|
// toolHandler is the function signature registered via AddTool.
|
|
type toolHandler = mcp.ToolHandler
|
|
|
|
func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
|
|
s := mcp.NewServer(&mcp.Implementation{Name: "oikos", Version: "dev"}, &mcp.ServerOptions{
|
|
Logger: slog.Default(),
|
|
})
|
|
|
|
register := func(tool *mcp.Tool, handler toolHandler) {
|
|
s.AddTool(tool, withActivityLogging(pool, agentID, tool.Name, handler))
|
|
}
|
|
|
|
register(&mcp.Tool{Name: "get_entity", Description: "Get an entity by slug or UUID",
|
|
InputSchema: objSchema(prop{"slug_or_id", "string", "Entity slug (e.g. host:hubris) or UUID"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
idOrSlug, _ := args["slug_or_id"].(string)
|
|
return queryEntity(ctx, pool, idOrSlug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "list_entities", Description: "List entities filtered by type, state, or search",
|
|
InputSchema: objSchema(
|
|
prop{"type", "string", "Filter by entity type"},
|
|
prop{"state", "string", "Filter by lifecycle state"},
|
|
prop{"q", "string", "Substring match on slug or name"},
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state, e.version, e.created_at, e.updated_at
|
|
FROM entities e
|
|
WHERE ($1::text IS NULL OR e.type = $1)
|
|
AND ($2::text IS NULL OR e.state = $2)
|
|
AND ($3::text IS NULL OR e.slug ILIKE '%'||$3||'%' OR e.name ILIKE '%'||$3||'%')
|
|
ORDER BY e.slug LIMIT $4`,
|
|
nStr(args["type"]), nStr(args["state"]), nStr(args["q"]), limit), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_relations", Description: "Get relationships for an entity",
|
|
InputSchema: objSchema(prop{"entity_id", "string", "Entity slug"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_id"].(string)
|
|
return queryRows(ctx, pool, `
|
|
SELECT r.type, src.slug AS source, tgt.slug AS target
|
|
FROM relationships r
|
|
JOIN entities src ON src.id = r.source_id
|
|
JOIN entities tgt ON tgt.id = r.target_id
|
|
WHERE (src.slug = $1 OR tgt.slug = $1) AND r.valid_to IS NULL
|
|
ORDER BY r.type`, slug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_blast_radius", Description: "Find entities affected if this entity goes down",
|
|
InputSchema: objSchema(
|
|
prop{"entity_id", "string", "Entity slug"},
|
|
prop{"depth", "integer", "Traversal depth (default 3)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_id"].(string)
|
|
depth := int(getFloat(args, "depth", 3))
|
|
return queryRows(ctx, pool,
|
|
"SELECT slug, CAST(depth AS int) FROM blast_radius((SELECT id FROM entities WHERE slug = $1), $2)",
|
|
slug, depth), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_health_summary", Description: "Current fleet health summary",
|
|
InputSchema: objSchema(),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, st.health, st.last_check_at
|
|
FROM entity_status st JOIN entities e ON e.id = st.entity_id
|
|
ORDER BY e.slug`), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_audit_trail", Description: "Query the audit log",
|
|
InputSchema: objSchema(prop{"entity_id", "string", "Filter by affected entity UUID"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
return queryRows(ctx, pool, `
|
|
SELECT id, ts, actor_type, action, entity_id::text, method, path, correlation_id
|
|
FROM audit_log
|
|
WHERE ($1::text IS NULL OR entity_id::text = $1)
|
|
ORDER BY ts DESC LIMIT 50`, nStr(args["entity_id"])), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "search_knowledge", Description: "Full-text search across documentation",
|
|
InputSchema: objSchema(prop{"query", "string", "Search terms"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
q := nStr(args["query"])
|
|
return queryRows(ctx, pool, `
|
|
SELECT id, title, LEFT(content, 500) AS preview
|
|
FROM knowledge_entities
|
|
WHERE ($1::text IS NULL OR title ILIKE '%'||$1||'%' OR content ILIKE '%'||$1||'%')
|
|
ORDER BY title LIMIT 20`, q), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "query_metrics", Description: "Query time-series metrics",
|
|
InputSchema: objSchema(prop{"hours", "integer", "Look-back window in hours (default 24)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
hours := int(getFloat(args, "hours", 24))
|
|
return queryRows(ctx, pool, `
|
|
SELECT time_bucket('1 hour', ts) AS bucket,
|
|
entity_id::text, metric,
|
|
ROUND(avg(value)::numeric, 2) AS avg,
|
|
ROUND(min(value)::numeric, 2) AS min,
|
|
ROUND(max(value)::numeric, 2) AS max
|
|
FROM metric_samples
|
|
WHERE ts > now() - make_interval(hours => $1)
|
|
GROUP BY bucket, entity_id, metric
|
|
ORDER BY bucket DESC LIMIT 100`, hours), nil
|
|
})
|
|
|
|
// ─── Phase 4: new tools ──────────────────────────────────────────
|
|
|
|
register(&mcp.Tool{Name: "get_signal_history", Description: "Query open and recent signals",
|
|
InputSchema: objSchema(
|
|
prop{"entity_slug", "string", "Filter by target entity slug"},
|
|
prop{"state", "string", "Filter by signal state (raised, resolved)"},
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return queryRows(ctx, pool, `
|
|
SELECT s.entity_id::text, s.kind, s.severity, s.state,
|
|
s.occurrence_count, e.slug AS target_slug,
|
|
s.first_seen_at, s.last_seen_at
|
|
FROM signals s
|
|
LEFT JOIN entities e ON e.id = s.target_entity_id
|
|
WHERE ($1::text IS NULL OR e.slug = $1)
|
|
AND ($2::text IS NULL OR s.state = $2)
|
|
ORDER BY s.last_seen_at DESC LIMIT $3`,
|
|
nStr(args["entity_slug"]), nStr(args["state"]), limit), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_patterns", Description: "List learned action patterns",
|
|
InputSchema: objSchema(
|
|
prop{"status", "string", "Filter by status (hypothesized, validated, active)"},
|
|
prop{"entity_type", "string", "Filter by applies_type"},
|
|
prop{"action", "string", "Filter by action"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
return queryRows(ctx, pool, `
|
|
SELECT p.entity_id::text, p.applies_type, p.action, p.pattern,
|
|
p.confidence, p.evidence_count, p.success_count, p.failure_count,
|
|
p.status, p.quarantined, p.version, p.last_validated_at
|
|
FROM patterns p
|
|
WHERE ($1::text IS NULL OR p.status = $1)
|
|
AND ($2::text IS NULL OR p.applies_type = $2)
|
|
AND ($3::text IS NULL OR p.action = $3)
|
|
ORDER BY p.applies_type, p.action`,
|
|
nStr(args["status"]), nStr(args["entity_type"]), nStr(args["action"])), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_skills", Description: "List available automation skills",
|
|
InputSchema: objSchema(
|
|
prop{"status", "string", "Filter by status (active, inactive, deprecated)"},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
return queryRows(ctx, pool, `
|
|
SELECT s.entity_id::text, s.version, s.name, LEFT(s.procedure::text, 300) AS procedure_preview,
|
|
s.applies_type, s.action, s.status, s.success_rate,
|
|
s.changed_by::text, s.change_reason, s.last_used_at
|
|
FROM skills s
|
|
WHERE ($1::text IS NULL OR s.status = $1)
|
|
ORDER BY s.name, s.version DESC`,
|
|
nStr(args["status"])), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "request_execution", Description: "Request a gated execution (Hermes-only mutation path). Supported actions: restart, systemctl, pct_exec, apt_upgrade.",
|
|
InputSchema: objSchema(
|
|
prop{"target", "string", "Target entity slug (e.g. lxc:caddy)"},
|
|
prop{"action", "string", "Action: restart, systemctl, pct_exec, apt_upgrade"},
|
|
prop{"params", "string", "Extra params: for systemctl use 'enable|disable|reload', for pct_exec use the shell command, for apt_upgrade use 'audit|upgrade'"},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
targetSlug, _ := args["target"].(string)
|
|
action, _ := args["action"].(string)
|
|
params, _ := args["params"].(string)
|
|
if targetSlug == "" || action == "" {
|
|
return textResult("error: target and action required"), nil
|
|
}
|
|
|
|
var targetID uuid.UUID
|
|
if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", targetSlug).Scan(&targetID); err != nil {
|
|
return textResult(fmt.Sprintf("target not found: %s", targetSlug)), nil
|
|
}
|
|
|
|
id, _ := uuid.NewV7()
|
|
correlationID := uuid.New().String()
|
|
|
|
// Write execution record
|
|
execSlug := "exec:" + targetSlug + ":" + id.String()[:8]
|
|
pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
|
|
id, execSlug, action+" on "+targetSlug)
|
|
pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, 'reversible_low', 'running', $4, $5)`,
|
|
id, targetID, action+":"+params, correlationID, agentID)
|
|
|
|
// Execute reversible actions immediately
|
|
switch action {
|
|
case "restart":
|
|
host, user, err := resolveHost(ctx, pool, targetSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve: %v", err)), nil
|
|
}
|
|
svc := strings.TrimPrefix(targetSlug, "lxc:")
|
|
out, err := sshExec(ctx, host, user, fmt.Sprintf("systemctl restart %s 2>&1; sleep 1; systemctl is-active %s", svc, svc))
|
|
result := fmt.Sprintf("restart %s: %s", svc, out)
|
|
if err != nil {
|
|
result = fmt.Sprintf("restart %s: ERROR %v", svc, err)
|
|
}
|
|
pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`,
|
|
id, fmt.Sprintf(`{"output":"%s"}`, strings.ReplaceAll(out, "\n", "\\n")))
|
|
return textResult(result), nil
|
|
|
|
case "systemctl":
|
|
host, user, err := resolveHost(ctx, pool, targetSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve: %v", err)), nil
|
|
}
|
|
svc := strings.TrimPrefix(targetSlug, "lxc:")
|
|
cmd := fmt.Sprintf("systemctl %s %s 2>&1; sleep 1; systemctl is-active %s", params, svc, svc)
|
|
if params == "enable" || params == "disable" {
|
|
// config_mutation — mark as pending for operator approval
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
return textResult(fmt.Sprintf("systemctl %s on %s requires approval — execution %s queued", params, svc, id)), nil
|
|
}
|
|
out, err := sshExec(ctx, host, user, cmd)
|
|
result := fmt.Sprintf("systemctl %s %s: %s", params, svc, out)
|
|
if err != nil {
|
|
result = fmt.Sprintf("systemctl %s %s: ERROR %v", params, svc, err)
|
|
}
|
|
pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`,
|
|
id, fmt.Sprintf(`{"output":"%s"}`, strings.ReplaceAll(out, "\n", "\\n")))
|
|
return textResult(result), nil
|
|
|
|
case "pct_exec":
|
|
var pveID string
|
|
if err := pool.QueryRow(ctx, "SELECT attributes->>'pve_id' FROM entities WHERE slug = $1", targetSlug).Scan(&pveID); err != nil || pveID == "" {
|
|
return textResult(fmt.Sprintf("LXC not found: %s", targetSlug)), nil
|
|
}
|
|
// Resolve Proxmox host
|
|
var hostSlug string
|
|
pool.QueryRow(ctx, "SELECT attributes->>'host' FROM entities WHERE slug = $1", targetSlug).Scan(&hostSlug)
|
|
if hostSlug == "" {
|
|
hostSlug = "host:hubris" // default
|
|
}
|
|
host, user, err := resolveHost(ctx, pool, hostSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve Proxmox host: %v", err)), nil
|
|
}
|
|
out, err := sshExec(ctx, host, user, fmt.Sprintf("pct exec %s -- %s 2>&1", pveID, params))
|
|
result := fmt.Sprintf("pct exec %s: %s", pveID, out)
|
|
if err != nil {
|
|
result = fmt.Sprintf("pct exec %s: ERROR %v", pveID, err)
|
|
}
|
|
pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`,
|
|
id, fmt.Sprintf(`{"output":"%s"}`, strings.ReplaceAll(out, "\n", "\\n")))
|
|
return textResult(result), nil
|
|
|
|
case "apt_upgrade":
|
|
if params == "audit" {
|
|
host, user, err := resolveHost(ctx, pool, targetSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve: %v", err)), nil
|
|
}
|
|
out, err := sshExec(ctx, host, user, "apt update -qq 2>&1 >/dev/null; apt list --upgradable 2>/dev/null | tail -n +2 | wc -l; apt list --upgradable 2>/dev/null | tail -n +2 | head -20")
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("apt audit error: %v", err)), nil
|
|
}
|
|
return textResult("apt audit:\n" + out), nil
|
|
}
|
|
// upgrade requires approval — queue
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
return textResult(fmt.Sprintf("apt_upgrade on %s requires approval — execution %s queued", targetSlug, id)), nil
|
|
|
|
default:
|
|
return textResult(fmt.Sprintf("unknown action: %s. Supported: restart, systemctl, pct_exec, apt_upgrade", action)), nil
|
|
}
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_trend", Description: "Get metric trends for an entity",
|
|
InputSchema: objSchema(
|
|
prop{"entity_id", "string", "Entity slug"},
|
|
prop{"days", "integer", "Look-back window in days (default 7)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_id"].(string)
|
|
days := int(getFloat(args, "days", 7))
|
|
return queryRows(ctx, pool, `
|
|
SELECT metric,
|
|
ROUND(avg(value)::numeric, 2) AS avg_val,
|
|
ROUND(stddev(value)::numeric, 2) AS std_val,
|
|
count(*) AS sample_count,
|
|
ROUND(regr_slope(value, EXTRACT(EPOCH FROM ts)::numeric)::numeric, 4) AS slope
|
|
FROM metric_samples ms
|
|
JOIN entities e ON e.id = ms.entity_id
|
|
WHERE e.slug = $1 AND ts >= now() - make_interval(days => $2)
|
|
GROUP BY metric
|
|
ORDER BY metric`, slug, days), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_event_timeline", Description: "Get recent events",
|
|
InputSchema: objSchema(
|
|
prop{"severity", "string", "Filter by severity (info, warn, error)"},
|
|
prop{"entity_slug", "string", "Filter by entity slug"},
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return queryRows(ctx, pool, `
|
|
SELECT ev.ts, ev.event_type, ev.severity, ev.actor, e.slug AS entity_slug,
|
|
ev.message, ev.correlation_id
|
|
FROM events ev
|
|
LEFT JOIN entities e ON e.id = ev.entity_id
|
|
WHERE ($1::text IS NULL OR ev.severity = $1)
|
|
AND ($2::text IS NULL OR e.slug = $2)
|
|
ORDER BY ev.ts DESC LIMIT $3`,
|
|
nStr(args["severity"]), nStr(args["entity_slug"]), limit), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_agent_activity", Description: "Agent self-inspection: query agent activity log",
|
|
InputSchema: objSchema(
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return queryRows(ctx, pool, `
|
|
SELECT id, ts, agent_id::text, session_id, activity_type, tool_name,
|
|
entity_id::text, left(input_summary, 200) AS input_summary,
|
|
left(output_summary, 200) AS output_summary,
|
|
duration_ms, token_count, success, correlation_id
|
|
FROM agent_activity
|
|
WHERE agent_id = $1
|
|
ORDER BY ts DESC LIMIT $2`, agentID, limit), nil
|
|
})
|
|
|
|
// ─── Phase 5: operational MCP tools ──────────────────────────────
|
|
|
|
register(&mcp.Tool{Name: "list_lxcs", Description: "List all LXC containers with ID, host, IP, and state",
|
|
InputSchema: objSchema(),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.name, e.attributes->>'pve_id' AS pve_id,
|
|
e.attributes->>'lan_ip' AS lan_ip,
|
|
st.health, st.last_check_at
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.type = 'lxc'
|
|
ORDER BY (e.attributes->>'pve_id')::int`), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "ping_service", Description: "Check if a service is reachable via HTTP",
|
|
InputSchema: objSchema(prop{"service_slug", "string", "Service entity slug"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("service_slug is required"), nil
|
|
}
|
|
rows, err := pool.Query(ctx, `
|
|
SELECT st.health, st.last_check_at, e.attributes->>'url' AS url
|
|
FROM entity_status st
|
|
JOIN entities e ON e.id = st.entity_id
|
|
WHERE e.slug = $1`, slug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("query error: %v", err)), nil
|
|
}
|
|
defer rows.Close()
|
|
if !rows.Next() {
|
|
return textResult(fmt.Sprintf("service not found: %s", slug)), nil
|
|
}
|
|
var health, lastCheck, url string
|
|
rows.Scan(&health, &lastCheck, &url)
|
|
if url == "" {
|
|
url = "(no URL in entity attributes)"
|
|
}
|
|
return textResult(fmt.Sprintf("health=%s last_check=%s url=%s", health, lastCheck, url)), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "tail_log", Description: "Get recent log lines from a service via journalctl",
|
|
InputSchema: objSchema(
|
|
prop{"service_slug", "string", "Service entity slug (e.g. lxc:caddy)"},
|
|
prop{"lines", "integer", "Number of lines (default 50)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
n := int(getFloat(args, "lines", 50))
|
|
if slug == "" {
|
|
return textResult("service_slug is required"), nil
|
|
}
|
|
host, user, err := resolveHost(ctx, pool, slug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve host: %v", err)), nil
|
|
}
|
|
svc := strings.TrimPrefix(slug, "lxc:")
|
|
out, err := sshExec(ctx, host, user, fmt.Sprintf("journalctl -u %s -n %d --no-pager 2>&1 || true", svc, n))
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("ssh: %v", err)), nil
|
|
}
|
|
return textResult(out), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_service_status", Description: "Check systemd service status on a host",
|
|
InputSchema: objSchema(
|
|
prop{"service_slug", "string", "Service entity slug (e.g. lxc:caddy)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("service_slug is required"), nil
|
|
}
|
|
host, user, err := resolveHost(ctx, pool, slug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve host: %v", err)), nil
|
|
}
|
|
svc := strings.TrimPrefix(slug, "lxc:")
|
|
out, err := sshExec(ctx, host, user,
|
|
fmt.Sprintf("systemctl is-active %s; systemctl is-enabled %s; systemctl show %s -p ActiveEnterTimestamp -p SubState 2>&1 || true", svc, svc, svc))
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("ssh: %v", err)), nil
|
|
}
|
|
return textResult(out), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_lxc_state", Description: "Get LXC container resource state from Proxmox host",
|
|
InputSchema: objSchema(
|
|
prop{"lxc_slug", "string", "LXC entity slug (e.g. lxc:caddy)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["lxc_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("lxc_slug is required"), nil
|
|
}
|
|
var pveID string
|
|
err := pool.QueryRow(ctx, "SELECT attributes->>'pve_id' FROM entities WHERE slug = $1", slug).Scan(&pveID)
|
|
if err != nil || pveID == "" {
|
|
return textResult(fmt.Sprintf("LXC not found or missing pve_id: %s", slug)), nil
|
|
}
|
|
// Resolve the Proxmox host — find the host that runs this LXC
|
|
var hostID uuid.UUID
|
|
err = pool.QueryRow(ctx, `
|
|
SELECT t.id FROM entities t
|
|
JOIN relationships r ON r.source_id = t.id
|
|
JOIN entities s ON s.id = r.target_id
|
|
WHERE s.slug = $1 AND r.type = 'hosts' AND r.valid_to IS NULL
|
|
LIMIT 1`, slug).Scan(&hostID)
|
|
if err != nil {
|
|
// Fallback: use the inventory host attribute if no relationship
|
|
var hostSlug string
|
|
err = pool.QueryRow(ctx, "SELECT attributes->>'host' FROM entities WHERE slug = $1", slug).Scan(&hostSlug)
|
|
if err != nil || hostSlug == "" {
|
|
return textResult(fmt.Sprintf("cannot resolve Proxmox host for %s", slug)), nil
|
|
}
|
|
var host, user string
|
|
host, user, err = resolveHost(ctx, pool, "host:"+hostSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve: %v", err)), nil
|
|
}
|
|
out, err2 := sshExec(ctx, host, user, fmt.Sprintf("pct status %s --verbose 2>&1 || true", pveID))
|
|
if err2 != nil {
|
|
return textResult(fmt.Sprintf("ssh: %v", err2)), nil
|
|
}
|
|
return textResult(out), nil
|
|
}
|
|
var hostSlug string
|
|
pool.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", hostID).Scan(&hostSlug)
|
|
host, user, err := resolveHost(ctx, pool, hostSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve host: %v", err)), nil
|
|
}
|
|
out, err := sshExec(ctx, host, user, fmt.Sprintf("pct status %s --verbose 2>&1 || true", pveID))
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("ssh: %v", err)), nil
|
|
}
|
|
return textResult(out), nil
|
|
})
|
|
|
|
return s
|
|
}
|
|
|
|
// withActivityLogging wraps a tool handler to record agent_activity rows.
|
|
func withActivityLogging(pool *db.Pool, agentID uuid.UUID, toolName string, next mcp.ToolHandler) mcp.ToolHandler {
|
|
if agentID == uuid.Nil {
|
|
return next
|
|
}
|
|
return func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
start := time.Now()
|
|
result, err := next(ctx, req)
|
|
duration := int(time.Since(start).Milliseconds())
|
|
|
|
// Build input summary (first 500 chars of args)
|
|
inputSummary := ""
|
|
if req != nil && len(req.Params.Arguments) > 0 {
|
|
inputSummary = string(req.Params.Arguments)
|
|
}
|
|
if len(inputSummary) > 500 {
|
|
inputSummary = inputSummary[:500]
|
|
}
|
|
|
|
// Build output summary
|
|
outputSummary := ""
|
|
success := err == nil
|
|
if result != nil {
|
|
for _, c := range result.Content {
|
|
if tc, ok := c.(*mcp.TextContent); ok {
|
|
outputSummary = tc.Text
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if err != nil {
|
|
outputSummary = err.Error()
|
|
success = false
|
|
}
|
|
if len(outputSummary) > 500 {
|
|
outputSummary = outputSummary[:500]
|
|
}
|
|
|
|
correlationID := uuid.New().String()
|
|
|
|
_, logErr := pool.Exec(ctx, `
|
|
INSERT INTO agent_activity
|
|
(agent_id, activity_type, tool_name, input_summary, output_summary,
|
|
duration_ms, success, correlation_id)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
|
|
agentID, "tool_call", toolName, inputSummary, outputSummary,
|
|
duration, success, correlationID)
|
|
if logErr != nil {
|
|
slog.Warn("mcp: log agent_activity", "error", logErr)
|
|
}
|
|
|
|
return result, err
|
|
}
|
|
}
|
|
|
|
// ─── Helpers ──────────────────────────────────────────────────────────
|
|
|
|
func argsMap(req *mcp.CallToolRequest) map[string]any {
|
|
if req == nil || len(req.Params.Arguments) == 0 {
|
|
return nil
|
|
}
|
|
var m map[string]any
|
|
json.Unmarshal(req.Params.Arguments, &m)
|
|
return m
|
|
}
|
|
|
|
func getFloat(m map[string]any, key string, def float64) float64 {
|
|
if m == nil {
|
|
return def
|
|
}
|
|
switch v := m[key].(type) {
|
|
case float64:
|
|
return v
|
|
case int:
|
|
return float64(v)
|
|
case json.Number:
|
|
f, err := v.Float64()
|
|
if err != nil {
|
|
return def
|
|
}
|
|
return f
|
|
}
|
|
return def
|
|
}
|
|
|
|
func nStr(v any) any {
|
|
if v == nil {
|
|
return nil
|
|
}
|
|
s, _ := v.(string)
|
|
if s == "" {
|
|
return nil
|
|
}
|
|
return s
|
|
}
|
|
|
|
func textResult(s string) *mcp.CallToolResult {
|
|
return &mcp.CallToolResult{
|
|
Content: []mcp.Content{&mcp.TextContent{Text: s}},
|
|
}
|
|
}
|
|
|
|
func queryEntity(ctx context.Context, pool *db.Pool, idOrSlug string) *mcp.CallToolResult {
|
|
var id uuid.UUID
|
|
if u, err := uuid.Parse(idOrSlug); err == nil {
|
|
id = u
|
|
} else {
|
|
pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", idOrSlug).Scan(&id)
|
|
}
|
|
if id == uuid.Nil {
|
|
return textResult(fmt.Sprintf("entity not found: %s", idOrSlug))
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT slug, type, name, state, attributes,
|
|
maintenance_until::text, version, created_at, updated_at
|
|
FROM entities WHERE id = $1`, id)
|
|
}
|
|
|
|
func queryRows(ctx context.Context, pool *db.Pool, query string, args ...any) *mcp.CallToolResult {
|
|
rows, err := pool.Query(ctx, query, args...)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err))
|
|
}
|
|
defer rows.Close()
|
|
|
|
cols := rows.FieldDescriptions()
|
|
var items []map[string]any
|
|
for rows.Next() {
|
|
vals, err := rows.Values()
|
|
if err != nil {
|
|
continue
|
|
}
|
|
m := make(map[string]any)
|
|
for i, col := range cols {
|
|
m[string(col.Name)] = fmt.Sprintf("%v", vals[i])
|
|
}
|
|
items = append(items, m)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err))
|
|
}
|
|
|
|
data, _ := json.MarshalIndent(items, "", " ")
|
|
return textResult(string(data))
|
|
}
|
|
|
|
// ─── SSH helpers ─────────────────────────────────────────────────────────
|
|
|
|
var (
|
|
sshUser string
|
|
sshKey []byte
|
|
sshPool = make(map[string]*ssh.Client)
|
|
sshPoolMu sync.Mutex
|
|
)
|
|
|
|
func initSSH() {
|
|
if sshUser == "" {
|
|
sshUser = os.Getenv("OIKOS_SSH_USER")
|
|
if sshUser == "" {
|
|
sshUser = "root"
|
|
}
|
|
}
|
|
keyPath := os.Getenv("OIKOS_SSH_KEY_PATH")
|
|
if keyPath == "" {
|
|
keyPath = "/etc/oikos/ssh_key"
|
|
}
|
|
if len(sshKey) == 0 {
|
|
var err error
|
|
sshKey, err = os.ReadFile(keyPath)
|
|
if err != nil {
|
|
slog.Warn("mcp ssh: cannot read key", "path", keyPath, "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func sshExec(ctx context.Context, host, user, command string) (string, error) {
|
|
initSSH()
|
|
if len(sshKey) == 0 {
|
|
return "", fmt.Errorf("no SSH key available")
|
|
}
|
|
if user == "" {
|
|
user = sshUser
|
|
}
|
|
|
|
addr := host + ":22"
|
|
signer, err := ssh.ParsePrivateKey(sshKey)
|
|
if err != nil {
|
|
return "", fmt.Errorf("parse key: %w", err)
|
|
}
|
|
|
|
cfg := &ssh.ClientConfig{
|
|
User: user,
|
|
Auth: []ssh.AuthMethod{ssh.PublicKeys(signer)},
|
|
HostKeyCallback: ssh.InsecureIgnoreHostKey(),
|
|
Timeout: 10 * time.Second,
|
|
}
|
|
|
|
client, err := ssh.Dial("tcp", addr, cfg)
|
|
if err != nil {
|
|
return "", fmt.Errorf("dial %s: %w", host, err)
|
|
}
|
|
defer client.Close()
|
|
|
|
session, err := client.NewSession()
|
|
if err != nil {
|
|
return "", fmt.Errorf("session: %w", err)
|
|
}
|
|
defer session.Close()
|
|
|
|
out, err := session.CombinedOutput(command)
|
|
if err != nil && out == nil {
|
|
return "", fmt.Errorf("exec: %w", err)
|
|
}
|
|
return strings.TrimSpace(string(out)), nil
|
|
}
|
|
|
|
func resolveHost(ctx context.Context, pool *db.Pool, entitySlug string) (hostIP string, sshUser string, err error) {
|
|
var attrs string
|
|
err = pool.QueryRow(ctx, "SELECT attributes::text FROM entities WHERE slug = $1", entitySlug).Scan(&attrs)
|
|
if err != nil {
|
|
return "", "", fmt.Errorf("entity not found: %s", entitySlug)
|
|
}
|
|
|
|
var m map[string]interface{}
|
|
if err := json.Unmarshal([]byte(attrs), &m); err != nil {
|
|
return "", "", fmt.Errorf("parse attributes: %w", err)
|
|
}
|
|
|
|
if ip, ok := m["lan_ip"].(string); ok && ip != "" {
|
|
return ip, sshUser, nil
|
|
}
|
|
if mesh, ok := m["mesh"].(map[string]interface{}); ok {
|
|
for _, proto := range []string{"netbird", "tailscale"} {
|
|
if p, ok := mesh[proto].(map[string]interface{}); ok {
|
|
if ip, ok := p["ip"].(string); ok && ip != "" {
|
|
return ip, sshUser, nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return "", "", fmt.Errorf("no IP found for %s", entitySlug)
|
|
} |