dtoro
8950bada44
Root cause of "running for 10+ minutes without stopping": a real production execution (TypeType pct_create) was found genuinely stuck 17+ minutes into a single blocking SSH call. The container's post_install script was looping on `getent hosts deb.debian.org`, waiting on a network that could never come up — the operator's static IP config used gw:192.168.8.1, but the actual gateway on that subnet is 192.168.8.2, so every network call hung instead of failing fast (packets dropped, not rejected). Two compounding bugs made this unrecoverable without manual intervention: 1. sshExec (both internal/httpapi/phase3.go and internal/mcp/server.go) had NO execution timeout — `session.CombinedOutput()` blocks until the remote command exits, with no deadline. A hung remote process blocks the Go goroutine forever; the execution can never leave 'running', and the operator has no way to make it stop. Fixed: both now race the SSH call against a 10-minute hard timeout, closing the session/client and returning a clear "timed out after 10m0s" error if exceeded. (The mcp/server.go copy also still had the original "swallowed non-zero exit" bug from before that fix was applied to httpapi's copy only — fixed here too.) 2. provisionScript's DNS-wait loop assumed `getent hosts` fails fast on no connectivity — it doesn't; a black-holed network can make each call hang far past the resolver's nominal timeout, so the documented "~90s" budget was never real. Wrapped every attempt in `timeout 3` so the wall-clock budget is now actually enforced (~2min worst case), and the failure message now suggests checking the net0 gateway. Also fixes the matching UI-side gap (operator's literal question: "is there a way to get more details? it has been running for 10+ minutes without stopping"): - InlineApproval's track() polling loop had its own ~6min ceiling and simply STOPPED polling after that — silently going stale before the backend (now correctly capped at 10min) could ever resolve. Raised to a 14min ceiling with margin, and added a distinct 'stalled' state if that's ever exceeded (explicitly says something's wrong, rather than freezing silently). - The running-card now shows live elapsed time (ticking, from the execution's created_at), the actual command being run, and the execution ID — previously just a static "this can take a minute" with zero information. Also added command display to the destructive pending- approval card for full transparency before confirming. Verified live end-to-end in a real browser (dev server proxying to production): queued a real command via chat, approved via the button, watched the elapsed-time counter tick in real time, and saw it transition to a completed card with real output once the command finished. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1291 lines
54 KiB
Go
1291 lines
54 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/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"html"
|
|
"io"
|
|
"log/slog"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"regexp"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/dtoro/oikos/internal/db"
|
|
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
|
"github.com/dtoro/oikos/internal/observability"
|
|
"github.com/dtoro/oikos/internal/policy"
|
|
"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 Nomos 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 e.slug, CAST(b.depth AS int) FROM blast_radius((SELECT id FROM entities WHERE slug = $1), $2) b JOIN entities e ON e.id = b.entity_id",
|
|
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
|
|
WHERE e.type <> 'check'
|
|
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 (PostgreSQL FTS with ts_rank ranking)",
|
|
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 ke.title, e.slug,
|
|
ts_rank(ke.search, plainto_tsquery('english', $1)) AS rank,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', $1),
|
|
'MaxWords=40, MinWords=15, ShortWord=3, MaxFragments=3,
|
|
FragmentDelimiter=" ... "') AS snippet,
|
|
ke.source, ke.tags
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
WHERE ke.search @@ plainto_tsquery('english', $1)
|
|
ORDER BY rank DESC
|
|
LIMIT 20`, q), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_entity_knowledge", Description: "All documents, investigations, and runbooks linked to an entity",
|
|
InputSchema: objSchema(prop{"entity_slug", "string", "Entity slug (e.g. lxc:jellyfin, service:caddy)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_slug"].(string)
|
|
return queryRows(ctx, pool, `
|
|
SELECT ke.title, ke.source, e.type AS kind, e.slug,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', '')) AS headline
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
JOIN relationships r ON r.source_id = ke.entity_id
|
|
JOIN entities target ON target.id = r.target_id
|
|
WHERE target.slug = $1
|
|
AND r.valid_to IS NULL
|
|
AND r.type IN ('documents', 'about')
|
|
UNION
|
|
SELECT ke.title, ke.source, e.type AS kind, e.slug,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', '')) AS headline
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
JOIN relationships r ON r.source_id = ke.entity_id
|
|
JOIN entity_types target_type ON target_type.name = (SELECT type FROM entities WHERE slug = $1)
|
|
JOIN entities ent ON ent.type = target_type.name AND ent.slug = $1
|
|
WHERE r.valid_to IS NULL
|
|
AND r.type = 'procedure-for'
|
|
ORDER BY 1`, slug), 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 (agent-only mutation path). Supported actions: restart, systemctl, pct_exec, apt_upgrade, pct_create. pct_create provisions a NEW LXC and installs its service in one approved step — you do NOT need follow-up pct_exec calls for package installs.",
|
|
InputSchema: objSchema(
|
|
prop{"target", "string", "Target entity slug. For pct_create this MUST be the Proxmox HOST that will run the container (e.g. host:strong) — NOT the new LXC's name. For restart/systemctl/apt_upgrade/pct_exec use the target service/LXC slug (e.g. lxc:caddy)."},
|
|
prop{"action", "string", "Action: restart, systemctl, pct_exec, apt_upgrade, pct_create"},
|
|
prop{"params", "string", "For systemctl: 'enable|disable|reload'. For pct_exec: the shell command. For apt_upgrade: 'audit|upgrade'. For pct_create: a JSON object string with keys: vmid (int, required, unused id), hostname (string, required), cores (int), memory (MB int), disk_gb (int), ip (CIDR e.g. 192.168.8.50/24), gw (gateway ip), storage (default local-lvm), template (optional — omit to auto-pick newest debian on the host), privileged (bool), nesting (bool), mounts ([]string of 'src,mp=/dst'), services ([]string of apt packages to install), post_install (string shell script run inside the container after create). Example: {\"vmid\":150,\"hostname\":\"typetype\",\"cores\":2,\"memory\":2048,\"disk_gb\":16,\"ip\":\"192.168.8.50/24\",\"gw\":\"192.168.8.2\",\"nesting\":true,\"services\":[\"docker.io\",\"git\"],\"post_install\":\"git clone https://github.com/x/y /opt/y && cd /opt/y && docker compose up -d\"}"},
|
|
),
|
|
}, 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
|
|
}
|
|
|
|
// Deduplicate: if a pending execution already exists for the same
|
|
// target+action, return the existing one instead of creating a
|
|
// duplicate. Prevents the LLM from re-requesting the same gated
|
|
// action in a tool-calling loop. Only blocks when a pending
|
|
// execution exists; completed/failed ones don't block.
|
|
if action == "systemctl" || action == "apt_upgrade" || action == "pct_create" {
|
|
execNamePrefix := action + " on " + targetSlug
|
|
var existingID string
|
|
err := pool.QueryRow(ctx, `
|
|
SELECT e.id::text FROM entities e
|
|
JOIN executions ex ON ex.entity_id = e.id
|
|
WHERE e.type = 'execution' AND e.name LIKE $1 AND ex.status = 'pending_approval'
|
|
ORDER BY e.created_at DESC LIMIT 1`, execNamePrefix+"%").Scan(&existingID)
|
|
if err == nil && existingID != "" {
|
|
return textResult(fmt.Sprintf("%s on %s is already queued for approval — execution %s. Wait for operator approval. Do not re-request.",
|
|
action, targetSlug, existingID)), nil
|
|
}
|
|
}
|
|
|
|
id, _ := uuid.NewV7()
|
|
correlationID := uuid.New().String()
|
|
|
|
// Full UUID, not a truncated prefix: UUIDv7's leading bytes encode a
|
|
// millisecond timestamp, so an 8-char prefix collides for real under
|
|
// back-to-back requests (observed live: two `run` calls seconds
|
|
// apart hit entities_slug_key). The full string is guaranteed unique.
|
|
execName := action + " on " + targetSlug + " (" + id.String() + ")"
|
|
execSlug := "exec:" + targetSlug + ":" + id.String()
|
|
_, err := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
|
|
id, execSlug, execName)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: failed to create execution: %v", err)), nil
|
|
}
|
|
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) ON CONFLICT DO NOTHING`,
|
|
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, jsonOut(out))
|
|
return textResult(result), nil
|
|
|
|
case "systemctl":
|
|
svc := strings.TrimPrefix(targetSlug, "lxc:")
|
|
if params == "enable" || params == "disable" {
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
createApproval(ctx, pool, id, targetID, "systemctl", svc+":"+params, "config_mutation")
|
|
return textResult(fmt.Sprintf("systemctl %s on %s requires approval — execution %s queued", params, svc, id)), nil
|
|
}
|
|
host, user, err := resolveHost(ctx, pool, targetSlug)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("resolve: %v", err)), nil
|
|
}
|
|
cmd := fmt.Sprintf("systemctl %s %s 2>&1; sleep 1; systemctl is-active %s", params, svc, svc)
|
|
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, jsonOut(out))
|
|
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, jsonOut(out))
|
|
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)
|
|
createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation")
|
|
return textResult(fmt.Sprintf("apt_upgrade on %s requires approval — execution %s queued", targetSlug, id)), nil
|
|
|
|
case "pct_create":
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
|
|
createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation")
|
|
return textResult(fmt.Sprintf("pct_create on %s requires approval — execution %s queued. The LXC will be provisioned once approved.", targetSlug, id)), nil
|
|
|
|
default:
|
|
return textResult(fmt.Sprintf("unknown action: %s. Supported: restart, systemctl, pct_exec, apt_upgrade, pct_create", action)), nil
|
|
}
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "run", Description: "Run ANY shell command against any host or LXC. This is the general execution primitive — prefer it over asking the operator to run something manually, and don't wait for a matching fixed action to exist. Every command is automatically risk-classified: read-only inspection (cat, systemctl status, docker ps, journalctl, df, git status, ...) runs immediately; anything that changes state requires operator approval (granted by the operator replying \"go ahead\"/\"yes\" in chat, or via the Approve button); commands matching a destructive pattern (rm -rf, dd, mkfs, pct/qm destroy, DROP TABLE, reboot, piping curl into a shell, ...) always require approval regardless of what you declare. You cannot talk your way past the destructive check by declaring a lower risk.",
|
|
InputSchema: objSchema(
|
|
prop{"target", "string", "Target entity slug: host:<slug> (e.g. host:strong) or lxc:<slug> (e.g. lxc:caddy). LXC commands run via pct exec on its Proxmox host automatically."},
|
|
prop{"command", "string", "The shell command to run. Can be a full script (multi-line, &&-chained). Runs as root."},
|
|
prop{"purpose", "string", "One sentence: why you're running this. Shown to the operator alongside the approval — be specific, this is what they're approving."},
|
|
prop{"declared_risk", "string", "Optional self-assessment: read_only, reversible_low, config_mutation, or destructive. This can only ESCALATE the automatic classification, never lower it — declaring a mutating command as read_only has no effect."},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
targetSlug, _ := args["target"].(string)
|
|
command, _ := args["command"].(string)
|
|
purpose, _ := args["purpose"].(string)
|
|
declaredRisk, _ := args["declared_risk"].(string)
|
|
if targetSlug == "" || command == "" {
|
|
return textResult("error: target and command are 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
|
|
}
|
|
|
|
riskClass := policy.ClassifyCommand(command, declaredRisk)
|
|
runParams, _ := json.Marshal(map[string]string{"command": command, "purpose": purpose})
|
|
actionCol := "run:" + string(runParams)
|
|
|
|
// Dedup: an identical pending command (same target, command, and
|
|
// purpose) blocks a re-request — stops a tool-calling loop from
|
|
// queuing the same approval repeatedly.
|
|
var existingID string
|
|
derr := pool.QueryRow(ctx, `
|
|
SELECT e.id::text FROM entities e
|
|
JOIN executions ex ON ex.entity_id = e.id
|
|
WHERE e.type = 'execution' AND ex.target_entity_id = $1
|
|
AND ex.action = $2 AND ex.status = 'pending_approval'
|
|
ORDER BY e.created_at DESC LIMIT 1`,
|
|
targetID, actionCol).Scan(&existingID)
|
|
if derr == nil && existingID != "" {
|
|
return textResult(fmt.Sprintf("An identical command is already queued for approval on %s — execution %s. Wait for the operator, don't re-request.", targetSlug, existingID)), nil
|
|
}
|
|
|
|
id, _ := uuid.NewV7()
|
|
correlationID := uuid.New().String()
|
|
// Full UUID, not a truncated prefix — see the matching comment on
|
|
// request_execution's exec slug generation above; the 8-char prefix
|
|
// collided for real under back-to-back requests.
|
|
execName := "run on " + targetSlug + " (" + id.String() + ")"
|
|
execSlug := "exec:" + targetSlug + ":" + id.String()
|
|
if _, err := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
|
|
id, execSlug, execName); err != nil {
|
|
return textResult(fmt.Sprintf("error: failed to create execution: %v", err)), nil
|
|
}
|
|
pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, $4, 'running', $5, $6) ON CONFLICT DO NOTHING`,
|
|
id, targetID, actionCol, riskClass, correlationID, agentID)
|
|
|
|
if riskClass == policy.RiskReadOnly {
|
|
host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
|
|
if rerr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error()))
|
|
return textResult(fmt.Sprintf("resolve target: %v", rerr)), nil
|
|
}
|
|
out, xerr := sshExec(ctx, host, user, wrap(command))
|
|
if xerr != nil {
|
|
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out))
|
|
return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out)), nil
|
|
}
|
|
pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out))
|
|
return textResult(fmt.Sprintf("run on %s (read_only, auto): %s", targetSlug, out)), nil
|
|
}
|
|
|
|
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class=$2 WHERE entity_id=$1`, id, riskClass)
|
|
createApproval(ctx, pool, id, targetID, "run", string(runParams), riskClass)
|
|
confirmNote := ""
|
|
if riskClass == policy.RiskDestructive {
|
|
confirmNote = " This is classified DESTRUCTIVE — flag that clearly to the operator; it needs explicit confirmation, not just a casual \"go ahead\"."
|
|
}
|
|
return textResult(fmt.Sprintf("run on %s requires approval (risk: %s) — execution %s queued.%s Present the command and purpose to the operator and wait; do not re-request.",
|
|
targetSlug, riskClass, id, confirmNote)), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "http_get", Description: "Fetch a public web page or raw file (e.g. a GitHub README/raw URL) and return sanitized text. Use this to research how to deploy a service before provisioning. HTTP/HTTPS only; body is truncated to ~16KB.",
|
|
InputSchema: objSchema(
|
|
prop{"url", "string", "Absolute http(s) URL to fetch"},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
rawURL, _ := args["url"].(string)
|
|
return httpGet(ctx, rawURL), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_execution_status", Description: "Check the status of a requested execution",
|
|
InputSchema: objSchema(
|
|
prop{"execution_id", "string", "Execution UUID (from request_execution output)"},
|
|
),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
execID, _ := args["execution_id"].(string)
|
|
if execID == "" {
|
|
return textResult("execution_id required"), nil
|
|
}
|
|
eid, err := uuid.Parse(execID)
|
|
if err != nil {
|
|
// Try finding by exec slug prefix
|
|
var found uuid.UUID
|
|
err2 := pool.QueryRow(ctx, "SELECT entity_id FROM executions WHERE entity_id::text LIKE $1 LIMIT 1", execID+"%").Scan(&found)
|
|
if err2 != nil {
|
|
return textResult(fmt.Sprintf("execution not found: %s", execID)), nil
|
|
}
|
|
eid = found
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.entity_id::text, e.action, e.risk_class, e.status,
|
|
e.result::text, e.duration_ms, e.started_at::text,
|
|
e.completed_at::text, e.correlation_id
|
|
FROM executions e
|
|
WHERE e.entity_id = $1`, eid), 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.type, ev.severity, ev.source, e.slug AS entity_slug,
|
|
ev.data::text AS 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
|
|
})
|
|
|
|
// ─── Client introspection tools (plan: client-lifecycle Phase 3) ──
|
|
|
|
register(&mcp.Tool{Name: "whoami", Description: "Get the current entity record, peers, and health for a host",
|
|
InputSchema: objSchema(prop{"hostname", "string", "Hostname of the calling machine"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
hostname, _ := args["hostname"].(string)
|
|
if hostname == "" {
|
|
return textResult("error: hostname required"), nil
|
|
}
|
|
slug := "ws:" + hostname
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
COALESCE(st.last_check_at::text, '') AS last_check,
|
|
e.attributes->>'mesh_ip' AS mesh_ip,
|
|
e.attributes->>'age_pubkey' AS age_pubkey,
|
|
e.enrolled_at
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.slug = $1
|
|
ORDER BY e.slug`, slug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "explain", Description: "Compact context card for a service: type, state, health, relations, risk",
|
|
InputSchema: objSchema(prop{"service_slug", "string", "Service entity slug (e.g. service:jellyfin, lxc:caddy)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
if slug == "" {
|
|
return textResult("error: service_slug required"), nil
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.name, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
COALESCE(st.last_check_at::text, '') AS last_check,
|
|
e.version, e.updated_at,
|
|
COALESCE(e.attributes::text, '{}') AS attrs
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.slug = $1`, slug), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "preflight", Description: "Risk classification for an action on a service",
|
|
InputSchema: objSchema(
|
|
prop{"service_slug", "string", "Entity slug"},
|
|
prop{"action", "string", "Planned action (restart, deploy, destroy, etc.)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["service_slug"].(string)
|
|
action, _ := args["action"].(string)
|
|
if slug == "" || action == "" {
|
|
return textResult("error: service_slug and action required"), nil
|
|
}
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.state,
|
|
CASE
|
|
WHEN $2 IN ('restart', 'logs', 'status') THEN 'reversible_low'
|
|
WHEN $2 IN ('deploy', 'upgrade', 'configure') THEN 'config_mutation'
|
|
WHEN $2 IN ('destroy', 'wipe', 'revoke') THEN 'destructive'
|
|
ELSE 'read_only'
|
|
END AS risk_class,
|
|
CASE
|
|
WHEN $2 IN ('read_only','reversible_low') THEN 'auto-act'
|
|
WHEN $2 = 'config_mutation' THEN 'operator-approval'
|
|
ELSE 'operator-approval+confirmation'
|
|
END AS approval
|
|
FROM entities e WHERE e.slug = $1`, slug, action), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_change_history", Description: "Last N change-ledger entries for an entity",
|
|
InputSchema: objSchema(
|
|
prop{"entity_slug", "string", "Entity slug"},
|
|
prop{"limit", "integer", "Max entries (default 20)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_slug"].(string)
|
|
limit := int(getFloat(args, "limit", 20))
|
|
return queryRows(ctx, pool, `
|
|
SELECT al.ts AS timestamp, al.actor_type, al.actor_id::text AS actor_label,
|
|
al.action, al.method, al.path,
|
|
al.detail::text AS details
|
|
FROM audit_log al
|
|
JOIN entities e ON e.id = al.entity_id
|
|
WHERE e.slug = $1
|
|
ORDER BY al.ts DESC
|
|
LIMIT $2`, slug, limit), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "get_state_snapshot", Description: "Last scheduler Observe-pass: fleet health, disk, drift count",
|
|
InputSchema: objSchema(),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return queryRows(ctx, pool, `
|
|
SELECT e.slug, e.type, e.state,
|
|
COALESCE(st.health, 'unknown') AS health,
|
|
COALESCE(st.last_check_at::text, '') AS last_check
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE e.state IS NOT NULL
|
|
OR st.health IS NOT NULL
|
|
ORDER BY st.health, e.slug
|
|
LIMIT 200
|
|
`), nil
|
|
})
|
|
|
|
register(&mcp.Tool{Name: "list_my_secrets", Description: "List secrets accessible to this client by public key",
|
|
InputSchema: objSchema(prop{"caller_pubkey", "string", "Age public key of the caller (optional)"}),
|
|
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
pubkey, _ := args["caller_pubkey"].(string)
|
|
// Match entities where age_pubkey attribute contains the caller's key.
|
|
query := `
|
|
SELECT e.slug, e.type, e.name,
|
|
e.attributes->>'age_pubkey' AS age_pubkey
|
|
FROM entities e
|
|
WHERE e.attributes->>'age_pubkey' IS NOT NULL`
|
|
var dbArgs []any
|
|
if pubkey != "" {
|
|
query += ` AND e.attributes->>'age_pubkey' = $1`
|
|
dbArgs = append(dbArgs, pubkey)
|
|
}
|
|
query += ` ORDER BY e.slug LIMIT 100`
|
|
return queryRows(ctx, pool, query, dbArgs...), 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}},
|
|
}
|
|
}
|
|
|
|
// jsonOut builds a valid {"output": "..."} JSON payload for an execution's
|
|
// result column. Command output contains quotes/backslashes/control chars, so
|
|
// it must be JSON-marshaled — a hand-built string fails the ::jsonb cast and
|
|
// silently drops the status update, leaving the execution stuck.
|
|
func jsonOut(out string) []byte {
|
|
b, _ := json.Marshal(map[string]any{"output": out})
|
|
return b
|
|
}
|
|
|
|
// jsonErr builds a valid {"error": "..."} JSON payload for an execution's
|
|
// result column — same rationale as jsonOut, for the failure path.
|
|
func jsonErr(format string, args ...any) []byte {
|
|
b, _ := json.Marshal(map[string]any{"error": fmt.Sprintf(format, args...)})
|
|
return b
|
|
}
|
|
|
|
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
|
|
items = make([]map[string]any, 0)
|
|
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)
|
|
}
|
|
}
|
|
}
|
|
|
|
// sshExecTimeout bounds how long a single remote command may run — see the
|
|
// matching constant/comment in httpapi/phase3.go. Without it, a hung remote
|
|
// command (piped install script stuck retrying DNS, etc.) blocks this
|
|
// goroutine forever with no way for the caller to ever get an answer.
|
|
const sshExecTimeout = 10 * time.Minute
|
|
|
|
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()
|
|
|
|
type result struct {
|
|
out []byte
|
|
err error
|
|
}
|
|
done := make(chan result, 1)
|
|
go func() {
|
|
out, err := session.CombinedOutput(command)
|
|
done <- result{out, err}
|
|
}()
|
|
|
|
select {
|
|
case r := <-done:
|
|
text := strings.TrimSpace(string(r.out))
|
|
// A non-zero exit MUST surface as an error — matching the fix
|
|
// applied to httpapi's sshExec (this copy still had the original
|
|
// bug: only erroring when there was no output at all, so a command
|
|
// that failed but printed something was silently reported as
|
|
// success).
|
|
if r.err != nil {
|
|
if text != "" {
|
|
return text, fmt.Errorf("%w: %s", r.err, text)
|
|
}
|
|
return text, fmt.Errorf("exec: %w", r.err)
|
|
}
|
|
return text, nil
|
|
case <-time.After(sshExecTimeout):
|
|
session.Close()
|
|
client.Close()
|
|
return "", fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host)
|
|
case <-ctx.Done():
|
|
session.Close()
|
|
client.Close()
|
|
return "", ctx.Err()
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
// htmlTagRe strips HTML tags for the naive text extraction in httpGet.
|
|
var htmlTagRe = regexp.MustCompile(`(?s)<(script|style)[^>]*>.*?</(script|style)>|<[^>]+>`)
|
|
|
|
// httpGet fetches a public URL and returns sanitized, size-capped text so the
|
|
// agent can read a service's README/site before provisioning. Guards: scheme
|
|
// allow-list, request timeout, 16KB body cap, and blocking of RFC1918/loopback
|
|
// hosts to avoid using the tool as an SSRF pivot into the private mesh.
|
|
func httpGet(ctx context.Context, rawURL string) *mcp.CallToolResult {
|
|
if rawURL == "" {
|
|
return textResult("error: url required")
|
|
}
|
|
u, err := url.Parse(strings.TrimSpace(rawURL))
|
|
if err != nil || (u.Scheme != "http" && u.Scheme != "https") {
|
|
return textResult("error: url must be an absolute http(s) URL")
|
|
}
|
|
if isPrivateHost(u.Hostname()) {
|
|
return textResult("error: refusing to fetch private/loopback address")
|
|
}
|
|
|
|
cctx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
|
defer cancel()
|
|
hreq, err := http.NewRequestWithContext(cctx, http.MethodGet, u.String(), nil)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err))
|
|
}
|
|
hreq.Header.Set("User-Agent", "oikos-nomos/1.0 (+homelab agent)")
|
|
hreq.Header.Set("Accept", "text/plain, text/html, application/json;q=0.9, */*;q=0.5")
|
|
|
|
client := &http.Client{Timeout: 20 * time.Second}
|
|
resp, err := client.Do(hreq)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: fetch failed: %v", err))
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
const cap = 256 * 1024 // read a bit extra pre-strip; final output capped below
|
|
body, _ := io.ReadAll(io.LimitReader(resp.Body, cap))
|
|
ct := resp.Header.Get("Content-Type")
|
|
text := sanitizeBody(ct, string(body))
|
|
return textResult(fmt.Sprintf("GET %s → %d %s\n\n%s", u.String(), resp.StatusCode, ct, text))
|
|
}
|
|
|
|
// sanitizeBody strips scripts/styles/tags from HTML, unescapes entities,
|
|
// collapses whitespace, and caps the result to ~16KB of readable text.
|
|
func sanitizeBody(contentType, raw string) string {
|
|
text := raw
|
|
if strings.Contains(contentType, "html") {
|
|
text = htmlTagRe.ReplaceAllString(text, " ")
|
|
text = html.UnescapeString(text)
|
|
text = strings.Join(strings.Fields(text), " ")
|
|
}
|
|
if len(text) > 16*1024 {
|
|
text = text[:16*1024] + "\n…[truncated]"
|
|
}
|
|
return text
|
|
}
|
|
|
|
// isPrivateHost reports whether host is loopback, link-local, or RFC1918.
|
|
func isPrivateHost(host string) bool {
|
|
host = strings.ToLower(host)
|
|
if host == "localhost" || strings.HasSuffix(host, ".local") || strings.HasSuffix(host, ".internal") {
|
|
return true
|
|
}
|
|
ip := net.ParseIP(host)
|
|
if ip == nil {
|
|
return false // hostname; DNS may still resolve private — acceptable for a homelab tool
|
|
}
|
|
return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsUnspecified()
|
|
}
|
|
|
|
// resolveExecTarget resolves any target slug (host: or lxc:) to the SSH
|
|
// endpoint that will actually run the command, and a wrap function that turns
|
|
// a plain shell command into whatever must actually be sent over that SSH
|
|
// connection: identity for a host, `pct exec <pve_id> -- ...` for an LXC.
|
|
//
|
|
// The lxc.attributes.host value is stored WITHOUT a "host:" prefix (e.g.
|
|
// "strong", not "host:strong") — see pct_create's entity registration. The
|
|
// pre-existing pct_exec handler queried resolveHost with that bare value
|
|
// directly, which can never match a "host:*" slug and always fails; this
|
|
// prefixes it correctly.
|
|
func resolveExecTarget(ctx context.Context, pool *db.Pool, targetSlug string) (host, user string, wrap func(cmd string) string, err error) {
|
|
if strings.HasPrefix(targetSlug, "host:") {
|
|
host, user, err = resolveHost(ctx, pool, targetSlug)
|
|
return host, user, func(cmd string) string { return cmd }, err
|
|
}
|
|
if strings.HasPrefix(targetSlug, "lxc:") {
|
|
var pveID, hostAttr string
|
|
// COALESCE the host column: many older LXC entities (seeded from
|
|
// inventory, not provisioned by pct_create) have pve_id but no host
|
|
// attribute at all. Scanning a SQL NULL into a plain string errors
|
|
// the whole row, wrongly reporting "missing pve_id" even when it was
|
|
// present — COALESCE avoids the NULL, "" is handled below.
|
|
if qerr := pool.QueryRow(ctx, "SELECT attributes->>'pve_id', COALESCE(attributes->>'host', '') FROM entities WHERE slug = $1", targetSlug).Scan(&pveID, &hostAttr); qerr != nil || pveID == "" {
|
|
return "", "", nil, fmt.Errorf("LXC not found or missing pve_id: %s", targetSlug)
|
|
}
|
|
hostSlug := hostAttr
|
|
if hostSlug == "" {
|
|
hostSlug = "hubris" // documented default Proxmox host when unset
|
|
}
|
|
if !strings.HasPrefix(hostSlug, "host:") {
|
|
hostSlug = "host:" + hostSlug
|
|
}
|
|
host, user, err = resolveHost(ctx, pool, hostSlug)
|
|
id := pveID
|
|
return host, user, func(cmd string) string {
|
|
b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
|
|
return fmt.Sprintf("pct exec %s -- bash -c 'echo %s | base64 -d | bash'", id, b64)
|
|
}, err
|
|
}
|
|
return "", "", nil, fmt.Errorf("unsupported target %q: must be host:<slug> or lxc:<slug>", targetSlug)
|
|
}
|
|
|
|
func createApproval(ctx context.Context, pool *db.Pool, execID, targetID uuid.UUID, action, params, riskClass string) {
|
|
p := map[string]any{"action": action, "params": params, "execution_id": execID.String()}
|
|
payload, _ := json.Marshal(p)
|
|
// approvals.entity_id is PK + FK to entities(id). Reuse the execution's
|
|
// entity (already inserted by request_execution) so the FK is satisfied —
|
|
// a fresh UUID here had no matching entities row, so the INSERT silently
|
|
// failed, orphaning the execution and never alerting the operator. One
|
|
// execution maps to at most one approval, so the 1:1 identity holds.
|
|
if _, err := pool.Exec(ctx, `
|
|
INSERT INTO approvals (entity_id, subject_entity_id, action, risk_class,
|
|
kind, payload, status, expires_at, created_at)
|
|
VALUES ($1, $2, $3, $4, 'execution', $5::jsonb, 'pending',
|
|
now() + interval '1 hour', now())`,
|
|
execID, targetID, action, riskClass, string(payload)); err != nil {
|
|
slog.Error("createApproval: insert approval", "error", err, "execution", execID)
|
|
return
|
|
}
|
|
if _, err := pool.Exec(ctx, `UPDATE executions SET approval_id = $1 WHERE entity_id = $1`, execID); err != nil {
|
|
slog.Error("createApproval: link approval to execution", "error", err, "execution", execID)
|
|
}
|
|
|
|
// Emit for SSE fan-out — the operator-facing moment: an agent-requested
|
|
// gated action is now awaiting a decision.
|
|
_ = observability.Event(ctx, sqlcgen.New(pool), "approval.created", &execID, "warning", "mcp", "",
|
|
map[string]any{"action": action, "params": params, "risk_class": riskClass})
|
|
} |