dtoro
75c0848a6f
E1: split monolithic files — cmd/nomos (main.go → server.go + mcp.go + workers.go),
internal/mcp/tools.go → entity_tools/ops_tools/knowledge_tools/analysis_tools,
internal/httpapi/impl.go → domain files (entities, events, signals, ontology,
fleet_health, client_context, client_lifecycle, entity_mutations, query_audit).
E2: migrate raw pool.Exec queries to sqlc (entities/relationships queries + generated).
E3: unify SSH — consolidate crypto/ssh dial into actuator/client.go (+client_test).
E4/E5: add tests — db/lifecycle, checkdefaults/build, ontology/preconditions, policy/risk.
243 lines
12 KiB
Go
243 lines
12 KiB
Go
package mcp
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
|
|
"github.com/dtoro/oikos/internal/db"
|
|
"github.com/dtoro/oikos/internal/policy"
|
|
"github.com/google/uuid"
|
|
"github.com/modelcontextprotocol/go-sdk/mcp"
|
|
)
|
|
|
|
func AnalysisTools(pool *db.Pool, agentID uuid.UUID, sec secretBackend) []toolReg {
|
|
return []toolReg{
|
|
{tool: &mcp.Tool{Name: "get_health_summary", Description: "Fleet health per entity — optionally filter by health state(s)",
|
|
InputSchema: objSchema(
|
|
prop{"health", "string", "Comma-separated health states to include (e.g. 'down,stale'). Omit for all."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
healthStr, _ := args["health"].(string)
|
|
query := `
|
|
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' AND e.state <> 'destroyed'`
|
|
if healthStr != "" {
|
|
query += ` AND st.health = ANY(string_to_array($1, ','))`
|
|
return queryRows(ctx, pool, query, healthStr), nil
|
|
}
|
|
query += ` ORDER BY e.slug`
|
|
return queryRows(ctx, pool, query), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_audit_trail", Description: "Query the audit log",
|
|
InputSchema: objSchema(prop{"entity_id", "string", "Filter by affected entity UUID"}),
|
|
}, handler: 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, session_id::text
|
|
FROM audit_log
|
|
WHERE ($1::text IS NULL OR entity_id::text = $1)
|
|
ORDER BY ts DESC LIMIT 50`, nStr(args["entity_id"])), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "query_metrics", Description: "Time-series metrics with bucketed avg/min/max over N hours",
|
|
InputSchema: objSchema(prop{"hours", "integer", "Look-back window in hours (default 24)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
hours := int(getFloat(args, "hours", 24))
|
|
return annotateJSONResult(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), "metric_chart"), nil
|
|
}},
|
|
// ─── Phase 4: new tools ──────────────────────────────────────────
|
|
|
|
{tool: &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)"}),
|
|
}, handler: 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
|
|
}},
|
|
{tool: &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"}),
|
|
}, handler: 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
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_skills", Description: "List available automation skills",
|
|
InputSchema: objSchema(
|
|
prop{"status", "string", "Filter by status (active, inactive, deprecated)"},
|
|
),
|
|
}, handler: 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
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_trend", Description: "Metric slope, variance, and averages for an entity over N days",
|
|
InputSchema: objSchema(
|
|
prop{"entity_id", "string", "Entity slug"},
|
|
prop{"days", "integer", "Look-back window in days (default 7)"}),
|
|
}, handler: 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
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_event_timeline", Description: "Recent events filtered by severity and entity slug",
|
|
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)"}),
|
|
}, handler: 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
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_agent_activity", Description: "Agent self-inspection: query agent activity log",
|
|
InputSchema: objSchema(
|
|
prop{"limit", "integer", "Max rows (default 50)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
limit := int(getFloat(args, "limit", 50))
|
|
return annotateJSONResult(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), "change_log"), nil
|
|
}},
|
|
// classify_command is the command-scoped preflight from
|
|
// plans/2026-07-20-session-review-ten-sessions.md P0.2. The
|
|
// existing `preflight` tool is entity/action-scoped — useless when
|
|
// the agent is composing a `run` command and needs to know whether
|
|
// the classifier will accept it before submitting. Without this,
|
|
// the agent has to retry with cosmetic variations until it finds
|
|
// one that passes (see sessions a51e2086, 8acea2e3 — three
|
|
// duplicate rclone sessions, all bouncing off the classifier).
|
|
// Call this BEFORE `run` whenever the classification is uncertain.
|
|
{tool: &mcp.Tool{Name: "classify_command", Description: "Pre-flight risk classification for a shell command BEFORE calling run. Returns the risk class (read_only / reversible_low / config_mutation / destructive) that `run` would assign. Use this when you're unsure whether a command will auto-execute or need approval — e.g. `pct exec`, `curl`, compound commands, or anything that might be mistaken for mutation. If this returns read_only, the same command will auto-execute via run with no approval; if it returns config_mutation, expect to need operator approval (or pre-frame the command so it classifies lower). Declared risk can only escalate, never de-escalate.",
|
|
InputSchema: objSchema(
|
|
prop{"command", "string", "The exact shell command you intend to pass to run."},
|
|
prop{"declared_risk", "string", "Optional self-assessment you would pass to run (read_only, reversible_low, config_mutation, destructive). Mirrors run's declared_risk parameter."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
command, _ := args["command"].(string)
|
|
declaredRisk, _ := args["declared_risk"].(string)
|
|
if command == "" {
|
|
return textResult("error: command is required"), nil
|
|
}
|
|
risk := policy.ClassifyCommand(command, declaredRisk)
|
|
note := ""
|
|
switch risk {
|
|
case policy.RiskReadOnly:
|
|
note = "auto-acts on `run` (no approval needed)."
|
|
case policy.RiskReversibleLow:
|
|
note = "auto-acts on `run` (no approval needed)."
|
|
case policy.RiskConfigMutation:
|
|
note = "requires operator approval on `run` (or loose assent window active)."
|
|
case policy.RiskDestructive:
|
|
note = "requires explicit operator confirmation on `run` (typed \"I confirm\" phrase)."
|
|
}
|
|
out, _ := json.Marshal(map[string]any{
|
|
"command": command,
|
|
"declared_risk": declaredRisk,
|
|
"risk_class": risk,
|
|
"note": note,
|
|
})
|
|
return textResult(string(out)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_ontology", Description: "Entity types, relationship types, and lifecycle definitions. Use this to understand the schema — what entity types exist, what relationships connect them, and what lifecycle states each type supports.",
|
|
InputSchema: objSchema(),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
etResult := queryRowsJSONSingle(ctx, pool, `
|
|
SELECT name, parent_type, is_abstract, domain, layer,
|
|
description, lifecycle_id, schema_version, status
|
|
FROM entity_types ORDER BY name`)
|
|
|
|
rtResult := queryRowsJSONSingle(ctx, pool, `
|
|
SELECT name, inverse, source_type, target_type,
|
|
cardinality, description
|
|
FROM relationship_types ORDER BY name`)
|
|
|
|
lcResult := queryRowsJSONSingle(ctx, pool, `
|
|
SELECT id, name, states, transitions::text
|
|
FROM lifecycles ORDER BY name`)
|
|
|
|
result := map[string]any{
|
|
"entity_types": etResult,
|
|
"relationship_types": rtResult,
|
|
"lifecycles": lcResult,
|
|
}
|
|
b, _ := json.MarshalIndent(result, "", " ")
|
|
return textResult(string(b)), nil
|
|
}},
|
|
{tool: &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"},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
rawURL, _ := args["url"].(string)
|
|
return httpGet(ctx, rawURL), nil
|
|
}},
|
|
}
|
|
}
|